catbird

package module

Go to main page

Versions in this module

v0

v0.2.0

Jun 26, 2026

Changes in this version

+ func MarkSeen(ctx context.Context, conn Conn, identity string, ids []int64) (int64, error)

+ func MarkSeenUntil(ctx context.Context, conn Conn, identity string, id int64) (int64, error)

+ func NotifyDurable(ctx context.Context, conn Conn, identity, topic, message string, ...) (int64, error)

+ func Render[T any](w *Wire, pattern string, ...)

type Client

+ func (c *Client) MarkSeen(ctx context.Context, identity string, ids []int64) (int64, error)

+ func (c *Client) MarkSeenUntil(ctx context.Context, identity string, id int64) (int64, error)

+ func (c *Client) NotifyDurable(ctx context.Context, identity, topic, message string, ...) (int64, error)

+ func (c *Client) UnseenNotifications(ctx context.Context, identity string, afterID int64, limit int) ([]Notification, error)

+ type Fragment struct

+ Data string

+ Event string

+ ID string

+ func (f *Fragment) Write(p []byte) (int, error)

type GCInfo

+ ExpiredNotificationsDeleted int

+ type Notification struct

+ CollapseKey string

+ CreatedAt time.Time

+ ExpiresAt time.Time

+ ID int64

+ Identity string

+ Message string

+ SeenAt time.Time

+ Topic string

+ func UnseenNotifications(ctx context.Context, conn Conn, identity string, afterID int64, limit int) ([]Notification, error)

+ type NotifyDurableOpts struct

+ CollapseKey string

+ ExpiresAt time.Time

+ type RenderHandler = func(r *http.Request, topic, message string) (Fragment, error)

type Wire

+ func (w *Wire) Render(pattern string, fn RenderHandler) *Wire

+ func (w *Wire) ServePoll(rw http.ResponseWriter, r *http.Request, token string)

v0.1.3

Jun 19, 2026

Changes in this version

+ type MigrationInfo struct

+ Applied bool

+ Name string

+ Version int

+ func MigrationStatus(ctx context.Context, db *sql.DB) ([]MigrationInfo, error)

v0.1.2

Jun 17, 2026

Changes in this version

+ func DeleteFlowSchedule(ctx context.Context, conn Conn, flowName string) (bool, error)

+ func DeleteTaskSchedule(ctx context.Context, conn Conn, taskName string) (bool, error)

type Client

+ func (c *Client) DeleteFlowSchedule(ctx context.Context, flowName string) (bool, error)

+ func (c *Client) DeleteTaskSchedule(ctx context.Context, taskName string) (bool, error)

v0.1.1

Apr 1, 2026

v0.1.0

Apr 1, 2026

Changes in this version

+ const StatusExpired

+ func BindFlow(ctx context.Context, conn Conn, flowName string, pattern string) error

+ func BindTask(ctx context.Context, conn Conn, taskName string, pattern string) error

+ func GetFlowRunID(ctx context.Context) (int64, error)

+ func GetTaskRunID(ctx context.Context) (int64, error)

+ func Notify(ctx context.Context, conn Conn, topic, message string, opts ...NotifyOpts) error

+ func Reader(ctx context.Context, conn Conn, queueName string, quantity int, ...) error

+ func RenderSSE[T any](w *Wire, pattern string, ...)

+ func UnbindFlow(ctx context.Context, conn Conn, flowName string, pattern string) (bool, error)

+ func UnbindTask(ctx context.Context, conn Conn, taskName string, pattern string) (bool, error)

type Client

+ func (c *Client) BindFlow(ctx context.Context, flowName string, pattern string) error

+ func (c *Client) BindTask(ctx context.Context, taskName string, pattern string) error

+ func (c *Client) Notify(ctx context.Context, topic, message string, opts ...NotifyOpts) error

+ func (c *Client) Reader(ctx context.Context, queueName string, quantity int, hideFor time.Duration, ...) error

+ func (c *Client) UnbindFlow(ctx context.Context, flowName string, pattern string) (bool, error)

+ func (c *Client) UnbindTask(ctx context.Context, taskName string, pattern string) (bool, error)

type Conn

+ func GetConn(ctx context.Context) (Conn, error)

type GCInfo

+ ExpiredFlowRuns int

+ ExpiredMessagesDeleted int

+ ExpiredTaskRuns int

+ StaleWireNodesDeleted int

+ type ListenHandler = func(ctx context.Context, topic, message string)

type Message

+ Body json.RawMessage

+ ConcurrencyKey string

+ ExpiresAt time.Time

+ type NotifyOpts struct

+ SentBy string

type PublishManyOpts

+ ConcurrencyKeys []string

