Add the append-only message log

The log of one chat stores each message as one JSON line, {i, kind,
text, size, date}, in main/YYYY-MM-DD.jsonl for the local day of the
write. Each append goes to the disk before it returns, and the log loads
all lines again at start. An exclusive lock on the chat makes sure that
only one process writes it. After a failed write the log refuses more
writes, because the end of the file is not known.

The kinds are user, unii, tool, echo, work, and note. Size is the UTF-8
byte length of the text. Tests use testify, so the Nix package gets a
vendor hash.

Issue: beb9a70

diff --git a/go.mod b/go.mod
index ca9050b925912a46f09be80d56e22c8716ea7fa7..2bd28835ab6fc8dd691cae902665a376834153a8 100644 (file)
--- a/go.mod
+++ b/go.mod
@@ -1,3 +1,7 @@
 module onechat
 
 go 1.26.5
 module onechat
 
 go 1.26.5

+
+require github.com/stretchr/testify v1.12.0
+
+require gopkg.in/yaml.v3 v3.0.1 // indirect

diff --git a/go.sum b/go.sum
new file mode 100644 (file)
index 0000000..054a196
--- /dev/null
+++ b/go.sum
@@ -0,0 +1,6 @@
+github.com/stretchr/testify v1.12.0 h1:K6Mr6jO9JICuend/5xzTM03ydSV3vdNRYAdPSukj8uI=
+github.com/stretchr/testify v1.12.0/go.mod h1:bOYBZb5qJ00vPzWfIqBUZPaxK8jWiXc6d3ErP4Ca9Gw=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

