Bind external events to permanent agent IDs

[onechat] / log.go

diff --git a/log.go b/log.go
index 22012e28eebd31ca4695b0e714960d7396f5e685..61f4d4c449e97288e92bc3734d5643b512237243 100644 (file)
--- a/log.go
+++ b/log.go
@@ -7,27 +7,44 @@ import (
        "fmt"
        "os"
        "path/filepath"
        "fmt"
        "os"
        "path/filepath"

-       "syscall"

        "time"
        "unicode/utf8"
 )
 
        "time"
        "unicode/utf8"
 )
 

-// errLocked tells that another process owns the chat.
-var errLocked = errors.New("another process owns the chat")
-
-// kinds holds the kinds that a message can have.
-var kinds = map[string]bool{
-       "user": true,
-       "unii": true,
-       "tool": true,
-       "echo": true,
-       "work": true,
-       "note": true,
+// logKinds holds the logKinds that a message can have.
+var logKinds = map[string]bool{
+       "user":  true,
+       "reply": true,
+       "tool":  true,
+       "echo":  true,
+       "work":  true,
+       "note":  true,

 }
 
 }
 

-// Message is one entry of the chat log. ID is the permanent position of the
-// message in the chat, from 0. Size is the UTF-8 byte length of Text. Date is
-// the time of the write in RFC 3339 format.
+// Log is the append-only message log of one chat. It writes one JSON line for
+// each message to main/YYYY-MM-DD.jsonl, named for the local day of the
+// message. It never edits or deletes a line. Each line goes to the operating
+// system before Append returns, so a crash of the process loses no message. An
+// exclusive lock on the lock file makes sure that only one process writes the
+// log. After a failed write, the end of the file is not known, so the log
+// refuses all writes until it is opened again. The caller serializes all calls.
+//
+// The log keeps no message in memory. The index index/main.idx holds the
+// position of message i in record i, so a read of a message is one read of the
+// index and one of its day file. At open the log indexes the lines after the
+// last indexed one, so a missing index is built again from the day files.
+type Log struct {
+       clock   func() time.Time
+       lock    *os.File
+       journal Journal
+       index   *Records
+       failure error
+}
+
+// Message is one entry of the chat log. ID is its position in the log, from 0.
+// A rebuild assigns positions in the merged log. Size is the UTF-8 byte length
+// of Text. Date is the message time in RFC 3339 format, supplied at append or
+// taken from the clock.

 type Message struct {
        ID   int    `json:"i"`
        Kind string `json:"kind"`
 type Message struct {
        ID   int    `json:"i"`
        Kind string `json:"kind"`


@@ -36,101 +53,154 @@ type Message struct {
        Date string `json:"date"`
 }
 
        Date string `json:"date"`
 }
 

-// Log is the append-only message log of one chat. It writes one JSON line for
-// each message to main/YYYY-MM-DD.jsonl, named for the local day of the write.
-// It never edits or deletes a line. Each write goes to the file and to the disk
-// before Append returns. An exclusive lock on the lock file makes sure that
-// only one process writes the log. After a failed write, the end of the file is
-// not known, so the log refuses all writes until it is opened again. The caller
-// serializes all calls.
-type Log struct {
-       clock    func() time.Time
-       lock     *os.File
-       journal  journal
-       messages []Message
-       failure  error
-}
-
-// openLog locks the chat in root and loads all of its messages. The clock gives
-// the date of each new message.
+// openLog locks the chat in root and indexes its new lines. The clock gives the
+// date of each new message.

 func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, error) {
 func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, error) {

-       err := os.MkdirAll(filepath.Join(root, "main"), 0o700)
-       if err != nil {
-               return nil, fmt.Errorf("create message directory: %w", err)
+       for _, dir := range []string{"main", "index"} {
+               err := os.MkdirAll(filepath.Join(root, dir), 0o700)
+               if err != nil {
+                       return nil, fmt.Errorf("create %s directory: %w", dir, err)
+               }

        }
 
        lock, err := lockFile(filepath.Join(root, "lock"))
        if err != nil {
        }
 
        lock, err := lockFile(filepath.Join(root, "lock"))
        if err != nil {

-               return nil, err
+               return nil, fmt.Errorf("lock message log: %w", err)

        }
 
        l := &Log{
                clock: clock,
                lock:  lock,
        }
 
        l := &Log{
                clock: clock,
                lock:  lock,

-               journal: journal{
+               journal: Journal{

                        dir: filepath.Join(root, "main"),
                },
        }
 
                        dir: filepath.Join(root, "main"),
                },
        }
 

-       err = l.load(ctx)
+       l.index, err = openRecords(filepath.Join(root, "index", "main.idx"), positionBytes)

        if err != nil {
        if err != nil {

-               return nil, errors.Join(fmt.Errorf("load message log: %w", err), l.Close())
+               return nil, errors.Join(err, l.Close())
+       }
+
+       err = l.catchUp(ctx)
+       if err != nil {
+               return nil, errors.Join(fmt.Errorf("index message log: %w", err), l.Close())

        }
 
        return l, nil
 }
 
        }
 
        return l, nil
 }
 

