git.giacomolli.uk - onechat/blob - journal.go

   1 package main
   2
   3 import (
   4         "bufio"
   5         "context"
   6         "errors"
   7         "fmt"
   8         "io"
   9         "os"
  10         "path/filepath"
  11         "slices"
  12         "strconv"
  13         "strings"
  14 )
  15
  16 // Position names one line of a day file: the day as YYYYMMDD, the byte offset
  17 // of the line, and its length with the newline.
  18 type Position struct {
  19         day    uint32
  20         offset uint64
  21         length uint32
  22 }
  23
  24 // Journal appends JSON lines to one file for each day in dir, named
  25 // YYYY-MM-DD.jsonl, and reads them back by their position. It keeps the file of
  26 // the last day open for writes and each file that it read open for reads.
  27 type Journal struct {
  28         dir     string
  29         file    *os.File
  30         day     uint32
  31         size    uint64
  32         readers map[uint32]*os.File
  33 }
  34
  35 // Write appends line to the file of day, YYYY-MM-DD, with one write to the
  36 // operating system, without a buffer. A crash of the process then loses
  37 // nothing. It returns the position of the line.
  38 func (j *Journal) Write(day string, line []byte) (Position, error) {
  39         number, err := journalDayNumber(day)
  40         if err != nil {
  41                 return Position{}, fmt.Errorf("parse journal day: %w", err)
  42         }
  43
  44         if number != j.day {
  45                 err = j.closeWriter()
  46                 if err != nil {
  47                         return Position{}, fmt.Errorf("close previous day writer: %w", err)
  48                 }
  49
  50                 j.file, err = os.OpenFile(filepath.Join(j.dir, day+".jsonl"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
  51                 if err != nil {
  52                         return Position{}, fmt.Errorf("open day file: %w", err)
  53                 }
  54
  55                 info, err := j.file.Stat()
  56                 if err != nil {
  57                         return Position{}, fmt.Errorf("read day file size: %w", err)
  58                 }
  59
  60                 j.day = number
  61                 j.size = uint64(info.Size())
  62         }
  63
  64         _, err = j.file.Write(line)
  65         if err != nil {
  66                 return Position{}, fmt.Errorf("write line: %w", err)
  67         }
  68
  69         place := Position{
  70                 day:    number,
  71                 offset: j.size,
  72                 length: uint32(len(line)),
  73         }
  74
  75         j.size += uint64(len(line))
  76
  77         return place, nil
  78 }
  79
  80 func (j *Journal) closeWriter() error {
  81         if j.file == nil {
  82                 return nil
  83         }
  84
  85         err := j.file.Close()
  86         j.file = nil
  87         j.day = 0
  88         if err != nil {
  89                 return fmt.Errorf("close day file: %w", err)
  90         }
  91
  92         return nil
  93 }
  94
  95 // Read returns the line at place without its newline.
  96 func (j *Journal) Read(place Position) ([]byte, error) {
  97         reader, open := j.readers[place.day]
  98         if !open {
  99                 var err error
 100
 101                 reader, err = os.Open(filepath.Join(j.dir, journalDayName(place.day)+".jsonl"))
 102                 if err != nil {
 103                         return nil, fmt.Errorf("open day file: %w", err)
 104                 }
 105
 106                 if j.readers == nil {
 107                         j.readers = make(map[uint32]*os.File)
 108                 }
 109
 110                 j.readers[place.day] = reader
 111         }
 112
 113         line := make([]byte, place.length)
 114         _, err := reader.ReadAt(line, int64(place.offset))
 115         if err != nil {
 116                 return nil, fmt.Errorf("read line of %s: %w", journalDayName(place.day), err)
 117         }
 118
 119         if len(line) == 0 || line[len(line)-1] != '\n' {
 120                 return nil, fmt.Errorf("line of %s at %d does not end its record", journalDayName(place.day), place.offset)
 121         }
 122
 123         return line[:len(line)-1], nil
 124 }
 125
 126 // Close closes the open day files.
 127 func (j *Journal) Close() error {
 128         err := j.closeWriter()
 129         for day, reader := range j.readers {
 130                 err = errors.Join(err, reader.Close())
 131
 132                 delete(j.readers, day)
 133         }
 134
 135         return err
 136 }
 137
 138 // Scan gives each line of the day files from place on to add, with its
 139 // position, oldest day first. The zero position starts at the first line. A
 140 // crash during a write can leave a part of a line at the end of the last day
 141 // file. Its write never returned, so Scan cuts it off.
 142 func (j *Journal) Scan(ctx context.Context, place Position, add func([]byte, Position) error) error {
 143         paths, err := filepath.Glob(filepath.Join(j.dir, "*.jsonl"))
 144         if err != nil {
 145                 return fmt.Errorf("list day files: %w", err)
 146         }
 147
 148         slices.Sort(paths)
 149
 150         for n, path := range paths {
 151                 day, err := journalDayNumber(strings.TrimSuffix(filepath.Base(path), ".jsonl"))
 152                 if err != nil {
 153                         return fmt.Errorf("parse day filename: %w", err)
 154                 }
 155
 156                 if day < place.day {
 157                         continue
 158                 }
 159
 160                 offset := uint64(0)
 161                 if day == place.day {
 162                         offset = place.offset
 163                 }
 164
 165                 err = journalScanFile(ctx, path, day, offset, n == len(paths)-1, add)
 166                 if err != nil {
 167                         return fmt.Errorf("scan day file: %w", err)
 168                 }
 169         }
 170
 171         return nil
 172 }
 173
 174 // journalDayNumber returns the day YYYY-MM-DD as the number YYYYMMDD.
 175 func journalDayNumber(day string) (uint32, error) {
 176         number, err := strconv.ParseUint(strings.ReplaceAll(day, "-", ""), 10, 32)
 177         if err != nil || len(day) != len("2006-01-02") {
 178                 return 0, fmt.Errorf("invalid day %q", day)
 179         }
 180
 181         return uint32(number), nil
 182 }
 183
 184 // journalDayName returns the day number YYYYMMDD as YYYY-MM-DD.
 185 func journalDayName(day uint32) string {
 186         return fmt.Sprintf("%04d-%02d-%02d", day/10000, day/100%100, day%100)
 187 }
 188
 189 // journalScanFile gives each line of the day file at path from offset on to
 190 // add. In the last day file, it cuts a part of a line at the end.
 191 func journalScanFile(ctx context.Context, path string, day uint32, offset uint64, last bool, add func([]byte, Position) error) (err error) {
 192         file, err := os.Open(path)
 193         if err != nil {
 194                 return fmt.Errorf("open day file: %w", err)
 195         }
 196
 197         defer func() {
 198                 err = errors.Join(err, file.Close())
 199         }()
 200
 201         _, err = file.Seek(int64(offset), io.SeekStart)
 202         if err != nil {
 203                 return fmt.Errorf("seek day file: %w", err)
 204         }
 205
 206         reader := bufio.NewReader(file)
 207         for {
 208                 if ctx.Err() != nil {
 209                         return ctx.Err()
 210                 }
 211
 212                 line, err := reader.ReadBytes('\n')
 213                 if errors.Is(err, io.EOF) && len(line) == 0 {
 214                         return nil
 215                 }
 216
 217                 if errors.Is(err, io.EOF) && last {
 218                         err = journalCutFile(path, offset)
 219                         if err != nil {
 220                                 return fmt.Errorf("cut incomplete day line: %w", err)
 221                         }
 222
 223                         return nil
 224                 }
 225
 226                 if errors.Is(err, io.EOF) {
 227                         return fmt.Errorf("%s ends in an incomplete line at %d", filepath.Base(path), offset)
 228                 }
 229
 230                 if err != nil {
 231                         return fmt.Errorf("read %s: %w", filepath.Base(path), err)
 232                 }
 233
 234                 place := Position{
 235                         day:    day,
 236                         offset: offset,
 237                         length: uint32(len(line)),
 238                 }
 239
 240                 err = add(line[:len(line)-1], place)
 241                 if err != nil {
 242                         return fmt.Errorf("%s at %d: %w", filepath.Base(path), offset, err)
 243                 }
 244
 245                 offset += uint64(len(line))
 246         }
 247 }
 248
 249 // journalCutFile cuts the file at path to size bytes.
 250 func journalCutFile(path string, size uint64) error {
 251         err := os.Truncate(path, int64(size))
 252         if err != nil {
 253                 return fmt.Errorf("cut incomplete line: %w", err)
 254         }
 255
 256         return nil
 257 }