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

broadcasts

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Oct 22, 2026 License: AGPL-3.0 Imports: 27 Imported by: 0

Documentation

Overview

Package broadcasts owns the Broadcast send/schedule state machine (ADR 0016): draft -> scheduled -> sending and clearing a schedule; the claim and the send job it needs are one transaction. Every surface that sends or schedules a Broadcast (the /site SPA today, /api and MCP next) calls it; none re-implements a transition. Handlers stay thin adapters: scope check, call, map errors.

Every Broadcast status write goes through the transition table in status.go (ADR 0046): a stateless table with a compare-and-swap write, so an illegal move is lifecycle.ErrInvalidTransition. Delivery owns the rest of the lifecycle: plan, pace, send one recipient through Outbound send (ADR 0015), give up, finalize. The queue adapter (river) only implements its Scheduler seam and decides how to retry.

Index

Constants

View Source 
const Queue = "broadcasts"

Queue is the river queue of the send jobs. A single blast fans out into one job per recipient; keeping them off the default queue stops a large broadcast from starving automation steps and welcome emails (the same isolation jobs.QueueWebhooks gives outbound webhooks).

Variables

View Source 
var (
	// ErrNotFound: no such Broadcast in this Workspace.
	ErrNotFound = errors.New("broadcasts: broadcast not found")
	// ErrNotSendable: the Broadcast is past draft/scheduled (sending, sent, failed).
	ErrNotSendable = errors.New("broadcasts: broadcast is not a draft or scheduled")
	// ErrNotArchivable: only a sent or failed Broadcast is archived; a draft is deleted and a
	// scheduled or sending one is still work in flight.
	ErrNotArchivable = errors.New("broadcasts: only a sent or failed broadcast can be archived")
	// ErrNotScheduled: the Broadcast is not scheduled, so there is no schedule to clear.
	ErrNotScheduled = errors.New("broadcasts: broadcast is not scheduled")
)
View Source 
var (
	// ErrNotDraft: the Broadcast is past draft (scheduled, sending, sent, failed) and
	// can no longer be edited, re-targeted or deleted through the authoring surface.
	ErrNotDraft = errors.New("broadcasts: broadcast is not a draft")
	// ErrSegmentNotFound: the audience Segment does not exist in this Workspace.
	ErrSegmentNotFound = errors.New("broadcasts: segment not found")
	// ErrIntegrationNotFound: the Integration does not exist in this Workspace or is archived.
	ErrIntegrationNotFound = errors.New("broadcasts: integration not found")
	// ErrBodyAndDoc: a write carries both a body and a document. A client writes one of them: the
	// document (the server builds the body from it) or the body (which clears the document).
	ErrBodyAndDoc = errors.New("broadcasts: write a body or a doc, not both")
)
View Source 
var ErrPlanFailed = errors.New("broadcasts: the plan failed the broadcast")

ErrPlanFailed marks a plan that failed the Broadcast for good (a template that does not render, a broken segment): the Broadcast is failed, and running it again cannot help.

Functions

func CancelPending 

func CancelPending(ctx context.Context, s *ent.Scoped, contactID int64, purger events.QueuePurger) error

CancelPending cancels a Contact's pending Broadcast work (Erasure, ADR 0021): the recipients whose send has not ended are removed with their queued send jobs, and each Broadcast left with nothing pending settles. A recipient whose send ended stays as an anonymous record. It runs inside the caller's transaction (s and purger come from the same one); a send job that is already running is not purged, and finds its recipient gone when it starts.

func Due 

func Due(ctx context.Context, s *ent.Scoped, id int64, jobScheduledAt *time.Time) (bool, error)

Due reports whether a queued send job should still run. A Broadcast that is back in draft was unscheduled after the job was queued. A delayed job (jobScheduledAt set) is also bound to the schedule it was enqueued for: unscheduling leaves it queued, so after a reschedule the Broadcast carries a different scheduled_at (or none, once sent now) and the superseded job must not send it early.

func Finalize 

func Finalize(ctx context.Context, s *ent.Scoped, broadcastID int64) error

Finalize flips a Broadcast to "sent" once none of its recipients' sends is still pending. The finalize move (sending -> sent, a compare-and-swap) makes concurrent finalizers (one per per-recipient job, or an Erasure that removed the last pending recipient) a no-op after the first, as is a Broadcast that is not sending at all, so sent_at is set exactly once.

func IsRetryLater 

func IsRetryLater(err error) bool

IsRetryLater reports whether err asks the adapter to try again rather than record a failure: a hold, a Deferral or a live claim.

