admin/memby
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.2.78

Browse Source
This commit is contained in:
ponzischeme89
2026-08-19 14:25:44 +12:00
parent 2b43b9ef12
commit 590e069366
83 changed files with 8948 additions and 1266 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/store/scheduler.go
+151 -23
View File
@@ -29,17 +29,44 @@ const (
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"`
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"`
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
@@ -55,24 +82,31 @@ type TaskSettings struct {
// 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) {
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, trigger, status)
VALUES ($1, $2, $3) RETURNING id`, taskID, trigger, TaskRunning).Scan(&id)
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.
func (s *Store) FinishTaskRun(ctx context.Context, id int64, status, detail, failure string) error {
// 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)
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)
}
@@ -103,7 +137,8 @@ func (s *Store) AbandonRunningTasks(ctx context.Context) (int64, error) {
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
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 {
@@ -113,9 +148,10 @@ func (s *Store) LatestTaskRuns(ctx context.Context) (map[string]TaskRun, error)
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 {
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
@@ -129,7 +165,8 @@ func (s *Store) TaskRuns(ctx context.Context, taskID string, limit int) ([]TaskR
limit = 50
}
rows, err := s.pool.Query(ctx, `
SELECT id, task_id, trigger, status, started_at, finished_at, duration_ms, detail, error
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
@@ -141,9 +178,10 @@ func (s *Store) TaskRuns(ctx context.Context, taskID string, limit int) ([]TaskR
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 {
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)
@@ -338,3 +376,93 @@ func (s *Store) PruneIntegrationDeliveries(ctx context.Context, keep int) (int64
}
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()
}