]> git.giacomolli.uk - onechat/blob - journal.go
Write each line to the OS without an fsync
[onechat] / journal.go
1 package main
2
3 import (
4         "bufio"
5         "context"
6         "encoding/json"
7         "errors"
8         "fmt"
9         "os"
10         "path/filepath"
11 )
12
13 // journal appends JSON lines to one file for each day in dir, named
14 // YYYY-MM-DD.jsonl. It keeps the file of the last day open.
15 type journal struct {
16         dir  string
17         file *os.File
18         day  string
19 }
20
21 // Write appends line to the file of day with one write to the operating system,
22 // without a buffer. A crash of the process then loses nothing.
23 func (j *journal) Write(day string, line []byte) error {
24         if day != j.day {
25                 err := j.Close()
26                 if err != nil {
27                         return err
28                 }
29
30                 j.file, err = os.OpenFile(filepath.Join(j.dir, day+".jsonl"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
31                 if err != nil {
32                         return fmt.Errorf("open day file: %w", err)
33                 }
34
35                 j.day = day
36         }
37
38         _, err := j.file.Write(line)
39         if err != nil {
40                 return fmt.Errorf("write line: %w", err)
41         }
42
43         return nil
44 }
45
46 // Close closes the open day file.
47 func (j *journal) Close() error {
48         if j.file == nil {
49                 return nil
50         }
51
52         err := j.file.Close()
53         j.file = nil
54         j.day = ""
55         if err != nil {
56                 return fmt.Errorf("close day file: %w", err)
57         }
58
59         return nil
60 }
61
62 // readLines decodes each line of the day files in dir, oldest day first, and
63 // gives it to add in file order.
64 func readLines[T any](ctx context.Context, dir string, add func(T) error) error {
65         paths, err := filepath.Glob(filepath.Join(dir, "*.jsonl"))
66         if err != nil {
67                 return fmt.Errorf("list day files: %w", err)
68         }
69
70         for _, path := range paths {
71                 err = readFile(ctx, path, add)
72                 if err != nil {
73                         return err
74                 }
75         }
76
77         return nil
78 }
79
80 func readFile[T any](ctx context.Context, path string, add func(T) error) (err error) {
81         file, err := os.Open(path)
82         if err != nil {
83                 return fmt.Errorf("open day file: %w", err)
84         }
85
86         defer func() {
87                 err = errors.Join(err, file.Close())
88         }()
89
90         scanner := bufio.NewScanner(file)
91         scanner.Buffer(nil, 1<<30)
92         for number := 1; scanner.Scan(); number++ {
93                 if ctx.Err() != nil {
94                         return ctx.Err()
95                 }
96
97                 var value T
98                 err = json.Unmarshal(scanner.Bytes(), &value)
99                 if err == nil {
100                         err = add(value)
101                 }
102
103                 if err != nil {
104                         return fmt.Errorf("%s line %d: %w", filepath.Base(path), number, err)
105                 }
106         }
107
108         err = scanner.Err()
109         if err != nil {
110                 return fmt.Errorf("read %s: %w", filepath.Base(path), err)
111         }
112
113         return nil
114 }