postgres

package
v0.13.0 Latest Latest
Warning

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

 Go to latest Published: Sep 11, 2026 License: MIT Imports: 18 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

View Source 
const DefaultBatch = 5000

DefaultBatch is the batch size when none is given. Large enough that a backlog clears in reasonable time, small enough that no single statement holds a lock long.

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 Day added in v0.11.2 

type Day struct {
	Date      time.Time
	Total     int
	Succeeded int
	Failed    int
	Open      int // running, retrying or queued
}

Day is one square of the calendar heatmap.

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 LoadDay added in v0.13.0 

type LoadDay struct {
	Date      time.Time
	Runs      int
	Rows      int64
	Records   int64
	Ignored   int64
	BytesIn   int64
	BytesOut  int64
	ExtractMs int64 // mean over the day
	LoadMs    int64 // mean over the day
}

LoadDay is one day of what a workflow's pipelines loaded.

Sums for the volumes and a MEAN for the durations, because the two answer different questions: "how much did this workflow move that day" is a total, and "is a run getting slower" is per run. A summed duration would climb whenever the schedule got denser and say nothing about the pipeline.

func (LoadDay) BytesPerRow added in v0.13.0 

func (d LoadDay) BytesPerRow() int64

BytesPerRow is the number the note actually asked for: a load whose size grows while its row count holds flat is a schema that gained a column.

Zero rows returns zero rather than dividing: a day with no rows has no answer to this, and an infinity on a chart is worse than a gap.

func (LoadDay) RowsPerSecond added in v0.13.0 

func (d LoadDay) RowsPerSecond() int64

