1package s3
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "fmt"
8 "io"
9 "net/url"
10 "path"
11 "sort"
12 "strings"
13 "time"
14
15 "github.com/aws/aws-sdk-go-v2/aws"
16 "github.com/aws/aws-sdk-go-v2/config"
17 "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager"
18 tmtypes "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager/types"
19 awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
20
21 "github.com/bnema/gordon/internal/domain"
22)
23
24const volumeBackupTimestampLayout = "20060102T150405Z"
25
26type volumeS3Client interface {
27 GetObject(ctx context.Context, params *awss3.GetObjectInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectOutput, error)
28 HeadObject(ctx context.Context, params *awss3.HeadObjectInput, optFns ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error)
29 ListObjectsV2(ctx context.Context, params *awss3.ListObjectsV2Input, optFns ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error)
30 DeleteObject(ctx context.Context, params *awss3.DeleteObjectInput, optFns ...func(*awss3.Options)) (*awss3.DeleteObjectOutput, error)
31}
32
33type volumeUploader interface {
34 UploadObject(ctx context.Context, input *transfermanager.UploadObjectInput, opts ...func(*transfermanager.Options)) (*transfermanager.UploadObjectOutput, error)
35}
36
37// VolumeBackupStorage stores volume archive backups in S3.
38type VolumeBackupStorage struct {
39 bucket string
40 prefix string
41 sseAlgorithm string
42 sseKMSKeyID string
43 client volumeS3Client
44 uploader volumeUploader
45}
46
47// NewVolumeBackupStorage creates an S3-backed volume backup storage adapter.
48func NewVolumeBackupStorage(ctx context.Context, cfg domain.VolumeBackupConfig) (*VolumeBackupStorage, error) {
49 awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(cfg.S3Region))
50 if err != nil {
51 return nil, fmt.Errorf("failed to load AWS config: %w", err)
52 }
53
54 client := awss3.NewFromConfig(awsCfg, func(o *awss3.Options) {
55 if cfg.S3Endpoint != "" {
56 o.BaseEndpoint = aws.String(cfg.S3Endpoint)
57 }
58 o.UsePathStyle = cfg.S3PathStyle
59 })
60 uploader := transfermanager.New(client, func(o *transfermanager.Options) {
61 o.Concurrency = 2
62 o.PartSizeBytes = 16 * 1024 * 1024
63 o.MultipartUploadThreshold = 16 * 1024 * 1024
64 o.FailTimeout = time.Minute
65 })
66
67 return NewVolumeBackupStorageWithClients(cfg, client, uploader), nil
68}
69
70// NewVolumeBackupStorageWithClients creates storage with injected clients for tests.
71func NewVolumeBackupStorageWithClients(cfg domain.VolumeBackupConfig, client volumeS3Client, uploader volumeUploader) *VolumeBackupStorage {
72 return &VolumeBackupStorage{
73 bucket: strings.TrimSpace(cfg.S3Bucket),
74 prefix: normalizeS3Prefix(cfg.S3Prefix),
75 sseAlgorithm: strings.TrimSpace(cfg.S3SSEAlgorithm),
76 sseKMSKeyID: strings.TrimSpace(cfg.S3SSEKMSKeyID),
77 client: client,
78 uploader: uploader,
79 }
80}
81
82// StoreVolumeArchive uploads a volume archive stream to S3 and returns its artifact reference.
83func (s *VolumeBackupStorage) StoreVolumeArchive(ctx context.Context, job domain.VolumeBackupJob, data io.Reader) (string, error) {
84 if s.bucket == "" {
85 return "", fmt.Errorf("s3 bucket is required")
86 }
87 if job.App == "" {
88 return "", fmt.Errorf("backup app is required")
89 }
90 if job.VolumeName == "" {
91 return "", fmt.Errorf("backup volume name is required")
92 }
93 if data == nil {
94 return "", fmt.Errorf("backup data is required")
95 }
96
97 key := s.objectKey(job)
98 compression := ""
99 if job.Metadata != nil {
100 compression = job.Metadata["compression"]
101 }
102 input := &transfermanager.UploadObjectInput{
103 Bucket: aws.String(s.bucket),
104 Key: aws.String(key),
105 Body: data,
106 ContentType: aws.String(contentTypeForCompression(compression)),
107 Metadata: map[string]string{
108 "gordon-app": job.App,
109 "gordon-service": job.Service,
110 "gordon-volume": job.VolumeName,
111 "gordon-container": job.ContainerName,
112 "gordon-mount-path": job.MountPath,
113 "gordon-compression": compression,
114 },
115 }
116 if s.sseAlgorithm != "" {
117 input.ServerSideEncryption = tmtypes.ServerSideEncryption(s.sseAlgorithm)
118 }
119 if s.sseKMSKeyID != "" {
120 input.SSEKMSKeyID = aws.String(s.sseKMSKeyID)
121 }
122
123 if _, err := s.uploader.UploadObject(ctx, input); err != nil {
124 return "", fmt.Errorf("failed to upload volume archive: %w", err)
125 }
126
127 return s.artifactRef(key), nil
128}
129
130// GetVolumeArchive retrieves a volume archive by artifact reference.
131func (s *VolumeBackupStorage) GetVolumeArchive(ctx context.Context, artifactRef string) (io.ReadCloser, error) {
132 key, err := s.keyFromArtifactRef(artifactRef)
133 if err != nil {
134 return nil, err
135 }
136 out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{
137 Bucket: aws.String(s.bucket),
138 Key: aws.String(key),
139 })
140 if err != nil {
141 return nil, fmt.Errorf("failed to get volume archive: %w", err)
142 }
143 return out.Body, nil
144}
145
146// ListVolumeArchives lists completed volume archive backups for a domain.
147func (s *VolumeBackupStorage) ListVolumeArchives(ctx context.Context, app string) ([]domain.VolumeBackupJob, error) {
148 listPrefix := s.appPrefix(app)
149 jobs := make([]domain.VolumeBackupJob, 0)
150 var token *string
151 for {
152 out, err := s.client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
153 Bucket: aws.String(s.bucket),
154 Prefix: aws.String(listPrefix),
155 ContinuationToken: token,
156 })
157 if err != nil {
158 return nil, fmt.Errorf("failed to list volume archives: %w", err)
159 }
160 for _, obj := range out.Contents {
161 if obj.Key == nil {
162 continue
163 }
164 job, ok := s.jobFromKey(app, *obj.Key)
165 if !ok {
166 continue
167 }
168 if obj.Size != nil {
169 job.SizeBytes = *obj.Size
170 }
171 s.hydrateVolumeBackupJobMetadata(ctx, &job, *obj.Key)
172 jobs = append(jobs, job)
173 }
174 if out.IsTruncated == nil || !*out.IsTruncated {
175 break
176 }
177 token = out.NextContinuationToken
178 }
179
180 sortVolumeBackupJobs(jobs)
181 return jobs, nil
182}
183
184// DeleteVolumeArchive deletes a volume archive by artifact reference.
185func (s *VolumeBackupStorage) hydrateVolumeBackupJobMetadata(ctx context.Context, job *domain.VolumeBackupJob, key string) {
186 out, err := s.client.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(key)})
187 if err != nil || out == nil || out.Metadata == nil {
188 return
189 }
190 job.Service = out.Metadata["gordon-service"]
191 job.ContainerName = out.Metadata["gordon-container"]
192 job.MountPath = out.Metadata["gordon-mount-path"]
193 if compression := out.Metadata["gordon-compression"]; compression != "" {
194 if job.Metadata == nil {
195 job.Metadata = make(map[string]string)
196 }
197 job.Metadata["compression"] = compression
198 }
199}
200
201func (s *VolumeBackupStorage) DeleteVolumeArchive(ctx context.Context, artifactRef string) error {
202 key, err := s.keyFromArtifactRef(artifactRef)
203 if err != nil {
204 return err
205 }
206 _, err = s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{
207 Bucket: aws.String(s.bucket),
208 Key: aws.String(key),
209 })
210 if err != nil {
211 return fmt.Errorf("failed to delete volume archive: %w", err)
212 }
213 return nil
214}
215
216// ApplyVolumeRetention deletes old completed volume archives according to the keep count.
217func (s *VolumeBackupStorage) ApplyVolumeRetention(ctx context.Context, app string, policy domain.VolumeBackupRetentionPolicy) (int, error) {
218 if policy.Keep < 0 {
219 return 0, fmt.Errorf("volume backup retention keep cannot be negative")
220 }
221 if policy.Keep == 0 {
222 return 0, nil
223 }
224
225 jobs, err := s.ListVolumeArchives(ctx, app)
226 if err != nil {
227 return 0, err
228 }
229
230 byVolume := make(map[string][]domain.VolumeBackupJob)
231 for _, job := range jobs {
232 byVolume[job.VolumeName] = append(byVolume[job.VolumeName], job)
233 }
234
235 deleted := 0
236 for _, group := range byVolume {
237 sortVolumeBackupJobs(group)
238 for i := policy.Keep; i < len(group); i++ {
239 if err := s.DeleteVolumeArchive(ctx, group[i].ArtifactRef); err != nil {
240 return deleted, err
241 }
242 deleted++
243 }
244 }
245 return deleted, nil
246}
247
248func (s *VolumeBackupStorage) objectKey(job domain.VolumeBackupJob) string {
249 started := job.StartedAt.UTC()
250 if started.IsZero() {
251 started = time.Now().UTC()
252 }
253 id := job.ID
254 if id == "" {
255 id = randomObjectSuffix()
256 }
257
258 ext := "tar.zst"
259 if job.Metadata != nil && job.Metadata["compression"] == string(domain.VolumeBackupCompressionGzip) {
260 ext = "tar.gz"
261 }
262 fileName := fmt.Sprintf("%s-%s.%s", started.Format(volumeBackupTimestampLayout), sanitizeS3KeyComponent(id), ext)
263 parts := []string{s.prefix, "apps", sanitizeS3KeyComponent(job.App), "volumes", sanitizeS3KeyComponent(job.VolumeName), fileName}
264 return joinS3Key(parts...)
265}
266
267// appPrefix is the object prefix of one app's volume archives, or of every
268// app when app is empty. Object keys are app-based; artifacts written by
269// older versions under a domain prefix are never read or migrated here.
270func (s *VolumeBackupStorage) appPrefix(app string) string {
271 if strings.TrimSpace(app) == "" {
272 return joinS3Key(s.prefix, "apps") + "/"
273 }
274 return joinS3Key(s.prefix, "apps", sanitizeS3KeyComponent(app), "volumes") + "/"
275}
276
277func (s *VolumeBackupStorage) artifactRef(key string) string {
278 return "s3://" + s.bucket + "/" + key
279}
280
281func (s *VolumeBackupStorage) keyFromArtifactRef(artifactRef string) (string, error) {
282 if strings.HasPrefix(artifactRef, "s3://") {
283 u, err := url.Parse(artifactRef)
284 if err != nil {
285 return "", fmt.Errorf("invalid s3 artifact ref: %w", err)
286 }
287 if u.Host != s.bucket {
288 return "", fmt.Errorf("artifact bucket %q does not match configured bucket", u.Host)
289 }
290 return strings.TrimPrefix(u.Path, "/"), nil
291 }
292 if strings.TrimSpace(artifactRef) == "" {
293 return "", fmt.Errorf("artifact ref is required")
294 }
295 return strings.TrimPrefix(artifactRef, "/"), nil
296}
297
298func (s *VolumeBackupStorage) jobFromKey(app, key string) (domain.VolumeBackupJob, bool) {
299 var rel string
300 if strings.TrimSpace(app) == "" {
301 rel = strings.TrimPrefix(key, joinS3Key(s.prefix, "apps")+"/")
302 parts := strings.Split(rel, "/")
303 if len(parts) != 4 || parts[1] != "volumes" {
304 return domain.VolumeBackupJob{}, false
305 }
306 app = parts[0]
307 rel = strings.Join(parts[2:], "/")
308 } else {
309 rel = strings.TrimPrefix(key, joinS3Key(s.prefix, "apps", sanitizeS3KeyComponent(app), "volumes")+"/")
310 }
311 parts := strings.Split(rel, "/")
312 if len(parts) != 2 {
313 return domain.VolumeBackupJob{}, false
314 }
315 volumeName := parts[0]
316 fileName := parts[1]
317 if volumeName == "" || fileName == "" {
318 return domain.VolumeBackupJob{}, false
319 }
320 started, id, ok := parseVolumeBackupFileName(fileName)
321 if !ok {
322 return domain.VolumeBackupJob{}, false
323 }
324 return domain.VolumeBackupJob{
325 ID: id,
326 App: app,
327 VolumeName: volumeName,
328 Type: domain.BackupTypeVolumeArchive,
329 Status: domain.BackupStatusCompleted,
330 StartedAt: started,
331 CompletedAt: started,
332 ArtifactRef: s.artifactRef(key),
333 }, true
334}
335
336func parseVolumeBackupFileName(fileName string) (time.Time, string, bool) {
337 trimmed := strings.TrimSuffix(strings.TrimSuffix(fileName, ".tar.zst"), ".tar.gz")
338 if trimmed == fileName {
339 return time.Time{}, "", false
340 }
341 timestampPart, id, ok := strings.Cut(trimmed, "-")
342 if !ok {
343 return time.Time{}, "", false
344 }
345 started, err := time.Parse(volumeBackupTimestampLayout, timestampPart)
346 if err != nil {
347 return time.Time{}, "", false
348 }
349 return started.UTC(), id, true
350}
351
352func sortVolumeBackupJobs(jobs []domain.VolumeBackupJob) {
353 sort.Slice(jobs, func(i, j int) bool {
354 if !jobs[i].StartedAt.Equal(jobs[j].StartedAt) {
355 return jobs[i].StartedAt.After(jobs[j].StartedAt)
356 }
357 return jobs[i].ArtifactRef < jobs[j].ArtifactRef
358 })
359}
360
361func normalizeS3Prefix(prefix string) string {
362 return strings.Trim(strings.TrimSpace(prefix), "/")
363}
364
365func joinS3Key(parts ...string) string {
366 clean := make([]string, 0, len(parts))
367 for _, part := range parts {
368 part = strings.Trim(part, "/")
369 if part != "" {
370 clean = append(clean, part)
371 }
372 }
373 return path.Join(clean...)
374}
375
376func sanitizeS3KeyComponent(input string) string {
377 clean := strings.TrimSpace(input)
378 if clean == "" || strings.Trim(clean, ".") == "" {
379 return "unknown"
380 }
381 clean = strings.Trim(clean, ".")
382 replacer := strings.NewReplacer("/", "_", "\\", "_", ":", "_", " ", "_")
383 return replacer.Replace(clean)
384}
385
386func contentTypeForCompression(compression string) string {
387 switch compression {
388 case string(domain.VolumeBackupCompressionGzip):
389 return "application/gzip"
390 case string(domain.VolumeBackupCompressionZstd):
391 return "application/zstd"
392 default:
393 return "application/octet-stream"
394 }
395}
396
397func randomObjectSuffix() string {
398 var b [8]byte
399 if _, err := rand.Read(b[:]); err != nil {
400 return fmt.Sprintf("%d", time.Now().UTC().UnixNano())
401 }
402 return hex.EncodeToString(b[:])
403}