Help us shape the future of Go! Take the 2026 Go Developer Survey

postgres

package
v0.42.0 Latest Latest
Warning

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

 Go to latest Published: Nov 3, 2026 License: MIT Imports: 27 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 (
	// DefaultEventPage is a page when none is asked for.
	DefaultEventPage = 50
	// MaxEventPage is the most one page reads.
	MaxEventPage = 200
)
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.

View Source 
const HistoryShown = 20

HistoryShown bounds the versions a page reads per writer: enough to see a configuration being tuned, not every change since the gateway existed.

Variables

View Source 
var ErrBadCursor = errors.New("not a cursor of the events feed")

ErrBadCursor is a cursor the feed did not write.

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.

View Source 
var ErrRunNotFound = errors.New("run not found")

ErrRunNotFound is a cancel asked of a run that does not exist.

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 APIManifest 

type APIManifest struct {
	Connection, Dialect string
	Cache               *catalog.EndpointsCache
}

APIManifest is an API's own settings, kept beside its endpoints by the same publish (screen-endpoints). Cache is nil for an API with none.

type Bucket 

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 

func (b Bucket) Total() int

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

type CancelOutcome 

type CancelOutcome int

CancelOutcome is what asking to cancel a run did. 

const (
	// CancelDone: the run was still waiting, and it is canceled now.
	CancelDone CancelOutcome = iota + 1
	// CancelRequested: a dispatcher holds the run, and stops it when it reads
	// the request.
	CancelRequested
	// CancelAlreadyEnded: nothing changed; Status says how it ended.
	CancelAlreadyEnded
)

type CancelResult 

type CancelResult struct {
	Outcome CancelOutcome
	Status  dom.Status
}

CancelResult is the outcome, and the run's status after it.

type CatalogEntry 

type CatalogEntry struct {
	// Target is the destination's identity -- or, for a legacy row, the label
	// its load carried before targets existed.
	Target string

	// Kind is the target's scheme (`bigquery`, `postgres`, `s3`…); empty for a
	// legacy label, which has none.
	Kind string

	// Legacy marks a label recovered by migration 00013, never a target.
	Legacy bool

	Writers []CatalogWriter

	// Recent is the rows of the destination's last loads, oldest first, across
	// every writer -- what the sparkline draws. Nil where the step did not say.
	Recent []*int64
}

CatalogEntry is one destination on /data: what wrote it and what it received.

type CatalogWriter 

type CatalogWriter struct {
	Workflow, Node string

	LastLoaded time.Time
	LastRows   *int64

	// The writer's schedule. HasSchedule is false for a workflow that only runs
	// by hand.
	HasSchedule    bool
	Cron, Timezone string
	Active         bool

	// Lag is the p90 of how long after its slot this writer's data landed, over
	// its last 20 landings. Zero when no slot is left to measure -- the runs
	// were purged -- which leaves the 10-minute floor as the grace.
	Lag time.Duration

	// RunInFlight is the id of a run of this workflow that is queued, running
	// or retrying, if there is one.
	RunInFlight *string

	// Config is how the step's pipeline was configured, read for the
	// destination's own page and never for the list. Nil for a step that never
	// described itself -- Python, dbt, a shell -- and for a gateway writer,
	// whose configuration is on Gateway.
	Config *WriterConfig

	// Gateway is set when the writer is a published gateway stream rather than
	// a workflow step; Workflow and Node are then empty, and nothing above is
	// measured -- a gateway's traffic is on its /metrics.
	Gateway *GatewayWriter
}

CatalogWriter is one workflow step landing on a destination, with what freshness needs to judge it.

type ConfigVersion 

type ConfigVersion struct {
	Hash   string
	At     time.Time
	Fields []catalog.ManifestField

	// RunID is the run that first wrote with it -- a step's versions only;
	// a gateway's come from a publish, which is no run.
	RunID string
}

ConfigVersion is a configuration as it was from At on.

type Day 

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 EndpointRepo 

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

EndpointRepo stores what brevis-sql-api files were published with.

func NewEndpointRepo 

func NewEndpointRepo(p *Pool) *EndpointRepo

func (*EndpointRepo) Get 

