1// Package filesystem implements storage adapters using the local filesystem.
2package filesystem
3
4import (
5 "crypto/sha256"
6 "crypto/sha512"
7 "encoding/hex"
8 "fmt"
9 "hash"
10 "io"
11 "os"
12 "path/filepath"
13 "strings"
14 "sync"
15 "time"
16
17 "github.com/bnema/zerowrap"
18 "github.com/google/uuid"
19
20 "github.com/bnema/gordon/internal/domain"
21 "github.com/bnema/gordon/pkg/validation"
22)
23
24// BlobStorage implements the BlobStorage interface using the local filesystem.
25type BlobStorage struct {
26 rootDir string
27 log zerowrap.Logger
28 uploadLocks sync.Map
29}
30
31// NewBlobStorage creates a new filesystem blob storage instance.
32func NewBlobStorage(rootDir string, log zerowrap.Logger) (*BlobStorage, error) {
33 // Create directory structure if it doesn't exist
34 dirs := []string{
35 filepath.Join(rootDir, "blobs"),
36 filepath.Join(rootDir, "uploads"),
37 }
38
39 for _, dir := range dirs {
40 if err := os.MkdirAll(dir, 0750); err != nil {
41 return nil, fmt.Errorf("failed to create directory %s: %w", dir, err)
42 }
43 }
44
45 log.Info().
46 Str(zerowrap.FieldLayer, "adapter").
47 Str(zerowrap.FieldAdapter, "filesystem").
48 Str("root_dir", rootDir).
49 Msg("blob storage initialized")
50
51 return &BlobStorage{
52 rootDir: rootDir,
53 log: log,
54 }, nil
55}
56
57// GetBlob retrieves a blob by digest.
58func (s *BlobStorage) GetBlob(digest string) (io.ReadCloser, error) {
59 blobPath, err := s.getBlobPath(digest)
60 if err != nil {
61 return nil, fmt.Errorf("invalid path: %w", err)
62 }
63
64 file, err := os.Open(blobPath)
65 if err != nil {
66 if os.IsNotExist(err) {
67 return nil, fmt.Errorf("blob not found: %s", digest)
68 }
69 return nil, fmt.Errorf("failed to open blob: %w", err)
70 }
71
72 return file, nil
73}
74
75// GetBlobPath returns the filesystem path to a blob.
76func (s *BlobStorage) GetBlobPath(digest string) (string, error) {
77 path, err := s.getBlobPath(digest)
78 if err != nil {
79 return "", fmt.Errorf("invalid path: %w", err)
80 }
81 if _, err := os.Stat(path); err != nil {
82 if os.IsNotExist(err) {
83 return "", fmt.Errorf("blob not found: %s", digest)
84 }
85 return "", err
86 }
87 return path, nil
88}
89
90// PutBlob stores a blob with the given digest.
91func (s *BlobStorage) PutBlob(digest string, data io.Reader, size int64) error {
92 blobPath, err := s.getBlobPath(digest)
93 if err != nil {
94 return fmt.Errorf("invalid path: %w", err)
95 }
96
97 // Create directory if it doesn't exist
98 if err := os.MkdirAll(filepath.Dir(blobPath), 0750); err != nil {
99 return fmt.Errorf("failed to create blob directory: %w", err)
100 }
101
102 // Create temporary file first
103 tmpPath := blobPath + ".tmp"
104 file, err := os.Create(tmpPath)
105 if err != nil {
106 return fmt.Errorf("failed to create temporary blob file: %w", err)
107 }
108 defer file.Close()
109
110 // Copy data to file
111 written, err := io.Copy(file, data)
112 if err != nil {
113 os.Remove(tmpPath)
114 return fmt.Errorf("failed to write blob data: %w", err)
115 }
116
117 // Verify size if provided
118 if size > 0 && written != size {
119 os.Remove(tmpPath)
120 return fmt.Errorf("blob size mismatch: expected %d, got %d", size, written)
121 }
122
123 // Move to final location
124 if err := os.Rename(tmpPath, blobPath); err != nil {
125 os.Remove(tmpPath)
126 return fmt.Errorf("failed to move blob to final location: %w", err)
127 }
128
129 s.log.Info().
130 Str(zerowrap.FieldLayer, "adapter").
131 Str(zerowrap.FieldAdapter, "filesystem").
132 Str("digest", digest).
133 Int64(zerowrap.FieldSize, written).
134 Msg("blob stored")
135
136 return nil
137}
138
139// DeleteBlob removes a blob by digest.
140// Returns the size in bytes of the removed blob.
141func (s *BlobStorage) DeleteBlob(digest string) (int64, error) {
142 blobPath, err := s.getBlobPath(digest)
143 if err != nil {
144 return 0, fmt.Errorf("invalid path: %w", err)
145 }
146
147 info, err := os.Stat(blobPath)
148 if err != nil {
149 if os.IsNotExist(err) {
150 return 0, fmt.Errorf("blob not found: %s", digest)
151 }
152 return 0, fmt.Errorf("failed to stat blob: %w", err)
153 }
154 size := info.Size()
155
156 if err := os.Remove(blobPath); err != nil {
157 return 0, fmt.Errorf("failed to delete blob: %w", err)
158 }
159
160 s.log.Info().
161 Str(zerowrap.FieldLayer, "adapter").
162 Str(zerowrap.FieldAdapter, "filesystem").
163 Str("digest", digest).
164 Int64(zerowrap.FieldSize, size).
165 Msg("blob deleted")
166
167 return size, nil
168}
169
170// BlobExists checks if a blob exists.
171func (s *BlobStorage) BlobExists(digest string) bool {
172 blobPath, err := s.getBlobPath(digest)
173 if err != nil {
174 return false // Invalid path means blob doesn't exist
175 }
176 _, err = os.Stat(blobPath)
177 return err == nil
178}
179
180// ListBlobs returns all blob digests.
181func (s *BlobStorage) ListBlobs() ([]string, error) {
182 blobsDir := filepath.Join(s.rootDir, "blobs")
183
184 if _, err := os.Stat(blobsDir); err != nil {
185 if os.IsNotExist(err) {
186 return []string{}, nil
187 }
188 return nil, fmt.Errorf("failed to access blobs directory: %w", err)
189 }
190
191 digests := make([]string, 0)
192 err := filepath.Walk(blobsDir, func(path string, info os.FileInfo, err error) error {
193 if err != nil {
194 return err
195 }
196
197 if info.IsDir() {
198 return nil
199 }
200
201 relPath, err := filepath.Rel(blobsDir, path)
202 if err != nil {
203 return err
204 }
205
206 parts := strings.Split(relPath, string(filepath.Separator))
207 if len(parts) < 2 {
208 return nil
209 }
210
211 digest := parts[0] + ":" + parts[len(parts)-1]
212 if err := validation.ValidateDigest(digest); err != nil {
213 return nil
214 }
215
216 digests = append(digests, digest)
217 return nil
218 })
219 if err != nil {
220 return nil, fmt.Errorf("failed to list blobs: %w", err)
221 }
222
223 return digests, nil
224}
225
226// GetBlobModTime returns the last modification time for a blob.
227func (s *BlobStorage) GetBlobModTime(digest string) (time.Time, error) {
228 blobPath, err := s.getBlobPath(digest)
229 if err != nil {
230 return time.Time{}, fmt.Errorf("invalid blob path: %w", err)
231 }
232
233 info, err := os.Stat(blobPath)
234 if err != nil {
235 if os.IsNotExist(err) {
236 return time.Time{}, fmt.Errorf("blob not found: %s", digest)
237 }
238 return time.Time{}, fmt.Errorf("failed to stat blob: %w", err)
239 }
240 return info.ModTime(), nil
241}
242
243// StartBlobUpload starts a new blob upload and returns the upload UUID.
244func (s *BlobStorage) StartBlobUpload(name string) (string, error) {
245 uploadID := uuid.New().String()
246
247 uploadPath, err := s.getUploadPath(uploadID)
248 if err != nil {
249 return "", fmt.Errorf("invalid upload path: %w", err)
250 }
251
252 // Create directory if it doesn't exist
253 if err := os.MkdirAll(filepath.Dir(uploadPath), 0750); err != nil {
254 return "", fmt.Errorf("failed to create upload directory: %w", err)
255 }
256
257 // Create empty upload file
258 file, err := os.Create(uploadPath)
259 if err != nil {
260 return "", fmt.Errorf("failed to create upload file: %w", err)
261 }
262 file.Close()
263
264 s.log.Info().
265 Str(zerowrap.FieldLayer, "adapter").
266 Str(zerowrap.FieldAdapter, "filesystem").
267 Str("uuid", uploadID).
268 Str("name", name).
269 Msg("blob upload started")
270
271 return uploadID, nil
272}
273
274// AppendBlobChunk appends data to an in-progress upload.
275func (s *BlobStorage) AppendBlobChunk(name, uuid string, data io.Reader, contentLength, maxBlobSize int64) (int64, error) {
276 uploadPath, err := s.getUploadPath(uuid)
277 if err != nil {
278 return 0, fmt.Errorf("invalid upload path: %w", err)
279 }
280
281 mu := s.getUploadLock(uuid)
282 mu.Lock()
283 defer mu.Unlock()
284
285 file, err := os.OpenFile(uploadPath, os.O_WRONLY|os.O_APPEND, 0600)
286 if err != nil {
287 if os.IsNotExist(err) {
288 s.cleanupUploadLock(uuid)
289 return 0, fmt.Errorf("upload not found: %s", uuid)
290 }
291 return 0, fmt.Errorf("failed to open upload file: %w", err)
292 }
293 defer file.Close()
294
295 fi, err := file.Stat()
296 if err != nil {
297 return 0, fmt.Errorf("failed to get file info: %w", err)
298 }
299 originalSize := fi.Size()
300
301 if maxBlobSize > 0 {
302 remaining := maxBlobSize - originalSize
303 if remaining < 0 || (contentLength >= 0 && contentLength > remaining) {
304 return 0, domain.ErrBlobSizeExceeded
305 }
306 data = io.LimitReader(data, remaining+1)
307 }
308
309 written, err := io.Copy(file, data)
310 if err != nil {
311 return 0, fmt.Errorf("failed to write chunk to upload file: %w", err)
312 }
313
314 if maxBlobSize > 0 && originalSize+written > maxBlobSize {
315 if err := file.Truncate(originalSize); err != nil {
316 return 0, fmt.Errorf("failed to truncate oversized upload: %w", err)
317 }
318 return 0, domain.ErrBlobSizeExceeded
319 }
320
321 fi, err = file.Stat()
322 if err != nil {
323 return 0, fmt.Errorf("failed to get file info: %w", err)
324 }
325
326 s.log.Debug().
327 Str(zerowrap.FieldLayer, "adapter").
328 Str(zerowrap.FieldAdapter, "filesystem").
329 Str("uuid", uuid).
330 Str("name", name).
331 Int64("chunk_size", written).
332 Int64("total_size", fi.Size()).
333 Msg("appended chunk to blob upload")
334
335 return fi.Size(), nil
336}
337
338// GetBlobUpload returns a writer for the upload.
339// The returned WriteCloser holds the per-upload lock; callers must Close it to release the lock.
340func (s *BlobStorage) GetBlobUpload(uuid string) (io.WriteCloser, error) {
341 uploadPath, err := s.getUploadPath(uuid)
342 if err != nil {
343 return nil, fmt.Errorf("invalid upload path: %w", err)
344 }
345
346 mu := s.getUploadLock(uuid)
347 mu.Lock()
348
349 file, err := os.OpenFile(uploadPath, os.O_WRONLY|os.O_APPEND, 0600)
350 if err != nil {
351 mu.Unlock()
352 if os.IsNotExist(err) {
353 s.cleanupUploadLock(uuid)
354 return nil, fmt.Errorf("upload not found: %s", uuid)
355 }
356 return nil, fmt.Errorf("failed to open upload file: %w", err)
357 }
358
359 return &lockedWriteCloser{file: file, mu: mu}, nil
360}
361
362// lockedWriteCloser wraps an io.WriteCloser and releases a mutex on Close.
363type lockedWriteCloser struct {
364 file io.WriteCloser
365 mu *sync.Mutex
366 closeOnce sync.Once
367 closeErr error
368}
369
370func (lw *lockedWriteCloser) Write(p []byte) (int, error) {
371 return lw.file.Write(p)
372}
373
374func (lw *lockedWriteCloser) Close() error {
375 lw.closeOnce.Do(func() {
376 defer lw.mu.Unlock()
377 lw.closeErr = lw.file.Close()
378 })
379 return lw.closeErr
380}
381
382// FinishBlobUpload completes an upload and moves it to blob storage.
383func (s *BlobStorage) FinishBlobUpload(uuid, digest string) error {
384 uploadPath, err := s.getUploadPath(uuid)
385 if err != nil {
386 return fmt.Errorf("invalid upload path: %w", err)
387 }
388
389 mu := s.getUploadLock(uuid)
390 mu.Lock()
391 var finalized bool
392 defer func() {
393 mu.Unlock()
394 if finalized {
395 s.cleanupUploadLock(uuid)
396 }
397 }()
398
399 blobPath, err := s.getBlobPath(digest)
400 if err != nil {
401 return fmt.Errorf("invalid blob path: %w", err)
402 }
403
404 // Verify digest before moving to ensure integrity
405 if err := s.verifyUploadDigest(uploadPath, digest); err != nil {
406 return fmt.Errorf("digest verification failed: %w", err)
407 }
408
409 // Create blob directory if it doesn't exist
410 if err := os.MkdirAll(filepath.Dir(blobPath), 0750); err != nil {
411 return fmt.Errorf("failed to create blob directory: %w", err)
412 }
413
414 // Update the timestamp before rename so it survives a restart even when the
415 // upload has been open longer than the pending-blob grace period.
416 now := time.Now()
417 if err := os.Chtimes(uploadPath, now, now); err != nil {
418 return fmt.Errorf("failed to mark blob as finalized: %w", err)
419 }
420
421 // Move upload file to blob location.
422 if err := os.Rename(uploadPath, blobPath); err != nil {
423 return fmt.Errorf("failed to move upload to blob location: %w", err)
424 }
425 finalized = true
426
427 s.log.Info().
428 Str(zerowrap.FieldLayer, "adapter").
429 Str(zerowrap.FieldAdapter, "filesystem").
430 Str("uuid", uuid).
431 Str("digest", digest).
432 Msg("blob upload finished")
433
434 return nil
435}
436
437// CancelBlobUpload cancels an in-progress upload.
438func (s *BlobStorage) CancelBlobUpload(uuid string) error {
439 uploadPath, err := s.getUploadPath(uuid)
440 if err != nil {
441 return fmt.Errorf("invalid upload path: %w", err)
442 }
443
444 mu := s.getUploadLock(uuid)
445 mu.Lock()
446 var finalized bool
447 defer func() {
448 mu.Unlock()
449 if finalized {
450 s.cleanupUploadLock(uuid)
451 }
452 }()
453
454 if err := os.Remove(uploadPath); err != nil {
455 if os.IsNotExist(err) {
456 s.cleanupUploadLock(uuid)
457 return fmt.Errorf("upload not found: %s", uuid)
458 }
459 return fmt.Errorf("failed to cancel upload: %w", err)
460 }
461 finalized = true
462
463 s.log.Info().
464 Str(zerowrap.FieldLayer, "adapter").
465 Str(zerowrap.FieldAdapter, "filesystem").
466 Str("uuid", uuid).
467 Msg("blob upload cancelled")
468
469 return nil
470}
471
472// CleanupStaleUploads removes upload files older than maxAge.
473func (s *BlobStorage) CleanupStaleUploads(maxAge time.Duration) (int, int64, error) {
474 uploadsDir := filepath.Join(s.rootDir, "uploads")
475
476 entries, err := os.ReadDir(uploadsDir)
477 if err != nil {
478 if os.IsNotExist(err) {
479 return 0, 0, nil
480 }
481 return 0, 0, fmt.Errorf("failed to read uploads directory: %w", err)
482 }
483
484 cutoff := time.Now().Add(-maxAge)
485 var removed int
486 var bytesReclaimed int64
487
488 for _, entry := range entries {
489 if entry.IsDir() {
490 continue
491 }
492
493 info, err := entry.Info()
494 if err != nil {
495 s.log.Warn().Err(err).Str("file", entry.Name()).Msg("failed to get file info for stale upload")
496 continue
497 }
498
499 if info.ModTime().Before(cutoff) {
500 path := filepath.Join(uploadsDir, entry.Name())
501
502 mu := s.getUploadLock(entry.Name())
503 mu.Lock()
504 currentInfo, err := os.Stat(path)
505 if err != nil {
506 mu.Unlock()
507 if !os.IsNotExist(err) {
508 s.log.Warn().Err(err).Str("file", entry.Name()).Msg("failed to stat stale upload")
509 }
510 continue
511 }
512 if !currentInfo.ModTime().Before(cutoff) {
513 mu.Unlock()
514 continue
515 }
516 size := currentInfo.Size()
517 if err := os.Remove(path); err != nil {
518 mu.Unlock()
519 s.log.Warn().Err(err).Str("file", entry.Name()).Msg("failed to remove stale upload")
520 continue
521 }
522
523 mu.Unlock()
524 s.cleanupUploadLock(entry.Name())
525 removed++
526 bytesReclaimed += size
527
528 s.log.Info().
529 Str(zerowrap.FieldLayer, "adapter").
530 Str(zerowrap.FieldAdapter, "filesystem").
531 Str("uuid", entry.Name()).
532 Int64("size", size).
533 Msg("removed stale upload")
534 }
535 }
536
537 return removed, bytesReclaimed, nil
538}
539
540// Helper methods for path generation with security validation
541
542func (s *BlobStorage) getBlobPath(digest string) (string, error) {
543 // Validate digest to prevent path traversal (defense in depth)
544 if err := validation.ValidateDigest(digest); err != nil {
545 return "", fmt.Errorf("invalid digest: %w", err)
546 }
547
548 // Split digest into directory structure (e.g., sha256:abc123... -> sha256/ab/abc123...)
549 parts := strings.SplitN(digest, ":", 2)
550 if len(parts) != 2 {
551 return "", fmt.Errorf("invalid digest format: missing algorithm separator")
552 }
553
554 algorithm := parts[0]
555 hashPart := parts[1]
556
557 var path string
558 // Create two-level directory structure for better performance
559 if len(hashPart) >= 2 {
560 path = filepath.Join(s.rootDir, "blobs", algorithm, hashPart[:2], hashPart)
561 } else {
562 path = filepath.Join(s.rootDir, "blobs", algorithm, hashPart)
563 }
564
565 // Verify the path stays within root directory
566 if err := validation.ValidatePathWithinRoot(s.rootDir, path); err != nil {
567 return "", fmt.Errorf("path validation failed: %w", err)
568 }
569
570 return path, nil
571}
572
573func (s *BlobStorage) getUploadPath(uuid string) (string, error) {
574 // Validate UUID to prevent path traversal
575 if err := validation.ValidateUUID(uuid); err != nil {
576 return "", fmt.Errorf("invalid UUID: %w", err)
577 }
578
579 path := filepath.Join(s.rootDir, "uploads", uuid)
580
581 // Verify the path stays within root directory
582 if err := validation.ValidatePathWithinRoot(s.rootDir, path); err != nil {
583 return "", fmt.Errorf("path validation failed: %w", err)
584 }
585
586 return path, nil
587}
588
589func (s *BlobStorage) getUploadLock(uuid string) *sync.Mutex {
590 v, _ := s.uploadLocks.LoadOrStore(uuid, &sync.Mutex{})
591 return v.(*sync.Mutex)
592}
593
594func (s *BlobStorage) cleanupUploadLock(uuid string) {
595 s.uploadLocks.Delete(uuid)
596}
597
598// verifyUploadDigest computes and verifies the digest of an uploaded file.
599func (s *BlobStorage) verifyUploadDigest(uploadPath, expectedDigest string) error {
600 parts := strings.SplitN(expectedDigest, ":", 2)
601 if len(parts) != 2 {
602 return fmt.Errorf("invalid digest format")
603 }
604
605 algorithm := parts[0]
606 expectedHash := parts[1]
607
608 var hasher hash.Hash
609 switch algorithm {
610 case "sha256":
611 hasher = sha256.New()
612 case "sha512":
613 hasher = sha512.New()
614 default:
615 return fmt.Errorf("unsupported hash algorithm: %s", algorithm)
616 }
617
618 file, err := os.Open(uploadPath)
619 if err != nil {
620 return fmt.Errorf("failed to open upload file: %w", err)
621 }
622 defer file.Close()
623
624 if _, err := io.Copy(hasher, file); err != nil {
625 return fmt.Errorf("failed to compute digest: %w", err)
626 }
627
628 actualHash := hex.EncodeToString(hasher.Sum(nil))
629 if actualHash != expectedHash {
630 return fmt.Errorf("digest mismatch: expected %s, got %s", expectedHash, actualHash)
631 }
632
633 return nil
634}