X-Git-Url: http://git.giacomolli.uk/onechat/blobdiff_plain/5023a5ff75de371f147392be54aec7737fca36a5..56d959f2d573eef1be5c04d0bf1659fa30d36e7c:/log.go

diff --git a/log.go b/log.go
index 22012e2..d34f459 100644
--- a/log.go
+++ b/log.go
@@ -7,56 +7,28 @@ import (
 	"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,
+// 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.
-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.
+// 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) {
-	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"))
@@ -67,39 +39,52 @@ func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, er
 	l := &Log{
 		clock: clock,
 		lock:  lock,
-		journal: journal{
+		journal: Journal{
 			dir: filepath.Join(root, "main"),
 		},
 	}
 
-	err = l.load(ctx)
+	l.index, err = openRecords(filepath.Join(root, "index", "main.idx"), positionBytes)
+	if err != nil {
+		return nil, errors.Join(err, l.Close())
+	}
+
+	err = l.catchUp(ctx)
 	if err != nil {
-		return nil, errors.Join(fmt.Errorf("load message log: %w", err), l.Close())
+		return nil, errors.Join(fmt.Errorf("index 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
-	})
+// 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 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
 }
 
 // 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
@@ -121,14 +106,14 @@ func (l *Log) Append(kind, text string) (Message, error) {
 
 	now := l.clock()
 	message := Message{
-		ID:   len(l.messages),
+		ID:   l.index.count,
 		Kind: kind,
 		Text: text,
 		Size: len(text),
 		Date: now.Format(time.RFC3339Nano),
 	}
 
-	err := validateMessage(message)
+	err := logValidateMessage(message)
 	if err != nil {
 		return Message{}, err
 	}
@@ -138,31 +123,111 @@ func (l *Log) Append(kind, text string) (Message, error) {
 		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)
 	}
 
-	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)
+	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{}, 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
+}
+
+// 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 err
+		}
+
+		last := decodePosition(record)
+		start = Position{
+			day:    last.day,
+			offset: last.offset + uint64(last.length),
+		}
+	}
+
+	return l.journal.Scan(ctx, start, func(line []byte, place Position) error {
+		var message Message
+		err := json.Unmarshal(line, &message)
+		if err != nil {
+			return err
+		}
+
+		if message.ID != l.index.count {
+			return fmt.Errorf("expected message %d, found %d", l.index.count, message.ID)
+		}
+
+		err = logValidateMessage(message)
+		if err != nil {
+			return fmt.Errorf("message %d: %w", message.ID, err)
+		}
+
+		return l.addIndex(place)
+	})
 }
 
-// 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] {
+func (l *Log) addIndex(place Position) error {
+	record := make([]byte, positionBytes)
+	encodePosition(record, place)
+
+	return l.index.Append(record)
+}
+
+// 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"`
+}
+
+// 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)
 	}
 
@@ -181,23 +246,3 @@ func validateMessage(message Message) error {
 
 	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
-}