X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/feda522eba3e2a27c91741c77d0d01f1840c0004..871743f6ab20d68dcede1673a9764730fda38308:/log.go diff --git a/log.go b/log.go index dfb6e70..6d0d478 100644 --- a/log.go +++ b/log.go @@ -7,16 +7,12 @@ 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{ +// logKinds holds the logKinds that a message can have. +var logKinds = map[string]bool{ "user": true, "reply": true, "tool": true, @@ -25,37 +21,6 @@ var kinds = map[string]bool{ "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) { @@ -74,7 +39,7 @@ 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"), }, } @@ -92,47 +57,24 @@ func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, er 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 +// 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 } // Close releases the lock. The log refuses all calls after Close. @@ -154,26 +96,45 @@ func (l *Log) Close() 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 { - 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) } + var err error + 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{ ID: l.index.count, Kind: kind, Text: text, Size: len(text), - Date: now.Format(time.RFC3339Nano), + Date: date, } - err := validateMessage(message) + err = logValidateMessage(message) if err != nil { - return Message{}, err + return Message{}, fmt.Errorf("validate dated message: %w", err) } line, err := json.Marshal(message) @@ -216,6 +177,7 @@ func (l *Log) Message(id int) (Message, error) { } var message Message + err = json.Unmarshal(line, &message) if err != nil { return Message{}, fmt.Errorf("decode message %d: %w", id, err) @@ -228,10 +190,66 @@ func (l *Log) Message(id int) (Message, error) { 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 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 = 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 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"` + 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) } @@ -250,23 +268,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 -}