func Pending 

func Pending() predicate.BroadcastRecipient

Pending matches the recipients whose send has not ended: no Outbound message of theirs has left pending. It is the one reading of "this recipient is still to be sent".

func Ratio 

func Ratio(num, denom int) float32

Ratio is num/denom as a float32 in [0,1]; a zero denominator yields 0.

func RecordClick 

func RecordClick(ctx context.Context, s *ent.Scoped, recipientID int64) (bool, error)

RecordClick marks the recipient as having clicked a link of the email.

func RecordOpen 

func RecordOpen(ctx context.Context, s *ent.Scoped, recipientID int64) (bool, error)

RecordOpen marks the recipient as having opened the email.

func RecordUnsubscribe 

func RecordUnsubscribe(ctx context.Context, s *ent.Scoped, broadcastID, contactID int64) (bool, error)

RecordUnsubscribe marks the Contact's recipient row of the Broadcast as having unsubscribed from it. A Contact that was never a recipient of it (a link to an erased row) matches nothing.

func StatsOf 

func StatsOf(ctx context.Context, s *ent.Scoped, ids ...int64) (map[int64]Stats, error)

StatsOf reads the figures of the given Broadcasts in two grouped queries, whatever their number. A Broadcast with no rows yet has zero figures.

Types

type DeferredError 

type DeferredError struct {
	// Delay is the wait the Outbound retry policy chose for this deferral (it was recorded on
	// the recipient so the backlog of later recipients can tell it is asleep).
	Delay time.Duration
}

DeferredError reports that Outbound send found the Integration's Send rate limit spent (outbound.Deferral). Nothing was consumed: the caller retries the same work after Delay.

func (*DeferredError) Error 

func (e *DeferredError) Error() string

type Delivery 

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

Delivery is the send side of the Broadcast lifecycle: one recipient at a time through Outbound send, the hold reason and the final settling of the Broadcast. A recipient has no delivery state of its own: how its send ended is its Outbound message's (ADR 0015), and it is pending until then. It holds no client: every call receives the Workspace-scoped client.

func NewDelivery 

func NewDelivery(out *outbound.Module) *Delivery

NewDelivery builds the delivery over the Outbound send module.

func (*Delivery) GiveUp 

func (d *Delivery) GiveUp(ctx context.Context, s *ent.Scoped, recipientID int64, cause error) error

GiveUp records the terminal failure of a recipient whose last attempt failed with cause: Outbound send ends its message as Failed with the cause as its reason (a send that already reached the provider stays sent), and the Broadcast settles if none is left pending. A recipient already decided keeps its outcome.

outbound.ErrInProgress means another attempt still holds a live claim and may yet reach the provider: nothing is failed, and the adapter waits out the lease and gives up again.

func (*Delivery) Pace 

func (d *Delivery) Pace(ctx context.Context, s *ent.Scoped, id int64, n int, now time.Time) ([]time.Time, error)

Pace spreads n recipient sends over time for a Broadcast whose Integration has a Send rate limit (ADR 0023): the i-th send is scheduled i steps after the start, a step being one over the effective rate, so the limiter is a guard and a Deferral the rare case. The start is now, or the Broadcast's scheduled time when that is later. It records the last scheduled time on the Broadcast, from which the progress estimate is derived, and returns nil (every send at once) when nothing limits the Integration. Pacing decides only when a send runs: eligibility is still checked per message at send time.

func (*Delivery) Plan 

func (d *Delivery) Plan(ctx context.Context, s *ent.Scoped, id int64) ([]int64, error)

Plan resolves the Broadcast's eligible audience, creates one pending recipient row per contact (idempotent through the unique (broadcast, contact) index, so a re-planned job converges) and returns the recipient ids to send. It refuses a Broadcast past its plan (lifecycle.ErrInvalidTransition) and holds one that cannot send now (*HeldError) before any row exists; a template that cannot render, or a broken segment, fails the Broadcast once, before any recipient row exists. An empty audience settles the Broadcast.

func (*Delivery) Run 

func (d *Delivery) Run(ctx context.Context, s *ent.Scoped, id int64, sched Scheduler, now time.Time) error

Run is the whole delivery sequence of a Broadcast: plan the audience, pace the recipients and hand each send to the scheduler. It returns *HeldError when the Broadcast is held (it keeps its status and records why, to be run again), lifecycle.ErrInvalidTransition when it is past its plan (a stale job), and a plain error on a fault. A plan that failed the Broadcast (see ErrPlanFailed) is recorded and returns nil. An empty audience settles the Broadcast and schedules nothing. Eligibility is still checked per message at send time, so an unsubscribe or Suppression that lands after planning is honoured.

