Function RehydrateScheduledReports

server-source-code/internal/queue/server.go:263–313

RehydrateScheduledReports enqueues delayed tasks for all enabled scheduled reports. Call this at startup so that event-driven report chains are seeded after a restart.

(qc *asynq.Client, defaultDB *database.DB, poolCache *hostctx.PoolCache, log *slog.Logger)

// RehydrateScheduledReports enqueues delayed tasks for all enabled scheduled reports.
// Call this at startup so that event-driven report chains are seeded after a restart.
func RehydrateScheduledReports(qc *asynq.Client, defaultDB *database.DB, poolCache *hostctx.PoolCache, log *slog.Logger) {
	if qc == nil || defaultDB == nil {
		return
	}
	ctx := context.Background()
	rehydrateDB := func(d *database.DB, host string) {
		now := time.Now()
		// Use a far-future cutoff to get ALL enabled reports with a next_run_at.
		rows, err := d.Queries.ListScheduledReportsDue(ctx, pgtime.From(now.Add(365*24*time.Hour)))
		if err != nil {
			if log != nil {
				log.Error("rehydrate scheduled reports: list failed", "error", err)
			}
			return
		}
		for _, r := range rows {
			runAt := r.NextRunAt.Time
			if !r.NextRunAt.Valid || runAt.Before(now) {
				// Compute actual next cron time instead of running immediately,
				// to avoid spurious duplicate runs on every restart.
				tz := r.Timezone
				if tz == "" {
					tz = "UTC"
				}
				if next, nerr := notifications.NextCronRun(r.CronExpr, tz, now); nerr == nil {
					runAt = next
				} else {
					runAt = now
				}
			}
			if err := EnqueueScheduledReportAt(qc, r.ID, host, runAt); err != nil {
				if log != nil {
					log.Error("rehydrate scheduled reports: enqueue failed", "report_id", r.ID, "error", err)
				}
			}
		}
		if log != nil && len(rows) > 0 {
			log.Info("rehydrate scheduled reports", "host", host, "count", len(rows))
		}
	}
	rehydrateDB(defaultDB, "")
	if poolCache != nil {
		for _, host := range poolCache.ListHosts() {
			d, err := poolCache.GetOrCreate(ctx, host)
			if err != nil || d == nil {
				continue
			}
			rehydrateDB(d, host)
		}
	}
}