memby/server/internal/store/scheduler.go

package store

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
)

// TaskRunRetention is how much run history a task keeps. Enough to see a pattern in an
// overnight job, not so much that a task running every five minutes fills the table.
const TaskRunRetention = 30 * 24 * time.Hour

// Run statuses.
const (
	TaskRunning = "running"
	TaskSuccess = "success"
	TaskFailed  = "failed"
	TaskSkipped = "skipped"
)

// Triggers. A run says how it came to happen, because "it has not run since Tuesday" and
// "it has only ever run when somebody pressed the button" are different problems.
const (
	TriggerSchedule = "schedule"
	TriggerManual   = "manual"
	TriggerStartup  = "startup"
)

// TaskRun is one execution.
type TaskRun struct {
	ID         int64      `json:"id"`
	TaskID     string     `json:"taskId"`
	Trigger    string     `json:"trigger"`
	Status     string     `json:"status"`
	StartedAt  time.Time  `json:"startedAt"`
	FinishedAt *time.Time `json:"finishedAt,omitempty"`
	DurationMS int64      `json:"durationMs"`
	Detail     string     `json:"detail,omitempty"`
	Error      string     `json:"error,omitempty"`
}

// TaskSettings is an operator's override for one task. IntervalSeconds of 0 means "the
// interval the task declares", so a task whose schedule is changed in code takes effect
// for every operator who never overrode it.
type TaskSettings struct {
	TaskID          string    `json:"taskId"`
	Enabled         bool      `json:"enabled"`
	IntervalSeconds int       `json:"intervalSeconds"`
	UpdatedAt       time.Time `json:"updatedAt"`
}

// BeginTaskRun opens a run and returns its id. The row exists before the work starts so a
// task killed by a restart leaves evidence it began — which is the only way to tell a job
// that hangs from one that was never scheduled.
func (s *Store) BeginTaskRun(ctx context.Context, taskID, trigger string) (int64, error) {
	var id int64
	err := s.pool.QueryRow(ctx, `
		INSERT INTO scheduled_task_runs (task_id, trigger, status)
		VALUES ($1, $2, $3) RETURNING id`, taskID, trigger, TaskRunning).Scan(&id)
	if err != nil {
		return 0, fmt.Errorf("store: begin task run: %w", err)
	}
	return id, nil
}

// FinishTaskRun closes a run with its outcome.
func (s *Store) FinishTaskRun(ctx context.Context, id int64, status, detail, failure string) error {
	_, err := s.pool.Exec(ctx, `
		UPDATE scheduled_task_runs
		SET status = $2, finished_at = now(), detail = $3, error = $4,
			duration_ms = GREATEST(0, EXTRACT(EPOCH FROM (now() - started_at)) * 1000)::bigint
		WHERE id = $1`, id, status, detail, failure)
	if err != nil {
		return fmt.Errorf("store: finish task run: %w", err)
	}
	return nil
}

// AbandonRunningTasks closes runs left open by a process that went away.
//
// Called once at start-up: a run in "running" with nothing running it is a lie the
// console would otherwise print for ever, and it is indistinguishable from a genuinely
// long job unless it is resolved at the one moment the answer is known — the moment the
// process that could have owned it has just started.
func (s *Store) AbandonRunningTasks(ctx context.Context) (int64, error) {
	tag, err := s.pool.Exec(ctx, `
		UPDATE scheduled_task_runs
		SET status = $1, finished_at = now(),
			error = 'interrupted by a server restart',
			duration_ms = GREATEST(0, EXTRACT(EPOCH FROM (now() - started_at)) * 1000)::bigint
		WHERE status = $2`, TaskFailed, TaskRunning)
	if err != nil {
		return 0, fmt.Errorf("store: abandon running tasks: %w", err)
	}
	return tag.RowsAffected(), nil
}