+ ExpiresAt time.Time

+ Priority int

type PublishOpts

+ ConcurrencyKey string

+ ExpiresAt time.Time

+ Priority int

+ type ReaderHandler = func(ctx context.Context, msg Message) error

type RunFlowOpts

+ ExpiresAt time.Time

type RunTaskOpts

+ ExpiresAt time.Time

+ type SSEEvent struct

+ Data string

+ Event string

+ ID string

+ func (e *SSEEvent) Write(p []byte) (int, error)

+ type SSERenderHandler = func(r *http.Request, topic, message string) (SSEEvent, error)

type SendManyOpts

+ ConcurrencyKeys []string

+ ExpiresAt time.Time

type SendOpts

+ ConcurrencyKey string

+ ExpiresAt time.Time

+ type TokenOpts struct

+ Identity string

+ ValidFor time.Duration

+ type Wire struct

+ func NewWire(pool *pgxpool.Pool, secret []byte) *Wire

+ func (w *Wire) ID() string

+ func (w *Wire) Listen(pattern string, fn ListenHandler) *Wire

+ func (w *Wire) Notify(ctx context.Context, topic, message string) error

+ func (w *Wire) Presence(ctx context.Context, topic string) ([]string, error)

+ func (w *Wire) RenderSSE(pattern string, fn SSERenderHandler) *Wire

+ func (w *Wire) ServeSSE(rw http.ResponseWriter, r *http.Request, token string)

+ func (w *Wire) Start(ctx context.Context) error

+ func (w *Wire) Token(topics []string, opts ...TokenOpts) string

+ func (w *Wire) WithLogger(logger *slog.Logger) *Wire

v0.0.16

Mar 23, 2026

Changes in this version

type Step

+ func (s *Step) IgnoreOutput(deps ...string) *Step

v0.0.15

Mar 18, 2026

Changes in this version

+ func ClearFlowRuns(ctx context.Context, conn Conn, flowName string) (int, error)

+ func ClearTaskRuns(ctx context.Context, conn Conn, taskName string) (int, error)

type Client

+ func (c *Client) ClearFlowRuns(ctx context.Context, flowName string) (int, error)

+ func (c *Client) ClearTaskRuns(ctx context.Context, taskName string) (int, error)

v0.0.14

Mar 15, 2026

Changes in this version

+ var ErrNotDefined = fmt.Errorf("catbird: not defined")

+ var ErrRunNotCompleted = fmt.Errorf("catbird: run not completed")

+ var ErrRunSkipped = fmt.Errorf("catbird: run skipped")

+ var ErrSignalNotDelivered = fmt.Errorf("catbird: signal not delivered")

type FlowRunInfo

+ Priority int

type Message

+ Priority int

type RunFlowOpts

+ Priority int

type RunTaskOpts

+ Priority int

type SendManyOpts

+ Priority int

type SendOpts

+ Priority int

type StepRunInfo

+ Priority int

type TaskRunInfo

+ Priority int

v0.0.13

Mar 14, 2026

Changes in this version

+ const CatchUpAll

+ const CatchUpOne

+ const CatchUpSkip

type FlowScheduleInfo

+ CatchUp string

type ScheduleOpt

+ func WithCatchUpAll() ScheduleOpt

+ func WithSkipCatchUp() ScheduleOpt

type TaskScheduleInfo

+ CatchUp string

v0.0.12

Mar 13, 2026

v0.0.11

Mar 5, 2026

v0.0.10

Mar 4, 2026

Changes in this version

type Worker

+ func (w *Worker) WithLogger(logger *slog.Logger) *Worker

+ func (w *Worker) WithShutdownTimeout(timeout time.Duration) *Worker

v0.0.9

Mar 4, 2026

v0.0.8

Mar 4, 2026

Changes in this version

+ const StatusCanceled

+ const StatusCanceling

+ const StatusQueued

+ const StatusWaitingForDependencies

+ const StatusWaitingForMapTasks

+ const StatusWaitingForSignal

+ var ErrNoFailedStepInput = fmt.Errorf("failed step input not available")

+ var ErrNoFailedStepSignal = fmt.Errorf("failed step signal input not available")

+ var ErrNoOutputCandidate = fmt.Errorf("no output candidate produced output")

+ var ErrNoRunContext = fmt.Errorf("no run context")

+ var ErrNotFound = fmt.Errorf("not found")

+ var ErrRunCanceled = fmt.Errorf("run canceled")

+ var ErrUnknownStepOutput = fmt.Errorf("unknown step output")

+ func Cancel(ctx context.Context, opts ...CancelOpts) error

+ func CancelFlowRun(ctx context.Context, conn Conn, flowName string, runID int64, ...) (bool, error)

