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