git.giacomolli.uk - onechat/blob - log.go

   1 package main
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "errors"
   7         "fmt"
   8         "os"
   9         "path/filepath"
  10         "syscall"
  11         "time"
  12         "unicode/utf8"
  13 )
  14
  15 // errLocked tells that another process owns the chat.
  16 var errLocked = errors.New("another process owns the chat")
  17
  18 // kinds holds the kinds that a message can have.
  19 var kinds = map[string]bool{
  20         "user":  true,
  21         "reply": true,
  22         "tool":  true,
  23         "echo":  true,
  24         "work":  true,
  25         "note":  true,
  26 }
  27
  28 // Message is one entry of the chat log. ID is the permanent position of the
  29 // message in the chat, from 0. Size is the UTF-8 byte length of Text. Date is
  30 // the time of the write in RFC 3339 format.
  31 type Message struct {
  32         ID   int    `json:"i"`
  33         Kind string `json:"kind"`
  34         Text string `json:"text"`
  35         Size int    `json:"size"`
  36         Date string `json:"date"`
  37 }
  38
  39 // Log is the append-only message log of one chat. It writes one JSON line for
  40 // each message to main/YYYY-MM-DD.jsonl, named for the local day of the write.
  41 // It never edits or deletes a line. Each line goes to the operating system
  42 // before Append returns, so a crash of the process loses no message. An
  43 // exclusive lock on the lock file makes sure that only one process writes the
  44 // log. After a failed write, the end of the file is not known, so the log
  45 // refuses all writes until it is opened again. The caller serializes all calls.
  46 type Log struct {
  47         clock    func() time.Time
  48         lock     *os.File
  49         journal  journal
  50         messages []Message
  51         failure  error
  52 }
  53
  54 // openLog locks the chat in root and loads all of its messages. The clock gives
  55 // the date of each new message.
  56 func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, error) {
  57         err := os.MkdirAll(filepath.Join(root, "main"), 0o700)
  58         if err != nil {
  59                 return nil, fmt.Errorf("create message directory: %w", err)
  60         }
  61
  62         lock, err := lockFile(filepath.Join(root, "lock"))
  63         if err != nil {
  64                 return nil, err
  65         }
  66
  67         l := &Log{
  68                 clock: clock,
  69                 lock:  lock,
  70                 journal: journal{
  71                         dir: filepath.Join(root, "main"),
  72                 },
  73         }
  74
  75         err = l.load(ctx)
  76         if err != nil {
  77                 return nil, errors.Join(fmt.Errorf("load message log: %w", err), l.Close())
  78         }
  79
  80         return l, nil
  81 }
  82
  83 func (l *Log) load(ctx context.Context) error {
  84         return readLines(ctx, l.journal.dir, func(message Message) error {
  85                 if message.ID != len(l.messages) {
  86                         return fmt.Errorf("expected message %d, found %d", len(l.messages), message.ID)
  87                 }
  88
  89                 err := validateMessage(message)
  90                 if err != nil {
  91                         return fmt.Errorf("message %d: %w", message.ID, err)
  92                 }
  93
  94                 l.messages = append(l.messages, message)
  95
  96                 return nil
  97         })
  98 }
  99
 100 // Close releases the lock. The log refuses all calls after Close.
 101 func (l *Log) Close() error {
 102         err := l.journal.Close()
 103         if l.lock != nil {
 104                 err = errors.Join(err, l.lock.Close())
 105                 l.lock = nil
 106         }
 107
 108         return err
 109 }
 110
 111 // Append writes a new message to the log and returns it with its ID, size, and
 112 // date.
 113 func (l *Log) Append(kind, text string) (Message, error) {
 114         if l.lock == nil {
 115                 return Message{}, os.ErrClosed
 116         }
 117
 118         if l.failure != nil {
 119                 return Message{}, fmt.Errorf("log needs a reopen after a failed write: %w", l.failure)
 120         }
 121
 122         now := l.clock()
 123         message := Message{
 124                 ID:   len(l.messages),
 125                 Kind: kind,
 126                 Text: text,
 127                 Size: len(text),
 128                 Date: now.Format(time.RFC3339Nano),
 129         }
 130
 131         err := validateMessage(message)
 132         if err != nil {
 133                 return Message{}, err
 134         }
 135
 136         line, err := json.Marshal(message)
 137         if err != nil {
 138                 return Message{}, fmt.Errorf("encode message: %w", err)
 139         }
 140
 141         err = l.journal.Write(now.Format(time.DateOnly), append(line, '\n'))
 142         if err != nil {
 143                 l.failure = err
 144                 return Message{}, fmt.Errorf("write message %d: %w", message.ID, err)
 145         }
 146
 147         l.messages = append(l.messages, message)
 148
 149         return message, nil
 150 }
 151
 152 // Len returns the number of messages in the log.
 153 func (l *Log) Len() int {
 154         return len(l.messages)
 155 }
 156
 157 // Message returns the message with the given ID.
 158 func (l *Log) Message(id int) Message {
 159         return l.messages[id]
 160 }
 161
 162 // validateMessage makes sure that a message has a known kind, valid UTF-8 text,
 163 // the size of its text, and a valid date.
 164 func validateMessage(message Message) error {
 165         if !kinds[message.Kind] {
 166                 return fmt.Errorf("unknown message kind %q", message.Kind)
 167         }
 168
 169         if !utf8.ValidString(message.Text) {
 170                 return errors.New("message text is not valid UTF-8")
 171         }
 172
 173         if message.Size != len(message.Text) {
 174                 return fmt.Errorf("message size %d is not the text size %d", message.Size, len(message.Text))
 175         }
 176
 177         _, err := time.Parse(time.RFC3339Nano, message.Date)
 178         if err != nil {
 179                 return fmt.Errorf("parse message date: %w", err)
 180         }
 181
 182         return nil
 183 }
 184
 185 // lockFile opens the file at path and takes an exclusive lock on it. The lock
 186 // stays until the file closes or the process ends.
 187 func lockFile(path string) (*os.File, error) {
 188         file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
 189         if err != nil {
 190                 return nil, fmt.Errorf("open lock file: %w", err)
 191         }
 192
 193         err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
 194         if errors.Is(err, syscall.EWOULDBLOCK) {
 195                 err = errLocked
 196         }
 197
 198         if err != nil {
 199                 return nil, errors.Join(fmt.Errorf("lock chat: %w", err), file.Close())
 200         }
 201
 202         return file, nil
 203 }