+ func CancelTaskRun(ctx context.Context, conn Conn, taskName string, runID int64, ...) (bool, error)

+ func CompleteEarly(ctx context.Context, output any, reason string) error

+ func PublishMany(ctx context.Context, conn Conn, topic string, payloads []any, ...) (int, error)

+ func PublishManyQuery(topic string, payloads []any, opts ...PublishManyOpts) (string, []any, error)

+ func PurgeFlowRuns(ctx context.Context, conn Conn, flowName string, olderThan time.Duration) (int, error)

+ func PurgeTaskRuns(ctx context.Context, conn Conn, taskName string, olderThan time.Duration) (int, error)

+ func SendMany(ctx context.Context, conn Conn, queueName string, payloads []any, ...) error

+ func SendManyQuery(queueName string, payloads []any, opts ...SendManyOpts) (string, []any, error)

+ type CancelOpts struct

+ Reason string

type Client

+ func (c *Client) CancelFlowRun(ctx context.Context, flowName string, runID int64, opts ...CancelOpts) (bool, error)

+ func (c *Client) CancelTaskRun(ctx context.Context, taskName string, runID int64, opts ...CancelOpts) (bool, error)

+ func (c *Client) PublishMany(ctx context.Context, topic string, payloads []any, opts ...PublishManyOpts) (int, error)

+ func (c *Client) PurgeFlowRuns(ctx context.Context, flowName string, olderThan time.Duration) (int, error)

+ func (c *Client) PurgeTaskRuns(ctx context.Context, taskName string, olderThan time.Duration) (int, error)

+ func (c *Client) SendMany(ctx context.Context, queueName string, payloads []any, opts ...SendManyOpts) error

type Flow

+ func (f *Flow) OnFail(fn any, opts ...HandlerOpt) *Flow

+ func (f *Flow) Output(stepName string) *Flow

+ func (f *Flow) OutputPriority(stepNames ...string) *Flow

+ func (f *Flow) RetentionPeriod(d time.Duration) *Flow

+ func (f *Flow) WithDescription(description string) *Flow

+ type FlowFailure struct

+ Attempts int

+ ConcurrencyKey string

+ ErrorMessage string

+ FailedAt time.Time

+ FailedStepInput json.RawMessage

+ FailedStepName string

+ FailedStepSignalInput json.RawMessage

+ FlowName string

+ FlowRunID int64

+ IdempotencyKey string

+ OnFailAttempts int

+ StartedAt time.Time

+ func (f FlowFailure) FailedStepInputAs(out any) error

+ func (f FlowFailure) FailedStepSignalAs(out any) error

+ func (f FlowFailure) Output(ctx context.Context, step string) (json.RawMessage, error)

+ func (f FlowFailure) OutputAs(ctx context.Context, step string, out any) error

type FlowInfo

+ Description string

+ OutputPriority []string

+ RetentionPeriod time.Duration

type FlowRunInfo

+ CancelReason string

+ CancelRequestedAt time.Time

+ CanceledAt time.Time

+ Headers json.RawMessage

+ func (r *FlowRunInfo) IsCompleted() bool

+ func (r *FlowRunInfo) IsDone() bool

+ type GCInfo struct

+ ExpiredQueuesDeleted int

+ FlowRunsPurged int

+ StaleWorkersDeleted int

+ TaskRunsPurged int

type HandlerOpt

+ func WithFullJitterBackoff(minDelay, maxDelay time.Duration) HandlerOpt

+ func WithTimeout(timeout time.Duration) HandlerOpt

type Message

+ Headers json.RawMessage

+ type PublishManyOpts struct

+ Headers []map[string]any

+ IdempotencyKeys []string

+ VisibleAt time.Time

type PublishOpts

+ Headers map[string]any

type QueueInfo

+ Description string

type QueueOpts

+ Description string

type RunFlowOpts

+ Headers map[string]any

type RunTaskOpts

+ Headers map[string]any

+ type SendManyOpts struct

+ Headers []map[string]any

+ IdempotencyKeys []string

+ Topic string

+ VisibleAt time.Time

type SendOpts

+ Headers map[string]any

type Step

+ func (s *Step) Do(fn any, opts ...HandlerOpt) *Step

+ func (s *Step) Generate(fn any) *Step

+ func (s *Step) MapFlowInput() *Step

+ func (s *Step) MapStepOutput(stepName string) *Step

+ func (s *Step) Reduce(initial any, fn any) *Step

+ func (s *Step) ReduceStep(stepName string) *Step

+ func (s *Step) WithCondition(condition string) *Step

+ func (s *Step) WithDescription(description string) *Step

+ func (s *Step) WithSignal() *Step

