1package backup
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "errors"
8 "fmt"
9 "io"
10 "sort"
11 "strings"
12 "time"
13
14 "github.com/bnema/zerowrap"
15
16 "github.com/bnema/gordon/internal/boundaries/out"
17 "github.com/bnema/gordon/internal/domain"
18)
19
20const backupExecTimeout = 30 * time.Minute
21
22// Service orchestrates declarative PostgreSQL app backups. Targets come
23// from the app's ACTIVE record: an app declares its databases and which
24// of them are backed up, and the app name is the backup identity.
25type Service struct {
26 runtime out.ContainerRuntime
27 storage out.BackupStorage
28 config domain.BackupConfig
29 log zerowrap.Logger
30 state out.AppStateReader
31 metrics out.BackupMetrics
32}
33
34// NewService creates a backup service.
35func NewService(
36 runtime out.ContainerRuntime,
37 storage out.BackupStorage,
38 config domain.BackupConfig,
39 log zerowrap.Logger,
40) *Service {
41 return &Service{
42 runtime: runtime,
43 storage: storage,
44 config: config,
45 log: log,
46 }
47}
48
49// WithAppState wires the ACTIVE app state target resolution reads.
50// Without it no target can be resolved.
51func (s *Service) WithAppState(state out.AppStateReader) *Service {
52 s.state = state
53 return s
54}
55
56// WithMetrics wires the backup metrics recorder. Without it no metric is
57// recorded.
58func (s *Service) WithMetrics(metrics out.BackupMetrics) *Service {
59 s.metrics = metrics
60 return s
61}
62
63// SeedMetrics reconciles backup metrics with the declared database
64// targets: it forgets targets that are no longer declared and reports the
65// newest stored dump of each declared one as its last success, so the
66// metric survives a restart. It counts no run.
67func (s *Service) SeedMetrics(ctx context.Context) error {
68 if s.metrics == nil {
69 return nil
70 }
71 asOf := time.Now()
72 targets, err := s.allTargets(ctx)
73 if err != nil {
74 return fmt.Errorf("seed backup metrics: %w", err)
75 }
76 refs := make([]domain.BackupTargetRef, 0, len(targets))
77 declared := make(map[domain.BackupTargetRef]struct{}, len(targets))
78 for _, target := range targets {
79 ref := databaseTargetRef(target)
80 refs = append(refs, ref)
81 declared[ref] = struct{}{}
82 }
83 s.metrics.RetainBackupTargets(domain.BackupKindDatabase, refs, asOf)
84
85 jobs, err := s.ListBackups(ctx, "")
86 if err != nil {
87 return fmt.Errorf("seed backup metrics: %w", err)
88 }
89 for _, job := range jobs {
90 ref := domain.BackupTargetRef{Kind: domain.BackupKindDatabase, App: job.App, Service: job.Service, Target: job.DBName}
91 if _, ok := declared[ref]; !ok {
92 continue
93 }
94 s.metrics.ObserveStoredBackup(domain.BackupRunRecord{
95 BackupTargetRef: ref, Succeeded: true, SizeBytes: job.SizeBytes, StartedAt: job.StartedAt,
96 })
97 }
98 return nil
99}
100
101func databaseTargetRef(target domain.DatabaseTarget) domain.BackupTargetRef {
102 return domain.BackupTargetRef{Kind: domain.BackupKindDatabase, App: target.App, Service: target.Service, Target: target.Database}
103}
104
105// Targets returns every declared database target of one app, ordered by
106// service then database.
107func (s *Service) Targets(ctx context.Context, app string) ([]domain.DatabaseTarget, error) {
108 if s.state == nil {
109 return nil, fmt.Errorf("backup: app state is not wired")
110 }
111 targets, _, err := declaredTargets(ctx, s.state, app)
112 if err != nil {
113 return nil, err
114 }
115 return targets, nil
116}
117
118// allTargets returns the declared database targets of every app.
119func (s *Service) allTargets(ctx context.Context) ([]domain.DatabaseTarget, error) {
120 if s.state == nil {
121 return nil, fmt.Errorf("backup: app state is not wired")
122 }
123 apps, err := appNames(ctx, s.state)
124 if err != nil {
125 return nil, err
126 }
127 var targets []domain.DatabaseTarget
128 for _, app := range apps {
129 appTargets, err := s.Targets(ctx, app)
130 if err != nil {
131 return nil, err
132 }
133 targets = append(targets, appTargets...)
134 }
135 return targets, nil
136}
137
138// RunBackup runs one declared database backup. The service and database
139// selectors are explicit; an omitted selector only succeeds when exactly
140// one compatible target exists.
141func (s *Service) RunBackup(ctx context.Context, app, service, database string) (*domain.BackupResult, error) {
142 targets, err := s.Targets(ctx, app)
143 if err != nil {
144 return nil, err
145 }
146 target, err := selectTarget(targets, "database", service, database, databaseTargetKey)
147 if err != nil {
148 return nil, err
149 }
150 return s.runTarget(ctx, target, "")
151}
152
153// RunForSchedule runs every declared database whose own schedule matches,
154// then applies retention under the canonical app identity. Apps and
155// databases that do not declare this schedule are never touched.
156func (s *Service) RunForSchedule(ctx context.Context, schedule domain.BackupSchedule) error {
157 if !isValidBackupSchedule(schedule) {
158 return fmt.Errorf("invalid backup schedule: %q", schedule)
159 }
160 targets, err := s.allTargets(ctx)
161 if err != nil {
162 return err
163 }
164 var firstErr error
165 apps := map[string]struct{}{}
166 for _, target := range targets {
167 if target.Schedule != schedule {
168 continue
169 }
170 if target.ContainerID == "" {
171 // Nothing to dump: the declared database has no live
172 // container. Reported per target, never guessed.
173 s.log.Warn().Str("app", target.App).Str("service", target.Service).
174 Str("database", target.Database).Msg("scheduled backup skipped: no active container")
175 continue
176 }
177 if _, err := s.runTarget(ctx, target, schedule); err != nil {
178 s.log.Error().Err(err).Str("app", target.App).Str("service", target.Service).
179 Str("database", target.Database).Msg("scheduled backup failed")
180 if firstErr == nil {
181 firstErr = fmt.Errorf("backup %s/%s/%s: %w", target.App, target.Service, target.Database, err)
182 }
183 continue
184 }
185 apps[target.App] = struct{}{}
186 }
187 for _, app := range sortedSet(apps) {
188 if _, err := s.storage.ApplyRetention(ctx, app, s.config.Retention); err != nil {
189 s.log.Error().Err(err).Str("app", app).Msg("scheduled backup retention failed")
190 if firstErr == nil {
191 firstErr = fmt.Errorf("apply retention for %s: %w", app, err)
192 }
193 }
194 }
195 return firstErr
196}
197
198// ListBackups lists stored backups of one app, or of every app when app
199// is empty. Storage is keyed by the canonical app name.
200func (s *Service) ListBackups(ctx context.Context, app string) ([]domain.BackupJob, error) {
201 if app != "" {
202 return s.storage.List(ctx, app, nil)
203 }
204 if s.state == nil {
205 return nil, fmt.Errorf("backup: app state is not wired")
206 }
207 apps, err := appNames(ctx, s.state)
208 if err != nil {
209 return nil, err
210 }
211 var jobs []domain.BackupJob
212 for _, name := range apps {
213 appJobs, err := s.storage.List(ctx, name, nil)
214 if err != nil {
215 return nil, err
216 }
217 jobs = append(jobs, appJobs...)
218 }
219 sort.Slice(jobs, func(i, j int) bool { return jobs[i].StartedAt.After(jobs[j].StartedAt) })
220 return jobs, nil
221}
222
223// Status returns stored backups plus the declared targets that have no
224// completed backup yet, so an operator can see what is configured and
225// what actually ran.
226func (s *Service) Status(ctx context.Context) ([]domain.BackupJob, error) {
227 jobs, err := s.ListBackups(ctx, "")
228 if err != nil {
229 return nil, err
230 }
231 completed := make(map[string]struct{}, len(jobs))
232 for _, job := range jobs {
233 completed[job.App+"\x00"+job.Service+"\x00"+job.DBName] = struct{}{}
234 }
235 targets, err := s.allTargets(ctx)
236 if err != nil {
237 return nil, err
238 }
239 for _, target := range targets {
240 if _, ok := completed[target.App+"\x00"+target.Service+"\x00"+target.Database]; ok {
241 continue
242 }
243 jobs = append(jobs, domain.BackupJob{
244 ID: "target:" + target.App + "/" + target.Service + "/" + target.Database,
245 App: target.App,
246 Service: target.Service,
247 DBName: target.Database,
248 Schedule: target.Schedule,
249 Type: domain.BackupTypeLogical,
250 Status: domain.BackupStatusPending,
251 Metadata: map[string]string{"declared_schedule": string(target.Schedule)},
252 })
253 }
254 sort.Slice(jobs, func(i, j int) bool {
255 if !jobs[i].StartedAt.Equal(jobs[j].StartedAt) {
256 return jobs[i].StartedAt.After(jobs[j].StartedAt)
257 }
258 return jobs[i].App < jobs[j].App
259 })
260 return jobs, nil
261}
262
263// runTarget runs one declared database backup through the service
264// container that hosts it and records the outcome in metrics.
265func (s *Service) runTarget(ctx context.Context, target domain.DatabaseTarget, schedule domain.BackupSchedule) (*domain.BackupResult, error) {
266 started := time.Now().UTC()
267 result, err := s.dumpTarget(ctx, target, schedule, started)
268 if s.metrics != nil {
269 run := domain.BackupRunRecord{
270 BackupTargetRef: databaseTargetRef(target),
271 Succeeded: err == nil,
272 Duration: time.Since(started),
273 StartedAt: started,
274 }
275 if result != nil {
276 run.SizeBytes = result.Job.SizeBytes
277 }
278 s.metrics.RecordBackupRun(ctx, run)
279 }
280 return result, err
281}
282
283func (s *Service) dumpTarget(ctx context.Context, target domain.DatabaseTarget, schedule domain.BackupSchedule, started time.Time) (*domain.BackupResult, error) {
284 if target.ContainerID == "" {
285 return nil, fmt.Errorf("backup: app %q service %q has no active container", target.App, target.Service)
286 }
287
288 execCtx, cancelExec := context.WithTimeout(ctx, backupExecTimeout)
289 defer cancelExec()
290 dumpPath := fmt.Sprintf("/tmp/gordon-backup-%d.bak", started.UnixNano())
291 defer s.cleanupDumpFile(target.ContainerID, dumpPath)
292
293 execResult, err := s.runtime.ExecInContainer(execCtx, target.ContainerID, []string{"sh", "-c", pgDumpToPathCommand(dumpPath, target.Database)})
294 if err != nil {
295 if errors.Is(execCtx.Err(), context.DeadlineExceeded) {
296 return nil, fmt.Errorf("pg_dump timed out after %s", backupExecTimeout)
297 }
298 return nil, err
299 }
300 if execResult.ExitCode != 0 {
301 return nil, fmt.Errorf("pg_dump failed with exit code %d: %s", execResult.ExitCode, string(execResult.Stderr))
302 }
303
304 dumpStream, err := s.runtime.CopyFromContainer(execCtx, target.ContainerID, dumpPath)
305 if err != nil {
306 if errors.Is(execCtx.Err(), context.DeadlineExceeded) {
307 return nil, fmt.Errorf("backup copy timed out after %s", backupExecTimeout)
308 }
309 return nil, err
310 }
311 defer dumpStream.Close()
312
313 counter := &byteCounter{}
314 path, err := s.storage.Store(ctx, target.App, target.Service, target.Database, schedule, started, io.TeeReader(dumpStream, counter))
315 if err != nil {
316 return nil, err
317 }
318
319 job := domain.BackupJob{
320 ID: newBackupJobID(started),
321 App: target.App,
322 Service: target.Service,
323 DBName: target.Database,
324 Schedule: schedule,
325 Type: domain.BackupTypeLogical,
326 Status: domain.BackupStatusCompleted,
327 StartedAt: started,
328 CompletedAt: time.Now().UTC(),
329 SizeBytes: counter.n,
330 FilePath: path,
331 }
332
333 return &domain.BackupResult{
334 Job: job,
335 Duration: time.Since(started),
336 }, nil
337}
338
339// databaseTargetKey reports the (service, database) identity of one
340// target for selector matching and ambiguity reporting.
341func databaseTargetKey(target domain.DatabaseTarget) (string, string) {
342 return target.Service, target.Database
343}
344
345func isValidBackupSchedule(schedule domain.BackupSchedule) bool {
346 switch schedule {
347 case domain.ScheduleHourly, domain.ScheduleDaily, domain.ScheduleWeekly, domain.ScheduleMonthly:
348 return true
349 default:
350 return false
351 }
352}
353
354// Restore restores a backup by ID.
355func (s *Service) Restore(context.Context, string, string) error {
356 return fmt.Errorf("backup restore not implemented yet")
357}
358
359// RestorePITR restores to a point in time.
360func (s *Service) RestorePITR(context.Context, string, time.Time) error {
361 return fmt.Errorf("pitr restore not implemented yet")
362}
363
364func newBackupJobID(started time.Time) string {
365 random := make([]byte, 4)
366 if _, err := rand.Read(random); err != nil {
367 return fmt.Sprintf("%s-%d", started.Format(time.RFC3339Nano), time.Now().UnixNano())
368 }
369 return fmt.Sprintf("%s-%s", started.Format(time.RFC3339Nano), hex.EncodeToString(random))
370}
371
372// pgDumpToPathCommand runs pg_dump inside the declared service container.
373// The database name comes from the app declaration (validated as a name),
374// while credentials keep coming from the container's own environment: the
375// manifest never carries secret values.
376func pgDumpToPathCommand(path, dbName string) string {
377 return fmt.Sprintf(
378 "pg_dump -Fc --dbname=%s --username=\"${POSTGRES_USER:-postgres}\" > %q",
379 shellSingleQuote(dbName), path,
380 )
381}
382
383// shellSingleQuote quotes one validated identifier for a single-quoted
384// shell word. Declared names are already restricted to safe characters;
385// this keeps that guarantee explicit across shells.
386func shellSingleQuote(value string) string {
387 return "'" + strings.ReplaceAll(value, "'", `'\''`) + "'"
388}
389
390func (s *Service) cleanupDumpFile(containerID, dumpPath string) {
391 cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
392 defer cancel()
393
394 _, _ = s.runtime.ExecInContainer(cleanupCtx, containerID, []string{"sh", "-c", fmt.Sprintf("rm -f %q", dumpPath)})
395}
396
397// byteCounter counts the bytes of one stream.
398type byteCounter struct {
399 n int64
400}
401
402func (c *byteCounter) Write(p []byte) (int, error) {
403 c.n += int64(len(p))
404 return len(p), nil
405}
406
407// sortedSet returns the keys of a set in sorted order.
408func sortedSet(set map[string]struct{}) []string {
409 keys := make([]string, 0, len(set))
410 for key := range set {
411 keys = append(keys, key)
412 }
413 sort.Strings(keys)
414 return keys
415}