Documentation
¶
Overview ¶
Package consumers holds real outbox consumers — the business-logic Processors the worker runs (ADR-016 §Class 2). They are kept OUT of pkg/worker so the core delivery loop stays dependency-light (e.g. excelize lives only here, not in the worker engine). A consumer reads an event, does its work, and writes results BACK through the engine HTTP API with a scoped service JWT (worker.EngineClient) — never the tenant DB directly.
Index ¶
Constants ¶
const DefaultEmailTopic = "email.send"
DefaultEmailTopic is the outbox topic the EmailProcessor consumes. A producer (a Class-1 handler, ctx.Enqueue) writes an "email.send" event in its business transaction; the user gets their HTTP response immediately and the email goes out async — never blocking the request (the whole point of the outbox).
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Email ¶
type Email struct {
To string
Subject string
HTML string
Text string // optional plaintext alternative
MessageID string // deterministic per outbox row — best-effort dedup hint
}
Email is one message to send. From/Date/Message-ID are filled by the SMTPSender (From from config; Message-ID by the EmailProcessor for idempotency). Text is an optional plaintext alternative — when set, the message is multipart/alternative so clients that can't render HTML still show something.
type EmailProcessor ¶
type EmailProcessor struct {
// contains filtered or unexported fields
}
EmailProcessor is the EMAIL-CONSUMER-V1 consumer: on an "email.send" outbox event it renders the named template with the payload data and sends it through an external SMTP provider (EmailSender). It writes nothing back to the engine — an email needs no write-back; the outbox row IS the record (sent ⇒ delivered to the provider; parked 'failed' ⇒ gave up after retries).
Idempotency / at-least-once: delivery is at-least-once, so a worker that crashes AFTER the provider accepts the message but BEFORE the outbox COMMIT will resend on redelivery. For TRANSACTIONAL email a rare duplicate is tolerable (a second "verify your email" is annoying, not harmful) and is the accepted trade-off here — far better than dropping the mail. As a free, standards-based mitigation every send carries a DETERMINISTIC Message-ID derived from the outbox Row.ID, so a well-behaved provider/MTA can dedupe the redelivery. For HARD idempotency, a consumer can record Row.ID in its own table and no-op the second delivery (the XLSX consumer's terminal-status check is the analogue); not done here by choice.
func NewEmailProcessor ¶
func NewEmailProcessor(sender EmailSender, log zerolog.Logger) *EmailProcessor
NewEmailProcessor builds the consumer. topic defaults to DefaultEmailTopic; the templates are the built-in demo set (verification, welcome) — swap them with WithTemplates for an app's own.
func (*EmailProcessor) Process ¶
Process implements worker.Processor. A foreign topic is acked (this is safe ONLY when the email worker is the sole consumer of those rows; for a shared outbox with multiple event types, compose consumers behind a Router instead — see Router). A malformed payload, unknown template, or empty recipient is a permanent error: it is returned (the worker retries to maxAttempts then parks the row 'failed', durably recording the bad event) and the SMTP send is never attempted. A send error (transient or 5xx) is returned so the row is retried.
func (*EmailProcessor) WithTopic ¶
func (p *EmailProcessor) WithTopic(topic string) *EmailProcessor
WithTopic overrides the consumed topic. Returns p for chaining.
type EmailSender ¶
EmailSender sends one Email. The interface is the seam the EmailProcessor talks to, so tests substitute a capturing mock and never touch a real SMTP server.
type FileOpener ¶
FileOpener resolves a job's file_ref to a readable stream for the tenant. The default opens file_ref as a local path; the VFS-backed opener (pkg/files) treats file_ref as a VFS file_id and streams the content-addressed blob — wiring the consumer to the real file store (FILES-V1) without coupling pkg/consumers to pkg/files. The caller closes the returned reader.
type Router ¶
type Router struct {
// contains filtered or unexported fields
}
Router is the COEXISTENCE mechanism for a shared outbox carrying multiple event types. It is a worker.Processor that dispatches each row to the registered consumer whose topic rule matches, and acks (logs) anything unmatched.
WHY a dispatcher and not one worker per mode: the outbox is one table drained by competing consumers under FOR UPDATE SKIP LOCKED, which distributes rows across ALL running workers regardless of topic. A single-purpose worker (xlsx-only, email-only) ACKS the topics it doesn't own — so if an xlsx worker and an email worker drain the SAME outbox, whichever claims a row first acks it, and the other consumer never sees it: silent event loss. The correct model for multiple event types is therefore ONE worker that handles ALL of them — a Router — scaled horizontally by running N identical Router workers (SKIP LOCKED still gives each a disjoint slice, no double-processing). Run single-mode workers only when they are the SOLE consumer of the outbox.
Matching is by exact topic or by a "prefix.*" suffix wildcard (e.g. "filejobs.*"); the first registered rule that matches wins, so register more specific rules first.
func (*Router) HandlePrefix ¶
HandlePrefix routes every topic starting with prefix (e.g. prefix "filejobs." matches "filejobs.created", "filejobs.updated") to proc. Returns r for chaining.
type SMTPConfig ¶
type SMTPConfig struct {
Host string
Port string
Username string
Password string
From string // envelope + header From, e.g. "App <no-reply@app.com>"
Timeout time.Duration // whole-dialog deadline; default 15s
// InsecureSkipVerify disables TLS cert verification — TEST ONLY.
InsecureSkipVerify bool
}
SMTPConfig points the SMTPSender at an EXTERNAL provider (Brevo, Resend, Mailgun, SES …) entirely through env-supplied values — no provider SDK, no hard-coded host. Self-hosting SMTP is deliberately NOT supported (deliverability, SPF/DKIM and IP reputation are a provider's job).
type SMTPSender ¶
type SMTPSender struct {
// contains filtered or unexported fields
}
SMTPSender sends mail over SMTP with STARTTLS (used automatically when the server advertises it) and AUTH PLAIN (only when a username is set). The whole dialog is bounded by Timeout, so a hung provider can never wedge the worker.
func NewSMTPSender ¶
func NewSMTPSender(cfg SMTPConfig) (*SMTPSender, error)
NewSMTPSender builds an SMTP sender. Host, Port and From are required; Username/ Password are optional (an open relay or a test server needs neither).
func (*SMTPSender) Send ¶
func (s *SMTPSender) Send(ctx context.Context, m Email) error
Send delivers m. A connection/timeout/transient SMTP error is returned so the outbox keeps the row pending for retry; a permanent rejection (e.g. 550) is also returned and is eventually parked 'failed' by the worker after maxAttempts.
type XLSXProcessor ¶
type XLSXProcessor struct {
// contains filtered or unexported fields
}
XLSXProcessor is the first real consumer (XLSX-CONSUMER-V1): the canonical FileJob pattern. On a "{resource}.created" event it fetches the job, STREAMS the referenced XLSX (excelize row iterator — never the whole file in RAM), computes an aggregate, and writes {status, result} back via the engine API using a scoped service JWT. A corrupt/invalid file is a PERMANENT failure (job → "failed", event acked); a transient engine error keeps the row pending for retry (at-least-once). Idempotent: a job already in a terminal state is skipped, so a redelivery never double-processes.
func NewXLSXProcessor ¶
func NewXLSXProcessor(client *worker.EngineClient, resource string, log zerolog.Logger) *XLSXProcessor
NewXLSXProcessor builds the consumer. resource defaults to "filejobs". The file source defaults to the local filesystem; use WithFileOpener to read from the VFS.
func (*XLSXProcessor) Process ¶
Process implements worker.Processor. Non-".created" topics and other resources are acked (logged) so this consumer never blocks the shared queue.
func (*XLSXProcessor) WithFileOpener ¶
func (p *XLSXProcessor) WithFileOpener(o FileOpener) *XLSXProcessor
WithFileOpener swaps the file source (e.g. the VFS). A nil opener is ignored so the default local-path source remains. Returns p for chaining.
type XLSXResult ¶
type XLSXResult struct {
Rows int `json:"rows"`
Total float64 `json:"total"`
ByType map[string]float64 `json:"by_type"`
}
XLSXResult is the demo aggregate written back as the job result: the FileJob pattern from the rt-api tax case — count rows, sum a numeric column, and sum it grouped by a type column.
func ProcessXLSX ¶
func ProcessXLSX(path string) (XLSXResult, error)
ProcessXLSX opens path and STREAMS its first sheet with the excelize row iterator (f.Rows — NOT f.GetRows, which materialises every row in RAM), computing the demo aggregate. It validates the header has the expected columns (tipo, monto) and returns a descriptive error on a corrupt/empty/missing-column file so the caller can record a "failed" job rather than crash.
func ProcessXLSXReader ¶
func ProcessXLSXReader(r io.Reader) (XLSXResult, error)
ProcessXLSXReader is ProcessXLSX over a stream rather than a path — the form the VFS uses (VFS.Get returns a reader). excelize.OpenReader buffers the zip archive (an .xlsx is a zip, so random access is inherent), but the per-row aggregation still STREAMS via the f.Rows iterator, never GetRows.
Source Files