type StepInfo

+ Description string

+ MapSourceStepName string

+ ReduceSourceStepName string

+ Signal bool

+ StepType StepType

type StepRunInfo

+ Attempts int

+ CanceledAt time.Time

+ CreatedAt time.Time

+ VisibleAt time.Time

+ func GetStep(ctx context.Context, stepName string) (*StepRunInfo, error)

+ func WaitForStep(ctx context.Context, stepName string, opts ...WaitOpts) (*StepRunInfo, error)

+ func (r *StepRunInfo) IsCompleted() bool

+ func (r *StepRunInfo) IsDone() bool

+ type StepType string

+ const StepTypeGenerator

+ const StepTypeMapper

+ const StepTypeNormal

+ const StepTypeReducer

type Task

+ func (t *Task) Do(fn any, opts ...HandlerOpt) *Task

+ func (t *Task) OnFail(fn any, opts ...HandlerOpt) *Task

+ func (t *Task) RetentionPeriod(d time.Duration) *Task

+ func (t *Task) WithCondition(condition string) *Task

+ func (t *Task) WithDescription(description string) *Task

+ type TaskFailure struct

+ Attempts int

+ ConcurrencyKey string

+ ErrorMessage string

+ FailedAt time.Time

+ IdempotencyKey string

+ OnFailAttempts int

+ StartedAt time.Time

+ TaskName string

+ TaskRunID int64

type TaskInfo

+ Description string

+ RetentionPeriod time.Duration

type TaskRunInfo

+ CancelReason string

+ CancelRequestedAt time.Time

+ CanceledAt time.Time

+ Headers json.RawMessage

+ func (r *TaskRunInfo) IsCompleted() bool

+ func (r *TaskRunInfo) IsDone() bool

type Worker

+ func (w *Worker) Logger(logger *slog.Logger) *Worker

+ func (w *Worker) ShutdownTimeout(timeout time.Duration) *Worker

v0.0.7

Feb 25, 2026

Changes in this version

type Step

+ func (s *Step) Map(stepName string) *Step

+ func (s *Step) MapInput() *Step

type StepInfo

+ HasSignal bool

+ IsMapStep bool

+ MapSource string

v0.0.6

Feb 25, 2026

Changes in this version

+ type FlowHandle struct

+ ID int64

+ Name string

+ func (h *FlowHandle) WaitForOutput(ctx context.Context, out any, opts ...WaitOpts) error

+ type FlowRunInfo struct

+ CompletedAt time.Time

+ ConcurrencyKey string

+ ErrorMessage string

+ FailedAt time.Time

+ ID int64

+ IdempotencyKey string

+ Input json.RawMessage

+ Output json.RawMessage

+ StartedAt time.Time

+ Status string

+ func (r *FlowRunInfo) OutputAs(out any) error

type Message

+ VisibleAt time.Time

type PublishOpts

+ VisibleAt *time.Time

+ type ReadPollOpts struct

+ PollFor time.Duration

+ PollInterval time.Duration

type RunFlowOpts

+ ConcurrencyKey string

+ IdempotencyKey string

+ VisibleAt time.Time

+ type RunTaskOpts struct

+ ConcurrencyKey string

+ IdempotencyKey string

+ VisibleAt time.Time

type SendOpts

+ VisibleAt time.Time

+ type TaskHandle struct

+ ID int64

+ Name string

+ func (h *TaskHandle) WaitForOutput(ctx context.Context, out any, opts ...WaitOpts) error

+ type TaskRunInfo struct

+ CompletedAt time.Time

+ ConcurrencyKey string

+ ErrorMessage string

+ FailedAt time.Time

+ ID int64

+ IdempotencyKey string

+ Input json.RawMessage

+ Output json.RawMessage

+ SkippedAt time.Time

+ StartedAt time.Time

+ Status string

+ func (r *TaskRunInfo) OutputAs(out any) error

+ type WaitOpts struct

+ PollFor time.Duration

+ PollInterval time.Duration

v0.0.5

Feb 24, 2026

Changes in this version

+ func CreateFlowSchedule(ctx context.Context, conn Conn, flowName, cronSpec string, opts *ScheduleOpts) error

+ func CreateTaskSchedule(ctx context.Context, conn Conn, taskName, cronSpec string, opts *ScheduleOpts) error

type Client

+ func (c *Client) CreateFlowSchedule(ctx context.Context, flowName, cronSpec string, opts *ScheduleOpts) error

+ func (c *Client) CreateTaskSchedule(ctx context.Context, taskName, cronSpec string, opts *ScheduleOpts) error

+ func (c *Client) ListFlowSchedules(ctx context.Context) ([]*FlowScheduleInfo, error)

