feat(backup): incremental backup system for ~/.obk data

config/config.go

@@ -38,6 +38,7 @@ type Config struct {
 	Slack        *SlackConfig        `yaml:"slack,omitempty"`
 	Scheduler    *SchedulerConfig    `yaml:"scheduler,omitempty"`
 	Tasks        *TasksConfig        `yaml:"tasks,omitempty"`
+	Backup       *BackupConfig       `yaml:"backup,omitempty"`
 }
 
 func (c *Config) ResolvedMode() Mode {
@@ -207,6 +208,25 @@ type TasksConfig struct {
 	Storage StorageConfig `yaml:"storage,omitempty"`
 }
 
+type BackupConfig struct {
+	Enabled     bool          `yaml:"enabled,omitempty"`
+	Schedule    string        `yaml:"schedule,omitempty"` // "6h", "12h", "24h", or "" (manual)
+	Destination string        `yaml:"destination,omitempty"` // "r2" or "gdrive"
+	R2          *R2Config     `yaml:"r2,omitempty"`
+	GDrive      *GDriveConfig `yaml:"gdrive,omitempty"`
+}
+
+type R2Config struct {
+	Bucket       string `yaml:"bucket,omitempty"`
+	Endpoint     string `yaml:"endpoint,omitempty"`
+	AccessKeyRef string `yaml:"access_key_ref,omitempty"` // e.g. "keychain:obk/r2-access-key"
+	SecretKeyRef string `yaml:"secret_key_ref,omitempty"` // e.g. "keychain:obk/r2-secret-key"
+}
+
+type GDriveConfig struct {
+	FolderID string `yaml:"folder_id,omitempty"`
+}
+
 type StorageConfig struct {
 	Driver string `yaml:"driver,omitempty"` // "sqlite" or "postgres"
 	DSN    string `yaml:"dsn,omitempty"`

config/paths.go

@@ -90,3 +90,15 @@ func CleanScratch(sessionID string) error {
 func LearningsDir() string {
 	return filepath.Join(Dir(), "learnings")
 }
+
+func BackupDir() string {
+	return filepath.Join(Dir(), "backup")
+}
+
+func BackupStagingDir() string {
+	return filepath.Join(BackupDir(), "staging")
+}
+
+func BackupLastManifestPath() string {
+	return filepath.Join(BackupDir(), "last_manifest.json")
+}

daemon/jobs/backup.go

@@ -0,0 +1,87 @@
+package jobs
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/http"
+
+	"github.com/riverqueue/river"
+
+	"github.com/73ai/openbotkit/config"
+	"github.com/73ai/openbotkit/oauth/google"
+	"github.com/73ai/openbotkit/provider"
+	backupsvc "github.com/73ai/openbotkit/service/backup"
+)
+
+type BackupArgs struct{}
+
+func (BackupArgs) Kind() string { return "backup" }
+
+type BackupWorker struct {
+	river.WorkerDefaults[BackupArgs]
+	Cfg *config.Config
+}
+
+func (w *BackupWorker) Work(ctx context.Context, job *river.Job[BackupArgs]) error {
+	if w.Cfg.Backup == nil || !w.Cfg.Backup.Enabled {
+		slog.Info("backup: not enabled, skipping")
+		return nil
+	}
+
+	if !config.IsSourceLinked("backup") {
+		slog.Info("backup: not linked, skipping")
+		return nil
+	}
+
+	slog.Info("starting backup job")
+
+	backend, err := backupsvc.ResolveBackend(ctx, backendOpts(w.Cfg))
+	if err != nil {
+		return fmt.Errorf("resolve backend: %w", err)
+	}
+
+	svc := backupsvc.New(backend, config.Dir())
+	result, err := svc.Run(ctx)
+	if err != nil {
+		return fmt.Errorf("backup: %w", err)
+	}
+
+	slog.Info("backup complete",
+		"changed", result.Changed,
+		"skipped", result.Skipped,
+		"uploaded", result.Uploaded,
+		"duration", result.Duration,
+	)
+	return nil
+}
+
+func backendOpts(cfg *config.Config) backupsvc.ResolveBackendOpts {
+	opts := backupsvc.ResolveBackendOpts{
+		ResolveCred: provider.ResolveAPIKey,
+		BackupDest:  cfg.Backup.Destination,
+		GoogleClient: func(ctx context.Context, gcfg backupsvc.GoogleClientConfig) (*http.Client, error) {
+			gp := google.New(google.Config{
+				CredentialsFile: cfg.GoogleCredentialsFile(),
+				TokenDBPath:     cfg.GoogleTokenDBPath(),
+			})
+			accounts, err := gp.Accounts(ctx)
+			if err != nil || len(accounts) == 0 {
+				return nil, fmt.Errorf("no Google account found")
+			}
+			return gp.Client(ctx, accounts[0], gcfg.Scopes)
+		},
+	}
+	if cfg.Backup.R2 != nil {
+		opts.R2Bucket = cfg.Backup.R2.Bucket
+		opts.R2Endpoint = cfg.Backup.R2.Endpoint
+		opts.R2AccessRef = cfg.Backup.R2.AccessKeyRef
+		opts.R2SecretRef = cfg.Backup.R2.SecretKeyRef
+	}
+	if cfg.Backup.GDrive != nil {
+		opts.GDriveFolderID = cfg.Backup.GDrive.FolderID
+	}
+	return opts
+}
+
+var _ river.Worker[BackupArgs] = (*BackupWorker)(nil)

daemon/river.go

@@ -42,26 +42,42 @@ func newRiverClient(ctx context.Context, cfg *config.Config, notifier *SyncNotif
 	})
 	river.AddWorker(workers, &jobs.ReminderWorker{})
 	river.AddWorker(workers, &jobs.ScheduledTaskWorker{Cfg: cfg})
+	river.AddWorker(workers, &jobs.BackupWorker{Cfg: cfg})
 
 	period, err := time.ParseDuration(cfg.Daemon.GmailSyncPeriod)
 	if err != nil {
 		period = 15 * time.Minute
 	}
 
+	periodicJobs := []*river.PeriodicJob{
+		river.NewPeriodicJob(
+			river.PeriodicInterval(period),
+			func() (river.JobArgs, *river.InsertOpts) {
+				return jobs.GmailSyncArgs{}, nil
+			},
+			&river.PeriodicJobOpts{RunOnStart: true},
+		),
+	}
+
+	if cfg.Backup != nil && cfg.Backup.Enabled && cfg.Backup.Schedule != "" {
+		backupPeriod, err := time.ParseDuration(cfg.Backup.Schedule)
+		if err == nil {
+			periodicJobs = append(periodicJobs, river.NewPeriodicJob(
+				river.PeriodicInterval(backupPeriod),
+				func() (river.JobArgs, *river.InsertOpts) {
+					return jobs.BackupArgs{}, nil
+				},
+				&river.PeriodicJobOpts{RunOnStart: false},
+			))
+		}
+	}
+
 	riverCfg := &river.Config{
 		Queues: map[string]river.QueueConfig{
 			river.QueueDefault: {MaxWorkers: 5},
 		},
-		Workers: workers,
-		PeriodicJobs: []*river.PeriodicJob{
-			river.NewPeriodicJob(
-				river.PeriodicInterval(period),
-				func() (river.JobArgs, *river.InsertOpts) {
-					return jobs.GmailSyncArgs{}, nil
-				},
-				&river.PeriodicJobOpts{RunOnStart: true},
-			),
-		},
+		Workers:      workers,
+		PeriodicJobs: periodicJobs,
 	}
 
 	client, err := river.NewClient(driver, riverCfg)