func (*Delivery) RunClaimed 

func (d *Delivery) RunClaimed(ctx context.Context, s *ent.Scoped, id int64, sched Scheduler, now time.Time) error

RunClaimed is Run for a job queued after the Broadcast was claimed (Send or Schedule): it refuses a draft with lifecycle.ErrInvalidTransition, so an unschedule that lands after Due has answered is not undone by the plan.

func (*Delivery) SendRecipient 

func (d *Delivery) SendRecipient(ctx context.Context, s *ent.Scoped, recipientID int64) error

SendRecipient hands one recipient's email to Outbound send, then settles the Broadcast if that was its last pending recipient. It is idempotent: a recipient already decided is left alone (a stale job is just that), and the Outbound idempotency key makes a retry after a committed send replay the recorded result.

A returned error is for the adapter to retry: *HeldError (the source is on hold), *DeferredError (its Send rate limit is spent), outbound.ErrInProgress (a live claim) or a fault; none consumed the recipient.

type Fields 

type Fields struct {
	Name      *string
	Subject   *string
	FromName  *string
	FromEmail *string
	Body      *string
	// Doc is the editor's JSON document, empty when the Broadcast is written as HTML. The body is
	// built from it in the same save (ADR 0040); a Body alone clears the stored document.
	Doc           json.RawMessage
	SegmentID     *int64
	IntegrationID *int64

	ClearFromName    bool
	ClearFromEmail   bool
	ClearSegment     bool
	ClearIntegration bool
}

Fields are the author-editable content of a Broadcast. A nil field is "not provided": Create falls back to the schema default, Update keeps the stored value. The Clear flags are JSON Merge Patch nulls for Update: they drop the stored value. SegmentID and IntegrationID reference rows of the same Workspace; the scoped client refuses a foreign one with ent.ErrNotInWorkspace.

type HeldError 

type HeldError struct{ Reason string }

HeldError reports that Outbound send put the source on hold (see outbound.Hold*). Nothing was consumed: the caller defers the same work.

func (*HeldError) Error 

func (e *HeldError) Error() string

type Module 

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

Module is the Broadcast send/schedule state machine. It holds no client: every call receives the Workspace-scoped client (ADR 0017) built by the entry point, so the module writes no Workspace predicate and cannot name another Workspace.

func New 

func New(bus *events.Bus) *Module

func (*Module) Archive 

func (m *Module) Archive(ctx context.Context, s *ent.Scoped, id int64) error

Archive discards a sent or failed Broadcast (ADR 0049): it leaves the lists but its Recipients, report, tracking and figures stay. A draft is purged (DeleteDraft) and a scheduled or sending Broadcast is work in flight, so both are ErrNotArchivable. Archiving an archived Broadcast changes nothing.

func (*Module) Create 

func (m *Module) Create(ctx context.Context, s *ent.Scoped, f Fields) (*ent.Broadcast, error)

Create makes a draft Broadcast. Nothing here schedules or sends. A subject, body or document that cannot render is emailrender.ErrInvalidContent; the body is stored sanitized.

func (*Module) DeleteDraft 

func (m *Module) DeleteDraft(ctx context.Context, s *ent.Scoped, id int64) error

DeleteDraft purges a draft Broadcast (ADR 0049). Past draft the Broadcast is history (its recipients and report) and is archived instead, never removed.

func (*Module) Get 

func (m *Module) Get(ctx context.Context, s *ent.Scoped, id int64) (*ent.Broadcast, error)

Get returns one Broadcast of the Workspace.

func (*Module) List 

func (m *Module) List(ctx context.Context, s *ent.Scoped, p pagination.Params, archived bool) (pagination.Page[*ent.Broadcast], error)

List returns one page of the Workspace's Broadcasts, newest first: the kept ones, or with archived the archived ones.

func (*Module) Report 

func (m *Module) Report(ctx context.Context, s *ent.Scoped, id int64) (Report, error)

Report reads the Broadcast's delivery figures (see Stats).

func (*Module) Schedule 

func (m *Module) Schedule(ctx context.Context, s *ent.Scoped, id int64, when time.Time) (*ent.Broadcast, error)

Schedule moves a draft or scheduled Broadcast to scheduled at `when` and queues a delayed job in the same transaction.

func (*Module) Send 

func (m *Module) Send(ctx context.Context, s *ent.Scoped, id int64) (*ent.Broadcast, error)