+ func (c *Client) ListTaskSchedules(ctx context.Context) ([]*TaskScheduleInfo, error)

+ type FlowScheduleInfo struct

+ CreatedAt time.Time

+ CronSpec string

+ Enabled bool

+ FlowName string

+ LastEnqueuedAt *time.Time

+ LastRunAt *time.Time

+ NextRunAt time.Time

+ UpdatedAt time.Time

+ func ListFlowSchedules(ctx context.Context, conn Conn) ([]*FlowScheduleInfo, error)

+ type ScheduleOpts struct

+ Input any

+ type TaskScheduleInfo struct

+ CreatedAt time.Time

+ CronSpec string

+ Enabled bool

+ LastEnqueuedAt *time.Time

+ LastRunAt *time.Time

+ NextRunAt time.Time

+ TaskName string

+ UpdatedAt time.Time

+ func ListTaskSchedules(ctx context.Context, conn Conn) ([]*TaskScheduleInfo, error)

v0.0.4

Feb 23, 2026

Changes in this version

+ func PublishQuery(topic string, payload any, opts *PublishOpts) (string, []any, error)

+ func RunFlowQuery(name string, input any, opts *RunFlowOpts) (string, []any, error)

+ func RunTaskQuery(name string, input any, opts *RunOpts) (string, []any, error)

+ func SendQuery(queue string, payload any, opts *SendOpts) (string, []any, error)

v0.0.3

Feb 23, 2026

Changes in this version

+ func EnqueuePublish(batch *pgx.Batch, topic string, payload any, opts PublishOpts) error

+ func Publish(ctx context.Context, conn Conn, topic string, payload any, opts *PublishOpts) error

+ type BackoffStrategy interface

+ NextDelay func(deliveryCount int) time.Duration

+ Validate func() error

+ type CircuitBreaker struct

+ func NewCircuitBreaker(failureThreshold int, openTimeout time.Duration) *CircuitBreaker

+ func (c *CircuitBreaker) Allow(now time.Time) (bool, time.Duration)

+ func (c *CircuitBreaker) RecordFailure(now time.Time)

+ func (c *CircuitBreaker) RecordSuccess()

+ func (c *CircuitBreaker) Validate() error

+ type CircuitBreakerStrategy interface

+ Allow func(now time.Time) (bool, time.Duration)

+ RecordFailure func(now time.Time)

+ RecordSuccess func()

+ Validate func() error

type Client

+ func (c *Client) Publish(ctx context.Context, topic string, payload any, opts *PublishOpts) error

type Flow

+ func (f *Flow) AddStep(step *Step) *Flow

+ func (f *Flow) Schedule(schedule string, inputFunc func(context.Context) (any, error)) *Flow

+ type FullJitterBackoff struct

+ MaxDelay time.Duration

+ MinDelay time.Duration

+ func NewFullJitterBackoff(minDelay, maxDelay time.Duration) *FullJitterBackoff

+ func (b *FullJitterBackoff) NextDelay(deliveryCount int) time.Duration

+ func (b *FullJitterBackoff) Validate() error

+ type HandlerOpts struct

+ Backoff BackoffStrategy

+ BatchSize int

+ CircuitBreaker CircuitBreakerStrategy

+ Concurrency int

+ MaxRetries int

+ Timeout time.Duration

+ type PublishOpts struct

+ DeliverAt *time.Time

+ IdempotencyKey string

+ type Queue struct

+ func NewQueue(name string, opts *QueueOpts) *Queue

type Step

+ func NewStep(name string) *Step

+ func (s *Step) Handler(fn any, opts *HandlerOpts) *Step

+ func (s *Step) Signal(signal bool) *Step

type Task

+ func (t *Task) Condition(condition string) *Task

+ func (t *Task) Handler(fn any, opts *HandlerOpts) *Task

+ func (t *Task) MarshalJSON() ([]byte, error)

+ func (t *Task) Schedule(schedule string, inputFunc func(context.Context) (any, error)) *Task

type Worker

+ func (w *Worker) AddFlow(f *Flow) *Worker

+ func (w *Worker) AddTask(t *Task) *Worker

+ type WorkerOpts struct

+ Logger *slog.Logger

+ ShutdownTimeout time.Duration

v0.0.2

Feb 13, 2026

v0.0.1

Feb 12, 2026

v0.0.0

Feb 12, 2026

Changes in this version

+ const SchemaVersion

+ const StatusCompleted

+ const StatusCreated

+ const StatusFailed

+ const StatusSkipped

+ const StatusStarted

+ var ErrRunFailed = fmt.Errorf("run failed")

+ func Bind(ctx context.Context, conn Conn, queue string, pattern string) error

