import (
"bufio"
"context"
- "encoding/json"
"errors"
"fmt"
+ "io"
"os"
"path/filepath"
+ "slices"
+ "strconv"
+ "strings"
)
-// journal appends JSON lines to one file for each day in dir, named
-// YYYY-MM-DD.jsonl. It keeps the file of the last day open.
-type journal struct {
- dir string
- file *os.File
- day string
+// Position names one line of a day file: the day as YYYYMMDD, the byte offset
+// of the line, and its length with the newline.
+type Position struct {
+ day uint32
+ offset uint64
+ length uint32
}
-// Write appends line to the file of day and flushes it to the disk.
-func (j *journal) Write(day string, line []byte) error {
- if day != j.day {
- err := j.Close()
+// Journal appends JSON lines to one file for each day in dir, named
+// YYYY-MM-DD.jsonl, and reads them back by their position. It keeps the file of
+// the last day open for writes and each file that it read open for reads.
+type Journal struct {
+ dir string
+ file *os.File
+ day uint32
+ size uint64
+ readers map[uint32]*os.File
+}
+
+// Write appends line to the file of day, YYYY-MM-DD, with one write to the
+// operating system, without a buffer. A crash of the process then loses
+// nothing. It returns the position of the line.
+func (j *Journal) Write(day string, line []byte) (Position, error) {
+ number, err := journalDayNumber(day)
+ if err != nil {
+ return Position{}, err
+ }
+
+ if number != j.day {
+ err = j.closeWriter()
if err != nil {
- return err
+ return Position{}, err
}
j.file, err = os.OpenFile(filepath.Join(j.dir, day+".jsonl"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
if err != nil {
- return fmt.Errorf("open day file: %w", err)
+ return Position{}, fmt.Errorf("open day file: %w", err)
+ }
+
+ info, err := j.file.Stat()
+ if err != nil {
+ return Position{}, fmt.Errorf("read day file size: %w", err)
+ }
+
+ j.day = number
+ j.size = uint64(info.Size())
+ }
+
+ _, err = j.file.Write(line)
+ if err != nil {
+ return Position{}, fmt.Errorf("write line: %w", err)
+ }
+
+ place := Position{
+ day: number,
+ offset: j.size,
+ length: uint32(len(line)),
+ }
+ j.size += uint64(len(line))
+
+ return place, nil
+}
+
+// Read returns the line at place without its newline.
+func (j *Journal) Read(place Position) ([]byte, error) {
+ reader, open := j.readers[place.day]
+ if !open {
+ var err error
+ reader, err = os.Open(filepath.Join(j.dir, journalDayName(place.day)+".jsonl"))
+ if err != nil {
+ return nil, fmt.Errorf("open day file: %w", err)
}
- j.day = day
+ if j.readers == nil {
+ j.readers = make(map[uint32]*os.File)
+ }
+
+ j.readers[place.day] = reader
}
- _, err := j.file.Write(line)
+ line := make([]byte, place.length)
+ _, err := reader.ReadAt(line, int64(place.offset))
if err != nil {
- return fmt.Errorf("write line: %w", err)
+ return nil, fmt.Errorf("read line of %s: %w", journalDayName(place.day), err)
+ }
+
+ if len(line) == 0 || line[len(line)-1] != '\n' {
+ return nil, fmt.Errorf("line of %s at %d does not end its record", journalDayName(place.day), place.offset)
}
- err = j.file.Sync()
+ return line[:len(line)-1], nil
+}
+
+// Close closes the open day files.
+func (j *Journal) Close() error {
+ err := j.closeWriter()
+ for day, reader := range j.readers {
+ err = errors.Join(err, reader.Close())
+ delete(j.readers, day)
+ }
+
+ return err
+}
+
+// Scan gives each line of the day files from place on to add, with its
+// position, oldest day first. The zero position starts at the first line. A
+// crash during a write can leave a part of a line at the end of the last day
+// file. Its write never returned, so Scan cuts it off.
+func (j *Journal) Scan(ctx context.Context, place Position, add func([]byte, Position) error) error {
+ paths, err := filepath.Glob(filepath.Join(j.dir, "*.jsonl"))
if err != nil {
- return fmt.Errorf("flush line: %w", err)
+ return fmt.Errorf("list day files: %w", err)
+ }
+
+ slices.Sort(paths)
+ for n, path := range paths {
+ day, err := journalDayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl"))
+ if err != nil {
+ return err
+ }
+
+ if day < place.day {
+ continue
+ }
+
+ offset := uint64(0)
+ if day == place.day {
+ offset = place.offset
+ }
+
+ err = journalScanFile(ctx, path, day, offset, n == len(paths)-1, add)
+ if err != nil {
+ return err
+ }
}
return nil
}
-// Close closes the open day file.
-func (j *journal) Close() error {
+func (j *Journal) closeWriter() error {
if j.file == nil {
return nil
}
err := j.file.Close()
j.file = nil
- j.day = ""
+ j.day = 0
if err != nil {
return fmt.Errorf("close day file: %w", err)
}
return nil
}
-// readLines decodes each line of the day files in dir, oldest day first, and
-// gives it to add in file order.
-func readLines[T any](ctx context.Context, dir string, add func(T) error) error {
- paths, err := filepath.Glob(filepath.Join(dir, "*.jsonl"))
- if err != nil {
- return fmt.Errorf("list day files: %w", err)
+// journalDayNumber returns the day YYYY-MM-DD as the number YYYYMMDD.
+func journalDayNumber(day string) (uint32, error) {
+ number, err := strconv.ParseUint(strings.ReplaceAll(day, "-", ""), 10, 32)
+ if err != nil || len(day) != len("2006-01-02") {
+ return 0, fmt.Errorf("invalid day %q", day)
}
- for _, path := range paths {
- err = readFile(ctx, path, add)
- if err != nil {
- return err
- }
- }
+ return uint32(number), nil
+}
- return nil
+// journalDayName returns the day number YYYYMMDD as YYYY-MM-DD.
+func journalDayName(day uint32) string {
+ return fmt.Sprintf("%04d-%02d-%02d", day/10000, day/100%100, day%100)
}
-func readFile[T any](ctx context.Context, path string, add func(T) error) (err error) {
+// journalScanFile gives each line of the day file at path from offset on to
+// add. In the last day file, it cuts a part of a line at the end.
+func journalScanFile(ctx context.Context, path string, day uint32, offset uint64, last bool, add func([]byte, Position) error) (err error) {
file, err := os.Open(path)
if err != nil {
return fmt.Errorf("open day file: %w", err)
err = errors.Join(err, file.Close())
}()
- scanner := bufio.NewScanner(file)
- scanner.Buffer(nil, 1<<30)
- for number := 1; scanner.Scan(); number++ {
+ _, err = file.Seek(int64(offset), io.SeekStart)
+ if err != nil {
+ return fmt.Errorf("seek day file: %w", err)
+ }
+
+ reader := bufio.NewReader(file)
+ for {
if ctx.Err() != nil {
return ctx.Err()
}
- var value T
- err = json.Unmarshal(scanner.Bytes(), &value)
- if err == nil {
- err = add(value)
+ line, err := reader.ReadBytes('\n')
+ if errors.Is(err, io.EOF) && len(line) == 0 {
+ return nil
+ }
+
+ if errors.Is(err, io.EOF) && last {
+ return journalCutFile(path, offset)
+ }
+
+ if errors.Is(err, io.EOF) {
+ return fmt.Errorf("%s ends in an incomplete line at %d", filepath.Base(path), offset)
+ }
+
+ if err != nil {
+ return fmt.Errorf("read %s: %w", filepath.Base(path), err)
+ }
+
+ place := Position{
+ day: day,
+ offset: offset,
+ length: uint32(len(line)),
}
+ err = add(line[:len(line)-1], place)
if err != nil {
- return fmt.Errorf("%s line %d: %w", filepath.Base(path), number, err)
+ return fmt.Errorf("%s at %d: %w", filepath.Base(path), offset, err)
}
+
+ offset += uint64(len(line))
}
+}
- err = scanner.Err()
+// journalCutFile cuts the file at path to size bytes.
+func journalCutFile(path string, size uint64) error {
+ err := os.Truncate(path, int64(size))
if err != nil {
- return fmt.Errorf("read %s: %w", filepath.Base(path), err)
+ return fmt.Errorf("cut incomplete line: %w", err)
}
return nil
projects / onechat / blobdiff