RowsPerSecond over the load phase. Zero when nothing was loaded or when the load was too fast to have measured a millisecond.

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

	// Published is what this step told the steps below it. Absent when it
	// published nothing, which is most steps.
	Published json.RawMessage `json:"saida,omitempty"`

	// Instances and Done are a MAPPED step's counts: how many instances there
	// are and how many finished well. Both zero for an unmapped step, which is
	// most of them, and the payload then carries neither field.
	Instances int `json:"instancias,omitempty"`
	Done      int `json:"instancias_ok,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 PruneReport added in v0.13.0 

type PruneReport struct {
	Trimmed int64
	Purged  int64
	DryRun  bool
}

PruneReport is what a prune did, or would do.

func (PruneReport) String added in v0.13.0 

func (r PruneReport) String() string

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) IndicatorsFor added in v0.11.2 

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

IndicatorsFor is Indicators scoped to ONE workflow, and the same query.

An empty slug means every workflow, which is what the dashboard asks for. Writing the aggregation twice -- once global, once per workflow -- would be two places for "what counts as finished", and the answer has already moved once: `Ratio` excludes what is still running, and a second copy would not know that.

func (*ReadRepo) Insights added in v0.8.0 

func (r *ReadRepo) Insights(ctx context.Context, from, to time.Time, env string) (notify.Report, error)

Insights aggregates a window for the periodic report.

It is a QUERY and not a rollup table, which is a departure from the plan worth stating. A rollup written as each run finishes would make this read cheap and would also add a write to the hottest path in the system, plus a second source of truth for numbers that already exist and can therefore drift from them. This runs once a week, off-peak, over an indexed window.

When it becomes slow, the rollup is the answer and the measurement is what says so. The duration is logged by the caller for exactly that reason.

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) LoadTrend added in v0.13.0 

func (r *ReadRepo) LoadTrend(ctx context.Context, workflow string, days int) ([]LoadDay, error)

LoadTrend returns one row per day a workflow loaded anything, oldest first.

Days with NO load are absent rather than zero, and that is deliberate in a way the calendar heatmap is not. The heatmap answers "did it run", so a blank day is information. This answers "is it getting slower", and a zero would not be a slow day -- it would be a day that is not in the series at all, dragging every average through a floor that never happened. The screen draws the gap.

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) RunsPerDay added in v0.11.2 

func (r *ReadRepo) RunsPerDay(ctx context.Context, workflow string, days int) ([]Day, error)

RunsPerDay returns one row per DAY that had a run, for one workflow.

Unlike RunsPerHour it does NOT fill the gaps with a generate_series, and the difference is the shape of the two charts. A bar chart with a missing hour compresses time and invents continuity; a calendar has a square for every day whether or not anything happened, so the empty ones are drawn by the renderer from the date range and never travel over the wire. Sending 365 mostly-zero rows to draw nothing would be the same picture at ten times the cost.

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 Retention added in v0.13.0 

type Retention struct {
	// TrimAfter empties log, etapas and saida on runs older than this.
	TrimAfter time.Duration

	// PurgeAfter deletes the run outright, and with it its task_runs, its queue
	// items and its alerts -- but NOT its row in load_metrics.
	//
	// Zero means never, which is a real answer: a run's status and duration are
	// small, and an installation may want them forever.
	PurgeAfter time.Duration

	// Batch is how many runs one statement touches. The whole point is that a
	// prune of a neglected database must not take a lock for a minute: a
	// hundred short transactions beat one long one, and an interrupted prune
	// leaves a consistent database with less work left to do.
	Batch int
}

Retention has TWO levels, because the cost and the meaning are not in the same place.

Measured on a year of hourly runs across forty workflows -- 350,000 runs: 

task_runs     339 MB   log 119, etapas 113, saida 20, indexes 35
runs          141 MB   auto_params 38, definicao 30, indexes 31
load_metrics   81 MB

Three quarters of the biggest table is BULK: the text a step printed and the JSON of its phases. Emptying those columns takes task_runs from 339 MB to 92 MB and deletes NOTHING -- every run still opens, with its steps, statuses, timings, exit codes and errors. What is lost is the log of a run from March, which is read by nobody, and its phase boxes.

So the first level TRIMS and the second DELETES, and the gap between them is where almost all the value is. A policy with only the second would be throwing away runs to reclaim space that was never in the run.

load_metrics is untouched by both. It has no foreign key for exactly this reason: the summary outlives the detail, which is the only thing that makes a year-long trend possible on a ninety-day log.

func (Retention) Validate added in v0.13.0 

func (p Retention) Validate() error

Validate refuses a policy that would destroy more than it says.

The checks are here rather than in the flag parsing because the scheduler could call this too one day, and a guard that lives in one caller is a guard the second caller does not have.

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) AlreadySucceeded added in v0.8.0 

func (r *RunRepo) AlreadySucceeded(ctx context.Context, runID uuid.UUID) (map[dom.StepKey]bool, error)

AlreadySucceeded returns the steps of THIS run that have already finished well, in any earlier attempt of it.

It is what makes a run's retry re-run only what failed, the way a cleared DAG run does in Airflow. Before it, a retry re-ran the whole graph: a workflow with an expensive step beside a flaky one paid for the expensive one on every attempt, and any step that was not idempotent did its work twice.

Per (run, node) and NOT per (workflow, node) -- that is StepHasSucceeded above, which answers a different question: whether this step has ever succeeded in an EARLIER run, which is what tells the SDK it is not the first. Confusing the two would make a step skip itself forever after its first good day.

func (*RunRepo) Attempt added in v0.8.0 

func (r *RunRepo) Attempt(ctx context.Context, id uuid.UUID, budget int,
	raise func(attempt int, gaveUp bool) []alerts.Pending,
) (attempt int, gaveUp bool, err error)

Attempt records the attempt that just finished and, when it was the LAST one, writes the alert it justifies -- in a single transaction.

That transaction is the whole design of the alerts feature. Before it, the dispatcher incremented the counter and then called Slack: a process that died between the two left a run out of attempts with nobody told, and no record that anybody should have been. Now either the run is recorded as having spent its last attempt and the alert exists, or neither happened.

`raise` is a builder rather than a value because building the message costs reads -- the run's details, the failing step's log -- and it may return nothing, which is what an installation with no channel configured does. It receives `gaveUp` because a step declaring `on_error.when: attempt` is announced on every failed attempt while the run-level alert waits for the last one.

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 investigating an incident needs.

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)

func (*RunRepo) FailedSteps added in v0.8.0 

func (r *RunRepo) FailedSteps(ctx context.Context, runID uuid.UUID) (map[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. FailedSteps returns EVERY step of this run that ended failed, newest first, with the end of its log.

FailedStep above answers "which step should the run's alert name", and one is the right answer there: an alert has to fit in a phone notification. This one exists for `on_error`, where each declaring step gets its own message -- and a run with two parallel branches can legitimately have two of them fail.

DISTINCT ON keeps the latest attempt per node. Without it a step that failed three times would produce three alerts saying the same thing.

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

func (r *RunRepo) IniciarTask(ctx context.Context, runID uuid.UUID, step dom.StepKey, 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) MarkSkipped added in v0.8.0 

func (r *RunRepo) MarkSkipped(ctx context.Context, runID uuid.UUID, step dom.StepKey,
	attempt int, reason string) error

MarkSkipped records a step whose trigger rule was not satisfied.

Not IniciarTask followed by TerminarTask, and the difference is the point: `iniciado_em` stays NULL. A skipped step never started, and stamping a start time would make it look like something that ran in zero seconds -- which is also what a step killed instantly looks like.

`erro` carries WHY. A skipped step with no explanation sends whoever is looking at the graph to trace edges by hand, and that column is where the screen already shows a step's last word.

ON CONFLICT because a run that retries re-evaluates every rule: attempt 0 of the second try overwrites attempt 0 of the first, and a step skipped once may well run the next time.

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) Prune added in v0.13.0 

func (r *RunRepo) Prune(ctx context.Context, p Retention, dry bool) (PruneReport, error)

Prune applies the retention policy.

`dry` reports what it would do and changes nothing, which is what makes the first run of this on a real installation something an operator can look at before believing.

func (*RunRepo) PublishedContext added in v0.8.0 

func (r *RunRepo) PublishedContext(ctx context.Context, runID uuid.UUID) (map[string]json.RawMessage, error)

PublishedContext is what the steps of a run have published so far.

DISTINCT ON keeps the latest attempt per step, which is the rule a retry needs: the previous attempt's output described work that did not finish, and the step below must not read it.

func (*RunRepo) RecordAuto added in v0.9.0 

func (r *RunRepo) RecordAuto(ctx context.Context, id uuid.UUID,
	window func(slot time.Time) (start, end time.Time),
) (dom.AutoParams, error)

RecordAuto writes the run's automatic params, and reads back what it needs to compute them in the SAME statement.

One round trip, and one transaction, because the parts are not independent: `started_at` is set by the transition that just happened, and `previous_error` is about the run before this one. Reading them separately would let a concurrent retry of that previous run change the answer between the two reads, and the run would carry a `previous_error` nobody can reproduce.

The previous run is the one with the closest EARLIER logical_date, falling back to creation order for a workflow with no schedule. A manual run in the middle of a nightly is not "the previous run" of that nightly.

func (*RunRepo) RecordContext added in v0.8.0 

func (r *RunRepo) RecordContext(ctx context.Context, runID uuid.UUID, step dom.StepKey,
	attempt int, published json.RawMessage) error

RecordContext stores what a step published, for the steps below it.

Written the moment the step publishes rather than when it ends: a run that resumes reads this back, and the process that would write it later is exactly the one that may not survive.

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) RecordLoad added in v0.13.0 

func (r *RunRepo) RecordLoad(ctx context.Context, runID uuid.UUID, step dom.StepKey,
	workflow string, n exec.LoadNumbers) error

RecordLoad keeps what one attempt of an SDK pipeline measured.

UPSERT rather than INSERT, and that is a retry rather than a race: attempt 2 of a step overwrites attempt 1's row. The trend asks "how much did this pipeline load that day", and a step that failed after loading 40,000 rows and then succeeded loading 48,000 loaded 48,000 -- counting both would invent 88,000 rows that never existed. The attempt is not in the key for exactly that reason; `task_runs` keeps the per-attempt history for whoever needs it.

func (*RunRepo) RecordStages added in v0.7.0 

func (r *RunRepo) RecordStages(ctx context.Context, runID uuid.UUID, step dom.StepKey,
	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, step dom.StepKey,
	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.