diff --git a/journal.go b/journal.go
new file mode 100644 (file)
index 0000000..3c2a7ef
--- /dev/null
+++ b/journal.go
@@ -0,0 +1,118 @@
+package main
+
+import (
+       "bufio"
+       "context"
+       "encoding/json"
+       "errors"
+       "fmt"
+       "os"
+       "path/filepath"
+)
+
+// journal appends JSON lines to one file for each day in dir, named
+// YYYY-MM-DD.jsonl. It keeps the file of the last day open.
+type journal struct {
+       dir  string
+       file *os.File
+       day  string
+}
+
+// Write appends line to the file of day and flushes it to the disk.
+func (j *journal) Write(day string, line []byte) error {
+       if day != j.day {
+               err := j.Close()
+               if err != nil {
+                       return err
+               }
+
+               j.file, err = os.OpenFile(filepath.Join(j.dir, day+".jsonl"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
+               if err != nil {
+                       return fmt.Errorf("open day file: %w", err)
+               }
+
+               j.day = day
+       }
+
+       _, err := j.file.Write(line)
+       if err != nil {
+               return fmt.Errorf("write line: %w", err)
+       }
+
+       err = j.file.Sync()
+       if err != nil {
+               return fmt.Errorf("flush line: %w", err)
+       }
+
+       return nil
+}
+
+// Close closes the open day file.
+func (j *journal) Close() error {
+       if j.file == nil {
+               return nil
+       }
+
+       err := j.file.Close()
+       j.file = nil
+       j.day = ""
+       if err != nil {
+               return fmt.Errorf("close day file: %w", err)
+       }
+
+       return nil
+}
+
+// readLines decodes each line of the day files in dir, oldest day first, and
+// gives it to add in file order.
+func readLines[T any](ctx context.Context, dir string, add func(T) error) error {
+       paths, err := filepath.Glob(filepath.Join(dir, "*.jsonl"))
+       if err != nil {
+               return fmt.Errorf("list day files: %w", err)
+       }
+
+       for _, path := range paths {
+               err = readFile(ctx, path, add)
+               if err != nil {
+                       return err
+               }
+       }
+
+       return nil
+}
+
+func readFile[T any](ctx context.Context, path string, add func(T) error) (err error) {
+       file, err := os.Open(path)
+       if err != nil {
+               return fmt.Errorf("open day file: %w", err)
+       }
+
+       defer func() {
+               err = errors.Join(err, file.Close())
+       }()
+
+       scanner := bufio.NewScanner(file)
+       scanner.Buffer(nil, 1<<30)
+       for number := 1; scanner.Scan(); number++ {
+               if ctx.Err() != nil {
+                       return ctx.Err()
+               }
+
+               var value T
+               err = json.Unmarshal(scanner.Bytes(), &value)
+               if err == nil {
+                       err = add(value)
+               }
+
+               if err != nil {
+                       return fmt.Errorf("%s line %d: %w", filepath.Base(path), number, err)
+               }
+       }
+
+       err = scanner.Err()
+       if err != nil {
+               return fmt.Errorf("read %s: %w", filepath.Base(path), err)
+       }
+
+       return nil
+}

diff --git a/log.go b/log.go
new file mode 100644 (file)
index 0000000..22012e2
--- /dev/null
+++ b/log.go
@@ -0,0 +1,203 @@
+package main
+
+import (
+       "context"
+       "encoding/json"
+       "errors"
+       "fmt"
+       "os"
+       "path/filepath"
+       "syscall"
+       "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,
+}
+
+// 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.
+type Message struct {
+       ID   int    `json:"i"`
+       Kind string `json:"kind"`
+       Text string `json:"text"`
+       Size int    `json:"size"`
+       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.
+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)
+       }
+
+       lock, err := lockFile(filepath.Join(root, "lock"))
+       if err != nil {
+               return nil, err
+       }
+
+       l := &Log{
+               clock: clock,
+               lock:  lock,
+               journal: journal{
+                       dir: filepath.Join(root, "main"),
+               },
+       }
+
+       err = l.load(ctx)
+       if err != nil {
+               return nil, errors.Join(fmt.Errorf("load message log: %w", err), l.Close())
+       }
+
+       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)
+               }
+
+               err := validateMessage(message)
+               if err != nil {
+                       return fmt.Errorf("message %d: %w", message.ID, err)
+               }
+
+               l.messages = append(l.messages, message)
+
+               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
+       }
+
+       return err
+}
+
+// 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) {
+       if l.lock == nil {
+               return Message{}, os.ErrClosed
+       }
+
+       if l.failure != nil {
+               return Message{}, fmt.Errorf("log needs a reopen after a failed write: %w", l.failure)
+       }
+
+       now := l.clock()
+       message := Message{
+               ID:   len(l.messages),
+               Kind: kind,
+               Text: text,
+               Size: len(text),
+               Date: now.Format(time.RFC3339Nano),
+       }
+
+       err := validateMessage(message)
+       if err != nil {
+               return Message{}, err
+       }
+
+       line, err := json.Marshal(message)
+       if err != nil {
+               return Message{}, fmt.Errorf("encode message: %w", err)
+       }
+
+       err = l.journal.Write(now.Format(time.DateOnly), append(line, '\n'))
+       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 len(l.messages)
+}
+
+// Message returns the message with the given ID.
+func (l *Log) Message(id int) Message {
+       return l.messages[id]
+}
+
+// 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] {
+               return fmt.Errorf("unknown message kind %q", message.Kind)
+       }
+
+       if !utf8.ValidString(message.Text) {
+               return errors.New("message text is not valid UTF-8")
+       }
+
+       if message.Size != len(message.Text) {
+               return fmt.Errorf("message size %d is not the text size %d", message.Size, len(message.Text))
+       }
+
+       _, err := time.Parse(time.RFC3339Nano, message.Date)
+       if err != nil {
+               return fmt.Errorf("parse message date: %w", err)
+       }
+
+       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
+}

