postgres

package
v0.7.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 7, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package postgres is the persistence adapter.

The plan (section 22) defines the database as the operational source of truth. Only the plumbing lives here: the connection pool, the migrations and the health check. The domain queries go into the phases that use them.

Index

Constants

This section is empty.

Variables

View Source 
var ErrJaExiste = errors.New("a run with this idempotency key already exists")

ErrJaExiste signals an idempotency-key collision. Typed so the caller can tell "I already created this" from a real error — the difference between a benign scheduler retry and a database failure.

Functions

func Migrate 

func Migrate(ctx context.Context, url, direction string) error

Migrate applies the embedded migrations. It runs through the `brevis migrate` subcommand, never in `serve`: starting the application and migrating the schema have a different blast radius, and joining the two turns a casual restart into a DDL.

Types

type Bucket added in v0.7.0 

type Bucket struct {
	Start        time.Time
	Succeeded    int
	Failed       int
	Running      int
	Queued       int
	MeanDuration time.Duration
}

Bucket is one column of the run chart.

func (Bucket) Total added in v0.7.0 

func (b Bucket) Total() int

Total sums the whole bucket — the column's height.

type Indicators added in v0.7.0 

type Indicators struct {
	Total        int
	Succeeded    int
	Failed       int
	Running      int
	Pending      int
	MeanDuration time.Duration
}

Indicators is the Overview's header.

func (Indicators) Ratio added in v0.7.0 

func (i Indicators) Ratio(part int) float64

Ratio returns `parte` as a percentage of everything already finished.

The denominator excludes what is still running: counting an in-flight run as a "non-success" makes the rate plunge during a burst of work and climb back on its own afterwards, with nothing having changed.

type NodeState added in v0.7.0 

type NodeState struct {
	NodeID     string `json:"node_id"`
	Status     string `json:"status"`
	Attempt    int    `json:"attempt"`
	ExitCode   *int   `json:"exit_code,omitempty"`
	Err        string `json:"erro,omitempty"`
	DurationMs int64  `json:"duracao_ms"`

	// Etapas are the phases announced by an SDK step. Empty for a step that is
	// not an SDK one -- and that step's screen stays exactly as it was.
	Stages []Stage `json:"etapas,omitempty"`

	// SdkVersao is the version the step announced, empty when it is not an SDK step.
	SdkVersion string `json:"sdk_versao,omitempty"`
}

NodeState is a step's state, for the UI.

type Pool 

type Pool struct {
	*pgxpool.Pool
}

Pool wraps pgxpool. The type exists so the rest of the system depends on something of ours, and not on the driver directly.

func New 

func New(ctx context.Context, url string) (*Pool, error)

New opens the pool and checks the connection before returning. A pool that only fails on first use turns a configuration error into a request error.

func (*Pool) Check 

func (p *Pool) Check(ctx context.Context) error

Check is the health contract: a ping with a deadline. Without a timeout, a slow database would make readiness hang instead of failing.

type ProjectSummary added in v0.7.0 

type ProjectSummary struct {
	Slug      string
	Name      string
	Workflows int
	Runs      int
	CreatedAt time.Time
}

ProjectSummary counts what exists under a project.

type ReadRepo added in v0.7.0 

type ReadRepo struct {
	// contains filtered or unexported fields
}

ReadRepo serves the UI.

func NewReadRepo added in v0.7.0 

func NewReadRepo(p *Pool) *ReadRepo

func (*ReadRepo) CountByStatus added in v0.7.0 

func (r *ReadRepo) CountByStatus(ctx context.Context) (map[string]int, error)

CountByStatus feeds the dashboard's cards.

func (*ReadRepo) CountRuns added in v0.7.0 

func (r *ReadRepo) CountRuns(ctx context.Context, f RunFilter) (int, error)

CountRuns returns the total for the SAME filter, so the pagination knows how many pages there are.

func (*ReadRepo) InFlight added in v0.7.0 

func (r *ReadRepo) InFlight(ctx context.Context, limite int) ([]RunSummary, error)

InFlight lists what is running or waiting its turn, oldest first.

The order is ascending on purpose: whatever has been in the queue longest is what deserves attention, and sorting by most recent would hide exactly that.

