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