diff --git a/log_test.go b/log_test.go
new file mode 100644 (file)
index 0000000..83e2aca
--- /dev/null
+++ b/log_test.go
@@ -0,0 +1,122 @@
+package main
+
+import (
+       "os"
+       "path/filepath"
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestLogReopenLoadsTheSameMessages(t *testing.T) {
+       root := t.TempDir()
+       clock := newTestClock()
+       log, err := openLog(t.Context(), root, clock.Now)
+       require.NoError(t, err)
+
+       var written []Message
+       for _, kind := range []string{"user", "unii", "tool", "echo", "work", "note"} {
+               message, err := log.Append(kind, kind+" text ✓")
+               require.NoError(t, err)
+
+               written = append(written, message)
+       }
+
+       require.Equal(t, 5, written[5].ID)
+       require.Equal(t, len("note text ✓"), written[5].Size)
+       require.NoError(t, log.Close())
+
+       log, err = openLog(t.Context(), root, clock.Now)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, log.Close())
+       }()
+
+       require.Equal(t, written, log.messages)
+
+       message, err := log.Append("user", "after the reopen")
+       require.NoError(t, err)
+       require.Equal(t, 6, message.ID)
+}
+
+func TestLogWritesOneFilePerDay(t *testing.T) {
+       root := t.TempDir()
+       clock := newTestClock()
+       log, err := openLog(t.Context(), root, clock.Now)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, log.Close())
+       }()
+
+       _, err = log.Append("user", "first day")
+       require.NoError(t, err)
+
+       clock.now = clock.now.Add(24 * time.Hour)
+       _, err = log.Append("user", "second day")
+       require.NoError(t, err)
+
+       first, err := os.ReadFile(filepath.Join(root, "main", "2026-01-01.jsonl"))
+       require.NoError(t, err)
+       require.Equal(t, `{"i":0,"kind":"user","text":"first day","size":9,"date":"2026-01-01T12:00:00Z"}`+"\n", string(first))
+
+       second, err := os.ReadFile(filepath.Join(root, "main", "2026-01-02.jsonl"))
+       require.NoError(t, err)
+       require.Equal(t, `{"i":1,"kind":"user","text":"second day","size":10,"date":"2026-01-02T12:00:00Z"}`+"\n", string(second))
+}
+
+func TestLogAllowsOneWriter(t *testing.T) {
+       root := t.TempDir()
+       clock := newTestClock()
+       log, err := openLog(t.Context(), root, clock.Now)
+       require.NoError(t, err)
+
+       _, err = openLog(t.Context(), root, clock.Now)
+       require.ErrorIs(t, err, errLocked)
+       require.NoError(t, log.Close())
+
+       log, err = openLog(t.Context(), root, clock.Now)
+       require.NoError(t, err)
+       require.NoError(t, log.Close())
+}
+
+func TestLogRejectsAnUnknownKind(t *testing.T) {
+       log, err := openLog(t.Context(), t.TempDir(), newTestClock().Now)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, log.Close())
+       }()
+
+       _, err = log.Append("talk", "text")
+       require.ErrorContains(t, err, `unknown message kind "talk"`)
+       require.Equal(t, 0, log.Len())
+}
+
+func TestLogRejectsAGapInTheIDs(t *testing.T) {
+       root := t.TempDir()
+       require.NoError(t, os.MkdirAll(filepath.Join(root, "main"), 0o700))
+
+       line := `{"i":1,"kind":"user","text":"x","size":1,"date":"2026-01-01T12:00:00Z"}` + "\n"
+       require.NoError(t, os.WriteFile(filepath.Join(root, "main", "2026-01-01.jsonl"), []byte(line), 0o600))
+
+       _, err := openLog(t.Context(), root, newTestClock().Now)
+       require.ErrorContains(t, err, "expected message 0, found 1")
+}
+
+// testClock is a clock that tests set by hand.
+type testClock struct {
+       now time.Time
+}
+
+func newTestClock() *testClock {
+       return &testClock{
+               now: time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC),
+       }
+}
+
+func (c *testClock) Now() time.Time {
+       return c.now
+}

diff --git a/nix/package.nix b/nix/package.nix
index 7b6623b6a9abe37423f0e0c4d216c8b9835bb3c9..47cb139cb5eeaa05b15fbc004a6a2c6c06fdb3c7 100644 (file)
--- a/nix/package.nix
+++ b/nix/package.nix
@@ -9,7 +9,7 @@ buildGoModule {
   inherit src;
 
   subPackages = [ "." ];
   inherit src;
 
   subPackages = [ "." ];

-  vendorHash = null;
+  vendorHash = "sha256-/h5U9uZRK5I2DkM5xETswrg0ouuLfMUz7LBOWP+edgw=";

 
   doInstallCheck = true;
   installCheckPhase = ''
 
   doInstallCheck = true;
   installCheckPhase = ''