func (r *EndpointRepo) Get(ctx context.Context, api, name string) (PublishedEndpoint, bool, error)

Get is one endpoint, and whether it is published.

func (*EndpointRepo) List 

func (r *EndpointRepo) List(ctx context.Context) ([]PublishedEndpoint, error)

List is every published endpoint, by API and path.

func (*EndpointRepo) Publish 

func (r *EndpointRepo) Publish(ctx context.Context, m catalog.Endpoints, at time.Time, actor string) error

Publish replaces one API's endpoints with the manifest's, in one transaction, under a lock of that API's own -- the gateway's lesson: two overlapping rollouts interleaving delete-then-insert put back a row the other had just put back, and failed on its key. Other APIs are untouched.

A publish that added, changed or removed an endpoint is one event in the feed, in the same transaction, with actor as who published it; one that changed nothing records nothing.

The manifest must already have been through catalog.ParseEndpoints.

func (*EndpointRepo) Reading 

func (r *EndpointRepo) Reading(ctx context.Context, target string) ([]PublishedEndpoint, error)

Reading is the endpoints whose SQL reads a target -- a model page's consumers beside its writers.

func (*EndpointRepo) Unpublish 

func (r *EndpointRepo) Unpublish(ctx context.Context, api, actor string) (int64, error)

Unpublish removes a decommissioned API, reporting how many endpoints went. Taking down an API that had endpoints is in the events feed.

type Event 

type Event struct {
	At      time.Time
	Kind    events.Kind
	Subject string
	// Actor is who did it: the console's user, "cli", or "" for a console
	// with no login and for a run.
	Actor string
	// RunID and Trigger are a run's; nil and "" on the rest.
	RunID   *uuid.UUID
	Trigger string
	Detail  map[string]any
	// contains filtered or unexported fields
}

Event is one thing that happened: an action from the events table, or a run starting or ending, read from runs.

type EventQuery 

type EventQuery struct {
	Before string
	Source events.Source
	Limit  int
}

EventQuery is a page of the feed. Before is the cursor a previous page returned, "" for the newest; Source narrows it to one source, "" for all.

type EventRepo 

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

EventRepo reads the console's events feed.

func NewEventRepo 

func NewEventRepo(p *Pool) *EventRepo

func (*EventRepo) RecentEvents 

func (r *EventRepo) RecentEvents(ctx context.Context, q EventQuery) ([]Event, string, error)

RecentEvents is a page of what happened, newest first, and the cursor of the next page -- "" when this was the last.

type FailingWorkflow 

type FailingWorkflow struct {
	Slug     string
	RunID    uuid.UUID
	FailedAt time.Time
	Streak   int
}

FailingWorkflow is a workflow whose last finished run failed.

type GatewayRepo 

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

GatewayRepo stores what gateways were published with.

func NewGatewayRepo 

func NewGatewayRepo(p *Pool) *GatewayRepo

func (*GatewayRepo) Publish 

func (r *GatewayRepo) Publish(ctx context.Context, m catalog.Manifest, at time.Time, actor string) error

Publish replaces one gateway's destinations and streams with the manifest's, in one transaction: a stream removed from the config is gone after this, and a failure leaves the previous publish whole rather than half of either. Other gateways' rows are not touched.

Each stream's configuration is stored once by its hash, and its history gains a line only when that hash differs from the last one recorded.

A stream that is new, or whose configuration changed, is in the events feed, in the same transaction, with actor as who published it -- the rule the history follows, so a republish that changes nothing records nothing.

The manifest must already have been through catalog.ParseManifest.

func (*GatewayRepo) Unpublish 

func (r *GatewayRepo) Unpublish(ctx context.Context, gateway, actor string) (int64, error)

Unpublish removes a decommissioned gateway from /data, reporting how many destinations went. Its configuration history stays: see migration 00018. Taking down a gateway that had streams is in the events feed.

type GatewayWriter 

type GatewayWriter struct {
	Name, Stream, Role, Kind, Note string
	Routes                         bool
	PublishedAt                    time.Time

	// Config is how the stream is configured, read for the destination's own
	// page and never for the list. Nil when the stream was published by a
	// version-1 manifest, which described nothing but where it lands.
	Config *WriterConfig
}

