Merge pull request #133 from 73ai/backup-failures

fix: unified scheduler with sleep/wake catch-up and backup failure alerts

Author: priyanshujain

daemon/jobs/backup.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 	"log/slog"
 	"net/http"
+	"time"
 
 	"github.com/riverqueue/river"
 
+	"github.com/73ai/openbotkit/channel"
 	"github.com/73ai/openbotkit/config"
 	"github.com/73ai/openbotkit/oauth/google"
 	"github.com/73ai/openbotkit/provider"
@@ -34,16 +36,33 @@ func (w *BackupWorker) Work(ctx context.Context, job *river.Job[BackupArgs]) err
 		return nil
 	}
 
+	schedule, err := time.ParseDuration(w.Cfg.Backup.Schedule)
+	if err == nil && schedule > 0 {
+		manifest, mErr := backupsvc.LoadManifest(config.BackupLastManifestPath())
+		if mErr == nil && manifest.ID != "" && time.Since(manifest.Timestamp) < schedule {
+			slog.Info("backup: last backup is recent, skipping",
+				"age", time.Since(manifest.Timestamp).Round(time.Minute),
+				"schedule", schedule)
+			return nil
+		}
+	}
+
 	slog.Info("starting backup job")
 
 	backend, err := backupsvc.ResolveBackend(ctx, backendOpts(w.Cfg))
 	if err != nil {
+		if job.Attempt >= job.MaxAttempts {
+			w.notifyFailure(ctx, err)
+		}
 		return fmt.Errorf("resolve backend: %w", err)
 	}
 
 	svc := backupsvc.New(backend, config.Dir())
 	result, err := svc.Run(ctx)
 	if err != nil {
+		if job.Attempt >= job.MaxAttempts {
+			w.notifyFailure(ctx, err)
+		}
 		return fmt.Errorf("backup: %w", err)
 	}
 
@@ -56,6 +75,25 @@ func (w *BackupWorker) Work(ctx context.Context, job *river.Job[BackupArgs]) err
 	return nil
 }
 
+func (w *BackupWorker) notifyFailure(ctx context.Context, backupErr error) {
+	if w.Cfg.Channels == nil {
+		return
+	}
+	tg := w.Cfg.Channels.Telegram
+	if tg == nil || tg.BotToken == "" {
+		return
+	}
+	pusher, err := channel.NewTelegramPusher(tg.BotToken, tg.OwnerID)
+	if err != nil {
+		slog.Error("backup: create telegram pusher", "error", err)
+		return
+	}
+	msg := fmt.Sprintf("Backup failed: %s", backupErr)
+	if err := pusher.Push(ctx, msg); err != nil {
+		slog.Error("backup: send failure alert", "error", err)
+	}
+}
+
 func backendOpts(cfg *config.Config) backupsvc.ResolveBackendOpts {
 	opts := backupsvc.ResolveBackendOpts{
 		ResolveCred: provider.ResolveAPIKey,

daemon/jobs/backup_test.go

@@ -0,0 +1,34 @@
+package jobs
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/73ai/openbotkit/config"
+)
+
+func TestBackupArgs_Kind(t *testing.T) {
+	args := BackupArgs{}
+	if args.Kind() != "backup" {
+		t.Errorf("Kind() = %q, want %q", args.Kind(), "backup")
+	}
+}
+
+func TestBackupWorker_notifyFailure_NoTelegram(t *testing.T) {
+	w := &BackupWorker{Cfg: &config.Config{}}
+	// Should not panic when Channels is nil.
+	w.notifyFailure(context.Background(), errors.New("test error"))
+}
+
+func TestBackupWorker_notifyFailure_EmptyBotToken(t *testing.T) {
+	w := &BackupWorker{
+		Cfg: &config.Config{
+			Channels: &config.ChannelsConfig{
+				Telegram: &config.TelegramConfig{BotToken: "", OwnerID: 0},
+			},
+		},
+	}
+	// Should return early when BotToken is empty.
+	w.notifyFailure(context.Background(), errors.New("test error"))
+}

daemon/river.go

@@ -59,19 +59,6 @@ func newRiverClient(ctx context.Context, cfg *config.Config, notifier *SyncNotif
 		),
 	}
 
-	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},