func (*ReadRepo) Indicators added in v0.7.0 

func (r *ReadRepo) Indicators(ctx context.Context, window time.Duration) (Indicators, error)

Indicators aggregates the recent window for the four cards at the top.

One query, with FILTER, rather than four: those would be four scans of the same table over the same time predicate.

func (*ReadRepo) LatestRuns added in v0.7.0 

func (r *ReadRepo) LatestRuns(ctx context.Context, limite int) ([]RunSummary, error)

LatestRuns lists the most recent runs.

func (*ReadRepo) Projects added in v0.7.0 

func (r *ReadRepo) Projects(ctx context.Context) ([]ProjectSummary, error)

Projects lists the projects with their totals.

func (*ReadRepo) QueueDepth added in v0.7.0 

func (r *ReadRepo) QueueDepth(ctx context.Context) (pending, claimed int, err error)

QueueDepth shows the queue on the dashboard.

func (*ReadRepo) Runs added in v0.7.0 

func (r *ReadRepo) Runs(ctx context.Context, f RunFilter) ([]RunSummary, error)

Runs lists runs with filtering and pagination.

func (*ReadRepo) RunsPerHour added in v0.7.0 

func (r *ReadRepo) RunsPerHour(ctx context.Context, horas int) ([]Bucket, error)

RunsPerHour returns one column per hour, the empty ones INCLUDED.

The left `generate_series` is the point: without it an hour with no run simply would not appear, and the chart would compress time, giving the impression of continuous activity where there was a gap.

func (*ReadRepo) Schedules added in v0.7.0 

func (r *ReadRepo) Schedules(ctx context.Context) ([]ScheduleSummary, error)

Schedules returns every schedule, active or not. The DAG list shows the paused ones too — hiding them from the screen would hide the reason nothing runs.

func (*ReadRepo) WorkflowRuns added in v0.7.0 

func (r *ReadRepo) WorkflowRuns(ctx context.Context, slug string, limite int) ([]RunSummary, error)

WorkflowRuns lists the runs of a single workflow, for its own screen.

func (*ReadRepo) Workflows added in v0.7.0 

func (r *ReadRepo) Workflows(ctx context.Context) ([]WorkflowSummary, error)

Workflows lists the published workflows with their schedule and last state.

type RunFilter added in v0.7.0 

type RunFilter struct {
	State    string
	Workflow string
	De       *time.Time
	Ate      *time.Time
	Limite   int
	Offset   int
}

RunFilter is the run screen's query. Empty fields do not filter.

type RunRepo 

type RunRepo struct {
	// contains filtered or unexported fields
}

RunRepo persiste Runs e TaskRuns.

func NewRunRepo 

func NewRunRepo(p *Pool) *RunRepo

func (*RunRepo) CountByStatus added in v0.7.0 

func (r *RunRepo) CountByStatus(ctx context.Context) (map[dom.Status]int, error)

CountByStatus is what PHASE 2's acceptance criterion measures.

func (*RunRepo) CountByTrigger added in v0.7.0 

func (r *RunRepo) CountByTrigger(ctx context.Context) (map[string]int, error)

CountByTrigger shows where the runs came from — telling a backfill from a scheduled run is what section 12 asks for while investigating an incident.

func (*RunRepo) Create added in v0.7.0 

func (r *RunRepo) Create(ctx context.Context, run dom.Run) (dom.Run, error)

Criar inserts the Run in CREATED.

A collision on the idempotency_key unique becomes ErrJaExiste, not a generic error: it is section 29's case — the scheduler died after creating and tries again on the way back up.

func (*RunRepo) FailedStep added in v0.7.0 

func (r *RunRepo) FailedStep(ctx context.Context, runID uuid.UUID) (string, string, error)

FailedStep returns the node and the output of the last attempt that failed.

`ORDER BY iniciado_em DESC` and not `attempt DESC`: in a graph with several steps, the highest attempt may belong to a step that had already failed and been superseded — what matters is what failed LAST, which is where the run stopped.

Absence is not an error: a run that died before any step started (a missing image, a cancelled queue) has no task_run at all, and the alert goes out without this part rather than not going out.

func (*RunRepo) Get added in v0.7.0 

func (r *RunRepo) Get(ctx context.Context, id uuid.UUID) (dom.Run, error)