+ func CreateFlow(ctx context.Context, conn Conn, flow *Flow) error

+ func CreateQueue(ctx context.Context, conn Conn, name string) error

+ func CreateQueueWithOpts(ctx context.Context, conn Conn, name string, opts QueueOpts) error

+ func CreateTask(ctx context.Context, conn Conn, task *Task) error

+ func Delete(ctx context.Context, conn Conn, queue string, id int64) (bool, error)

+ func DeleteMany(ctx context.Context, conn Conn, queue string, ids []int64) error

+ func DeleteQueue(ctx context.Context, conn Conn, name string) (bool, error)

+ func Dispatch(ctx context.Context, conn Conn, topic string, payload any) error

+ func DispatchWithOpts(ctx context.Context, conn Conn, topic string, payload any, opts DispatchOpts) error

+ func EnqueueDispatch(batch *pgx.Batch, topic string, payload any, opts DispatchOpts) error

+ func EnqueueSend(batch *pgx.Batch, queue string, payload any, opts SendOpts) error

+ func GC(ctx context.Context, conn Conn) error

+ func Hide(ctx context.Context, conn Conn, queue string, id int64, hideFor time.Duration) (bool, error)

+ func HideMany(ctx context.Context, conn Conn, queue string, ids []int64, ...) error

+ func MigrateDownTo(ctx context.Context, db *sql.DB, version int) error

+ func MigrateUpTo(ctx context.Context, db *sql.DB, version int) error

+ func Send(ctx context.Context, conn Conn, queue string, payload any) error

+ func SendWithOpts(ctx context.Context, conn Conn, queue string, payload any, opts SendOpts) error

+ func SignalFlow(ctx context.Context, conn Conn, flowName string, flowRunID int64, ...) error

+ func Unbind(ctx context.Context, conn Conn, queue string, pattern string) error

+ type Client struct

+ Conn Conn

+ func New(conn Conn) *Client

+ func (c *Client) Bind(ctx context.Context, queue string, pattern string) error

+ func (c *Client) CreateFlow(ctx context.Context, flow *Flow) error

+ func (c *Client) CreateQueue(ctx context.Context, name string) error

+ func (c *Client) CreateQueueWithOpts(ctx context.Context, name string, opts QueueOpts) error

+ func (c *Client) CreateTask(ctx context.Context, task *Task) error

+ func (c *Client) Delete(ctx context.Context, queue string, id int64) (bool, error)

+ func (c *Client) DeleteMany(ctx context.Context, queue string, ids []int64) error

+ func (c *Client) DeleteQueue(ctx context.Context, name string) (bool, error)

+ func (c *Client) Dispatch(ctx context.Context, topic string, payload any) error

+ func (c *Client) DispatchWithOpts(ctx context.Context, topic string, payload any, opts DispatchOpts) error

+ func (c *Client) GC(ctx context.Context) error

+ func (c *Client) GetFlow(ctx context.Context, name string) (*FlowInfo, error)

+ func (c *Client) GetFlowRun(ctx context.Context, name string, id int64) (*RunInfo, error)

+ func (c *Client) GetFlowRunSteps(ctx context.Context, flowName string, flowRunID int64) ([]*StepRunInfo, error)

+ func (c *Client) GetQueue(ctx context.Context, name string) (*QueueInfo, error)

+ func (c *Client) GetTask(ctx context.Context, name string) (*TaskInfo, error)

+ func (c *Client) GetTaskRun(ctx context.Context, name string, id int64) (*RunInfo, error)

+ func (c *Client) Hide(ctx context.Context, queue string, id int64, hideFor time.Duration) (bool, error)

+ func (c *Client) HideMany(ctx context.Context, queue string, ids []int64, hideFor time.Duration) error

+ func (c *Client) ListFlowRuns(ctx context.Context, name string) ([]*RunInfo, error)

+ func (c *Client) ListFlows(ctx context.Context) ([]*FlowInfo, error)

+ func (c *Client) ListQueues(ctx context.Context) ([]*QueueInfo, error)

+ func (c *Client) ListTaskRuns(ctx context.Context, name string) ([]*RunInfo, error)

+ func (c *Client) ListTasks(ctx context.Context) ([]*TaskInfo, error)

+ func (c *Client) ListWorkers(ctx context.Context) ([]*WorkerInfo, error)

+ func (c *Client) NewWorker(ctx context.Context, opts ...WorkerOpt) (*Worker, error)

+ func (c *Client) Read(ctx context.Context, queue string, quantity int, hideFor time.Duration) ([]Message, error)

+ func (c *Client) ReadPoll(ctx context.Context, queue string, quantity int, ...) ([]Message, error)

