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