Buscar le um Run.

func (*RunRepo) IncrementAttempt added in v0.7.0 

func (r *RunRepo) IncrementAttempt(ctx context.Context, id uuid.UUID) (int, error)

IncrementAttempt bumps the counter when requeuing for a retry.

func (*RunRepo) IniciarTask 

func (r *RunRepo) IniciarTask(ctx context.Context, runID uuid.UUID, nodeID string, attempt int) error

IniciarTask records a step's start.

`ON CONFLICT DO UPDATE` on the (run, node, attempt) key: re-running the same step on the same attempt is idempotent, which matters when the dispatcher recovers an item from a dead worker and redoes it.

func (*RunRepo) LogsDaRun 

func (r *RunRepo) LogsDaRun(ctx context.Context, runID uuid.UUID) ([]StepLog, error)

LogsDaRun returns the output of every attempt of every step, in execution order.

EVERY attempt, not only the last: when a step passes on the second, what explains the first failure is precisely in the attempt the screen would discard.

func (*RunRepo) NodeStates added in v0.7.0 

func (r *RunRepo) NodeStates(ctx context.Context, runID uuid.UUID) (map[string]NodeState, error)

NodeStates returns each node's state on its LAST attempt.

`DISTINCT ON` rather than max(attempt) in a subselect: the most recent attempt is the one that matters on screen, and an old attempt that failed must not paint the node red after the retry succeeded.

func (*RunRepo) RecordError added in v0.7.0 

func (r *RunRepo) RecordError(ctx context.Context, id uuid.UUID, msg string) error

RecordError stores the cause of the failure.

func (*RunRepo) RecordStages added in v0.7.0 

func (r *RunRepo) RecordStages(ctx context.Context, runID uuid.UUID, nodeID string,
	attempt int, sdkVersion string, stages json.RawMessage) error

RecordStages records the advance of an SDK step's phases.

It overwrites the whole array rather than appending: the runner's collector already keeps ONE entry per phase, with its current state, and the screen wants four boxes rather than a diary.

func (*RunRepo) StepHasSucceeded added in v0.7.0 

func (r *RunRepo) StepHasSucceeded(ctx context.Context, workflowSlug, nodeID string, exceto uuid.UUID) (bool, error)

StepHasSucceeded answers whether this step, in this workflow, has ever finished well before -- in any earlier run.

It is what decides whether the current run is that step's FIRST, information that goes into the step's environment and that the SDK uses to create the destination table. The alternative would be for the SDK to infer it from "the table does not exist", and then somebody drops the table by mistake and the next run believes it is the first.

Per (workflow, step), not per workflow: a workflow with three fetchers writing to three tables would create only the first step's if the answer covered the whole workflow.

`exceto` is the current run, excluded so the attempt in progress does not count as an earlier success.

func (*RunRepo) TerminarTask 

func (r *RunRepo) TerminarTask(ctx context.Context, runID uuid.UUID, nodeID string,
	attempt int, status dom.Status, exit *int, failure string, log string) error

TerminarTask records the outcome.

func (*RunRepo) Transicionar 

func (r *RunRepo) Transicionar(ctx context.Context, id uuid.UUID, para dom.Status) error

Transicionar applies the state change, validating BEFORE writing.

The validation happens against the state read inside the transaction, with FOR UPDATE: reading outside it would let two dispatchers both read "queued" and both write "running".

type RunSummary added in v0.7.0 

type RunSummary struct {
	ID           string
	WorkflowSlug string
	Status       string
	TriggerType  string
	Attempt      int
	LogicalDate  *time.Time
	CreatedAt    time.Time
	StartedAt    *time.Time
	Duration     *time.Duration
	Err          string
}

RunSummary is one row of the run list.

type ScheduleRepo 

type ScheduleRepo struct {
	// contains filtered or unexported fields
}

ScheduleRepo le e atualiza agendas.

func NewScheduleRepo 

func NewScheduleRepo(p *Pool) *ScheduleRepo

func (*ScheduleRepo) Active added in v0.7.0 

func (r *ScheduleRepo) Active(ctx context.Context) ([]sch.Schedule, error)

Ativas lists the schedules the scheduler has to evaluate.