// LatestTaskRuns is the most recent run per task, in one query — the console draws this
// beside every task and a query per task would grow with the registry.
func (s *Store) LatestTaskRuns(ctx context.Context) (map[string]TaskRun, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT DISTINCT ON (task_id)
			id, task_id, trigger, status, started_at, finished_at, duration_ms, detail, error
		FROM scheduled_task_runs
		ORDER BY task_id, started_at DESC, id DESC`)
	if err != nil {
		return nil, fmt.Errorf("store: latest task runs: %w", err)
	}
	defer rows.Close()
	latest := map[string]TaskRun{}
	for rows.Next() {
		var run TaskRun
		if err := rows.Scan(&run.ID, &run.TaskID, &run.Trigger, &run.Status,
			&run.StartedAt, &run.FinishedAt, &run.DurationMS, &run.Detail,
			&run.Error); err != nil {
			return nil, fmt.Errorf("store: scan task run: %w", err)
		}
		latest[run.TaskID] = run
	}
	return latest, rows.Err()
}

// TaskRuns is the history for one task, or for every task when taskID is blank.
func (s *Store) TaskRuns(ctx context.Context, taskID string, limit int) ([]TaskRun, error) {
	if limit <= 0 || limit > 500 {
		limit = 50
	}
	rows, err := s.pool.Query(ctx, `
		SELECT id, task_id, trigger, status, started_at, finished_at, duration_ms, detail, error
		FROM scheduled_task_runs
		WHERE ($1 = '' OR task_id = $1)
		ORDER BY started_at DESC, id DESC
		LIMIT $2`, taskID, limit)
	if err != nil {
		return nil, fmt.Errorf("store: list task runs: %w", err)
	}
	defer rows.Close()
	runs := []TaskRun{}
	for rows.Next() {
		var run TaskRun
		if err := rows.Scan(&run.ID, &run.TaskID, &run.Trigger, &run.Status,
			&run.StartedAt, &run.FinishedAt, &run.DurationMS, &run.Detail,
			&run.Error); err != nil {
			return nil, fmt.Errorf("store: scan task run: %w", err)
		}
		runs = append(runs, run)
	}
	return runs, rows.Err()
}

// TaskSettingsAll returns every override. A task with no row is absent from the map, and
// absence means "as declared" — see TaskSettings.
func (s *Store) TaskSettingsAll(ctx context.Context) (map[string]TaskSettings, error) {
	rows, err := s.pool.Query(ctx,
		`SELECT task_id, enabled, interval_seconds, updated_at FROM scheduled_task_settings`)
	if err != nil {
		return nil, fmt.Errorf("store: task settings: %w", err)
	}
	defer rows.Close()
	settings := map[string]TaskSettings{}
	for rows.Next() {
		var row TaskSettings
		if err := rows.Scan(&row.TaskID, &row.Enabled, &row.IntervalSeconds,
			&row.UpdatedAt); err != nil {
			return nil, fmt.Errorf("store: scan task settings: %w", err)
		}
		settings[row.TaskID] = row
	}
	return settings, rows.Err()
}

// SetTaskSettings records an operator's override.
func (s *Store) SetTaskSettings(ctx context.Context, settings TaskSettings) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO scheduled_task_settings (task_id, enabled, interval_seconds, updated_at)
		VALUES ($1, $2, $3, now())
		ON CONFLICT (task_id) DO UPDATE SET
			enabled = EXCLUDED.enabled,
			interval_seconds = EXCLUDED.interval_seconds,
			updated_at = now()`,
		settings.TaskID, settings.Enabled, settings.IntervalSeconds)
	if err != nil {
		return fmt.Errorf("store: set task settings: %w", err)
	}
	return nil
}

