X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/5023a5ff75de371f147392be54aec7737fca36a5..effd32b0f2e4d146b219b1727601b0b7f6593638:/journal.go

diff --git a/journal.go b/journal.go
index 3c2a7ef..bb1897c 100644
--- a/journal.go
+++ b/journal.go
@@ -3,59 +3,166 @@ package main
 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 = day
+		j.day = number
+		j.size = uint64(info.Size())
 	}
 
-	_, err := j.file.Write(line)
+	_, err = j.file.Write(line)
 	if err != nil {
-		return fmt.Errorf("write line: %w", err)
+		return Position{}, fmt.Errorf("write line: %w", err)
 	}
 
-	err = j.file.Sync()
+	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)
+		}
+
+		if j.readers == nil {
+			j.readers = make(map[uint32]*os.File)
+		}
+
+		j.readers[place.day] = reader
+	}
+
+	line := make([]byte, place.length)
+	_, err := reader.ReadAt(line, int64(place.offset))
 	if err != nil {
-		return fmt.Errorf("flush 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)
+	}
+
+	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("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)
 	}
@@ -63,25 +170,24 @@ func (j *journal) Close() error {
 	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)
@@ -91,27 +197,54 @@ func readFile[T any](ctx context.Context, path string, add func(T) error) (err e
 		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