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