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// StartBlobUpload starts a new blob upload and returns the upload UUID.
227func (s *BlobStorage) StartBlobUpload(name string) (string, error) {
228 uploadID := uuid.New().String()
229
230 uploadPath, err := s.getUploadPath(uploadID)
231 if err != nil {
232 return "", fmt.Errorf("invalid upload path: %w", err)
233 }
234
235 // Create directory if it doesn't exist
236 if err := os.MkdirAll(filepath.Dir(uploadPath), 0750); err != nil {
237 return "", fmt.Errorf("failed to create upload directory: %w", err)
238 }
239
240 // Create empty upload file
241 file, err := os.Create(uploadPath)
242 if err != nil {
243 return "", fmt.Errorf("failed to create upload file: %w", err)
244 }
245 file.Close()
246
247 s.log.Info().
248 Str(zerowrap.FieldLayer, "adapter").
249 Str(zerowrap.FieldAdapter, "filesystem").
250 Str("uuid", uploadID).
251 Str("name", name).
252 Msg("blob upload started")
253
254 return uploadID, nil
255}
256
257// AppendBlobChunk appends data to an in-progress upload.
258func (s *BlobStorage) AppendBlobChunk(name, uuid string, data io.Reader, contentLength, maxBlobSize int64) (int64, error) {
259 uploadPath, err := s.getUploadPath(uuid)
260 if err != nil {
261 return 0, fmt.Errorf("invalid upload path: %w", err)
262 }
263
264 mu := s.getUploadLock(uuid)
265 mu.Lock()
266 defer mu.Unlock()
267
268 file, err := os.OpenFile(uploadPath, os.O_WRONLY|os.O_APPEND, 0600)
269 if err != nil {
270 if os.IsNotExist(err) {
271 s.cleanupUploadLock(uuid)
272 return 0, fmt.Errorf("upload not found: %s", uuid)
273 }
274 return 0, fmt.Errorf("failed to open upload file: %w", err)
275 }
276 defer file.Close()
277
278 fi, err := file.Stat()
279 if err != nil {
280 return 0, fmt.Errorf("failed to get file info: %w", err)
281 }
282 originalSize := fi.Size()
283
284 if maxBlobSize > 0 {
285 remaining := maxBlobSize - originalSize
286 if remaining < 0 || (contentLength >= 0 && contentLength > remaining) {
287 return 0, domain.ErrBlobSizeExceeded
288 }
289 data = io.LimitReader(data, remaining+1)
290 }
291
292 written, err := io.Copy(file, data)
293 if err != nil {
294 return 0, fmt.Errorf("failed to write chunk to upload file: %w", err)
295 }
296
297 if maxBlobSize > 0 && originalSize+written > maxBlobSize {
298 if err := file.Truncate(originalSize); err != nil {
299 return 0, fmt.Errorf("failed to truncate oversized upload: %w", err)
300 }
301 return 0, domain.ErrBlobSizeExceeded
302 }
303
304 fi, err = file.Stat()
305 if err != nil {
306 return 0, fmt.Errorf("failed to get file info: %w", err)
307 }
308
309 s.log.Debug().
310 Str(zerowrap.FieldLayer, "adapter").
311 Str(zerowrap.FieldAdapter, "filesystem").
312 Str("uuid", uuid).
313 Str("name", name).
314 Int64("chunk_size", written).
315 Int64("total_size", fi.Size()).
316 Msg("appended chunk to blob upload")
317
318 return fi.Size(), nil
319}
320
321// GetBlobUpload returns a writer for the upload.
322// The returned WriteCloser holds the per-upload lock; callers must Close it to release the lock.
323func (s *BlobStorage) GetBlobUpload(uuid string) (io.WriteCloser, error) {
324 uploadPath, err := s.getUploadPath(uuid)
325 if err != nil {
326 return nil, fmt.Errorf("invalid upload path: %w", err)
327 }
328
329 mu := s.getUploadLock(uuid)
330 mu.Lock()
331
332 file, err := os.OpenFile(uploadPath, os.O_WRONLY|os.O_APPEND, 0600)
333 if err != nil {
334 mu.Unlock()
335 if os.IsNotExist(err) {
336 s.cleanupUploadLock(uuid)
337 return nil, fmt.Errorf("upload not found: %s", uuid)
338 }
339 return nil, fmt.Errorf("failed to open upload file: %w", err)
340 }
341
342 return &lockedWriteCloser{file: file, mu: mu}, nil
343}
344
345// lockedWriteCloser wraps an io.WriteCloser and releases a mutex on Close.
346type lockedWriteCloser struct {
347 file io.WriteCloser
348 mu *sync.Mutex
349 closeOnce sync.Once
350 closeErr error
351}
352
353func (lw *lockedWriteCloser) Write(p []byte) (int, error) {
354 return lw.file.Write(p)
355}
356
357func (lw *lockedWriteCloser) Close() error {
358 lw.closeOnce.Do(func() {
359 defer lw.mu.Unlock()
360 lw.closeErr = lw.file.Close()
361 })
362 return lw.closeErr
363}
364
365// FinishBlobUpload completes an upload and moves it to blob storage.
366func (s *BlobStorage) FinishBlobUpload(uuid, digest string) error {
367 uploadPath, err := s.getUploadPath(uuid)
368 if err != nil {
369 return fmt.Errorf("invalid upload path: %w", err)
370 }
371
372 mu := s.getUploadLock(uuid)
373 mu.Lock()
374 var finalized bool
375 defer func() {
376 mu.Unlock()
377 if finalized {
378 s.cleanupUploadLock(uuid)
379 }
380 }()
381
382 blobPath, err := s.getBlobPath(digest)
383 if err != nil {
384 return fmt.Errorf("invalid blob path: %w", err)
385 }
386
387 // Verify digest before moving to ensure integrity
388 if err := s.verifyUploadDigest(uploadPath, digest); err != nil {
389 return fmt.Errorf("digest verification failed: %w", err)
390 }
391
392 // Create blob directory if it doesn't exist
393 if err := os.MkdirAll(filepath.Dir(blobPath), 0750); err != nil {
394 return fmt.Errorf("failed to create blob directory: %w", err)
395 }
396
397 // Move upload file to blob location
398 if err := os.Rename(uploadPath, blobPath); err != nil {
399 return fmt.Errorf("failed to move upload to blob location: %w", err)
400 }
401 finalized = true
402
403 s.log.Info().
404 Str(zerowrap.FieldLayer, "adapter").
405 Str(zerowrap.FieldAdapter, "filesystem").
406 Str("uuid", uuid).
407 Str("digest", digest).
408 Msg("blob upload finished")
409
410 return nil
411}
412
413// CancelBlobUpload cancels an in-progress upload.
414func (s *BlobStorage) CancelBlobUpload(uuid string) error {
415 uploadPath, err := s.getUploadPath(uuid)
416 if err != nil {
417 return fmt.Errorf("invalid upload path: %w", err)
418 }
419
420 mu := s.getUploadLock(uuid)
421 mu.Lock()
422 var finalized bool
423 defer func() {
424 mu.Unlock()
425 if finalized {
426 s.cleanupUploadLock(uuid)
427 }
428 }()
429
430 if err := os.Remove(uploadPath); err != nil {
431 if os.IsNotExist(err) {
432 s.cleanupUploadLock(uuid)
433 return fmt.Errorf("upload not found: %s", uuid)
434 }
435 return fmt.Errorf("failed to cancel upload: %w", err)
436 }
437 finalized = true
438
439 s.log.Info().
440 Str(zerowrap.FieldLayer, "adapter").
441 Str(zerowrap.FieldAdapter, "filesystem").
442 Str("uuid", uuid).
443 Msg("blob upload cancelled")
444
445 return nil
446}
447
448// CleanupStaleUploads removes upload files older than maxAge.
449func (s *BlobStorage) CleanupStaleUploads(maxAge time.Duration) (int, int64, error) {
450 uploadsDir := filepath.Join(s.rootDir, "uploads")
451
452 entries, err := os.ReadDir(uploadsDir)
453 if err != nil {
454 if os.IsNotExist(err) {
455 return 0, 0, nil
456 }
457 return 0, 0, fmt.Errorf("failed to read uploads directory: %w", err)
458 }
459
460 cutoff := time.Now().Add(-maxAge)
461 var removed int
462 var bytesReclaimed int64
463
464 for _, entry := range entries {
465 if entry.IsDir() {
466 continue
467 }
468
469 info, err := entry.Info()
470 if err != nil {
471 s.log.Warn().Err(err).Str("file", entry.Name()).Msg("failed to get file info for stale upload")
472 continue
473 }
474
475 if info.ModTime().Before(cutoff) {
476 path := filepath.Join(uploadsDir, entry.Name())
477 size := info.Size()
478
479 mu := s.getUploadLock(entry.Name())
480 mu.Lock()
481
482 if err := os.Remove(path); err != nil {
483 mu.Unlock()
484 s.log.Warn().Err(err).Str("file", entry.Name()).Msg("failed to remove stale upload")
485 continue
486 }
487
488 mu.Unlock()
489 s.cleanupUploadLock(entry.Name())
490 removed++
491 bytesReclaimed += size
492
493 s.log.Info().
494 Str(zerowrap.FieldLayer, "adapter").
495 Str(zerowrap.FieldAdapter, "filesystem").
496 Str("uuid", entry.Name()).
497 Int64("size", size).
498 Msg("removed stale upload")
499 }
500 }
501
502 return removed, bytesReclaimed, nil
503}
504
505// Helper methods for path generation with security validation
506
507func (s *BlobStorage) getBlobPath(digest string) (string, error) {
508 // Validate digest to prevent path traversal (defense in depth)
509 if err := validation.ValidateDigest(digest); err != nil {
510 return "", fmt.Errorf("invalid digest: %w", err)
511 }
512
513 // Split digest into directory structure (e.g., sha256:abc123... -> sha256/ab/abc123...)
514 parts := strings.SplitN(digest, ":", 2)
515 if len(parts) != 2 {
516 return "", fmt.Errorf("invalid digest format: missing algorithm separator")
517 }
518
519 algorithm := parts[0]
520 hashPart := parts[1]
521
522 var path string
523 // Create two-level directory structure for better performance
524 if len(hashPart) >= 2 {
525 path = filepath.Join(s.rootDir, "blobs", algorithm, hashPart[:2], hashPart)
526 } else {
527 path = filepath.Join(s.rootDir, "blobs", algorithm, hashPart)
528 }
529
530 // Verify the path stays within root directory
531 if err := validation.ValidatePathWithinRoot(s.rootDir, path); err != nil {
532 return "", fmt.Errorf("path validation failed: %w", err)
533 }
534
535 return path, nil
536}
537
538func (s *BlobStorage) getUploadPath(uuid string) (string, error) {
539 // Validate UUID to prevent path traversal
540 if err := validation.ValidateUUID(uuid); err != nil {
541 return "", fmt.Errorf("invalid UUID: %w", err)
542 }
543
544 path := filepath.Join(s.rootDir, "uploads", uuid)
545
546 // Verify the path stays within root directory
547 if err := validation.ValidatePathWithinRoot(s.rootDir, path); err != nil {
548 return "", fmt.Errorf("path validation failed: %w", err)
549 }
550
551 return path, nil
552}
553
554func (s *BlobStorage) getUploadLock(uuid string) *sync.Mutex {
555 v, _ := s.uploadLocks.LoadOrStore(uuid, &sync.Mutex{})
556 return v.(*sync.Mutex)
557}
558
559func (s *BlobStorage) cleanupUploadLock(uuid string) {
560 s.uploadLocks.Delete(uuid)
561}
562
563// verifyUploadDigest computes and verifies the digest of an uploaded file.
564func (s *BlobStorage) verifyUploadDigest(uploadPath, expectedDigest string) error {
565 parts := strings.SplitN(expectedDigest, ":", 2)
566 if len(parts) != 2 {
567 return fmt.Errorf("invalid digest format")
568 }
569
570 algorithm := parts[0]
571 expectedHash := parts[1]
572
573 var hasher hash.Hash
574 switch algorithm {
575 case "sha256":
576 hasher = sha256.New()
577 case "sha512":
578 hasher = sha512.New()
579 default:
580 return fmt.Errorf("unsupported hash algorithm: %s", algorithm)
581 }
582
583 file, err := os.Open(uploadPath)
584 if err != nil {
585 return fmt.Errorf("failed to open upload file: %w", err)
586 }
587 defer file.Close()
588
589 if _, err := io.Copy(hasher, file); err != nil {
590 return fmt.Errorf("failed to compute digest: %w", err)
591 }
592
593 actualHash := hex.EncodeToString(hasher.Sum(nil))
594 if actualHash != expectedHash {
595 return fmt.Errorf("digest mismatch: expected %s, got %s", expectedHash, actualHash)
596 }
597
598 return nil
599}