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