0.1.38 gateway
This commit is contained in:
ponzischeme89
@@ -0,0 +1,340 @@
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user