GatewayWriter is a gateway stream as its published manifest describes it.

type Grain 

type Grain int

Grain is a chart's bucket: an hour or a day. A closed type, mapped to the SQL literal here -- a value from a URL never reaches the query. 

const (
	ByHour Grain = iota
	ByDay
)

type Holder 

type Holder struct {
	RunID    uuid.UUID
	Workflow string
	Since    time.Time // when it was claimed
}

Holder is a run occupying one of the slots.

type Indicators 

type Indicators struct {
	Total        int
	Succeeded    int
	Failed       int
	Running      int
	Pending      int
	MeanDuration time.Duration
	// Retried are the runs on their second attempt or later: a failure that
	// healed, which does not show as a failure (screen-overview-v2).
	Retried int
}

Indicators is the Overview's header.

func (Indicators) Ratio 

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 Landed 

type Landed struct {
	Target   string
	Workflow string
	LoadedAt time.Time
	Recorded time.Time
}

Landed is one landing, as the trigger loop needs it.

Four fields and not the row: the loop matches a target, cuts a window from an instant, and names the writer in the refusal when a workflow would trigger itself. Rows and bytes are the catalog's business.

type LoadDay 

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 

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 

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 

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, opts ...PoolOptions) (*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 PoolOptions 

type PoolOptions struct {
	// MaxConns caps the pool. Zero keeps pgx's default, max(4, NumCPU).
	MaxConns int
	// StatementTimeout makes the server abort any statement on this pool that
	// runs longer. Zero leaves the URL's own setting (or none) untouched.
	StatementTimeout time.Duration
}

PoolOptions are the limits an installation can put on the pool. The zero value is the previous behavior: the driver's default size and no timeout.

type ProjectSummary 

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

ProjectSummary counts what exists under a project.

type PruneReport 

type PruneReport struct {
	Trimmed int64
	Purged  int64
	// Events is how many of the events feed's rows went with the runs.
	Events int64
	DryRun bool
}

PruneReport is what a prune did, or would do.

func (PruneReport) String 

func (r PruneReport) String() string

type PublishedEndpoint 

type PublishedEndpoint struct {
	API         string
	Name        string
	Path        string
	Tenant      string
	Reads       []string
	Endpoint    catalog.Endpoint
	PublishedAt time.Time

	// Manifest is the endpoint's API as its file declares it: the
	// connection, the dialect and the cache. Zero for an API published
	// before the manifest was kept, until it is published again.
	Manifest APIManifest
}

PublishedEndpoint is one row: an endpoint as its API last published it.

type ReadRepo 

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

ReadRepo serves the UI.

func NewReadRepo 

func NewReadRepo(p *Pool) *ReadRepo

func (*ReadRepo) Catalog 

func (r *ReadRepo) Catalog(ctx context.Context) ([]CatalogEntry, error)

Catalog lists every destination the ecosystem has landed on.

Two round trips and no more, whatever the number of destinations:

  1. The latest landing of each writer, with its schedule, its lag and any run in flight. DISTINCT ON is ordered as landings_writer_idx is, so the latest per writer is read off the index. The lag reads the writer's last 20 landings from that same index and each one's slot from runs by primary key: `runs` has no index by workflow, and a p90 of run durations per workflow would scan it -- which is why the grace is measured on the landing (decided with the spec's owner, 2026-10-07).
  2. The last 30 loads of each destination, through landings_target_idx with a LIMIT per target.

Status is NOT computed here. This knows rows and times; what "late" means is the domain's (catalog.Freshness), and the page applies it.

func (*ReadRepo) CatalogTarget 

func (r *ReadRepo) CatalogTarget(ctx context.Context, target string) (*TargetDetail, error)

CatalogTarget reads one destination, or nil when nothing ever landed there. A legacy label is looked up as stored, so its page opens like any other.

func (*ReadRepo) Connects 

func (r *ReadRepo) Connects(ctx context.Context, since time.Time) (map[string][]string, error)

Connects is, per workflow, the scheme of every target its runs landed since `since` -- `bigquery://acme/bronze/orders` is `bigquery` -- once each, in alphabetical order: what the Workflows list draws as what a workflow talks to. A target with no scheme, a legacy table name, says nothing about where it lives and is left out. The schemes are as written; naming their logos is the screen's.

func (*ReadRepo) CountByStatus 

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

CountByStatus feeds the dashboard's cards.

func (*ReadRepo) CountRuns 

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

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

CountRunsByState counts the filter's runs per state, IGNORING its own state: the Runs list's state filter says, beside each choice, what its link would show with the rest of the filter as it is. One GROUP BY over the same predicate the list uses.

func (*ReadRepo) CountRunsInFlight 

func (r *ReadRepo) CountRunsInFlight(ctx context.Context) (int, error)

CountRunsInFlight is how many runs are queued, running or retrying: the sidebar badge's number, drawn on every page. It names the states so the partial index on runs(status) serves it; counting every state (as CountRunsByState does) scans the whole table, finished runs included.

func (*ReadRepo) InFlight 

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 

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 

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 

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

func (r *ReadRepo) LastStepRuns(ctx context.Context, slug string) (map[string]StepRun, error)

LastStepRuns is, for each step of a workflow, its last attempt in the newest run that ran it: a step the last run skipped keeps what it did before. The workflow page's graph draws it on each card, and a mapped step's instances are one row each -- the newest is the one shown.

func (*ReadRepo) LatestRuns 

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

LatestRuns lists the most recent runs.

func (*ReadRepo) LoadTrend 

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 

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

Projects lists the projects with their totals.

func (*ReadRepo) QueueDepth 

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

QueueDepth shows the queue on the dashboard.

func (*ReadRepo) RecentStates 

func (r *ReadRepo) RecentStates(ctx context.Context, slugs []string, n int) (map[string][]RecentRun, error)

RecentStates is the last n runs of each workflow asked, oldest to newest so a strip reads left to right like time. ONE query for a page rather than one per row; a workflow that never ran is absent from the map.

func (*ReadRepo) Runs 

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

Runs lists runs with filtering and pagination.

func (*ReadRepo) RunsPer 

func (r *ReadRepo) RunsPer(ctx context.Context, g Grain, n int) ([]Bucket, error)

RunsPer returns one column per hour or per day, 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) RunsPerDay 

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

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

func (r *ReadRepo) StepConfigs(ctx context.Context) (map[StepKey]*WriterConfig, error)

StepConfigs is every step's current configuration on each destination it writes, in one query, for the loads list's origin column (screen-loads) -- the catalog read leaves them out, and one CatalogTarget per row would be a query per load. The newest line of step_config_history is the current one, as on the destination's page; no history here. A step that never described itself is absent.

func (*ReadRepo) StepConnects 

func (r *ReadRepo) StepConnects(ctx context.Context, slug string, since time.Time) (map[string][]string, error)

StepConnects is Connects by step: the scheme of every target each step of a workflow landed since `since`, once each, in order.

func (*ReadRepo) StreamConfigs 

func (r *ReadRepo) StreamConfigs(ctx context.Context) (map[StreamKey]*WriterConfig, error)

StreamConfigs is every published stream's current configuration and notes, in one query, for the streams list -- the catalog read leaves them out, and one CatalogTarget per card would be a query per stream. No history: that is the stream's own page. A version-1 stream, which described nothing, is absent.

func (*ReadRepo) WorkflowHistory 

func (r *ReadRepo) WorkflowHistory(ctx context.Context, slug string, day *time.Time, limit int) ([]RunSummary, error)

WorkflowHistory is a workflow's runs, newest first, with the values each was given: the workflow page's Runs tab. With a day, only that day's -- a UTC day by when the run was created, the same day the calendar counts it in.

func (*ReadRepo) Workflows 

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

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

type RecentRun 

type RecentRun struct {
	ID     string
	Status string
	At     time.Time
}

RecentRun is one square of a workflow's recent-runs strip.

type Retention 

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 rows in load_metrics and landings.
	//
	// 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 and landings are untouched by both. Neither has a 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 -- and the only thing that keeps a table on /data after the runs that loaded it are gone.

func (Retention) Validate 

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 

type RunFilter struct {
	State string
	// States is several at once -- "in flight" is three -- and is ORed with
	// nothing: State and States are not meant together.
	States []string
	// Workflow is a piece of the slug, matched without case. The whole slug
	// other screens link with is a piece of itself, so their links still land.
	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) AdvanceLandingCursor 

func (r *RunRepo) AdvanceLandingCursor(ctx context.Context, at time.Time) error

AdvanceLandingCursor moves it, and ONLY FORWARD.

A cycle that read less far than the last one must not rewind it. Rewinding would replay a window already seen, and the only reason that would be harmless is the idempotency key -- relying on that for CORRECTNESS rather than for safety is how a real bug becomes invisible.

func (*RunRepo) AlreadySucceeded 

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

func (*RunRepo) Attempt 

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

func (r *RunRepo) CancelRequested(ctx context.Context, ids []uuid.UUID) ([]uuid.UUID, error)

CancelRequested returns which of these runs someone asked to cancel. The dispatcher asks it about the runs it holds, every CancelPoll: one query by primary key, however many there are.

func (*RunRepo) CountByStatus 

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

CountByStatus is what PHASE 2's acceptance criterion measures.

func (*RunRepo) CountByTrigger 

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 

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 

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

func (*RunRepo) FailedSteps 

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

func (r *RunRepo) FailingWorkflows(ctx context.Context) ([]FailingWorkflow, error)

FailingWorkflows lists the workflows failing now, worst first: the longest streak, then the most recent failure. One query for the whole list.

func (*RunRepo) FailureStreak 

func (r *RunRepo) FailureStreak(ctx context.Context, slug string) (int, error)

FailureStreak is how many times in a row the workflow failed, counted back from its newest finished run to its last success; 0 when the last one passed or it never ran.

func (*RunRepo) Get 

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

Buscar le um Run.

func (*RunRepo) Handovers 

func (r *RunRepo) Handovers(ctx context.Context, runID uuid.UUID) (map[dom.StepKey]execution.Handover, 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. Handovers returns what a stopped scheduler left running in this run (#125): the rows with a handover beside them. No status filter, and not by omission: a row its reattach or a later attempt ended has no handover -- TerminarTask and IniciarTask clear it -- so a handover IS a row still running, and asking twice would be a second rule to keep in agreement.

The row's own attempt is the one used: it is what the next scheduler continues, and the one the handover was recorded against.

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

func (r *RunRepo) LandingCursor(ctx context.Context) (time.Time, bool, error)

LandingCursor is how far the landings have been read. `ok` is false when nothing has planted one yet, which is not an error: it is the first cycle.

func (*RunRepo) LandingsSince 

func (r *RunRepo) LandingsSince(ctx context.Context, cursor time.Time) ([]Landed, error)

LandingsSince reads everything the engine recorded after `cursor`.

BY recorded_at AND NEVER BY loaded_at, which is why 00015 exists: `loaded_at` is the step's own clock and a landing dated before the cursor is an ordinary thing, not a broken one.

The caller passes a cursor already moved BACK by an overlap. A row whose transaction began before the cursor and committed after it is behind a cursor already advanced, and no column fixes that -- the overlap plus the run's unique idempotency key does, by making a second sighting create nothing.

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

func (r *RunRepo) MarkDraining(ctx context.Context, id uuid.UUID) (time.Duration, error)

MarkDraining records that the run went back to the queue because every agent of a host was draining (#146), and returns how long ago that FIRST happened.

The first instant is kept, not the latest: the dispatcher sends a run back only while this is under its ceiling, and a run refused every thirty seconds would otherwise never reach it. Measured by the database's clock on both ends, so no process's clock is compared with another's. Spending an attempt or succeeding clears it.

func (*RunRepo) MarkSkipped 

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 

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 

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 

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 

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 

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 

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

RecordError stores the cause of the failure.

func (*RunRepo) RecordHandover 

func (r *RunRepo) RecordHandover(ctx context.Context, runID uuid.UUID, step dom.StepKey,
	attempt int, h execution.Handover) error

TerminarTask records the outcome. RecordHandover writes where a handed-over step goes on running (#125). The row stays as it is -- `running` -- because the step is.

func (*RunRepo) RecordLandings 

func (r *RunRepo) RecordLandings(ctx context.Context, runID uuid.UUID, step dom.StepKey,
	workflow string, landings []exec.Landing) error

RecordLandings keeps what one attempt declared it wrote.

One statement for every target, through unnest, rather than one round trip per landing: a step may declare up to a hundred, and this runs as the step ends, while the next one waits for its slot.

The configuration a landing was written with (Landing.Config) is stored once by its hash, in the same transaction, and the landing points at it.

UPSERT, and the attempt is not in the key, for RecordLoad's reason: attempt 2 replaces attempt 1's numbers for a target it lands again. A target only attempt 1 landed is left alone, because it did land -- the rows are in the warehouse whatever happened to the process afterwards.

func (*RunRepo) RecordLoad 

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 

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

func (r *RunRepo) RequestCancel(ctx context.Context, id uuid.UUID, actor string) (CancelResult, error)

RequestCancel cancels a run that is still waiting, or asks the dispatcher that holds it to.

The run's row is locked first, the lock Transicionar takes, so a cancel and a dispatcher starting the run are serialized: whichever comes second sees the other. A run is waiting when its item is unclaimed, or when it has none yet (`created`); its item is deleted in this same transaction, because an item left behind would be claimed and refused `running` forever. A claimed item is a dispatcher's, even before it marks the run `running`.

A failed run has ended: a failure that will be retried is back in the queue in the same breath. canceled_by keeps who asked first.

func (*RunRepo) StepHasSucceeded 

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

func (r *RunRepo) Stranded(ctx context.Context, key string) (id uuid.UUID, status dom.Status, found bool, err error)

Stranded finds the run behind an idempotency key that never reached the queue: still `created` or `queued` and with no queue_items row.

Create, Transicionar and Enqueue are three statements, so a process that dies between them leaves exactly this run -- and the queue's orphan sweep only looks at CLAIMED items, so nothing else would ever see it. A run that is running, finished or already queued is not stranded; found is false for it.

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

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

func (*RunRepo) WaitingOn 

func (r *RunRepo) WaitingOn(ctx context.Context, runID uuid.UUID) (*Wait, error)

WaitingOn says whether a run waits on its concurrency limit, and on what. Nil when it does not -- not in the queue, already claimed, not yet available, no limit, or next in line with a slot free -- because a page that names the wrong reason is worse than one that names none.

IT IS THE CLAIM'S ARITHMETIC, written a second time (internal/queue's Claim is an UPDATE and cannot be asked). The same key, COALESCE('g:' || group, 'w:' || slug); in flight means claimed, not `running`; the same order, priority DESC, available_at, id; and a run waits when what is in flight plus what is ahead of it leaves no slot. A change to the claim is a change here.

type RunSummary 

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

	// Params are the values the run was given. Read only by WorkflowHistory:
	// the lists that do not show them do not pay for them.
	Params map[string]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 

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

Ativas lists the schedules the scheduler has to evaluate.

func (*ScheduleRepo) AdvanceSlot 

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

func (*ScheduleRepo) Of 

func (r *ScheduleRepo) Of(ctx context.Context, slug string) (sch.Schedule, bool, error)

Of is a workflow's schedule, active or paused; false when it has none.

func (*ScheduleRepo) SetActive 

func (r *ScheduleRepo) SetActive(ctx context.Context, slug string, active1 bool, actor string) (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.

actor is who did it, for the events feed: the console's operator, or "cli".

func (*ScheduleRepo) Toggle 

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

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

type ScheduleSummary 

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

ScheduleSummary is the minimum needed to compute the next trigger.

type Stage 

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 StepKey 

type StepKey struct{ Workflow, Node, Target string }

StepKey names one step writing one destination: a step that writes two has a configuration for each.

type StepLog 

type StepLog struct {
	NodeID string
	// MapIndex is the instance of a mapped step, run.Unmapped for the rest.
	MapIndex   int
	Attempt    int
	Status     string
	ExitCode   *int
	Err        string
	Log        string
	DurationMs int64
	// StartedAt is zero for a step that never started.
	StartedAt time.Time
	// Output is what the step published for the steps below it, the same
	// value the graph shows on its node; nil when it published nothing.
	Output json.RawMessage
}

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

type StepRun 

type StepRun struct {
	Status     string
	DurationMs int64
}

StepRun is how a step went the last time it ran.

type StreamKey 

type StreamKey struct{ Gateway, Path string }

StreamKey names a gateway stream: the gateway and the stream's path.

type Subscription 

type Subscription struct {
	Slug    string
	Trigger wf.Trigger
}

Subscription is one workflow and what starts it.

type TargetDetail 

type TargetDetail struct {
	CatalogEntry
	Loads []TargetLoad
}

TargetDetail is one destination for its own page: the entry as the list shows it, and its last loads one by one.

type TargetLoad 

type TargetLoad struct {
	RunID          string
	Workflow, Node string
	LoadedAt       time.Time
	Rows, Bytes    *int64
}

TargetLoad is one landing on the destination, newest first.

type Wait 

type Wait struct {
	// Group is empty when the run contends only with its own workflow, under
	// `concurrency: N`.
	Group   string
	Limit   int
	Holders []Holder // claimed with the same key, oldest first
	Ahead   int      // queued with the same key, that the claim hands out first
}

Wait is what holds a queued run back when it is its concurrency limit: the group or the workflow, the limit, the runs in the slots and how many of the same key the claim takes first (#95).

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 

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

Definition le o grafo publicado.

func (*WorkflowRepo) Definitions 

func (r *WorkflowRepo) Definitions(ctx context.Context) (map[string]wf.Workflow, error)

Definitions is every published workflow's current definition, by slug, in one query: what a page that shows many workflows reads instead of one Definition per row.

func (*WorkflowRepo) Podar 

func (r *WorkflowRepo) Podar(ctx context.Context, projeto uuid.UUID, manter []string, actor 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.

Each removal is in the events feed, written by the same statement that deletes, with actor as who did it.

func (*WorkflowRepo) Publicar 

func (r *WorkflowRepo) Publicar(ctx context.Context, w wf.Workflow, projeto uuid.UUID, actor string) 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.

A publish that changed something -- a new workflow, a new graph, a new schedule -- is in the events feed, in the same transaction, with actor as who did it. One that changed nothing is not: a CI republishing everything on every deploy must not fill the feed.

func (*WorkflowRepo) PublishedGroupMembers 

func (r *WorkflowRepo) PublishedGroupMembers(ctx context.Context, projeto uuid.UUID,
	publishing []string, prune bool) ([]wf.Workflow, error)

PublishedGroupMembers returns the published workflows in a concurrency group that a publish of `publishing` into `projeto` leaves in place: their slug, group and limit, and nothing else.

What `publish` sees is the files on its command line. The other members of their groups are here, and CheckGroups has to hold against them too, or publishing a folder one file at a time is a way around it (#32). Left out: the workflows this publish replaces, and with `prune` every other workflow of the project, which the prune removes. Other projects' stay members.

func (*WorkflowRepo) Subscriptions 

func (r *WorkflowRepo) Subscriptions(ctx context.Context) ([]Subscription, error)

Subscriptions are the workflows that declare a trigger.

READ FROM THE PUBLISHED DOCUMENT, and only from there. `definicao` is what `publish` stored, so the trigger the scheduler acts on is the one in the file that was published -- a column of its own would be a second place for it to disagree with the document.

FILTERED IN SQL AND NOT IN GO. Almost no workflow has a trigger, and walking every definition to find the two that do would read the whole table every cycle. The partial index in 00016 serves exactly this WHERE.

THE CASE IS AT THE SOURCE, not a WHERE the planner may evaluate late -- the rule 00013 states for the same reason. A workflow with no trigger stores `"OnLanded": null`, Go's spelling of a nil slice, and `jsonb_array_length` on a scalar RAISES rather than returning zero.

type WorkflowSummary 

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.

type WriterConfig 

type WriterConfig struct {
	Hash   string
	Fields []catalog.ManifestField

	// Notes are the config file's comments, by setting path; "" is the
	// stream's own.
	Notes map[string]string

	// History is every configuration the writer has had, newest first and
	// the current one included, one per CHANGE -- at most HistoryShown.
	History []ConfigVersion
}

WriterConfig is one writer's configuration as last published.

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.