func (*ScheduleRepo) AdvanceSlot added in v0.7.0 

func (r *ScheduleRepo) AdvanceSlot(ctx context.Context, slug string, slot time.Time) error

func (*ScheduleRepo) SetActive added in v0.7.0 

func (r *ScheduleRepo) SetActive(ctx context.Context, slug string, active1 bool) (bool, error)

AdvanceSlot records how far the schedule has been materialized.

The condition `ultimo_slot IS NULL OR ultimo_slot < $2` makes the operation idempotent and safe under concurrency: two schedulers evaluating the same schedule never make the marker go backwards. SetActive pauses or resumes a schedule and returns the resulting state.

It returns rather than only writing because the UI toggles without knowing the current value: without the return, the screen would need a second query and would be open to a race between two operators clicking at the same time.

Pausing does NOT cancel what is already queued: materialized runs are accepted work, and discarding them on a pause would surprise somebody who only wanted to stop creating new ones.

func (*ScheduleRepo) Toggle added in v0.7.0 

func (r *ScheduleRepo) Toggle(ctx context.Context, slug string) (bool, error)

Alternar flips the current state in a single round trip to the database.

type ScheduleSummary added in v0.7.0 

type ScheduleSummary struct {
	WorkflowSlug string
	Cron         string
	Timezone     string
	Active       bool
}

ScheduleSummary is the minimum needed to compute the next trigger.

type Stage added in v0.7.0 

type Stage struct {
	Index   int            `json:"indice"`
	Name    string         `json:"nome"`
	State   string         `json:"estado"`
	Ms      *int64         `json:"ms,omitempty"`
	At      string         `json:"em"`
	Numbers map[string]any `json:"numeros,omitempty"`
}

Etapa is one phase of an SDK step, for the screen.

type StepLog added in v0.7.0 

type StepLog struct {
	NodeID     string
	Attempt    int
	Status     string
	ExitCode   *int
	Err        string
	Log        string
	DurationMs int64
}

StepLog is one attempt's output, for the run's screen.

type WorkflowRepo 

type WorkflowRepo struct {
	// contains filtered or unexported fields
}

WorkflowRepo persiste a definicao publicada de um workflow.

func NewWorkflowRepo 

func NewWorkflowRepo(p *Pool) *WorkflowRepo

func (*WorkflowRepo) Definition added in v0.7.0 

func (r *WorkflowRepo) Definition(ctx context.Context, slug string) (wf.Workflow, error)

Definition le o grafo publicado.

func (*WorkflowRepo) Podar 

func (r *WorkflowRepo) Podar(ctx context.Context, projeto uuid.UUID, manter []string) ([]string, error)

Podar removes from the project the workflows that are NOT in the list, along with their agendas. Devolve os slugs removidos.

It exists because publishing only ever added: taking a file out of the folder took nothing out of the database, and the scheduler went on materializing runs for a workflow nobody could see any more. With 15-minute schedules, that is invisible work running forever.

The history (`runs`) is NOT deleted: it references the slug as text, not through a foreign key, precisely so it survives the definition's removal. Deleting the runs along with it would be deleting the evidence of what happened.

func (*WorkflowRepo) Publicar 

func (r *WorkflowRepo) Publicar(ctx context.Context, w wf.Workflow, projeto uuid.UUID) error

Publicar writes the workflow and its schedule in one transaction.

Both things together, and not in separate calls: publishing the graph without the schedule would leave a workflow that never fires, and the schedule without the graph would make the scheduler create runs for something that does not exist.

type WorkflowSummary added in v0.7.0 

type WorkflowSummary struct {
	Slug        string
	Name        string
	Project     string
	Cron        string
	Timezone    string
	Catchup     bool
	Active      bool
	HasSchedule bool
	LastSlot    *time.Time
	LastStatus  string
	TotalRuns   int

	// From the last run — the list's "Latest Run" column.
	LastRunID *string
	LastRunAt *time.Time

	Tags []string

	// ProximaRun does not come from the database: it is computed from the cron,
	// in the consumer. Storing it would demand a recompute on every schedule
	// change and living with a stale value in between.
	NextRun *time.Time
}

WorkflowSummary joins the workflow, its schedule and the last run's state.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.