Send moves a draft or scheduled Broadcast to sending (dropping any schedule) and queues its send job in the same transaction: either both happen or neither does.

func (*Module) SetAudience 

func (m *Module) SetAudience(ctx context.Context, s *ent.Scoped, id int64, segmentID *int64) (*ent.Broadcast, error)

SetAudience points a draft at a Segment, or at all active contacts when segmentID is nil. It never schedules or sends.

func (*Module) Unarchive 

func (m *Module) Unarchive(ctx context.Context, s *ent.Scoped, id int64) error

Unarchive brings an archived Broadcast back to the lists. Restoring a Broadcast that is not archived changes nothing.

func (*Module) Unschedule 

func (m *Module) Unschedule(ctx context.Context, s *ent.Scoped, id int64) (*ent.Broadcast, error)

Unschedule returns a scheduled Broadcast to draft and drops its schedule. The delayed job stays queued; when it fires it finds a draft and does nothing.

func (*Module) Update 

func (m *Module) Update(ctx context.Context, s *ent.Scoped, id int64, f Fields) (*ent.Broadcast, error)

Update edits a draft. The UPDATE only matches a draft row, so a Broadcast that was claimed for sending in the meantime is never edited.

type Progress 

type Progress struct {
	// Processed counts recipients already decided: sent, skipped or failed.
	Processed int
	// Remaining counts recipients still pending.
	Remaining int
	// EstimatedCompletion is when the last remaining recipient is expected to go out;
	// nil when it cannot be known (a hold, or no Send rate limit to pace by).
	EstimatedCompletion *time.Time
}

Progress is how far a sending Broadcast has got (ADR 0023). A Broadcast stays "sending" while it is paced, held or deferred, so the progress is derived from its recipients, not from a status.

func ProgressOf 

func ProgressOf(ctx context.Context, s *ent.Scoped, b *ent.Broadcast, now time.Time) (*Progress, error)

ProgressOf computes the progress of a Broadcast, or nil when it is not sending. The estimate is the last scheduled recipient job while that is still ahead; once it has passed (jobs were deferred, or the limit was lowered mid-send) it falls back to the remaining recipients at the Integration's current rate. A hold has no estimate: the source is blocked, not busy, and nothing says when it lifts.

type Report 

type Report struct {
	Status       broadcast.Status
	HoldReason   *string
	Recipients   int
	Sent         int
	Skipped      int
	Failed       int
	Opened       int
	Clicked      int
	Unsubscribed int
	OpenRate     float32
	ClickRate    float32
}

Report is a Broadcast's delivery report. Rates are ratios in [0,1]; a zero denominator yields 0.

type Scheduler 

type Scheduler interface {
	Schedule(ctx context.Context, broadcastID int64, sends []Send) error
}

Scheduler is the seam of the Broadcast lifecycle: "schedule this recipient's send at time T". The river adapter inserts a job per send at its time; the synchronous adapter sends in order at once. The module computes the times either way, so both run one sequence.

type Send 

type Send struct {
	RecipientID int64
	At          time.Time
}

Send is one recipient's send to schedule. A zero At means as soon as possible: the Broadcast is not paced.

type SendArgs 

type SendArgs struct {
	BroadcastID int64 `json:"broadcast_id"`
	// ScheduledAt is the time a delayed job was queued for; nil for an immediate send.
	ScheduledAt *time.Time `json:"scheduled_at,omitempty"`
}

SendArgs is the fan-out (plan) job payload: which Broadcast to dispatch. The worker plans the audience and queues one job per recipient, so no single job carries the whole send.

func (SendArgs) Kind 

func (SendArgs) Kind() string

type Stats 

type Stats struct {
	// Recipients counts the Broadcast's recipient rows.
	Recipients int
	// Sent, Skipped and Failed count the recipients whose message ended that way.
	Sent, Skipped, Failed int
	// Pending counts the recipients whose send has not ended: no message yet, or one still
	// claimed. It is Recipients less the three above.
	Pending int
	// Opened, Clicked and Unsubscribed count the recipients with that engagement.
	Opened, Clicked, Unsubscribed int
}

Stats is a Broadcast's delivery figures. They are never stored: the audience is its recipient rows, how each send ended is its Outbound message (ADR 0015), and engagement is the timestamps on the recipient, so the figures cannot drift from what the rows say and are live while the Broadcast is still sending.

func (Stats) Processed 

func (st Stats) Processed() int

Processed counts the recipients whose send has ended.

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.