1package backup
2
3import (
4 "context"
5 "fmt"
6 "io"
7 "sort"
8 "sync"
9 "time"
10
11 "github.com/bnema/zerowrap"
12
13 "github.com/bnema/gordon/internal/boundaries/out"
14 "github.com/bnema/gordon/internal/domain"
15)
16
17// VolumeService orchestrates volume archive backups.
18type VolumeService struct {
19 runtime out.ContainerRuntime
20 exporter out.VolumeArchiveExporter
21 storage out.VolumeBackupStorage
22 config domain.VolumeBackupConfig
23 log zerowrap.Logger
24
25 mu sync.Mutex
26 recent map[string]domain.VolumeBackupJob
27}
28
29// NewVolumeService creates a volume backup service.
30func NewVolumeService(runtime out.ContainerRuntime, exporter out.VolumeArchiveExporter, storage out.VolumeBackupStorage, config domain.VolumeBackupConfig, log zerowrap.Logger) *VolumeService {
31 return &VolumeService{
32 runtime: runtime,
33 exporter: exporter,
34 storage: storage,
35 config: config,
36 log: log,
37 recent: make(map[string]domain.VolumeBackupJob),
38 }
39}
40
41// ListVolumeBackups lists completed volume backups for a domain, or all domains when empty.
42func (s *VolumeService) ListVolumeBackups(ctx context.Context, domainName string) ([]domain.VolumeBackupJob, error) {
43 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
44 zerowrap.FieldLayer: "usecase",
45 zerowrap.FieldUseCase: "ListVolumeBackups",
46 "domain": domainName,
47 })
48 jobs, err := s.storage.ListVolumeArchives(ctx, domainName)
49 if err != nil {
50 return nil, fmt.Errorf("list volume archives: %w", err)
51 }
52 return jobs, nil
53}
54
55// VolumeBackupStatus returns completed backup artifacts plus current/recent in-memory job state.
56func (s *VolumeService) VolumeBackupStatus(ctx context.Context) ([]domain.VolumeBackupJob, error) {
57 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
58 zerowrap.FieldLayer: "usecase",
59 zerowrap.FieldUseCase: "VolumeBackupStatus",
60 })
61 jobs, err := s.storage.ListVolumeArchives(ctx, "")
62 if err != nil {
63 return nil, fmt.Errorf("list volume backup status: %w", err)
64 }
65
66 s.mu.Lock()
67 for _, job := range s.recent {
68 if job.Status == domain.BackupStatusRunning || job.Status == domain.BackupStatusFailed {
69 jobs = append(jobs, job)
70 }
71 }
72 s.mu.Unlock()
73
74 sort.Slice(jobs, func(i, j int) bool {
75 if !jobs[i].StartedAt.Equal(jobs[j].StartedAt) {
76 return jobs[i].StartedAt.After(jobs[j].StartedAt)
77 }
78 return jobs[i].VolumeName < jobs[j].VolumeName
79 })
80 return jobs, nil
81}
82
83// RunVolumeBackups runs volume backups for all eligible targets, optionally scoped to a domain and volume.
84func (s *VolumeService) RunVolumeBackups(ctx context.Context, domainName, volumeName string) ([]domain.VolumeBackupJob, error) {
85 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
86 zerowrap.FieldLayer: "usecase",
87 zerowrap.FieldUseCase: "RunVolumeBackups",
88 "domain": domainName,
89 "volume": volumeName,
90 })
91 log := zerowrap.FromCtx(ctx)
92 if !s.config.Enabled {
93 return []domain.VolumeBackupJob{}, nil
94 }
95
96 containers, err := s.runtime.ListContainers(ctx, true)
97 if err != nil {
98 return nil, fmt.Errorf("failed to list containers for volume backups: %w", err)
99 }
100 targets := SelectVolumeBackupTargetsForScope(containers, s.config.VolumePrefix, domainName, volumeName)
101 log.Info().
102 Str("domain", domainName).
103 Str("volume", volumeName).
104 Int("targets", len(targets)).
105 Msg("selected volume backup targets")
106 if len(targets) == 0 {
107 return []domain.VolumeBackupJob{}, nil
108 }
109
110 results, firstErr := s.runVolumeBackupTargets(ctx, targets)
111 if err := s.applyRetentionForSuccessfulVolumeBackups(ctx, results); err != nil && firstErr == nil {
112 firstErr = err
113 }
114 return results, firstErr
115}
116
117func (s *VolumeService) runVolumeBackupTargets(ctx context.Context, targets []domain.VolumeBackupTarget) ([]domain.VolumeBackupJob, error) {
118 runCtx, cancel := context.WithCancel(ctx)
119 defer cancel()
120
121 sem := make(chan struct{}, max(s.config.MaxConcurrency, 1))
122 results := make([]domain.VolumeBackupJob, 0, len(targets))
123 var wg sync.WaitGroup
124 var mu sync.Mutex
125 var firstErr error
126
127launchLoop:
128 for _, target := range targets {
129 select {
130 case <-runCtx.Done():
131 firstErr = runCtx.Err()
132 break launchLoop
133 case sem <- struct{}{}:
134 }
135 wg.Add(1)
136 go func(target domain.VolumeBackupTarget) {
137 defer wg.Done()
138 defer func() { <-sem }()
139 job := s.runVolumeBackup(runCtx, target)
140 mu.Lock()
141 results = append(results, job)
142 mu.Unlock()
143 }(target)
144 }
145 if firstErr != nil {
146 cancel()
147 }
148 wg.Wait()
149 return results, firstVolumeBackupError(results, firstErr)
150}
151
152func (s *VolumeService) applyRetentionForSuccessfulVolumeBackups(ctx context.Context, results []domain.VolumeBackupJob) error {
153 successDomains, failedDomains := volumeBackupResultDomains(results)
154 for domainName := range successDomains {
155 if _, failed := failedDomains[domainName]; failed {
156 continue
157 }
158 deleted, err := s.storage.ApplyVolumeRetention(ctx, domainName, s.config.Retention)
159 if err != nil {
160 return fmt.Errorf("apply volume backup retention for %s: %w", domainName, err)
161 }
162 log := zerowrap.FromCtx(ctx)
163 log.Info().
164 Str("domain", domainName).
165 Int("deleted", deleted).
166 Msg("volume backup retention applied")
167 }
168 return nil
169}
170
171func firstVolumeBackupError(results []domain.VolumeBackupJob, firstErr error) error {
172 if firstErr != nil {
173 return firstErr
174 }
175 for _, job := range results {
176 if job.Status == domain.BackupStatusFailed {
177 return fmt.Errorf("volume backup failed for %s/%s: %s", job.Domain, job.VolumeName, job.Error)
178 }
179 }
180 return nil
181}
182
183func volumeBackupResultDomains(results []domain.VolumeBackupJob) (map[string]struct{}, map[string]struct{}) {
184 successDomains := make(map[string]struct{})
185 failedDomains := make(map[string]struct{})
186 for _, job := range results {
187 switch job.Status {
188 case domain.BackupStatusCompleted:
189 successDomains[job.Domain] = struct{}{}
190 case domain.BackupStatusFailed:
191 failedDomains[job.Domain] = struct{}{}
192 }
193 }
194 return successDomains, failedDomains
195}
196
197func (s *VolumeService) runVolumeBackup(ctx context.Context, target domain.VolumeBackupTarget) domain.VolumeBackupJob {
198 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
199 "domain": target.Domain,
200 "volume": target.VolumeName,
201 "container_name": target.ContainerName,
202 "mount_path": target.MountPath,
203 })
204 log := zerowrap.FromCtx(ctx)
205 started := time.Now().UTC()
206 job := domain.VolumeBackupJob{
207 ID: newBackupJobID(started),
208 Domain: target.Domain,
209 ContainerName: target.ContainerName,
210 ContainerID: target.ContainerID,
211 VolumeName: target.VolumeName,
212 MountPath: target.MountPath,
213 Type: domain.BackupTypeVolumeArchive,
214 Status: domain.BackupStatusRunning,
215 StartedAt: started,
216 Metadata: map[string]string{
217 "compression": string(s.config.Compression),
218 },
219 }
220 s.remember(job)
221
222 timeout := s.config.Timeout
223 if timeout <= 0 {
224 timeout = 2 * time.Hour
225 }
226 exportCtx, cancel := context.WithTimeout(ctx, timeout)
227 defer cancel()
228
229 archive, err := s.exporter.ExportVolumeArchive(exportCtx, domain.VolumeArchiveRequest{
230 VolumeName: target.VolumeName,
231 MountPath: target.MountPath,
232 Compression: s.config.Compression,
233 HelperImage: s.config.HelperImage,
234 })
235 if err != nil {
236 log.Error().Err(err).Msg("volume archive export failed")
237 return s.failJob(ctx, job, err)
238 }
239 defer archive.Stream.Close()
240
241 counter := &byteCounter{}
242 artifactRef, err := s.storage.StoreVolumeArchive(exportCtx, job, io.TeeReader(archive.Stream, counter))
243 if err != nil {
244 log.Error().Err(err).Msg("volume archive upload failed")
245 return s.failJob(ctx, job, err)
246 }
247
248 job.Status = domain.BackupStatusCompleted
249 job.CompletedAt = time.Now().UTC()
250 job.SizeBytes = counter.n
251 job.ArtifactRef = artifactRef
252 log.Info().
253 Int64("size_bytes", job.SizeBytes).
254 Dur("duration", job.CompletedAt.Sub(job.StartedAt)).
255 Msg("volume backup completed")
256 s.forget(job)
257 return job
258}
259
260func (s *VolumeService) failJob(ctx context.Context, job domain.VolumeBackupJob, err error) domain.VolumeBackupJob {
261 job.Status = domain.BackupStatusFailed
262 job.CompletedAt = time.Now().UTC()
263 job.Error = err.Error()
264 log := zerowrap.FromCtx(ctx)
265 log.Error().Err(err).Msg("volume backup failed")
266 s.remember(job)
267 return job
268}
269
270func (s *VolumeService) remember(job domain.VolumeBackupJob) {
271 s.mu.Lock()
272 defer s.mu.Unlock()
273 s.recent[job.Domain+"/"+job.VolumeName] = job
274 if len(s.recent) > 100 {
275 for k := range s.recent {
276 delete(s.recent, k)
277 break
278 }
279 }
280}
281
282func (s *VolumeService) forget(job domain.VolumeBackupJob) {
283 s.mu.Lock()
284 defer s.mu.Unlock()
285 delete(s.recent, job.Domain+"/"+job.VolumeName)
286}