1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3// Package cilogs implements ports.CILogStore with append-only files under
4// $ERMITE_DATA_DIR/ci/logs/<run id>/<job id>/<attempt>.log.
5//
6// A file holds one line per log line, in GitHub's raw log shape: an RFC 3339
7// timestamp with nanoseconds, a space, then the text, behind a one-character
8// marker: "-" for a line, "+" for a piece that continues the line before it
9// (it keeps its own line number):
10//
11// - 2026-10-12T10:00:00.123456789Z text
12// - 2026-10-12T10:00:00.123456789Z more text of the same line
13//
14// The file is the truth for bytes and line counts: the metadata store is never
15// written per chunk. Counts are cached per open log and rebuilt from the file
16// after a restart.
17package cilogs
18
19import (
20 "bufio"
21 "bytes"
22 "context"
23 "errors"
24 "fmt"
25 "io"
26 "io/fs"
27 "os"
28 "path/filepath"
29 "strconv"
30 "sync"
31 "time"
32
33 "bnema.dev/ermite/internal/domain"
34 "bnema.dev/ermite/internal/ports"
35)
36
37// Dir is the directory of the logs under the data directory.
38func Dir(dataDir string) string { return filepath.Join(dataDir, "ci", "logs") }
39
40// timeLayout is RFC 3339 with nanoseconds, always in UTC and with all nine
41// digits, so that every prefix has the same length.
42const timeLayout = "2006-01-02T15:04:05.000000000Z"
43
44// prefixLen is the length of "- " + timestamp + " ".
45const prefixLen = 2 + len(timeLayout) + 1
46
47// Store is the file-backed log store.
48type Store struct {
49 root string
50
51 mu sync.Mutex
52 logs map[ports.CILogRef]*logState
53}
54
55var _ ports.CILogStore = (*Store)(nil)
56
57// logState is what is known of one log file.
58type logState struct {
59 mu sync.Mutex
60 loaded bool
61 count int64
62 size int64
63 offsets []int64 // byte offset of every 1024th line: index i is line i*1024
64 changed chan struct{}
65}
66
67const offsetEvery = 1024
68
69// New returns a store rooted at dir, which need not exist yet.
70func New(dir string) *Store {
71 return &Store{root: dir, logs: make(map[ports.CILogRef]*logState)}
72}
73
74func (s *Store) path(ref ports.CILogRef) string {
75 return filepath.Join(s.root, strconv.FormatInt(ref.RunID, 10), ref.JobID, strconv.Itoa(int(ref.Attempt))+".log")
76}
77
78func (s *Store) state(ref ports.CILogRef) *logState {
79 s.mu.Lock()
80 defer s.mu.Unlock()
81 st, ok := s.logs[ref]
82 if !ok {
83 st = &logState{changed: make(chan struct{})}
84 s.logs[ref] = st
85 }
86 return st
87}
88
89// validRef rejects a reference that could escape the directory.
90func validRef(ref ports.CILogRef) bool {
91 if ref.RunID < 1 || ref.Attempt < 1 {
92 return false
93 }
94 _, _, _, ok := domain.ParseJobID(ref.JobID)
95 return ok
96}
97
98// load reads the file to learn its line count and size, once. The caller
99// holds st.mu.
100func (s *Store) load(ref ports.CILogRef, st *logState) error {
101 if st.loaded {
102 return nil
103 }
104 f, err := os.Open(s.path(ref))
105 if errors.Is(err, fs.ErrNotExist) {
106 st.loaded = true
107 return nil
108 }
109 if err != nil {
110 return fmt.Errorf("open log: %w", err)
111 }
112 defer func() { _ = f.Close() }()
113 r := bufio.NewReaderSize(f, 1<<16)
114 var count, pos int64
115 for {
116 line, err := r.ReadBytes('\n')
117 if len(line) > 0 && line[len(line)-1] == '\n' {
118 if count%offsetEvery == 0 {
119 st.offsets = append(st.offsets, pos)
120 }
121 count++
122 pos += int64(len(line))
123 } else if len(line) > 0 {
124 // A torn last line from a crash: cut it, it was never acknowledged.
125 if terr := os.Truncate(s.path(ref), pos); terr != nil {
126 return fmt.Errorf("repair log: %w", terr)
127 }
128 }
129 if err != nil {
130 if errors.Is(err, io.EOF) {
131 break
132 }
133 return fmt.Errorf("read log: %w", err)
134 }
135 }
136 st.count, st.size, st.loaded = count, pos, true
137 return nil
138}
139
140func encodeLine(l domain.CILogLine) []byte {
141 b := make([]byte, 0, prefixLen+len(l.Text)+1)
142 if l.Continued {
143 b = append(b, '+', ' ')
144 } else {
145 b = append(b, '-', ' ')
146 }
147 b = l.Time.UTC().AppendFormat(b, timeLayout)
148 b = append(b, ' ')
149 b = append(b, l.Text...)
150 return append(b, '\n')
151}
152
153func decodeLine(b []byte) (domain.CILogLine, error) {
154 b = bytes.TrimSuffix(b, []byte{'\n'})
155 if len(b) < prefixLen || (b[0] != '-' && b[0] != '+') {
156 return domain.CILogLine{}, errors.New("malformed log line")
157 }
158 t, err := time.Parse(timeLayout, string(b[2:2+len(timeLayout)]))
159 if err != nil {
160 return domain.CILogLine{}, fmt.Errorf("malformed log time: %w", err)
161 }
162 return domain.CILogLine{Time: t, Text: string(b[prefixLen:]), Continued: b[0] == '+'}, nil
163}
164
165// Append appends lines at offset.
166func (s *Store) Append(ctx context.Context, ref ports.CILogRef, offset int64, lines []domain.CILogLine, limit int64) (int64, int, error) {
167 return s.append(ctx, ref, offset, lines, limit, false)
168}
169
170// AppendForced appends lines whatever the limit.
171func (s *Store) AppendForced(ctx context.Context, ref ports.CILogRef, lines []domain.CILogLine) (int64, error) {
172 next, _, err := s.append(ctx, ref, -1, lines, 0, true)
173 return next, err
174}
175
176func (s *Store) append(_ context.Context, ref ports.CILogRef, offset int64, lines []domain.CILogLine, limit int64, force bool) (int64, int, error) {
177 if !validRef(ref) {
178 return 0, 0, domain.ErrNotFound
179 }
180 st := s.state(ref)
181 st.mu.Lock()
182 defer st.mu.Unlock()
183 if err := s.load(ref, st); err != nil {
184 return 0, 0, err
185 }
186 if offset < 0 {
187 offset = st.count
188 }
189 if offset > st.count {
190 return 0, 0, &ports.ErrCILogGap{Count: st.count}
191 }
192 skip := st.count - offset
193 if skip >= int64(len(lines)) {
194 return st.count, 0, nil
195 }
196 fresh := lines[skip:]
197 var buf bytes.Buffer
198 for _, l := range fresh {
199 buf.Write(encodeLine(l))
200 }
201 if !force && st.size+int64(buf.Len()) > limit {
202 return st.count, 0, ports.ErrCILogLimit
203 }
204 if err := os.MkdirAll(filepath.Dir(s.path(ref)), 0o700); err != nil {
205 return 0, 0, fmt.Errorf("create log directory: %w", err)
206 }
207 f, err := os.OpenFile(s.path(ref), os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0o600)
208 if err != nil {
209 return 0, 0, fmt.Errorf("open log: %w", err)
210 }
211 n, werr := f.Write(buf.Bytes())
212 cerr := f.Close()
213 if werr != nil || cerr != nil {
214 // Cut what was written so that the file keeps whole lines.
215 _ = os.Truncate(s.path(ref), st.size)
216 return 0, 0, fmt.Errorf("write log: %w", errors.Join(werr, cerr))
217 }
218 pos := st.size
219 for i, l := range fresh {
220 if (st.count+int64(i))%offsetEvery == 0 {
221 st.offsets = append(st.offsets, pos)
222 }
223 pos += int64(len(encodeLine(l)))
224 }
225 st.count += int64(len(fresh))
226 st.size += int64(n)
227 close(st.changed)
228 st.changed = make(chan struct{})
229 return st.count, len(fresh), nil
230}
231
232// Count returns the number of lines held.
233func (s *Store) Count(_ context.Context, ref ports.CILogRef) (int64, error) {
234 if !validRef(ref) {
235 return 0, nil
236 }
237 st := s.state(ref)
238 st.mu.Lock()
239 defer st.mu.Unlock()
240 if err := s.load(ref, st); err != nil {
241 return 0, err
242 }
243 return st.count, nil
244}
245
246// Read returns lines from line number from.
247func (s *Store) Read(_ context.Context, ref ports.CILogRef, from int64, maxLines, maxBytes int) ([]domain.CILogLine, error) {
248 if !validRef(ref) || from < 0 {
249 return nil, domain.ErrNotFound
250 }
251 st := s.state(ref)
252 st.mu.Lock()
253 if err := s.load(ref, st); err != nil {
254 st.mu.Unlock()
255 return nil, err
256 }
257 count, size := st.count, st.size
258 var start int64
259 var first int64
260 if idx := from / offsetEvery; idx < int64(len(st.offsets)) {
261 start, first = st.offsets[idx], idx*offsetEvery
262 }
263 st.mu.Unlock()
264 if from > count {
265 return nil, domain.ErrOutOfRange
266 }
267 if _, err := os.Stat(s.path(ref)); errors.Is(err, fs.ErrNotExist) {
268 if from == 0 && count == 0 {
269 return nil, domain.ErrNotFound
270 }
271 return nil, domain.ErrNotFound
272 }
273 if from == count {
274 return nil, nil
275 }
276 f, err := os.Open(s.path(ref))
277 if err != nil {
278 return nil, fmt.Errorf("open log: %w", err)
279 }
280 defer func() { _ = f.Close() }()
281 // Never read past the size seen under the lock: a concurrent append is
282 // for the next call.
283 r := bufio.NewReaderSize(io.NewSectionReader(f, start, size-start), 1<<16)
284 var out []domain.CILogLine
285 total := 0
286 for n := first; n < count; n++ {
287 raw, err := r.ReadBytes('\n')
288 if err != nil {
289 return nil, fmt.Errorf("read log: %w", err)
290 }
291 if n < from {
292 continue
293 }
294 l, err := decodeLine(raw)
295 if err != nil {
296 return nil, fmt.Errorf("log line %d: %w", n, err)
297 }
298 if len(out) > 0 && (len(out) >= maxLines || total+len(l.Text) > maxBytes) {
299 break
300 }
301 total += len(l.Text)
302 out = append(out, l)
303 }
304 return out, nil
305}
306
307// Changed returns a channel closed at the next append.
308func (s *Store) Changed(ref ports.CILogRef) <-chan struct{} {
309 st := s.state(ref)
310 st.mu.Lock()
311 defer st.mu.Unlock()
312 return st.changed
313}
314
315// Reset empties the log.
316func (s *Store) Reset(_ context.Context, ref ports.CILogRef) error {
317 if !validRef(ref) {
318 return nil
319 }
320 st := s.state(ref)
321 st.mu.Lock()
322 defer st.mu.Unlock()
323 if err := os.Remove(s.path(ref)); err != nil && !errors.Is(err, fs.ErrNotExist) {
324 return fmt.Errorf("remove log: %w", err)
325 }
326 st.loaded, st.count, st.size, st.offsets = true, 0, 0, nil
327 close(st.changed)
328 st.changed = make(chan struct{})
329 return nil
330}
331
332// RemoveRuns deletes the logs of the runs.
333func (s *Store) RemoveRuns(_ context.Context, runIDs []int64) error {
334 var errs []error
335 for _, id := range runIDs {
336 if id < 1 {
337 continue
338 }
339 if err := os.RemoveAll(filepath.Join(s.root, strconv.FormatInt(id, 10))); err != nil {
340 errs = append(errs, err)
341 }
342 }
343 s.mu.Lock()
344 for ref := range s.logs {
345 for _, id := range runIDs {
346 if ref.RunID == id {
347 delete(s.logs, ref)
348 }
349 }
350 }
351 s.mu.Unlock()
352 return errors.Join(errs...)
353}
354
355// RunIDs lists the runs that have a log directory.
356func (s *Store) RunIDs(_ context.Context) ([]int64, error) {
357 entries, err := os.ReadDir(s.root)
358 if errors.Is(err, fs.ErrNotExist) {
359 return nil, nil
360 }
361 if err != nil {
362 return nil, fmt.Errorf("list logs: %w", err)
363 }
364 var ids []int64
365 for _, e := range entries {
366 if id, err := strconv.ParseInt(e.Name(), 10, 64); err == nil && e.IsDir() {
367 ids = append(ids, id)
368 }
369 }
370 return ids, nil
371}