-func (l *Log) load(ctx context.Context) error {
-       return readLines(ctx, l.journal.dir, func(message Message) error {
-               if message.ID != len(l.messages) {
-                       return fmt.Errorf("expected message %d, found %d", len(l.messages), message.ID)
+// Close releases the lock. The log refuses all calls after Close.
+func (l *Log) Close() error {
+       err := l.journal.Close()
+       if l.index != nil {
+               err = errors.Join(err, l.index.Close())
+               l.index = nil
+       }
+
+       if l.lock != nil {
+               err = errors.Join(err, l.lock.Close())
+               l.lock = nil
+       }
+
+       return err
+}
+
+// catchUp indexes the lines after the last indexed line.
+func (l *Log) catchUp(ctx context.Context) error {
+       start := Position{}
+       if l.index.count > 0 {
+               record, _, err := l.index.Read(l.index.count - 1)
+               if err != nil {
+                       return fmt.Errorf("read last log position: %w", err)
+               }
+
+               last := decodePosition(record)
+               start = Position{
+                       day:    last.day,
+                       offset: last.offset + uint64(last.length),
+               }
+       }
+
+       err := l.journal.Scan(ctx, start, func(line []byte, place Position) error {
+               var message Message
+
+               err := json.Unmarshal(line, &message)
+               if err != nil {
+                       return fmt.Errorf("decode log message: %w", err)
+               }
+
+               if message.ID != l.index.count {
+                       return fmt.Errorf("expected message %d, found %d", l.index.count, message.ID)

                }
 
                }
 

-               err := validateMessage(message)
+               err = logValidateMessage(message)

                if err != nil {
                        return fmt.Errorf("message %d: %w", message.ID, err)
                }
 
                if err != nil {
                        return fmt.Errorf("message %d: %w", message.ID, err)
                }
 

-               l.messages = append(l.messages, message)
+               err = l.addIndex(place)
+               if err != nil {
+                       return fmt.Errorf("index scanned message: %w", err)
+               }

 
                return nil
        })
 
                return nil
        })

+       if err != nil {
+               return fmt.Errorf("scan message journal: %w", err)
+       }
+
+       return nil

 }
 
 }
 

