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"
)

// RunCounts is what a run did, in numbers.
//
// Four figures rather than a free map, because these are the four questions an operator
// asks of any piece of batch work — how much did it look at, how much did it change, how
// much did it decline, how much went wrong — and a schema-free bag would let two
// integrations answer them under different names. Anything an integration counts beyond
// these belongs in the run's own sentence.
//
// Processed is the load-bearing one: a run reporting zero processed counted nothing at
// all, and the console draws no figures rather than four zeroes.
type RunCounts struct {
	Processed int `json:"processed"`
	Changed   int `json:"changed"`
	Skipped   int `json:"skipped"`
	Failed    int `json:"failed"`
}

// Counted reports whether this run counted anything worth printing.
func (c RunCounts) Counted() bool {
	return c.Processed != 0 || c.Changed != 0 || c.Skipped != 0 || c.Failed != 0
}

// TaskRun is one execution.
type TaskRun struct {
	ID     int64  `json:"id"`
	TaskID string `json:"taskId"`
	// IntegrationID names the external service this run belongs to, and is empty for the
	// gateway's own housekeeping. It is what lets the integrations area read operational
	// history out of the scheduler's own table instead of keeping a second one.
	IntegrationID string     `json:"integrationId,omitempty"`
	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"`
	Counts        RunCounts  `json:"counts"`
}

// 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, integrationID, trigger string,
) (int64, error) {
	var id int64
	err := s.pool.QueryRow(ctx, `
		INSERT INTO scheduled_task_runs (task_id, integration_id, trigger, status)
		VALUES ($1, $2, $3, $4) RETURNING id`,
		taskID, integrationID, 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 and whatever it counted.
func (s *Store) FinishTaskRun(
	ctx context.Context, id int64, status, detail, failure string, counts RunCounts,
) error {
	_, err := s.pool.Exec(ctx, `
		UPDATE scheduled_task_runs
		SET status = $2, finished_at = now(), detail = $3, error = $4,
			processed = $5, changed = $6, skipped = $7, failed = $8,
			duration_ms = GREATEST(0, EXTRACT(EPOCH FROM (now() - started_at)) * 1000)::bigint
		WHERE id = $1`, id, status, detail, failure,
		counts.Processed, counts.Changed, counts.Skipped, counts.Failed)
	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, integration_id, trigger, status, started_at, finished_at,
			duration_ms, detail, error, processed, changed, skipped, failed
		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.IntegrationID, &run.Trigger,
			&run.Status, &run.StartedAt, &run.FinishedAt, &run.DurationMS, &run.Detail,
			&run.Error, &run.Counts.Processed, &run.Counts.Changed,
			&run.Counts.Skipped, &run.Counts.Failed); 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, integration_id, trigger, status, started_at, finished_at,
			duration_ms, detail, error, processed, changed, skipped, failed
		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.IntegrationID, &run.Trigger,
			&run.Status, &run.StartedAt, &run.FinishedAt, &run.DurationMS, &run.Detail,
			&run.Error, &run.Counts.Processed, &run.Counts.Changed,
			&run.Counts.Skipped, &run.Counts.Failed); 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
}

// --- integration run history ------------------------------------------------------
//
// The same rows as above, read with a different question in mind. An integration run *is*
// a scheduled task run: giving the integrations area a table of its own would mean two
// schedulers, two retention jobs and two places one piece of work could be recorded as
// having failed. What differs is only the axis — by service rather than by job.

// IntegrationRuns is the operational history for one external service, newest first.
func (s *Store) IntegrationRuns(
	ctx context.Context, integrationID string, limit int,
) ([]TaskRun, error) {
	if limit <= 0 || limit > 500 {
		limit = 50
	}
	rows, err := s.pool.Query(ctx, `
		SELECT id, task_id, integration_id, trigger, status, started_at, finished_at,
			duration_ms, detail, error, processed, changed, skipped, failed
		FROM scheduled_task_runs
		WHERE integration_id <> '' AND ($1 = '' OR integration_id = $1)
		ORDER BY started_at DESC, id DESC
		LIMIT $2`, integrationID, limit)
	if err != nil {
		return nil, fmt.Errorf("store: list integration runs: %w", err)
	}
	defer rows.Close()
	runs := []TaskRun{}
	for rows.Next() {
		var run TaskRun
		if err := rows.Scan(&run.ID, &run.TaskID, &run.IntegrationID, &run.Trigger,
			&run.Status, &run.StartedAt, &run.FinishedAt, &run.DurationMS, &run.Detail,
			&run.Error, &run.Counts.Processed, &run.Counts.Changed,
			&run.Counts.Skipped, &run.Counts.Failed); err != nil {
			return nil, fmt.Errorf("store: scan integration run: %w", err)
		}
		runs = append(runs, run)
	}
	return runs, rows.Err()
}

// IntegrationRunSummary answers the overview's whole row for one service.
//
// Last success and last failure are both carried because they are different questions and
// the answer to one is not the absence of the other: a service that failed an hour ago and
// has worked since is healthy, and one that succeeded last week and has failed every hour
// since is not. Neither is derivable from a single "last run".
type IntegrationRunSummary struct {
	IntegrationID string     `json:"integrationId"`
	LastSuccessAt *time.Time `json:"lastSuccessAt,omitempty"`
	LastFailureAt *time.Time `json:"lastFailureAt,omitempty"`
	LastError     string     `json:"lastError,omitempty"`
	Runs          int        `json:"runs"`
	Failures      int        `json:"failures"`
}

// IntegrationRunSummaries is every service's summary in one query, because the overview
// draws one per row and a query per integration would grow with the catalogue.
func (s *Store) IntegrationRunSummaries(
	ctx context.Context,
) (map[string]IntegrationRunSummary, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT integration_id,
			max(started_at) FILTER (WHERE status = $1),
			max(started_at) FILTER (WHERE status = $2),
			(array_remove(array_agg(error ORDER BY started_at DESC)
				FILTER (WHERE status = $2), ''))[1],
			count(*), count(*) FILTER (WHERE status = $2)
		FROM scheduled_task_runs
		WHERE integration_id <> ''
		GROUP BY integration_id`, TaskSuccess, TaskFailed)
	if err != nil {
		return nil, fmt.Errorf("store: integration run summaries: %w", err)
	}
	defer rows.Close()
	summaries := map[string]IntegrationRunSummary{}
	for rows.Next() {
		var summary IntegrationRunSummary
		var lastError *string
		if err := rows.Scan(&summary.IntegrationID, &summary.LastSuccessAt,
			&summary.LastFailureAt, &lastError, &summary.Runs,
			&summary.Failures); err != nil {
			return nil, fmt.Errorf("store: scan integration run summary: %w", err)
		}
		if lastError != nil {
			summary.LastError = *lastError
		}
		summaries[summary.IntegrationID] = summary
	}
	return summaries, rows.Err()
}