// 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.
+// 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
- messages []Message
- failure error
+ clock func() time.Time
+ lock *os.File
+ journal journal
+ index *records
+ 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"))
},
}
- 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)
+// 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 := validateMessage(message)
+ err = validateMessage(message)
if err != nil {
return fmt.Errorf("message %d: %w", message.ID, err)
}
- l.messages = append(l.messages, message)
-
- return nil
+ return l.addIndex(place)
})
}
+func (l *Log) addIndex(place position) error {
+ record := make([]byte, positionBytes)
+ encodePosition(record, place)
+
+ return l.index.Append(record)
+}
+
// 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
now := l.clock()
message := Message{
- ID: len(l.messages),
+ ID: l.index.count,
Kind: kind,
Text: text,
Size: len(text),
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
}
// validateMessage makes sure that a message has a known kind, valid UTF-8 text,
projects / onechat / blobdiff