1package backup
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "fmt"
8 "io"
9 "sort"
10 "sync"
11 "time"
12
13 "github.com/bnema/zerowrap"
14
15 "github.com/bnema/gordon/internal/boundaries/out"
16 "github.com/bnema/gordon/internal/domain"
17)
18
19// VolumeService orchestrates declarative app volume archive backups.
20// Targets come from the app's ACTIVE record: a service declares its
21// volumes and which of them are backed up, and the app name is the
22// backup identity. No container label or attachment is ever consulted.
23type VolumeService struct {
24 exporter out.VolumeArchiveExporter
25 storage out.VolumeBackupStorage
26 config domain.VolumeBackupConfig
27 log zerowrap.Logger
28 state out.AppStateReader
29
30 mu sync.Mutex
31 recent map[string]domain.VolumeBackupJob
32}
33
34// NewVolumeService creates a volume backup service.
35func NewVolumeService(exporter out.VolumeArchiveExporter, storage out.VolumeBackupStorage, config domain.VolumeBackupConfig, log zerowrap.Logger) *VolumeService {
36 return &VolumeService{
37 exporter: exporter,
38 storage: storage,
39 config: config,
40 log: log,
41 recent: make(map[string]domain.VolumeBackupJob),
42 }
43}
44
45// WithAppState wires the ACTIVE app state target resolution reads.
46func (s *VolumeService) WithAppState(state out.AppStateReader) *VolumeService {
47 s.state = state
48 return s
49}
50
51// VolumeTargets returns every declared volume target of one app.
52func (s *VolumeService) VolumeTargets(ctx context.Context, app string) ([]domain.VolumeBackupTarget, error) {
53 if s.state == nil {
54 return nil, fmt.Errorf("backup: app state is not wired")
55 }
56 _, targets, err := declaredTargets(ctx, s.state, app)
57 if err != nil {
58 return nil, err
59 }
60 return targets, nil
61}
62
63// allVolumeTargets returns the declared volume targets of every app.
64func (s *VolumeService) allVolumeTargets(ctx context.Context) ([]domain.VolumeBackupTarget, error) {
65 if s.state == nil {
66 return nil, fmt.Errorf("backup: app state is not wired")
67 }
68 apps, err := appNames(ctx, s.state)
69 if err != nil {
70 return nil, err
71 }
72 var targets []domain.VolumeBackupTarget
73 for _, app := range apps {
74 appTargets, err := s.VolumeTargets(ctx, app)
75 if err != nil {
76 return nil, err
77 }
78 targets = append(targets, appTargets...)
79 }
80 return targets, nil
81}
82
83// ListVolumeBackups lists completed volume backups of one app, or every
84// app when app is empty.
85func (s *VolumeService) ListVolumeBackups(ctx context.Context, app string) ([]domain.VolumeBackupJob, error) {
86 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
87 zerowrap.FieldLayer: "usecase",
88 zerowrap.FieldUseCase: "ListVolumeBackups",
89 "app": app,
90 })
91 jobs, err := s.storage.ListVolumeArchives(ctx, app)
92 if err != nil {
93 return nil, fmt.Errorf("list volume archives: %w", err)
94 }
95 return jobs, nil
96}
97
98// VolumeBackupStatus returns completed backup artifacts plus current or
99// recent in-memory jobs.
100func (s *VolumeService) VolumeBackupStatus(ctx context.Context) ([]domain.VolumeBackupJob, error) {
101 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
102 zerowrap.FieldLayer: "usecase",
103 zerowrap.FieldUseCase: "VolumeBackupStatus",
104 })
105 jobs, err := s.storage.ListVolumeArchives(ctx, "")
106 if err != nil {
107 return nil, fmt.Errorf("list volume backup status: %w", err)
108 }
109
110 s.mu.Lock()
111 for _, job := range s.recent {
112 if job.Status == domain.BackupStatusRunning || job.Status == domain.BackupStatusFailed {
113 jobs = append(jobs, job)
114 }
115 }
116 s.mu.Unlock()
117
118 sortVolumeJobs(jobs)
119 return jobs, nil
120}
121
122// RunVolumeBackups runs the declared volume backups of one app. service
123// and volume are explicit selectors: an omitted selector succeeds only
124// when exactly one compatible target exists.
125func (s *VolumeService) RunVolumeBackups(ctx context.Context, app, service, volume string) ([]domain.VolumeBackupJob, error) {
126 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
127 zerowrap.FieldLayer: "usecase",
128 zerowrap.FieldUseCase: "RunVolumeBackups",
129 "app": app,
130 "volume": volume,
131 })
132 if !s.config.Enabled {
133 return []domain.VolumeBackupJob{}, nil
134 }
135 targets, err := s.VolumeTargets(ctx, app)
136 if err != nil {
137 return nil, err
138 }
139 target, err := selectTarget(targets, "volume", service, volume, volumeTargetKey)
140 if err != nil {
141 return nil, err
142 }
143 job := s.runVolumeBackup(ctx, target)
144 if job.Status == domain.BackupStatusCompleted {
145 if _, err := s.storage.ApplyVolumeRetention(ctx, job.App, s.config.Retention); err != nil {
146 return []domain.VolumeBackupJob{job}, fmt.Errorf("apply volume backup retention for %s: %w", job.App, err)
147 }
148 }
149 if job.Status == domain.BackupStatusFailed {
150 return []domain.VolumeBackupJob{job}, fmt.Errorf("volume backup failed for %s/%s: %s", job.App, job.VolumeName, job.Error)
151 }
152 return []domain.VolumeBackupJob{job}, nil
153}
154
155// RunVolumeBackupsForSchedule runs every declared volume backup of every
156// app under the installation's schedule, then applies retention. Volume
157// declarations carry no schedule of their own: the installation preset is
158// the schedule, and the label is only used for the caller's own filter.
159func (s *VolumeService) RunVolumeBackupsForSchedule(ctx context.Context, schedule domain.BackupSchedule) error {
160 if !s.config.Enabled {
161 return nil
162 }
163 if schedule != "" && !isValidBackupSchedule(schedule) {
164 return fmt.Errorf("invalid backup schedule: %q", schedule)
165 }
166 targets, err := s.allVolumeTargets(ctx)
167 if err != nil {
168 return err
169 }
170 jobs := s.runVolumeBackupsConcurrently(ctx, targets)
171 var firstErr error
172 apps := map[string]struct{}{}
173 for _, job := range jobs {
174 if job.Status != domain.BackupStatusCompleted {
175 if firstErr == nil {
176 firstErr = fmt.Errorf("volume backup %s/%s: %s", job.App, job.VolumeName, job.Error)
177 }
178 continue
179 }
180 apps[job.App] = struct{}{}
181 }
182 // Retention runs only after every archive finished, so a pass never
183 // deletes an app's older archive while its replacement is uploading.
184 for _, app := range sortedSet(apps) {
185 if _, err := s.storage.ApplyVolumeRetention(ctx, app, s.config.Retention); err != nil {
186 if firstErr == nil {
187 firstErr = fmt.Errorf("apply volume backup retention for %s: %w", app, err)
188 }
189 }
190 }
191 return firstErr
192}
193
194// runVolumeBackupsConcurrently runs every target with at most
195// MaxConcurrency exports in flight and returns the jobs in target order.
196func (s *VolumeService) runVolumeBackupsConcurrently(ctx context.Context, targets []domain.VolumeBackupTarget) []domain.VolumeBackupJob {
197 limit := s.config.MaxConcurrency
198 if limit < 1 {
199 limit = 1
200 }
201 jobs := make([]domain.VolumeBackupJob, len(targets))
202 sem := make(chan struct{}, limit)
203 var wg sync.WaitGroup
204 for i, target := range targets {
205 wg.Add(1)
206 sem <- struct{}{}
207 go func(i int, target domain.VolumeBackupTarget) {
208 defer wg.Done()
209 defer func() { <-sem }()
210 jobs[i] = s.runVolumeBackup(ctx, target)
211 }(i, target)
212 }
213 wg.Wait()
214 return jobs
215}
216
217func (s *VolumeService) runVolumeBackup(ctx context.Context, target domain.VolumeBackupTarget) domain.VolumeBackupJob {
218 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
219 "app": target.App,
220 "service": target.Service,
221 "volume": target.VolumeName,
222 "runtime_volume": target.RuntimeVolumeName,
223 "mount_path": target.MountPath,
224 })
225 log := zerowrap.FromCtx(ctx)
226 started := time.Now().UTC()
227 job := domain.VolumeBackupJob{
228 ID: newVolumeBackupJobID(),
229 App: target.App,
230 Service: target.Service,
231 VolumeName: target.VolumeName,
232 RuntimeVolumeName: target.RuntimeVolumeName,
233 MountPath: target.MountPath,
234 Type: domain.BackupTypeVolumeArchive,
235 Status: domain.BackupStatusRunning,
236 StartedAt: started,
237 Metadata: map[string]string{
238 "compression": string(s.config.Compression),
239 },
240 }
241 s.remember(job)
242
243 timeout := s.config.Timeout
244 if timeout <= 0 {
245 timeout = 2 * time.Hour
246 }
247 exportCtx, cancel := context.WithTimeout(ctx, timeout)
248 defer cancel()
249
250 archive, err := s.exporter.ExportVolumeArchive(exportCtx, domain.VolumeArchiveRequest{
251 VolumeName: target.RuntimeVolumeName,
252 MountPath: target.MountPath,
253 Compression: s.config.Compression,
254 HelperImage: s.config.HelperImage,
255 })
256 if err != nil {
257 log.Error().Err(err).Msg("volume archive export failed")
258 return s.failJob(ctx, job, err)
259 }
260 defer archive.Stream.Close()
261
262 counter := &byteCounter{}
263 artifactRef, err := s.storage.StoreVolumeArchive(exportCtx, job, io.TeeReader(archive.Stream, counter))
264 if err != nil {
265 log.Error().Err(err).Msg("volume archive upload failed")
266 return s.failJob(ctx, job, err)
267 }
268
269 job.Status = domain.BackupStatusCompleted
270 job.CompletedAt = time.Now().UTC()
271 job.SizeBytes = counter.n
272 job.ArtifactRef = artifactRef
273 log.Info().
274 Int64("size_bytes", job.SizeBytes).
275 Dur("duration", job.CompletedAt.Sub(job.StartedAt)).
276 Msg("volume backup completed")
277 s.forget(job)
278 return job
279}
280
281// newVolumeBackupJobID returns a short random job ID. Storage keys already
282// carry the start timestamp, so the ID only has to make each run unique.
283func newVolumeBackupJobID() string {
284 random := make([]byte, 4)
285 if _, err := rand.Read(random); err != nil {
286 return fmt.Sprintf("%x", time.Now().UnixNano())
287 }
288 return hex.EncodeToString(random)
289}
290
291// volumeTargetKey reports the (service, volume) identity of one target.
292func volumeTargetKey(target domain.VolumeBackupTarget) (string, string) {
293 return target.Service, target.VolumeName
294}
295
296func (s *VolumeService) failJob(ctx context.Context, job domain.VolumeBackupJob, err error) domain.VolumeBackupJob {
297 job.Status = domain.BackupStatusFailed
298 job.CompletedAt = time.Now().UTC()
299 job.Error = err.Error()
300 log := zerowrap.FromCtx(ctx)
301 log.Error().Err(err).Msg("volume backup failed")
302 s.remember(job)
303 return job
304}
305
306func (s *VolumeService) remember(job domain.VolumeBackupJob) {
307 s.mu.Lock()
308 defer s.mu.Unlock()
309 s.recent[job.App+"/"+job.VolumeName] = job
310 if len(s.recent) > 100 {
311 for k := range s.recent {
312 delete(s.recent, k)
313 break
314 }
315 }
316}
317
318func (s *VolumeService) forget(job domain.VolumeBackupJob) {
319 s.mu.Lock()
320 defer s.mu.Unlock()
321 delete(s.recent, job.App+"/"+job.VolumeName)
322}
323
324func sortVolumeJobs(jobs []domain.VolumeBackupJob) {
325 sort.Slice(jobs, func(i, j int) bool {
326 if !jobs[i].StartedAt.Equal(jobs[j].StartedAt) {
327 return jobs[i].StartedAt.After(jobs[j].StartedAt)
328 }
329 return jobs[i].VolumeName < jobs[j].VolumeName
330 })
331}