-// Close releases the lock. The log refuses all calls after Close.
-func (l *Log) Close() error {
-       err := l.journal.Close()
-       if l.lock != nil {
-               err = errors.Join(err, l.lock.Close())
-               l.lock = nil
-       }
+func (l *Log) addIndex(place Position) error {
+       record := make([]byte, positionBytes)
+       encodePosition(record, place)

 
 

-       return err
+       return l.index.Append(record)

 }
 
 // Append writes a new message to the log and returns it with its ID, size, and
 // date.
 func (l *Log) Append(kind, text string) (Message, error) {
 }
 
 // Append writes a new message to the log and returns it with its ID, size, and
 // date.
 func (l *Log) Append(kind, text string) (Message, error) {

+       return l.AppendDated(kind, text, "")
+}
+
+// AppendDated writes a message with date in RFC 3339 format. An empty date uses
+// the clock. The day file uses the local day of the given date.
+func (l *Log) AppendDated(kind, text, date string) (Message, error) {

        if l.lock == nil {
        if l.lock == nil {

-               return Message{}, os.ErrClosed
+               return Message{}, fmt.Errorf("append dated message: %w", os.ErrClosed)

        }
 
        if l.failure != nil {
                return Message{}, fmt.Errorf("log needs a reopen after a failed write: %w", l.failure)
        }
 
        }
 
        if l.failure != nil {
                return Message{}, fmt.Errorf("log needs a reopen after a failed write: %w", l.failure)
        }
 

+       var err error
+

        now := l.clock()
        now := l.clock()

+       if date == "" {
+               date = now.Format(time.RFC3339Nano)
+       } else {
+               now, err = time.Parse(time.RFC3339Nano, date)
+               if err != nil {
+                       return Message{}, fmt.Errorf("parse message date: %w", err)
+               }
+
+               now = now.In(time.Local)
+       }
+

        message := Message{
        message := Message{

-               ID:   len(l.messages),
+               ID:   l.index.count,

                Kind: kind,
                Text: text,
                Size: len(text),
                Kind: kind,
                Text: text,
                Size: len(text),

-               Date: now.Format(time.RFC3339Nano),
+               Date: date,

        }
 
        }
 

-       err := validateMessage(message)
+       err = logValidateMessage(message)

        if err != nil {
        if err != nil {

-               return Message{}, err
+               return Message{}, fmt.Errorf("validate dated message: %w", err)

        }
 
        line, err := json.Marshal(message)
        }
 
        line, err := json.Marshal(message)


@@ -138,31 +208,58 @@ func (l *Log) Append(kind, text string) (Message, error) {
                return Message{}, fmt.Errorf("encode message: %w", err)
        }
 
                return Message{}, fmt.Errorf("encode message: %w", err)
        }
 

-       err = l.journal.Write(now.Format(time.DateOnly), append(line, '\n'))
+       place, err := l.journal.Write(now.Format(time.DateOnly), append(line, '\n'))
+       if err == nil {
+               err = l.addIndex(place)
+       }
+

        if err != nil {
                l.failure = err
                return Message{}, fmt.Errorf("write message %d: %w", message.ID, err)
        }
 
        if err != nil {
                l.failure = err
                return Message{}, fmt.Errorf("write message %d: %w", message.ID, err)
        }
 

-       l.messages = append(l.messages, message)
-

        return message, nil
 }
 
 // Len returns the number of messages in the log.
 func (l *Log) Len() int {
        return message, nil
 }
 
 // Len returns the number of messages in the log.
 func (l *Log) Len() int {

-       return len(l.messages)
+       return l.index.count

 }
 
 }
 

-// Message returns the message with the given ID.
-func (l *Log) Message(id int) Message {
-       return l.messages[id]
+// Message reads the message with the given ID from its day file.
+func (l *Log) Message(id int) (Message, error) {
+       record, found, err := l.index.Read(id)
+       if err != nil {
+               return Message{}, fmt.Errorf("read message position: %w", err)
+       }
+
+       if !found {
+               return Message{}, fmt.Errorf("no message %d", id)
+       }
+
+       line, err := l.journal.Read(decodePosition(record))
+       if err != nil {
+               return Message{}, fmt.Errorf("read message %d: %w", id, err)
+       }
+
+       var message Message
+
+       err = json.Unmarshal(line, &message)
+       if err != nil {
+               return Message{}, fmt.Errorf("decode message %d: %w", id, err)
+       }
+
+       if message.ID != id {
+               return Message{}, fmt.Errorf("index of message %d points to message %d", id, message.ID)
+       }
+
+       return message, nil

 }
 
 }
 

-// validateMessage makes sure that a message has a known kind, valid UTF-8 text,
-// the size of its text, and a valid date.
-func validateMessage(message Message) error {
-       if !kinds[message.Kind] {
+// logValidateMessage makes sure that a message has a known kind, valid UTF-8
+// text, the size of its text, and a valid date.
+func logValidateMessage(message Message) error {
+       if !logKinds[message.Kind] {

                return fmt.Errorf("unknown message kind %q", message.Kind)
        }
 
                return fmt.Errorf("unknown message kind %q", message.Kind)
        }
 


@@ -181,23 +278,3 @@ func validateMessage(message Message) error {
 
        return nil
 }
 
        return nil
 }

-
-// lockFile opens the file at path and takes an exclusive lock on it. The lock
-// stays until the file closes or the process ends.
-func lockFile(path string) (*os.File, error) {
-       file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
-       if err != nil {
-               return nil, fmt.Errorf("open lock file: %w", err)
-       }
-
-       err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
-       if errors.Is(err, syscall.EWOULDBLOCK) {
-               err = errLocked
-       }
-
-       if err != nil {
-               return nil, errors.Join(fmt.Errorf("lock chat: %w", err), file.Close())
-       }
-
-       return file, nil
-}