// TaskLastSuccess is when each task last finished cleanly, which is what the scheduler
// restores its clock from after a restart — without it a container replaced at 3am would
// re-run every overnight job the moment it came up.
func (s *Store) TaskLastSuccess(ctx context.Context) (map[string]time.Time, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT task_id, max(started_at) FROM scheduled_task_runs
		WHERE status = $1 GROUP BY task_id`, TaskSuccess)
	if err != nil {
		return nil, fmt.Errorf("store: task last success: %w", err)
	}
	defer rows.Close()
	last := map[string]time.Time{}
	for rows.Next() {
		var taskID string
		var at time.Time
		if err := rows.Scan(&taskID, &at); err != nil {
			return nil, fmt.Errorf("store: scan task last success: %w", err)
		}
		last[taskID] = at
	}
	return last, rows.Err()
}

// PruneTaskRuns drops run history past the retention period.
func (s *Store) PruneTaskRuns(ctx context.Context, retention time.Duration) (int64, error) {
	if retention <= 0 {
		retention = TaskRunRetention
	}
	tag, err := s.pool.Exec(ctx,
		`DELETE FROM scheduled_task_runs WHERE started_at < now() - $1::interval`,
		fmt.Sprintf("%d seconds", int64(retention.Seconds())))
	if err != nil {
		return 0, fmt.Errorf("store: prune task runs: %w", err)
	}
	return tag.RowsAffected(), nil
}

// --- integration delivery history -------------------------------------------------

// IntegrationDelivery is one attempt to hand an event to an external service.
type IntegrationDelivery struct {
	ID            int64     `json:"id"`
	IntegrationID string    `json:"integrationId"`
	EventType     string    `json:"eventType"`
	AttemptedAt   time.Time `json:"attemptedAt"`
	Success       bool      `json:"success"`
	StatusCode    int       `json:"statusCode"`
	DurationMS    int64     `json:"durationMs"`
	Error         string    `json:"error,omitempty"`
}

// IntegrationHealth is the pair of questions an operator actually has about a webhook:
// is it working, and if not, what did it say. Both are "last", not "count", because a
// destination that failed once an hour ago and has worked since is healthy.
type IntegrationHealth struct {
	IntegrationID string     `json:"integrationId"`
	LastSuccess   *time.Time `json:"lastSuccess,omitempty"`
	LastFailure   *time.Time `json:"lastFailure,omitempty"`
	LastError     string     `json:"lastError,omitempty"`
	Deliveries    int        `json:"deliveries"`
	Failures      int        `json:"failures"`
}

func (s *Store) RecordIntegrationDelivery(ctx context.Context, delivery IntegrationDelivery) error {
	_, err := s.pool.Exec(ctx, `
		INSERT INTO integration_deliveries (
			integration_id, event_type, success, status_code, duration_ms, error
		) VALUES ($1, $2, $3, $4, $5, $6)`,
		delivery.IntegrationID, delivery.EventType, delivery.Success,
		delivery.StatusCode, delivery.DurationMS, delivery.Error)
	if err != nil {
		return fmt.Errorf("store: record integration delivery: %w", err)
	}
	return nil
}

func (s *Store) IntegrationDeliveries(ctx context.Context, integrationID string, limit int) ([]IntegrationDelivery, error) {
	if limit <= 0 || limit > 200 {
		limit = 25
	}
	rows, err := s.pool.Query(ctx, `
		SELECT id, integration_id, event_type, attempted_at, success, status_code,
			duration_ms, error
		FROM integration_deliveries
		WHERE ($1 = '' OR integration_id = $1)
		ORDER BY attempted_at DESC, id DESC
		LIMIT $2`, integrationID, limit)
	if err != nil {
		return nil, fmt.Errorf("store: list integration deliveries: %w", err)
	}
	defer rows.Close()
	deliveries := []IntegrationDelivery{}
	for rows.Next() {
		var delivery IntegrationDelivery
		if err := rows.Scan(&delivery.ID, &delivery.IntegrationID, &delivery.EventType,
			&delivery.AttemptedAt, &delivery.Success, &delivery.StatusCode,
			&delivery.DurationMS, &delivery.Error); err != nil {
			return nil, fmt.Errorf("store: scan integration delivery: %w", err)
		}
		deliveries = append(deliveries, delivery)
	}
	return deliveries, rows.Err()
}

func (s *Store) IntegrationHealthFor(ctx context.Context, integrationID string) (IntegrationHealth, error) {
	health := IntegrationHealth{IntegrationID: integrationID}
	var lastError *string
	err := s.pool.QueryRow(ctx, `
		SELECT max(attempted_at) FILTER (WHERE success),
			max(attempted_at) FILTER (WHERE NOT success),
			(SELECT error FROM integration_deliveries
			 WHERE integration_id = $1 AND NOT success
			 ORDER BY attempted_at DESC LIMIT 1),
			count(*), count(*) FILTER (WHERE NOT success)
		FROM integration_deliveries WHERE integration_id = $1`, integrationID).
		Scan(&health.LastSuccess, &health.LastFailure, &lastError,
			&health.Deliveries, &health.Failures)
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return health, fmt.Errorf("store: integration health: %w", err)
	}
	if lastError != nil {
		health.LastError = *lastError
	}
	return health, nil
}

// PruneIntegrationDeliveries keeps the newest `keep` attempts per integration.
//
// It is bounded per integration rather than by age, because the useful property of this
// table is "the last few attempts for each destination" — an age cut would empty it
// entirely for a webhook that fires once a month, which is the one whose last delivery an
// operator most wants to see.
func (s *Store) PruneIntegrationDeliveries(ctx context.Context, keep int) (int64, error) {
	if keep <= 0 {
		keep = 100
	}
	tag, err := s.pool.Exec(ctx, `
		DELETE FROM integration_deliveries WHERE id IN (
			SELECT id FROM (
				SELECT id, row_number() OVER (
					PARTITION BY integration_id ORDER BY attempted_at DESC, id DESC
				) AS position
				FROM integration_deliveries
			) ranked WHERE position > $1
		)`, keep)
	if err != nil {
		return 0, fmt.Errorf("store: prune integration deliveries: %w", err)
	}
	return tag.RowsAffected(), nil
}