1// Package filesystem implements storage adapters using the local filesystem.
2package filesystem
3
4import (
5 "fmt"
6 "io"
7 "os"
8 "path/filepath"
9 "strings"
10 "time"
11
12 "github.com/bnema/zerowrap"
13)
14
15// BlobStorage implements the BlobStorage interface using the local filesystem.
16type BlobStorage struct {
17 rootDir string
18 log zerowrap.Logger
19}
20
21// NewBlobStorage creates a new filesystem blob storage instance.
22func NewBlobStorage(rootDir string, log zerowrap.Logger) (*BlobStorage, error) {
23 // Create directory structure if it doesn't exist
24 dirs := []string{
25 filepath.Join(rootDir, "blobs"),
26 filepath.Join(rootDir, "uploads"),
27 }
28
29 for _, dir := range dirs {
30 if err := os.MkdirAll(dir, 0750); err != nil {
31 return nil, fmt.Errorf("failed to create directory %s: %w", dir, err)
32 }
33 }
34
35 log.Info().
36 Str(zerowrap.FieldLayer, "adapter").
37 Str(zerowrap.FieldAdapter, "filesystem").
38 Str("root_dir", rootDir).
39 Msg("blob storage initialized")
40
41 return &BlobStorage{
42 rootDir: rootDir,
43 log: log,
44 }, nil
45}
46
47// GetBlob retrieves a blob by digest.
48func (s *BlobStorage) GetBlob(digest string) (io.ReadCloser, error) {
49 blobPath := s.getBlobPath(digest)
50
51 file, err := os.Open(blobPath)
52 if err != nil {
53 if os.IsNotExist(err) {
54 return nil, fmt.Errorf("blob not found: %s", digest)
55 }
56 return nil, fmt.Errorf("failed to open blob: %w", err)
57 }
58
59 return file, nil
60}
61
62// GetBlobPath returns the filesystem path to a blob.
63func (s *BlobStorage) GetBlobPath(digest string) (string, error) {
64 path := s.getBlobPath(digest)
65 if _, err := os.Stat(path); err != nil {
66 if os.IsNotExist(err) {
67 return "", fmt.Errorf("blob not found: %s", digest)
68 }
69 return "", err
70 }
71 return path, nil
72}
73
74// PutBlob stores a blob with the given digest.
75func (s *BlobStorage) PutBlob(digest string, data io.Reader, size int64) error {
76 blobPath := s.getBlobPath(digest)
77
78 // Create directory if it doesn't exist
79 if err := os.MkdirAll(filepath.Dir(blobPath), 0750); err != nil {
80 return fmt.Errorf("failed to create blob directory: %w", err)
81 }
82
83 // Create temporary file first
84 tmpPath := blobPath + ".tmp"
85 file, err := os.Create(tmpPath)
86 if err != nil {
87 return fmt.Errorf("failed to create temporary blob file: %w", err)
88 }
89 defer file.Close()
90
91 // Copy data to file
92 written, err := io.Copy(file, data)
93 if err != nil {
94 os.Remove(tmpPath)
95 return fmt.Errorf("failed to write blob data: %w", err)
96 }
97
98 // Verify size if provided
99 if size > 0 && written != size {
100 os.Remove(tmpPath)
101 return fmt.Errorf("blob size mismatch: expected %d, got %d", size, written)
102 }
103
104 // Move to final location
105 if err := os.Rename(tmpPath, blobPath); err != nil {
106 os.Remove(tmpPath)
107 return fmt.Errorf("failed to move blob to final location: %w", err)
108 }
109
110 s.log.Info().
111 Str(zerowrap.FieldLayer, "adapter").
112 Str(zerowrap.FieldAdapter, "filesystem").
113 Str("digest", digest).
114 Int64(zerowrap.FieldSize, written).
115 Msg("blob stored")
116
117 return nil
118}
119
120// DeleteBlob removes a blob by digest.
121func (s *BlobStorage) DeleteBlob(digest string) error {
122 blobPath := s.getBlobPath(digest)
123
124 if err := os.Remove(blobPath); err != nil {
125 if os.IsNotExist(err) {
126 return fmt.Errorf("blob not found: %s", digest)
127 }
128 return fmt.Errorf("failed to delete blob: %w", err)
129 }
130
131 s.log.Info().
132 Str(zerowrap.FieldLayer, "adapter").
133 Str(zerowrap.FieldAdapter, "filesystem").
134 Str("digest", digest).
135 Msg("blob deleted")
136
137 return nil
138}
139
140// BlobExists checks if a blob exists.
141func (s *BlobStorage) BlobExists(digest string) bool {
142 blobPath := s.getBlobPath(digest)
143 _, err := os.Stat(blobPath)
144 return err == nil
145}
146
147// StartBlobUpload starts a new blob upload and returns the upload UUID.
148func (s *BlobStorage) StartBlobUpload(name string) (string, error) {
149 // Generate UUID-like upload ID using timestamp
150 uuid := fmt.Sprintf("%d-%s", time.Now().UnixNano(), strings.ReplaceAll(name, "/", "_"))
151
152 uploadPath := s.getUploadPath(uuid)
153
154 // Create directory if it doesn't exist
155 if err := os.MkdirAll(filepath.Dir(uploadPath), 0750); err != nil {
156 return "", fmt.Errorf("failed to create upload directory: %w", err)
157 }
158
159 // Create empty upload file
160 file, err := os.Create(uploadPath)
161 if err != nil {
162 return "", fmt.Errorf("failed to create upload file: %w", err)
163 }
164 file.Close()
165
166 s.log.Info().
167 Str(zerowrap.FieldLayer, "adapter").
168 Str(zerowrap.FieldAdapter, "filesystem").
169 Str("uuid", uuid).
170 Str("name", name).
171 Msg("blob upload started")
172
173 return uuid, nil
174}
175
176// AppendBlobChunk appends data to an in-progress upload.
177func (s *BlobStorage) AppendBlobChunk(name, uuid string, chunk []byte) (int64, error) {
178 uploadPath := s.getUploadPath(uuid)
179
180 file, err := os.OpenFile(uploadPath, os.O_WRONLY|os.O_APPEND, 0600)
181 if err != nil {
182 if os.IsNotExist(err) {
183 return 0, fmt.Errorf("upload not found: %s", uuid)
184 }
185 return 0, fmt.Errorf("failed to open upload file: %w", err)
186 }
187 defer file.Close()
188
189 written, err := file.Write(chunk)
190 if err != nil {
191 return 0, fmt.Errorf("failed to write chunk to upload file: %w", err)
192 }
193
194 fi, err := file.Stat()
195 if err != nil {
196 return 0, fmt.Errorf("failed to get file info: %w", err)
197 }
198
199 s.log.Debug().
200 Str(zerowrap.FieldLayer, "adapter").
201 Str(zerowrap.FieldAdapter, "filesystem").
202 Str("uuid", uuid).
203 Str("name", name).
204 Int("chunk_size", written).
205 Int64("total_size", fi.Size()).
206 Msg("appended chunk to blob upload")
207
208 return fi.Size(), nil
209}
210
211// GetBlobUpload returns a writer for the upload.
212func (s *BlobStorage) GetBlobUpload(uuid string) (io.WriteCloser, error) {
213 uploadPath := s.getUploadPath(uuid)
214
215 file, err := os.OpenFile(uploadPath, os.O_WRONLY|os.O_APPEND, 0600)
216 if err != nil {
217 if os.IsNotExist(err) {
218 return nil, fmt.Errorf("upload not found: %s", uuid)
219 }
220 return nil, fmt.Errorf("failed to open upload file: %w", err)
221 }
222
223 return file, nil
224}
225
226// FinishBlobUpload completes an upload and moves it to blob storage.
227func (s *BlobStorage) FinishBlobUpload(uuid, digest string) error {
228 uploadPath := s.getUploadPath(uuid)
229 blobPath := s.getBlobPath(digest)
230
231 // Create blob directory if it doesn't exist
232 if err := os.MkdirAll(filepath.Dir(blobPath), 0750); err != nil {
233 return fmt.Errorf("failed to create blob directory: %w", err)
234 }
235
236 // Move upload file to blob location
237 if err := os.Rename(uploadPath, blobPath); err != nil {
238 return fmt.Errorf("failed to move upload to blob location: %w", err)
239 }
240
241 s.log.Info().
242 Str(zerowrap.FieldLayer, "adapter").
243 Str(zerowrap.FieldAdapter, "filesystem").
244 Str("uuid", uuid).
245 Str("digest", digest).
246 Msg("blob upload finished")
247
248 return nil
249}
250
251// CancelBlobUpload cancels an in-progress upload.
252func (s *BlobStorage) CancelBlobUpload(uuid string) error {
253 uploadPath := s.getUploadPath(uuid)
254
255 if err := os.Remove(uploadPath); err != nil {
256 if os.IsNotExist(err) {
257 return fmt.Errorf("upload not found: %s", uuid)
258 }
259 return fmt.Errorf("failed to cancel upload: %w", err)
260 }
261
262 s.log.Info().
263 Str(zerowrap.FieldLayer, "adapter").
264 Str(zerowrap.FieldAdapter, "filesystem").
265 Str("uuid", uuid).
266 Msg("blob upload cancelled")
267
268 return nil
269}
270
271// Helper methods for path generation
272
273func (s *BlobStorage) getBlobPath(digest string) string {
274 // Split digest into directory structure (e.g., sha256:abc123... -> sha256/ab/abc123...)
275 parts := strings.SplitN(digest, ":", 2)
276 if len(parts) != 2 {
277 // Fallback if digest format is unexpected
278 return filepath.Join(s.rootDir, "blobs", digest)
279 }
280
281 algorithm := parts[0]
282 hash := parts[1]
283
284 // Create two-level directory structure for better performance
285 if len(hash) >= 2 {
286 return filepath.Join(s.rootDir, "blobs", algorithm, hash[:2], hash)
287 }
288
289 return filepath.Join(s.rootDir, "blobs", algorithm, hash)
290}
291
292func (s *BlobStorage) getUploadPath(uuid string) string {
293 return filepath.Join(s.rootDir, "uploads", uuid)
294}