package postgres import ( "context" "database/sql" "encoding/json" "fmt" "time" "switchboard-core/models" "switchboard-core/store" ) type ScheduledTaskStore struct{} func NewScheduledTaskStore() *ScheduledTaskStore { return &ScheduledTaskStore{} } func (s *ScheduledTaskStore) Create(ctx context.Context, t *models.ScheduledTask) error { params, _ := json.Marshal(t.TemplateParams) if len(params) == 0 { params = []byte("{}") } return DB.QueryRowContext(ctx, `INSERT INTO scheduled_tasks (name, description, creator_id, run_as, cron_expr, next_fire_at, enabled, script, template_id, template_params) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING id, created_at, updated_at`, t.Name, t.Description, t.CreatorID, t.RunAs, t.CronExpr, t.NextFireAt, t.Enabled, t.Script, nilIfEmpty(t.TemplateID), params). Scan(&t.ID, &t.CreatedAt, &t.UpdatedAt) } func (s *ScheduledTaskStore) GetByID(ctx context.Context, id string) (*models.ScheduledTask, error) { var t models.ScheduledTask var params []byte err := DB.QueryRowContext(ctx, `SELECT id, name, description, creator_id, run_as, cron_expr, next_fire_at, last_fire_at, enabled, script, COALESCE(template_id,''), template_params, fire_count, COALESCE(last_error,''), last_duration_ms, created_at, updated_at FROM scheduled_tasks WHERE id = $1`, id). Scan(&t.ID, &t.Name, &t.Description, &t.CreatorID, &t.RunAs, &t.CronExpr, &t.NextFireAt, &t.LastFireAt, &t.Enabled, &t.Script, &t.TemplateID, ¶ms, &t.FireCount, &t.LastError, &t.LastDurationMs, &t.CreatedAt, &t.UpdatedAt) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } t.TemplateParams = json.RawMessage(params) return &t, nil } func (s *ScheduledTaskStore) Update(ctx context.Context, t *models.ScheduledTask) error { params, _ := json.Marshal(t.TemplateParams) if len(params) == 0 { params = []byte("{}") } _, err := DB.ExecContext(ctx, `UPDATE scheduled_tasks SET name = $1, description = $2, run_as = $3, cron_expr = $4, next_fire_at = $5, enabled = $6, script = $7, template_id = $8, template_params = $9, updated_at = NOW() WHERE id = $10`, t.Name, t.Description, t.RunAs, t.CronExpr, t.NextFireAt, t.Enabled, t.Script, nilIfEmpty(t.TemplateID), params, t.ID) return err } func (s *ScheduledTaskStore) Delete(ctx context.Context, id string) error { _, err := DB.ExecContext(ctx, `DELETE FROM scheduled_tasks WHERE id = $1`, id) return err } func (s *ScheduledTaskStore) List(ctx context.Context, opts store.ScheduledTaskListOptions) ([]models.ScheduledTask, int, error) { where := "1=1" args := []any{} n := 0 if opts.CreatorID != "" { n++ where += fmt.Sprintf(" AND creator_id = $%d", n) args = append(args, opts.CreatorID) } if opts.Enabled != nil { n++ where += fmt.Sprintf(" AND enabled = $%d", n) args = append(args, *opts.Enabled) } var total int countArgs := make([]any, len(args)) copy(countArgs, args) err := DB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduled_tasks WHERE "+where, countArgs...).Scan(&total) if err != nil { return nil, 0, err } limit := opts.Limit if limit <= 0 { limit = 50 } n++ query := fmt.Sprintf( `SELECT id, name, description, creator_id, run_as, cron_expr, next_fire_at, last_fire_at, enabled, script, COALESCE(template_id,''), template_params, fire_count, COALESCE(last_error,''), last_duration_ms, created_at, updated_at FROM scheduled_tasks WHERE %s ORDER BY created_at DESC LIMIT $%d OFFSET $%d`, where, n, n+1) args = append(args, limit, opts.Offset) rows, err := DB.QueryContext(ctx, query, args...) if err != nil { return nil, 0, err } defer rows.Close() tasks := scanScheduledTaskRows(rows) return tasks, total, rows.Err() } func (s *ScheduledTaskStore) ListByCreator(ctx context.Context, creatorID string) ([]models.ScheduledTask, error) { rows, err := DB.QueryContext(ctx, `SELECT id, name, description, creator_id, run_as, cron_expr, next_fire_at, last_fire_at, enabled, script, COALESCE(template_id,''), template_params, fire_count, COALESCE(last_error,''), last_duration_ms, created_at, updated_at FROM scheduled_tasks WHERE creator_id = $1 ORDER BY created_at DESC`, creatorID) if err != nil { return nil, err } defer rows.Close() return scanScheduledTaskRowsErr(rows) } func (s *ScheduledTaskStore) ListEnabled(ctx context.Context) ([]models.ScheduledTask, error) { rows, err := DB.QueryContext(ctx, `SELECT id, name, description, creator_id, run_as, cron_expr, next_fire_at, last_fire_at, enabled, script, COALESCE(template_id,''), template_params, fire_count, COALESCE(last_error,''), last_duration_ms, created_at, updated_at FROM scheduled_tasks WHERE enabled = true ORDER BY created_at`) if err != nil { return nil, err } defer rows.Close() return scanScheduledTaskRowsErr(rows) } func (s *ScheduledTaskStore) SetEnabled(ctx context.Context, id string, enabled bool) error { _, err := DB.ExecContext(ctx, `UPDATE scheduled_tasks SET enabled = $1, updated_at = NOW() WHERE id = $2`, enabled, id) return err } func (s *ScheduledTaskStore) UpdateFireState(ctx context.Context, id string, lastFireAt time.Time, nextFireAt *time.Time, lastError string, durationMs int) error { _, err := DB.ExecContext(ctx, `UPDATE scheduled_tasks SET last_fire_at = $1, next_fire_at = $2, last_error = $3, last_duration_ms = $4, fire_count = fire_count + 1, updated_at = NOW() WHERE id = $5`, lastFireAt, nextFireAt, nilIfEmpty(lastError), durationMs, id) return err } // ── Helpers ────────────────────────────────── func scanScheduledTaskRows(rows *sql.Rows) []models.ScheduledTask { var tasks []models.ScheduledTask for rows.Next() { t := scanOneScheduledTask(rows) if t != nil { tasks = append(tasks, *t) } } if tasks == nil { tasks = []models.ScheduledTask{} } return tasks } func scanScheduledTaskRowsErr(rows *sql.Rows) ([]models.ScheduledTask, error) { var tasks []models.ScheduledTask for rows.Next() { var t models.ScheduledTask var params []byte if err := rows.Scan(&t.ID, &t.Name, &t.Description, &t.CreatorID, &t.RunAs, &t.CronExpr, &t.NextFireAt, &t.LastFireAt, &t.Enabled, &t.Script, &t.TemplateID, ¶ms, &t.FireCount, &t.LastError, &t.LastDurationMs, &t.CreatedAt, &t.UpdatedAt); err != nil { return nil, err } t.TemplateParams = json.RawMessage(params) tasks = append(tasks, t) } if tasks == nil { tasks = []models.ScheduledTask{} } return tasks, nil } func scanOneScheduledTask(rows *sql.Rows) *models.ScheduledTask { var t models.ScheduledTask var params []byte if err := rows.Scan(&t.ID, &t.Name, &t.Description, &t.CreatorID, &t.RunAs, &t.CronExpr, &t.NextFireAt, &t.LastFireAt, &t.Enabled, &t.Script, &t.TemplateID, ¶ms, &t.FireCount, &t.LastError, &t.LastDurationMs, &t.CreatedAt, &t.UpdatedAt); err != nil { return nil } t.TemplateParams = json.RawMessage(params) return &t }