+ func (c *Client) RunFlow(ctx context.Context, name string, input any, opts ...RunOpts) (*RunHandle, error)

+ func (c *Client) RunFlowWithOpts(ctx context.Context, name string, input any, opts RunFlowOpts) (*RunHandle, error)

+ func (c *Client) RunTask(ctx context.Context, name string, input any, opts ...RunOpts) (*RunHandle, error)

+ func (c *Client) RunTaskWithOpts(ctx context.Context, name string, input any, opts RunOpts) (*RunHandle, error)

+ func (c *Client) Send(ctx context.Context, queue string, payload any) error

+ func (c *Client) SendWithOpts(ctx context.Context, queue string, payload any, opts SendOpts) error

+ func (c *Client) SignalFlow(ctx context.Context, flowName string, flowRunID int64, stepName string, ...) error

+ func (c *Client) Unbind(ctx context.Context, queue string, pattern string) error

+ type Conn interface

+ Exec func(context.Context, string, ...any) (pgconn.CommandTag, error)

+ Query func(context.Context, string, ...any) (pgx.Rows, error)

+ QueryRow func(context.Context, string, ...any) pgx.Row

+ type DispatchOpts struct

+ DeliverAt *time.Time

+ IdempotencyKey string

+ type Flow struct

+ Name string

+ Steps []*Step

+ func NewFlow(name string, opts ...FlowOpt) *Flow

+ type FlowInfo struct

+ CreatedAt time.Time

+ Name string

+ Steps []StepInfo

+ func GetFlow(ctx context.Context, conn Conn, name string) (*FlowInfo, error)

+ func ListFlows(ctx context.Context, conn Conn) ([]*FlowInfo, error)

+ type FlowOpt func(*Flow)

+ func InitialStepWithSignal[In, SigIn, Out any](name string, fn func(context.Context, In, SigIn) (Out, error), ...) FlowOpt

+ func InitialStep[In, Out any](name string, fn func(context.Context, In) (Out, error), opts ...HandlerOpt) FlowOpt

+ func StepWithDependencyAndSignal[In, SigIn, Dep1Out, Out any](name string, dep1 *StepDependency, ...) FlowOpt

+ func StepWithDependency[In, Dep1Out, Out any](name string, dep1 *StepDependency, ...) FlowOpt

+ func StepWithThreeDependenciesAndSignal[In, SigIn, Dep1Out, Dep2Out, Dep3Out, Out any](name string, dep1 *StepDependency, dep2 *StepDependency, dep3 *StepDependency, ...) FlowOpt

+ func StepWithThreeDependencies[In, Dep1Out, Dep2Out, Dep3Out, Out any](name string, dep1 *StepDependency, dep2 *StepDependency, dep3 *StepDependency, ...) FlowOpt

+ func StepWithTwoDependenciesAndSignal[In, SigIn, Dep1Out, Dep2Out, Out any](name string, dep1 *StepDependency, dep2 *StepDependency, ...) FlowOpt

+ func StepWithTwoDependencies[In, Dep1Out, Dep2Out, Out any](name string, dep1 *StepDependency, dep2 *StepDependency, ...) FlowOpt

+ type HandlerOpt func(*handlerOpts)

+ func WithBackoff(minDelay, maxDelay time.Duration) HandlerOpt

+ func WithBatchSize(n int) HandlerOpt

+ func WithCircuitBreaker(failureThreshold int, openTimeout time.Duration) HandlerOpt

+ func WithConcurrency(n int) HandlerOpt

+ func WithCondition(expr string) HandlerOpt

+ func WithMaxDuration(d time.Duration) HandlerOpt

+ func WithMaxRetries(n int) HandlerOpt

+ type Message struct

+ CreatedAt time.Time

+ DeliverAt time.Time

+ Deliveries int

+ ID int64

+ IdempotencyKey string

+ Payload json.RawMessage

+ Topic string

+ func Read(ctx context.Context, conn Conn, queue string, quantity int, ...) ([]Message, error)

+ func ReadPoll(ctx context.Context, conn Conn, queue string, quantity int, ...) ([]Message, error)

+ type Optional struct

+ IsSet bool

+ Value T

+ type QueueInfo struct

+ CreatedAt time.Time

+ ExpiresAt time.Time

+ Name string

+ Unlogged bool

+ func GetQueue(ctx context.Context, conn Conn, name string) (*QueueInfo, error)

+ func ListQueues(ctx context.Context, conn Conn) ([]*QueueInfo, error)

+ type QueueOpts struct

+ ExpiresAt time.Time

+ Unlogged bool

+ type RunFlowOpts = RunOpts

+ type RunHandle struct

+ ID int64

+ Name string

