1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3// Package ciartifacts implements ports.CIArtifactFiles under
4// $ERMITE_DATA_DIR/ci/artifacts, keyed by the global run id like the logs (a
5// repository rename never moves them):
6//
7// <run:020d>/<id:020d> the content of a committed artifact
8// <run:020d>/<id:020d>.blocks/<f> the staged blocks of a pending one
9// <run:020d>/.tmp-<id:020d>-* a content being assembled
10//
11// Paths are made of ids and of block file names the use case builds from
12// hexadecimal characters: an artifact name never reaches the file system.
13// Files are 0600 in 0700 directories, synced before they count, and a content
14// appears by rename, whole or not at all.
15package ciartifacts
16
17import (
18 "context"
19 "crypto/sha256"
20 "encoding/hex"
21 "errors"
22 "fmt"
23 "io"
24 "io/fs"
25 "os"
26 "path/filepath"
27 "slices"
28 "strconv"
29 "strings"
30
31 "bnema.dev/ermite/internal/domain"
32 "bnema.dev/ermite/internal/ports"
33)
34
35// Dir is the directory of the artifacts under the data directory.
36func Dir(dataDir string) string { return filepath.Join(dataDir, "ci", "artifacts") }
37
38const (
39 idFormat = "%020d"
40 blocksSuffix = ".blocks"
41 tmpPrefix = ".tmp-"
42 // maxFileName bounds a block file name (well under the 255 bytes of a
43 // name on Linux).
44 maxFileName = 200
45)
46
47// Store is the file-backed artifact store.
48type Store struct{ root string }
49
50var _ ports.CIArtifactFiles = (*Store)(nil)
51
52// New returns a store rooted at dir, which need not exist yet.
53func New(dir string) *Store { return &Store{root: dir} }
54
55func validRef(ref ports.CIArtifactRef) bool { return ref.RunID > 0 && ref.ID > 0 }
56
57// validFile accepts the block file names of the use case: lowercase
58// hexadecimal characters and dots, not starting with a dot.
59func validFile(name string) bool {
60 if name == "" || len(name) > maxFileName || name[0] == '.' {
61 return false
62 }
63 return strings.Trim(name, "0123456789abcdef.") == ""
64}
65
66func (s *Store) runDir(runID int64) string {
67 return filepath.Join(s.root, fmt.Sprintf(idFormat, runID))
68}
69
70func (s *Store) contentPath(ref ports.CIArtifactRef) string {
71 return filepath.Join(s.runDir(ref.RunID), fmt.Sprintf(idFormat, ref.ID))
72}
73
74func (s *Store) blocksDir(ref ports.CIArtifactRef) string {
75 return s.contentPath(ref) + blocksSuffix
76}
77
78// syncDir makes the entries of a directory durable.
79func syncDir(dir string) error {
80 d, err := os.Open(dir)
81 if err != nil {
82 return err
83 }
84 serr := d.Sync()
85 return errors.Join(serr, d.Close())
86}
87
88// ctxReader stops a copy when its context ends.
89type ctxReader struct {
90 ctx context.Context
91 r io.Reader
92}
93
94func (c ctxReader) Read(p []byte) (int, error) {
95 if err := c.ctx.Err(); err != nil {
96 return 0, err
97 }
98 return c.r.Read(p)
99}
100
101// PutBlock writes a new block file, created exclusively.
102func (s *Store) PutBlock(ctx context.Context, ref ports.CIArtifactRef, file string, src io.Reader) (int64, error) {
103 if !validRef(ref) || !validFile(file) {
104 return 0, fmt.Errorf("%w: block file %q", domain.ErrInvalid, file)
105 }
106 dir := s.blocksDir(ref)
107 if err := os.MkdirAll(dir, 0o700); err != nil {
108 return 0, fmt.Errorf("create block directory: %w", err)
109 }
110 path := filepath.Join(dir, file)
111 f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
112 if err != nil {
113 return 0, fmt.Errorf("create block: %w", err)
114 }
115 n, werr := io.Copy(f, ctxReader{ctx, src})
116 var serr error
117 if werr == nil {
118 serr = f.Sync()
119 }
120 cerr := f.Close()
121 if err := errors.Join(werr, serr, cerr); err != nil {
122 _ = os.Remove(path)
123 if werr != nil {
124 return 0, werr // the source's error, as it is
125 }
126 return 0, fmt.Errorf("write block: %w", err)
127 }
128 if err := syncDir(dir); err != nil {
129 _ = os.Remove(path)
130 return 0, fmt.Errorf("sync block directory: %w", err)
131 }
132 return n, nil
133}
134
135// RemoveBlock deletes one block file.
136func (s *Store) RemoveBlock(_ context.Context, ref ports.CIArtifactRef, file string) error {
137 if !validRef(ref) || !validFile(file) {
138 return nil
139 }
140 if err := os.Remove(filepath.Join(s.blocksDir(ref), file)); err != nil && !errors.Is(err, fs.ErrNotExist) {
141 return fmt.Errorf("remove block: %w", err)
142 }
143 return nil
144}
145
146// Assemble writes the blocks into a temporary file, syncs it and renames it
147// over the content.
148func (s *Store) Assemble(ctx context.Context, ref ports.CIArtifactRef, files []string) (int64, string, error) {
149 if !validRef(ref) {
150 return 0, "", domain.ErrNotFound
151 }
152 for _, f := range files {
153 if !validFile(f) {
154 return 0, "", fmt.Errorf("%w: block file %q", domain.ErrInvalid, f)
155 }
156 }
157 dir := s.runDir(ref.RunID)
158 if err := os.MkdirAll(dir, 0o700); err != nil {
159 return 0, "", fmt.Errorf("create artifact directory: %w", err)
160 }
161 out, err := os.CreateTemp(dir, tmpPrefix+fmt.Sprintf(idFormat, ref.ID)+"-*")
162 if err != nil {
163 return 0, "", fmt.Errorf("create artifact: %w", err)
164 }
165 tmp := out.Name()
166 h := sha256.New()
167 size, err := s.concat(ctx, ref, files, io.MultiWriter(out, h))
168 if err == nil {
169 err = out.Sync()
170 }
171 if cerr := out.Close(); err == nil {
172 err = cerr
173 }
174 if err == nil {
175 err = os.Rename(tmp, s.contentPath(ref))
176 }
177 if err != nil {
178 _ = os.Remove(tmp)
179 if errors.Is(err, domain.ErrNotFound) || ctx.Err() != nil {
180 return 0, "", err
181 }
182 return 0, "", fmt.Errorf("assemble artifact: %w", err)
183 }
184 if err := syncDir(dir); err != nil {
185 return 0, "", fmt.Errorf("sync artifact directory: %w", err)
186 }
187 return size, "sha256:" + hex.EncodeToString(h.Sum(nil)), nil
188}
189
190func (s *Store) concat(ctx context.Context, ref ports.CIArtifactRef, files []string, w io.Writer) (int64, error) {
191 var size int64
192 for _, name := range files {
193 f, err := os.Open(filepath.Join(s.blocksDir(ref), name))
194 if errors.Is(err, fs.ErrNotExist) {
195 return 0, fmt.Errorf("%w: block file %s", domain.ErrNotFound, name)
196 }
197 if err != nil {
198 return 0, err
199 }
200 n, err := io.Copy(w, ctxReader{ctx, f})
201 _ = f.Close()
202 if err != nil {
203 return 0, err
204 }
205 size += n
206 }
207 return size, nil
208}
209
210// Open opens the content at offset.
211func (s *Store) Open(_ context.Context, ref ports.CIArtifactRef, offset int64) (io.ReadCloser, error) {
212 if !validRef(ref) || offset < 0 {
213 return nil, domain.ErrNotFound
214 }
215 f, err := os.Open(s.contentPath(ref))
216 if errors.Is(err, fs.ErrNotExist) {
217 return nil, domain.ErrNotFound
218 }
219 if err != nil {
220 return nil, fmt.Errorf("open artifact: %w", err)
221 }
222 if _, err := f.Seek(offset, io.SeekStart); err != nil {
223 _ = f.Close()
224 return nil, fmt.Errorf("seek artifact: %w", err)
225 }
226 return f, nil
227}
228
229// Remove deletes the content, the blocks and any assembly in progress.
230func (s *Store) Remove(ctx context.Context, ref ports.CIArtifactRef) error {
231 if !validRef(ref) {
232 return nil
233 }
234 var errs []error
235 if err := os.Remove(s.contentPath(ref)); err != nil && !errors.Is(err, fs.ErrNotExist) {
236 errs = append(errs, err)
237 }
238 errs = append(errs, s.RemoveBlocks(ctx, ref))
239 tmps, _ := filepath.Glob(filepath.Join(s.runDir(ref.RunID), tmpPrefix+fmt.Sprintf(idFormat, ref.ID)+"-*"))
240 for _, t := range tmps {
241 if err := os.Remove(t); err != nil && !errors.Is(err, fs.ErrNotExist) {
242 errs = append(errs, err)
243 }
244 }
245 if err := errors.Join(errs...); err != nil {
246 return fmt.Errorf("remove artifact: %w", err)
247 }
248 return nil
249}
250
251// RemoveBlocks deletes the staging directory of the artifact.
252func (s *Store) RemoveBlocks(_ context.Context, ref ports.CIArtifactRef) error {
253 if !validRef(ref) {
254 return nil
255 }
256 if err := os.RemoveAll(s.blocksDir(ref)); err != nil {
257 return fmt.Errorf("remove blocks: %w", err)
258 }
259 return nil
260}
261
262// RemoveRuns deletes the directories of the runs.
263func (s *Store) RemoveRuns(_ context.Context, runIDs []int64) error {
264 var errs []error
265 for _, id := range runIDs {
266 if id < 1 {
267 continue
268 }
269 if err := os.RemoveAll(s.runDir(id)); err != nil {
270 errs = append(errs, err)
271 }
272 }
273 return errors.Join(errs...)
274}
275
276// RunIDs lists the run directories.
277func (s *Store) RunIDs(_ context.Context) ([]int64, error) {
278 entries, err := os.ReadDir(s.root)
279 if errors.Is(err, fs.ErrNotExist) {
280 return nil, nil
281 }
282 if err != nil {
283 return nil, fmt.Errorf("list artifact directories: %w", err)
284 }
285 var ids []int64
286 for _, e := range entries {
287 if id, ok := parseID(e.Name()); ok && e.IsDir() {
288 ids = append(ids, id)
289 }
290 }
291 return ids, nil
292}
293
294// ArtifactIDs lists the artifacts that have a content, blocks or an assembly
295// in progress in the run directory.
296func (s *Store) ArtifactIDs(_ context.Context, runID int64) ([]int64, error) {
297 if runID < 1 {
298 return nil, nil
299 }
300 entries, err := os.ReadDir(s.runDir(runID))
301 if errors.Is(err, fs.ErrNotExist) {
302 return nil, nil
303 }
304 if err != nil {
305 return nil, fmt.Errorf("list artifacts: %w", err)
306 }
307 var ids []int64
308 for _, e := range entries {
309 name := e.Name()
310 if rest, ok := strings.CutPrefix(name, tmpPrefix); ok {
311 name, _, _ = strings.Cut(rest, "-")
312 } else {
313 name = strings.TrimSuffix(name, blocksSuffix)
314 }
315 if id, ok := parseID(name); ok && !slices.Contains(ids, id) {
316 ids = append(ids, id)
317 }
318 }
319 slices.Sort(ids)
320 return ids, nil
321}
322
323// parseID reads a zero-padded id.
324func parseID(name string) (int64, bool) {
325 if len(name) != 20 {
326 return 0, false
327 }
328 id, err := strconv.ParseInt(name, 10, 64)
329 return id, err == nil && id > 0
330}