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 //
  47 // The log keeps no message in memory. The index index/main.idx holds the
  48 // position of message i in record i, so a read of a message is one read of the
  49 // index and one of its day file. At open the log indexes the lines after the
  50 // last indexed one, so a missing index is built again from the day files.
  51 type Log struct {
  52         clock   func() time.Time
  53         lock    *os.File
  54         journal journal
  55         index   *records
  56         failure error
  57 }
  58
  59 // openLog locks the chat in root and indexes its new lines. The clock gives the
  60 // date of each new message.
  61 func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, error) {
  62         for _, dir := range []string{"main", "index"} {
  63                 err := os.MkdirAll(filepath.Join(root, dir), 0o700)
  64                 if err != nil {
  65                         return nil, fmt.Errorf("create %s directory: %w", dir, err)
  66                 }
  67         }
  68
  69         lock, err := lockFile(filepath.Join(root, "lock"))
  70         if err != nil {
  71                 return nil, err
  72         }
  73
  74         l := &Log{
  75                 clock: clock,
  76                 lock:  lock,
  77                 journal: journal{
  78                         dir: filepath.Join(root, "main"),
  79                 },
  80         }
  81
  82         l.index, err = openRecords(filepath.Join(root, "index", "main.idx"), positionBytes)
  83         if err != nil {
  84                 return nil, errors.Join(err, l.Close())
  85         }
  86
  87         err = l.catchUp(ctx)
  88         if err != nil {
  89                 return nil, errors.Join(fmt.Errorf("index message log: %w", err), l.Close())
  90         }
  91
  92         return l, nil
  93 }
  94
  95 // catchUp indexes the lines after the last indexed line.
  96 func (l *Log) catchUp(ctx context.Context) error {
  97         start := position{}
  98         if l.index.count > 0 {
  99                 record, _, err := l.index.Read(l.index.count - 1)
 100                 if err != nil {
 101                         return err
 102                 }
 103
 104                 last := decodePosition(record)
 105                 start = position{
 106                         day:    last.day,
 107                         offset: last.offset + uint64(last.length),
 108                 }
 109         }
 110
 111         return l.journal.Scan(ctx, start, func(line []byte, place position) error {
 112                 var message Message
 113                 err := json.Unmarshal(line, &message)
 114                 if err != nil {
 115                         return err
 116                 }
 117
 118                 if message.ID != l.index.count {
 119                         return fmt.Errorf("expected message %d, found %d", l.index.count, message.ID)
 120                 }
 121
 122                 err = validateMessage(message)
 123                 if err != nil {
 124                         return fmt.Errorf("message %d: %w", message.ID, err)
 125                 }
 126
 127                 return l.addIndex(place)
 128         })
 129 }
 130
 131 func (l *Log) addIndex(place position) error {
 132         record := make([]byte, positionBytes)
 133         encodePosition(record, place)
 134
 135         return l.index.Append(record)
 136 }
 137
 138 // Close releases the lock. The log refuses all calls after Close.
 139 func (l *Log) Close() error {
 140         err := l.journal.Close()
 141         if l.index != nil {
 142                 err = errors.Join(err, l.index.Close())
 143                 l.index = nil
 144         }
 145
 146         if l.lock != nil {
 147                 err = errors.Join(err, l.lock.Close())
 148                 l.lock = nil
 149         }
 150
 151         return err
 152 }
 153
 154 // Append writes a new message to the log and returns it with its ID, size, and
 155 // date.
 156 func (l *Log) Append(kind, text string) (Message, error) {
 157         if l.lock == nil {
 158                 return Message{}, os.ErrClosed
 159         }
 160
 161         if l.failure != nil {
 162                 return Message{}, fmt.Errorf("log needs a reopen after a failed write: %w", l.failure)
 163         }
 164
 165         now := l.clock()
 166         message := Message{
 167                 ID:   l.index.count,
 168                 Kind: kind,
 169                 Text: text,
 170                 Size: len(text),
 171                 Date: now.Format(time.RFC3339Nano),
 172         }
 173
 174         err := validateMessage(message)
 175         if err != nil {
 176                 return Message{}, err
 177         }
 178
 179         line, err := json.Marshal(message)
 180         if err != nil {
 181                 return Message{}, fmt.Errorf("encode message: %w", err)
 182         }
 183
 184         place, err := l.journal.Write(now.Format(time.DateOnly), append(line, '\n'))
 185         if err == nil {
 186                 err = l.addIndex(place)
 187         }
 188
 189         if err != nil {
 190                 l.failure = err
 191                 return Message{}, fmt.Errorf("write message %d: %w", message.ID, err)
 192         }
 193
 194         return message, nil
 195 }
 196
 197 // Len returns the number of messages in the log.
 198 func (l *Log) Len() int {
 199         return l.index.count
 200 }
 201
 202 // Message reads the message with the given ID from its day file.
 203 func (l *Log) Message(id int) (Message, error) {
 204         record, found, err := l.index.Read(id)
 205         if err != nil {
 206                 return Message{}, err
 207         }
 208
 209         if !found {
 210                 return Message{}, fmt.Errorf("no message %d", id)
 211         }
 212
 213         line, err := l.journal.Read(decodePosition(record))
 214         if err != nil {
 215                 return Message{}, fmt.Errorf("read message %d: %w", id, err)
 216         }
 217
 218         var message Message
 219         err = json.Unmarshal(line, &message)
 220         if err != nil {
 221                 return Message{}, fmt.Errorf("decode message %d: %w", id, err)
 222         }
 223
 224         if message.ID != id {
 225                 return Message{}, fmt.Errorf("index of message %d points to message %d", id, message.ID)
 226         }
 227
 228         return message, nil
 229 }
 230
 231 // validateMessage makes sure that a message has a known kind, valid UTF-8 text,
 232 // the size of its text, and a valid date.
 233 func validateMessage(message Message) error {
 234         if !kinds[message.Kind] {
 235                 return fmt.Errorf("unknown message kind %q", message.Kind)
 236         }
 237
 238         if !utf8.ValidString(message.Text) {
 239                 return errors.New("message text is not valid UTF-8")
 240         }
 241
 242         if message.Size != len(message.Text) {
 243                 return fmt.Errorf("message size %d is not the text size %d", message.Size, len(message.Text))
 244         }
 245
 246         _, err := time.Parse(time.RFC3339Nano, message.Date)
 247         if err != nil {
 248                 return fmt.Errorf("parse message date: %w", err)
 249         }
 250
 251         return nil
 252 }
 253
 254 // lockFile opens the file at path and takes an exclusive lock on it. The lock
 255 // stays until the file closes or the process ends.
 256 func lockFile(path string) (*os.File, error) {
 257         file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
 258         if err != nil {
 259                 return nil, fmt.Errorf("open lock file: %w", err)
 260         }
 261
 262         err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
 263         if errors.Is(err, syscall.EWOULDBLOCK) {
 264                 err = errLocked
 265         }
 266
 267         if err != nil {
 268                 return nil, errors.Join(fmt.Errorf("lock chat: %w", err), file.Close())
 269         }
 270
 271         return file, nil
 272 }