core/server/store/postgres/scheduled_tasks.go

package postgres

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"

	"armature/models"
	"armature/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, &params, &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, &params, &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, &params, &t.FireCount,
		&t.LastError, &t.LastDurationMs, &t.CreatedAt, &t.UpdatedAt); err != nil {
		return nil
	}
	t.TemplateParams = json.RawMessage(params)
	return &t
}