X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/f5c601f1801aaa276f8c335fcb51c373edc0b50c..e86df5fad9aa2f280aad845adeefd239b0e944ae:/log.go?ds=sidebyside diff --git a/log.go b/log.go index 2b9f32f..61f4d4c 100644 --- a/log.go +++ b/log.go @@ -65,7 +65,7 @@ func openLog(ctx context.Context, root string, clock func() time.Time) (*Log, er lock, err := lockFile(filepath.Join(root, "lock")) if err != nil { - return nil, err + return nil, fmt.Errorf("lock message log: %w", err) } l := &Log{ @@ -105,6 +105,60 @@ func (l *Log) Close() error { return err } +// 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 fmt.Errorf("read last log position: %w", err) + } + + last := decodePosition(record) + start = Position{ + day: last.day, + offset: last.offset + uint64(last.length), + } + } + + err := 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) + } + + err = l.addIndex(place) + if err != nil { + return fmt.Errorf("index scanned message: %w", err) + } + + return nil + }) + if err != nil { + return fmt.Errorf("scan message journal: %w", err) + } + + return nil +} + +func (l *Log) addIndex(place Position) error { + record := make([]byte, positionBytes) + encodePosition(record, place) + + return l.index.Append(record) +} + // 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) { @@ -176,7 +230,7 @@ func (l *Log) Len() int { func (l *Log) Message(id int) (Message, error) { record, found, err := l.index.Read(id) if err != nil { - return Message{}, err + return Message{}, fmt.Errorf("read message position: %w", err) } if !found { @@ -202,50 +256,6 @@ func (l *Log) Message(id int) (Message, error) { 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 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) -} - // 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 {