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         "time"
  11         "unicode/utf8"
  12 )
  13
  14 // kinds holds the kinds that a message can have.
  15 var kinds = map[string]bool{
  16         "user":  true,
  17         "reply": true,
  18         "tool":  true,
  19         "echo":  true,
  20         "work":  true,
  21         "note":  true,
  22 }
  23
  24 // Message is one entry of the chat log. ID is the permanent position of the
  25 // message in the chat, from 0. Size is the UTF-8 byte length of Text. Date is
  26 // the time of the write in RFC 3339 format.
  27 type Message struct {
  28         ID   int    `json:"i"`
  29         Kind string `json:"kind"`
  30         Text string `json:"text"`
  31         Size int    `json:"size"`
  32         Date string `json:"date"`
  33 }
  34
  35 // Log is the append-only message log of one chat. It writes one JSON line for
  36 // each message to main/YYYY-MM-DD.jsonl, named for the local day of the write.
  37 // It never edits or deletes a line. Each line goes to the operating system
  38 // before Append returns, so a crash of the process loses no message. An
  39 // exclusive lock on the lock file makes sure that only one process writes the
  40 // log. After a failed write, the end of the file is not known, so the log
  41 // refuses all writes until it is opened again. The caller serializes all calls.
  42 //
  43 // The log keeps no message in memory. The index index/main.idx holds the
  44 // position of message i in record i, so a read of a message is one read of the
  45 // index and one of its day file. At open the log indexes the lines after the
  46 // last indexed one, so a missing index is built again from the day files.
  47 type Log struct {
  48         clock   func() time.Time
  49         lock    *os.File
  50         journal journal
  51         index   *records
  52         failure error
  53 }
  54
  55 // openLog locks the chat in root and indexes its new lines. The clock gives the
  56 // date of each new message.
  57 func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, error) {
  58         for _, dir := range []string{"main", "index"} {
  59                 err := os.MkdirAll(filepath.Join(root, dir), 0o700)
  60                 if err != nil {
  61                         return nil, fmt.Errorf("create %s directory: %w", dir, err)
  62                 }
  63         }
  64
  65         lock, err := lockFile(filepath.Join(root, "lock"))
  66         if err != nil {
  67                 return nil, err
  68         }
  69
  70         l := &Log{
  71                 clock: clock,
  72                 lock:  lock,
  73                 journal: journal{
  74                         dir: filepath.Join(root, "main"),
  75                 },
  76         }
  77
  78         l.index, err = openRecords(filepath.Join(root, "index", "main.idx"), positionBytes)
  79         if err != nil {
  80                 return nil, errors.Join(err, l.Close())
  81         }
  82
  83         err = l.catchUp(ctx)
  84         if err != nil {
  85                 return nil, errors.Join(fmt.Errorf("index message log: %w", err), l.Close())
  86         }
  87
  88         return l, nil
  89 }
  90
  91 // catchUp indexes the lines after the last indexed line.
  92 func (l *Log) catchUp(ctx context.Context) error {
  93         start := position{}
  94         if l.index.count > 0 {
  95                 record, _, err := l.index.Read(l.index.count - 1)
  96                 if err != nil {
  97                         return err
  98                 }
  99
 100                 last := decodePosition(record)
 101                 start = position{
 102                         day:    last.day,
 103                         offset: last.offset + uint64(last.length),
 104                 }
 105         }
 106
 107         return l.journal.Scan(ctx, start, func(line []byte, place position) error {
 108                 var message Message
 109                 err := json.Unmarshal(line, &message)
 110                 if err != nil {
 111                         return err
 112                 }
 113
 114                 if message.ID != l.index.count {
 115                         return fmt.Errorf("expected message %d, found %d", l.index.count, message.ID)
 116                 }
 117
 118                 err = validateMessage(message)
 119                 if err != nil {
 120                         return fmt.Errorf("message %d: %w", message.ID, err)
 121                 }
 122
 123                 return l.addIndex(place)
 124         })
 125 }
 126
 127 func (l *Log) addIndex(place position) error {
 128         record := make([]byte, positionBytes)
 129         encodePosition(record, place)
 130
 131         return l.index.Append(record)
 132 }
 133
 134 // Close releases the lock. The log refuses all calls after Close.
 135 func (l *Log) Close() error {
 136         err := l.journal.Close()
 137         if l.index != nil {
 138                 err = errors.Join(err, l.index.Close())
 139                 l.index = nil
 140         }
 141
 142         if l.lock != nil {
 143                 err = errors.Join(err, l.lock.Close())
 144                 l.lock = nil
 145         }
 146
 147         return err
 148 }
 149
 150 // Append writes a new message to the log and returns it with its ID, size, and
 151 // date.
 152 func (l *Log) Append(kind, text string) (Message, error) {
 153         if l.lock == nil {
 154                 return Message{}, os.ErrClosed
 155         }
 156
 157         if l.failure != nil {
 158                 return Message{}, fmt.Errorf("log needs a reopen after a failed write: %w", l.failure)
 159         }
 160
 161         now := l.clock()
 162         message := Message{
 163                 ID:   l.index.count,
 164                 Kind: kind,
 165                 Text: text,
 166                 Size: len(text),
 167                 Date: now.Format(time.RFC3339Nano),
 168         }
 169
 170         err := validateMessage(message)
 171         if err != nil {
 172                 return Message{}, err
 173         }
 174
 175         line, err := json.Marshal(message)
 176         if err != nil {
 177                 return Message{}, fmt.Errorf("encode message: %w", err)
 178         }
 179
 180         place, err := l.journal.Write(now.Format(time.DateOnly), append(line, '\n'))
 181         if err == nil {
 182                 err = l.addIndex(place)
 183         }
 184
 185         if err != nil {
 186                 l.failure = err
 187                 return Message{}, fmt.Errorf("write message %d: %w", message.ID, err)
 188         }
 189
 190         return message, nil
 191 }
 192
 193 // Len returns the number of messages in the log.
 194 func (l *Log) Len() int {
 195         return l.index.count
 196 }
 197
 198 // Message reads the message with the given ID from its day file.
 199 func (l *Log) Message(id int) (Message, error) {
 200         record, found, err := l.index.Read(id)
 201         if err != nil {
 202                 return Message{}, err
 203         }
 204
 205         if !found {
 206                 return Message{}, fmt.Errorf("no message %d", id)
 207         }
 208
 209         line, err := l.journal.Read(decodePosition(record))
 210         if err != nil {
 211                 return Message{}, fmt.Errorf("read message %d: %w", id, err)
 212         }
 213
 214         var message Message
 215         err = json.Unmarshal(line, &message)
 216         if err != nil {
 217                 return Message{}, fmt.Errorf("decode message %d: %w", id, err)
 218         }
 219
 220         if message.ID != id {
 221                 return Message{}, fmt.Errorf("index of message %d points to message %d", id, message.ID)
 222         }
 223
 224         return message, nil
 225 }
 226
 227 // validateMessage makes sure that a message has a known kind, valid UTF-8 text,
 228 // the size of its text, and a valid date.
 229 func validateMessage(message Message) error {
 230         if !kinds[message.Kind] {
 231                 return fmt.Errorf("unknown message kind %q", message.Kind)
 232         }
 233
 234         if !utf8.ValidString(message.Text) {
 235                 return errors.New("message text is not valid UTF-8")
 236         }
 237
 238         if message.Size != len(message.Text) {
 239                 return fmt.Errorf("message size %d is not the text size %d", message.Size, len(message.Text))
 240         }
 241
 242         _, err := time.Parse(time.RFC3339Nano, message.Date)
 243         if err != nil {
 244                 return fmt.Errorf("parse message date: %w", err)
 245         }
 246
 247         return nil
 248 }