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