"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{
+// logKinds holds the logKinds that a message can have.
+var logKinds = map[string]bool{
"user": true,
"reply": true,
"tool": 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 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
-}
-
// 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) {
l := &Log{
clock: clock,
lock: lock,
- journal: journal{
+ journal: Journal{
dir: filepath.Join(root, "main"),
},
}
return l, 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 = validateMessage(message)
- if err != nil {
- return fmt.Errorf("message %d: %w", message.ID, err)
- }
-
- return l.addIndex(place)
- })
-}
-
-func (l *Log) addIndex(place position) error {
- record := make([]byte, positionBytes)
- encodePosition(record, place)
-
- return l.index.Append(record)
+// 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.
Date: now.Format(time.RFC3339Nano),
}
- err := validateMessage(message)
+ err := logValidateMessage(message)
if err != nil {
return Message{}, err
}
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] {
+// 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)
+ })
+}
+
+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)
}
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
-}
projects / onechat / blobdiff