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.Domain == "" {
88 return "", fmt.Errorf("backup domain 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-domain": job.Domain,
109 "gordon-volume": job.VolumeName,
110 "gordon-container": job.ContainerName,
111 "gordon-mount-path": job.MountPath,
112 "gordon-compression": compression,
113 },
114 }
115 if s.sseAlgorithm != "" {
116 input.ServerSideEncryption = tmtypes.ServerSideEncryption(s.sseAlgorithm)
117 }
118 if s.sseKMSKeyID != "" {
119 input.SSEKMSKeyID = aws.String(s.sseKMSKeyID)
120 }
121
122 if _, err := s.uploader.UploadObject(ctx, input); err != nil {
123 return "", fmt.Errorf("failed to upload volume archive: %w", err)
124 }
125
126 return s.artifactRef(key), nil
127}
128
129// GetVolumeArchive retrieves a volume archive by artifact reference.
130func (s *VolumeBackupStorage) GetVolumeArchive(ctx context.Context, artifactRef string) (io.ReadCloser, error) {
131 key, err := s.keyFromArtifactRef(artifactRef)
132 if err != nil {
133 return nil, err
134 }
135 out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{
136 Bucket: aws.String(s.bucket),
137 Key: aws.String(key),
138 })
139 if err != nil {
140 return nil, fmt.Errorf("failed to get volume archive: %w", err)
141 }
142 return out.Body, nil
143}
144
145// ListVolumeArchives lists completed volume archive backups for a domain.
146func (s *VolumeBackupStorage) ListVolumeArchives(ctx context.Context, domainName string) ([]domain.VolumeBackupJob, error) {
147 listPrefix := s.domainPrefix(domainName)
148 jobs := make([]domain.VolumeBackupJob, 0)
149 var token *string
150 for {
151 out, err := s.client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
152 Bucket: aws.String(s.bucket),
153 Prefix: aws.String(listPrefix),
154 ContinuationToken: token,
155 })
156 if err != nil {
157 return nil, fmt.Errorf("failed to list volume archives: %w", err)
158 }
159 for _, obj := range out.Contents {
160 if obj.Key == nil {
161 continue
162 }
163 job, ok := s.jobFromKey(domainName, *obj.Key)
164 if !ok {
165 continue
166 }
167 if obj.Size != nil {
168 job.SizeBytes = *obj.Size
169 }
170 s.hydrateVolumeBackupJobMetadata(ctx, &job, *obj.Key)
171 jobs = append(jobs, job)
172 }
173 if out.IsTruncated == nil || !*out.IsTruncated {
174 break
175 }
176 token = out.NextContinuationToken
177 }
178
179 sortVolumeBackupJobs(jobs)
180 return jobs, nil
181}
182
183// DeleteVolumeArchive deletes a volume archive by artifact reference.
184func (s *VolumeBackupStorage) hydrateVolumeBackupJobMetadata(ctx context.Context, job *domain.VolumeBackupJob, key string) {
185 out, err := s.client.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(key)})
186 if err != nil || out == nil || out.Metadata == nil {
187 return
188 }
189 job.ContainerName = out.Metadata["gordon-container"]
190 job.MountPath = out.Metadata["gordon-mount-path"]
191 if compression := out.Metadata["gordon-compression"]; compression != "" {
192 if job.Metadata == nil {
193 job.Metadata = make(map[string]string)
194 }
195 job.Metadata["compression"] = compression
196 }
197}
198
199func (s *VolumeBackupStorage) DeleteVolumeArchive(ctx context.Context, artifactRef string) error {
200 key, err := s.keyFromArtifactRef(artifactRef)
201 if err != nil {
202 return err
203 }
204 _, err = s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{
205 Bucket: aws.String(s.bucket),
206 Key: aws.String(key),
207 })
208 if err != nil {
209 return fmt.Errorf("failed to delete volume archive: %w", err)
210 }
211 return nil
212}
213
214// ApplyVolumeRetention deletes old completed volume archives according to the keep count.
215func (s *VolumeBackupStorage) ApplyVolumeRetention(ctx context.Context, domainName string, policy domain.VolumeBackupRetentionPolicy) (int, error) {
216 if policy.Keep < 0 {
217 return 0, fmt.Errorf("volume backup retention keep cannot be negative")
218 }
219 if policy.Keep == 0 {
220 return 0, nil
221 }
222
223 jobs, err := s.ListVolumeArchives(ctx, domainName)
224 if err != nil {
225 return 0, err
226 }
227
228 byVolume := make(map[string][]domain.VolumeBackupJob)
229 for _, job := range jobs {
230 byVolume[job.VolumeName] = append(byVolume[job.VolumeName], job)
231 }
232
233 deleted := 0
234 for _, group := range byVolume {
235 sortVolumeBackupJobs(group)
236 for i := policy.Keep; i < len(group); i++ {
237 if err := s.DeleteVolumeArchive(ctx, group[i].ArtifactRef); err != nil {
238 return deleted, err
239 }
240 deleted++
241 }
242 }
243 return deleted, nil
244}
245
246func (s *VolumeBackupStorage) objectKey(job domain.VolumeBackupJob) string {
247 started := job.StartedAt.UTC()
248 if started.IsZero() {
249 started = time.Now().UTC()
250 }
251 id := job.ID
252 if id == "" {
253 id = randomObjectSuffix()
254 }
255
256 ext := "tar.zst"
257 if job.Metadata != nil && job.Metadata["compression"] == string(domain.VolumeBackupCompressionGzip) {
258 ext = "tar.gz"
259 }
260 fileName := fmt.Sprintf("%s-%s.%s", started.Format(volumeBackupTimestampLayout), sanitizeS3KeyComponent(id), ext)
261 parts := []string{s.prefix, "domains", sanitizeS3KeyComponent(job.Domain), "volumes", sanitizeS3KeyComponent(job.VolumeName), fileName}
262 return joinS3Key(parts...)
263}
264
265func (s *VolumeBackupStorage) domainPrefix(domainName string) string {
266 if strings.TrimSpace(domainName) == "" {
267 return joinS3Key(s.prefix, "domains") + "/"
268 }
269 return joinS3Key(s.prefix, "domains", sanitizeS3KeyComponent(domainName), "volumes") + "/"
270}
271
272func (s *VolumeBackupStorage) artifactRef(key string) string {
273 return "s3://" + s.bucket + "/" + key
274}
275
276func (s *VolumeBackupStorage) keyFromArtifactRef(artifactRef string) (string, error) {
277 if strings.HasPrefix(artifactRef, "s3://") {
278 u, err := url.Parse(artifactRef)
279 if err != nil {
280 return "", fmt.Errorf("invalid s3 artifact ref: %w", err)
281 }
282 if u.Host != s.bucket {
283 return "", fmt.Errorf("artifact bucket %q does not match configured bucket", u.Host)
284 }
285 return strings.TrimPrefix(u.Path, "/"), nil
286 }
287 if strings.TrimSpace(artifactRef) == "" {
288 return "", fmt.Errorf("artifact ref is required")
289 }
290 return strings.TrimPrefix(artifactRef, "/"), nil
291}
292
293func (s *VolumeBackupStorage) jobFromKey(domainName, key string) (domain.VolumeBackupJob, bool) {
294 var rel string
295 if strings.TrimSpace(domainName) == "" {
296 rel = strings.TrimPrefix(key, joinS3Key(s.prefix, "domains")+"/")
297 parts := strings.Split(rel, "/")
298 if len(parts) != 4 || parts[1] != "volumes" {
299 return domain.VolumeBackupJob{}, false
300 }
301 domainName = parts[0]
302 rel = strings.Join(parts[2:], "/")
303 } else {
304 rel = strings.TrimPrefix(key, joinS3Key(s.prefix, "domains", sanitizeS3KeyComponent(domainName), "volumes")+"/")
305 }
306 parts := strings.Split(rel, "/")
307 if len(parts) != 2 {
308 return domain.VolumeBackupJob{}, false
309 }
310 volumeName := parts[0]
311 fileName := parts[1]
312 if volumeName == "" || fileName == "" {
313 return domain.VolumeBackupJob{}, false
314 }
315 started, id, ok := parseVolumeBackupFileName(fileName)
316 if !ok {
317 return domain.VolumeBackupJob{}, false
318 }
319 return domain.VolumeBackupJob{
320 ID: id,
321 Domain: domainName,
322 VolumeName: volumeName,
323 Type: domain.BackupTypeVolumeArchive,
324 Status: domain.BackupStatusCompleted,
325 StartedAt: started,
326 CompletedAt: started,
327 ArtifactRef: s.artifactRef(key),
328 }, true
329}
330
331func parseVolumeBackupFileName(fileName string) (time.Time, string, bool) {
332 trimmed := strings.TrimSuffix(strings.TrimSuffix(fileName, ".tar.zst"), ".tar.gz")
333 if trimmed == fileName {
334 return time.Time{}, "", false
335 }
336 timestampPart, id, ok := strings.Cut(trimmed, "-")
337 if !ok {
338 return time.Time{}, "", false
339 }
340 started, err := time.Parse(volumeBackupTimestampLayout, timestampPart)
341 if err != nil {
342 return time.Time{}, "", false
343 }
344 return started.UTC(), id, true
345}
346
347func sortVolumeBackupJobs(jobs []domain.VolumeBackupJob) {
348 sort.Slice(jobs, func(i, j int) bool {
349 if !jobs[i].StartedAt.Equal(jobs[j].StartedAt) {
350 return jobs[i].StartedAt.After(jobs[j].StartedAt)
351 }
352 return jobs[i].ArtifactRef < jobs[j].ArtifactRef
353 })
354}
355
356func normalizeS3Prefix(prefix string) string {
357 return strings.Trim(strings.TrimSpace(prefix), "/")
358}
359
360func joinS3Key(parts ...string) string {
361 clean := make([]string, 0, len(parts))
362 for _, part := range parts {
363 part = strings.Trim(part, "/")
364 if part != "" {
365 clean = append(clean, part)
366 }
367 }
368 return path.Join(clean...)
369}
370
371func sanitizeS3KeyComponent(input string) string {
372 clean := strings.TrimSpace(input)
373 if clean == "" || strings.Trim(clean, ".") == "" {
374 return "unknown"
375 }
376 clean = strings.Trim(clean, ".")
377 replacer := strings.NewReplacer("/", "_", "\\", "_", ":", "_", " ", "_")
378 return replacer.Replace(clean)
379}
380
381func contentTypeForCompression(compression string) string {
382 switch compression {
383 case string(domain.VolumeBackupCompressionGzip):
384 return "application/gzip"
385 case string(domain.VolumeBackupCompressionZstd):
386 return "application/zstd"
387 default:
388 return "application/octet-stream"
389 }
390}
391
392func randomObjectSuffix() string {
393 var b [8]byte
394 if _, err := rand.Read(b[:]); err != nil {
395 return fmt.Sprintf("%d", time.Now().UTC().UnixNano())
396 }
397 return hex.EncodeToString(b[:])
398}