+ func RunFlow(ctx context.Context, conn Conn, name string, input any, opts ...RunOpts) (*RunHandle, error)

+ func RunFlowWithOpts(ctx context.Context, conn Conn, name string, input any, opts RunFlowOpts) (*RunHandle, error)

+ func RunTask(ctx context.Context, conn Conn, name string, input any, opts ...RunOpts) (*RunHandle, error)

+ func RunTaskWithOpts(ctx context.Context, conn Conn, name string, input any, opts RunOpts) (*RunHandle, error)

+ func (h *RunHandle) WaitForOutput(ctx context.Context, out any) error

+ type RunInfo struct

+ CompletedAt time.Time

+ ConcurrencyKey string

+ ErrorMessage string

+ FailedAt time.Time

+ ID int64

+ IdempotencyKey string

+ Input json.RawMessage

+ Output json.RawMessage

+ SkippedAt time.Time

+ StartedAt time.Time

+ Status string

+ func GetFlowRun(ctx context.Context, conn Conn, name string, id int64) (*RunInfo, error)

+ func GetTaskRun(ctx context.Context, conn Conn, name string, id int64) (*RunInfo, error)

+ func ListFlowRuns(ctx context.Context, conn Conn, name string) ([]*RunInfo, error)

+ func ListTaskRuns(ctx context.Context, conn Conn, name string) ([]*RunInfo, error)

+ func (r *RunInfo) OutputAs(out any) error

+ type RunOpts struct

+ ConcurrencyKey string

+ IdempotencyKey string

+ type ScheduleOpt func(*scheduleEntry)

+ func WithInput[T any](inputFn func(context.Context) (T, error)) ScheduleOpt

+ type Scheduler struct

+ func NewScheduler(conn Conn, logger *slog.Logger) *Scheduler

+ func (s *Scheduler) AddFlow(flowName string, schedule string, opts ...ScheduleOpt)

+ func (s *Scheduler) AddTask(taskName string, schedule string, opts ...ScheduleOpt)

+ func (s *Scheduler) Start(ctx context.Context) error

+ func (s *Scheduler) Stop(ctx context.Context)

+ type SendOpts struct

+ DeliverAt time.Time

+ IdempotencyKey string

+ Topic string

+ type Step struct

+ Condition string

+ DependsOn []*StepDependency

+ HasSignal bool

+ Name string

+ type StepDependency struct

+ Name string

+ Optional bool

+ func Dependency(name string) *StepDependency

+ func OptionalDependency(name string) *StepDependency

+ type StepDependencyInfo struct

+ Name string

+ type StepHandlerInfo struct

+ FlowName string

+ StepName string

+ type StepInfo struct

+ DependsOn []StepDependencyInfo

+ Name string

+ type StepOpt func(*Step)

+ type StepRunInfo struct

+ CompletedAt time.Time

+ ErrorMessage string

+ FailedAt time.Time

+ ID int64

+ Output json.RawMessage

+ SkippedAt time.Time

+ StartedAt time.Time

+ Status string

+ StepName string

+ func GetFlowRunSteps(ctx context.Context, conn Conn, flowName string, flowRunID int64) ([]*StepRunInfo, error)

+ type Task struct

+ Name string

+ func NewTask[In, Out any](name string, fn func(context.Context, In) (Out, error), opts ...HandlerOpt) *Task

+ type TaskHandlerInfo struct

+ TaskName string

+ type TaskInfo struct

+ CreatedAt time.Time

+ Name string

+ func GetTask(ctx context.Context, conn Conn, name string) (*TaskInfo, error)

+ func ListTasks(ctx context.Context, conn Conn) ([]*TaskInfo, error)

+ type Worker struct

+ func NewWorker(ctx context.Context, conn Conn, opts ...WorkerOpt) (*Worker, error)

+ func (w *Worker) Start(ctx context.Context) error

+ type WorkerInfo struct

+ ID string

+ LastHeartbeatAt time.Time

+ StartedAt time.Time

+ StepHandlers []*StepHandlerInfo

+ TaskHandlers []*TaskHandlerInfo

+ func ListWorkers(ctx context.Context, conn Conn) ([]*WorkerInfo, error)

+ type WorkerOpt func(*Worker)

+ func WithFlow(f *Flow) WorkerOpt

+ func WithGracefulShutdown(d time.Duration) WorkerOpt

+ func WithLogger(l *slog.Logger) WorkerOpt

+ func WithScheduledFlow(flowName string, schedule string, opts ...ScheduleOpt) WorkerOpt

+ func WithScheduledTask(taskName string, schedule string, opts ...ScheduleOpt) WorkerOpt

+ func WithTask(t *Task) WorkerOpt

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL