1package backup
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "errors"
8 "fmt"
9 "io"
10 "sort"
11 "strings"
12 "sync"
13 "time"
14
15 "github.com/bnema/zerowrap"
16
17 "github.com/bnema/gordon/internal/boundaries/in"
18 "github.com/bnema/gordon/internal/boundaries/out"
19 "github.com/bnema/gordon/internal/domain"
20)
21
22const backupExecTimeout = 30 * time.Minute
23
24// Service orchestrates backup operations.
25type Service struct {
26 runtime out.ContainerRuntime
27 storage out.BackupStorage
28 containerSvc in.ContainerService
29 config domain.BackupConfig
30 log zerowrap.Logger
31}
32
33// NewService creates a backup service.
34func NewService(
35 runtime out.ContainerRuntime,
36 storage out.BackupStorage,
37 containerSvc in.ContainerService,
38 config domain.BackupConfig,
39 log zerowrap.Logger,
40) *Service {
41 return &Service{
42 runtime: runtime,
43 storage: storage,
44 containerSvc: containerSvc,
45 config: config,
46 log: log,
47 }
48}
49
50// ListBackups returns backups for a domain.
51func (s *Service) ListBackups(ctx context.Context, domainName string) ([]domain.BackupJob, error) {
52 return s.storage.List(ctx, domainName, nil)
53}
54
55// DetectDatabases inspects attachments and returns detected DBs.
56func (s *Service) DetectDatabases(ctx context.Context, domainName string) ([]domain.DBInfo, error) {
57 attachments := s.containerSvc.ListAttachments(ctx, domainName)
58 dbs := make([]domain.DBInfo, 0, len(attachments))
59 for _, attachment := range attachments {
60 if db, ok := detectDatabaseFromAttachment(domainName, attachment); ok {
61 dbs = append(dbs, db)
62 }
63 }
64 return dbs, nil
65}
66
67// RunBackup triggers a logical PostgreSQL backup for a detected DB.
68func (s *Service) RunBackup(ctx context.Context, domainName, dbName string) (*domain.BackupResult, error) {
69 return s.runBackup(ctx, domainName, dbName, domain.BackupSchedule(""))
70}
71
72// RunForSchedule executes backups for all detected databases and applies retention.
73func (s *Service) RunForSchedule(ctx context.Context, schedule domain.BackupSchedule) error {
74 if !isValidBackupSchedule(schedule) {
75 return fmt.Errorf("invalid backup schedule: %q", schedule)
76 }
77
78 routes := s.containerSvc.List(ctx)
79 domainNames := make([]string, 0, len(routes))
80 for domainName := range routes {
81 domainNames = append(domainNames, domainName)
82 }
83 sort.Strings(domainNames)
84
85 var firstErr error
86 for _, domainName := range domainNames {
87 dbs, err := s.DetectDatabases(ctx, domainName)
88 if err != nil {
89 s.log.Error().Err(err).Str("domain", domainName).Msg("scheduled backup database detection failed")
90 if firstErr == nil {
91 firstErr = fmt.Errorf("detect databases for %s: %w", domainName, err)
92 }
93 continue
94 }
95
96 for _, db := range dbs {
97 if _, err := s.runBackupForDB(ctx, domainName, db, schedule); err != nil {
98 s.log.Error().Err(err).Str("domain", domainName).Str("db", db.Name).Msg("scheduled backup failed")
99 if firstErr == nil {
100 firstErr = fmt.Errorf("run backup for %s/%s: %w", domainName, db.Name, err)
101 }
102 }
103 }
104
105 if _, err := s.storage.ApplyRetention(ctx, domainName, s.config.Retention); err != nil {
106 s.log.Error().Err(err).Str("domain", domainName).Msg("scheduled backup retention failed")
107 if firstErr == nil {
108 firstErr = fmt.Errorf("apply retention for %s: %w", domainName, err)
109 }
110 }
111 }
112
113 return firstErr
114}
115
116func (s *Service) runBackup(ctx context.Context, domainName, dbName string, schedule domain.BackupSchedule) (*domain.BackupResult, error) {
117 dbs, err := s.DetectDatabases(ctx, domainName)
118 if err != nil {
119 return nil, err
120 }
121
122 db, err := selectDatabase(dbs, dbName)
123 if err != nil {
124 return nil, err
125 }
126
127 return s.runBackupForDB(ctx, domainName, db, schedule)
128}
129
130func (s *Service) runBackupForDB(ctx context.Context, domainName string, db domain.DBInfo, schedule domain.BackupSchedule) (*domain.BackupResult, error) {
131 started := time.Now().UTC()
132
133 if db.Type != domain.DBTypePostgreSQL {
134 return nil, fmt.Errorf("unsupported database type: %s", db.Type)
135 }
136
137 execCtx, cancelExec := context.WithTimeout(ctx, backupExecTimeout)
138 defer cancelExec()
139 dumpPath := fmt.Sprintf("/tmp/gordon-backup-%d.bak", started.UnixNano())
140 defer s.cleanupDumpFile(db.ContainerID, dumpPath)
141
142 execResult, err := s.runtime.ExecInContainer(execCtx, db.ContainerID, []string{"sh", "-c", pgDumpToPathCommand(dumpPath, db.Name)})
143 if err != nil {
144 if errors.Is(execCtx.Err(), context.DeadlineExceeded) {
145 return nil, fmt.Errorf("pg_dump timed out after %s", backupExecTimeout)
146 }
147 return nil, err
148 }
149 if execResult.ExitCode != 0 {
150 return nil, fmt.Errorf("pg_dump failed with exit code %d: %s", execResult.ExitCode, string(execResult.Stderr))
151 }
152
153 dumpStream, err := s.runtime.CopyFromContainer(execCtx, db.ContainerID, dumpPath)
154 if err != nil {
155 if errors.Is(execCtx.Err(), context.DeadlineExceeded) {
156 return nil, fmt.Errorf("backup copy timed out after %s", backupExecTimeout)
157 }
158 return nil, err
159 }
160 defer dumpStream.Close()
161
162 counter := &byteCounter{}
163 path, err := s.storage.Store(ctx, domainName, db.Name, schedule, started, io.TeeReader(dumpStream, counter))
164 if err != nil {
165 return nil, err
166 }
167
168 job := domain.BackupJob{
169 ID: newBackupJobID(started),
170 Domain: domainName,
171 DBName: db.Name,
172 Type: domain.BackupTypeLogical,
173 Status: domain.BackupStatusCompleted,
174 StartedAt: started,
175 CompletedAt: time.Now().UTC(),
176 SizeBytes: counter.n,
177 FilePath: path,
178 }
179
180 return &domain.BackupResult{
181 Job: job,
182 Duration: time.Since(started),
183 }, nil
184}
185
186func isValidBackupSchedule(schedule domain.BackupSchedule) bool {
187 switch schedule {
188 case domain.ScheduleHourly, domain.ScheduleDaily, domain.ScheduleWeekly, domain.ScheduleMonthly:
189 return true
190 default:
191 return false
192 }
193}
194
195// Restore restores a backup by ID.
196func (s *Service) Restore(context.Context, string, string) error {
197 return fmt.Errorf("backup restore not implemented yet")
198}
199
200// RestorePITR restores to a point in time.
201func (s *Service) RestorePITR(context.Context, string, time.Time) error {
202 return fmt.Errorf("pitr restore not implemented yet")
203}
204
205// Status returns aggregate backup status for all managed domains.
206func (s *Service) Status(ctx context.Context) ([]domain.BackupJob, error) {
207 routes := s.containerSvc.List(ctx)
208 domainNames := make([]string, 0, len(routes))
209 for domainName := range routes {
210 domainNames = append(domainNames, domainName)
211 }
212 sort.Strings(domainNames)
213
214 const maxWorkers = 4
215 sem := make(chan struct{}, maxWorkers)
216 results := make([][]domain.BackupJob, len(domainNames))
217 errCh := make(chan error, 1)
218 var wg sync.WaitGroup
219
220 for i, domainName := range domainNames {
221 select {
222 case <-ctx.Done():
223 return nil, ctx.Err()
224 default:
225 }
226
227 select {
228 case sem <- struct{}{}:
229 case <-ctx.Done():
230 return nil, ctx.Err()
231 }
232
233 wg.Add(1)
234 go func(idx int, name string) {
235 defer wg.Done()
236 defer func() { <-sem }()
237
238 domainJobs, err := s.ListBackups(ctx, name)
239 if err != nil {
240 select {
241 case errCh <- err:
242 default:
243 }
244 return
245 }
246 results[idx] = domainJobs
247 }(i, domainName)
248 }
249
250 wg.Wait()
251 select {
252 case err := <-errCh:
253 return nil, err
254 default:
255 }
256
257 jobs := make([]domain.BackupJob, 0)
258 for _, domainJobs := range results {
259 jobs = append(jobs, domainJobs...)
260 }
261
262 return jobs, nil
263}
264
265func newBackupJobID(started time.Time) string {
266 random := make([]byte, 4)
267 if _, err := rand.Read(random); err != nil {
268 return fmt.Sprintf("%s-%d", started.Format(time.RFC3339Nano), time.Now().UnixNano())
269 }
270 return fmt.Sprintf("%s-%s", started.Format(time.RFC3339Nano), hex.EncodeToString(random))
271}
272
273func selectDatabase(dbs []domain.DBInfo, requested string) (domain.DBInfo, error) {
274 if len(dbs) == 0 {
275 return domain.DBInfo{}, fmt.Errorf("no supported database attachments detected")
276 }
277
278 if requested == "" {
279 if len(dbs) == 1 {
280 return dbs[0], nil
281 }
282 return domain.DBInfo{}, fmt.Errorf("multiple database attachments detected; please specify one")
283 }
284
285 for _, db := range dbs {
286 if strings.EqualFold(db.Name, requested) {
287 return db, nil
288 }
289 }
290
291 return domain.DBInfo{}, fmt.Errorf("database %q not found for domain", requested)
292}
293
294func postgresDumpCommand(_ string) string {
295 return "pg_dump -Fc --dbname=\"${POSTGRES_DB:-postgres}\" --username=\"${POSTGRES_USER:-postgres}\""
296}
297
298func pgDumpToPathCommand(path, dbName string) string {
299 return fmt.Sprintf("%s > %q", postgresDumpCommand(dbName), path)
300}
301
302func (s *Service) cleanupDumpFile(containerID, dumpPath string) {
303 cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
304 defer cancel()
305
306 _, _ = s.runtime.ExecInContainer(cleanupCtx, containerID, []string{"sh", "-c", fmt.Sprintf("rm -f %q", dumpPath)})
307}
308
309type byteCounter struct {
310 n int64
311}
312
313func (c *byteCounter) Write(p []byte) (int, error) {
314 c.n += int64(len(p))
315 return len(p), nil
316}