Documentation
¶
Overview ¶
Package tasks wires Hanzo Notify into hanzoai/tasks as a durable execution substrate.
One workflow type, NotifySendWorkflow, takes a SendRequest and walks the same code path the sync HTTP handler does:
resolve provider → render template → call library Send → write message row → write meter row → set status
The worker registers itself on a single task queue ("notify-send"). Async POST /v1/notify/send Dispatch hands off to ExecuteWorkflow; the sync path calls the same internal SendOnce function directly.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Activities ¶
type Activities struct {
// contains filtered or unexported fields
}
Activities groups the activity-level side effects so the worker can register a typed receiver and call sites can pass a mock during tests.
The app field is the narrow core.App interface, not *base.Base — the activity only needs storage methods (FindRecordById / Save) and tests can therefore drive it with a tests.TestApp without booting a full daemon.
Two resolvers, one decision: when a chainResolver is wired and the caller did NOT pin a specific provider (in.Provider == ""), Deliver uses the per-channel chain (Plivo→Twilio, SES API→SES SMTP, …). When either the chain resolver is absent OR the caller pinned a provider id, Deliver falls back to the single-provider tenant.Resolver — that keeps the legacy "force this exact provider" path working for tests and the platform UI's manual probes.
func NewActivities ¶
func NewActivities(app core.App, resolver *tenant.Resolver) *Activities
NewActivities returns a fresh Activities bound to the app + resolver. chain may be nil — single-provider behavior is preserved.
func NewActivitiesWithChain ¶ added in v1.6.17
func NewActivitiesWithChain(app core.App, resolver *tenant.Resolver, chain *tenant.ChainResolver) *Activities
NewActivitiesWithChain returns an Activities that uses chain when the caller did not pin a provider id, and falls back to resolver otherwise. Either chain or resolver may be nil; at least one must be non-nil for sends to succeed.
func (*Activities) Deliver ¶
func (a *Activities) Deliver(ctx context.Context, in SendInput) (SendResult, error)
Deliver is the only activity. It does five things in order:
- Look up the message row (created by the route handler before dispatch).
- Resolve a Notifier via tenant.Resolver.
- Call the library Send.
- Update message status (sent / failed).
- Write a meter row.
Idempotency on retries: step 1 reloads the row; if it's already in a terminal state (sent / failed), the activity returns the current result without re-sending. This handles tasks server replays without double-firing the SMS.
type Config ¶
type Config struct {
// Address is the tasks server ZAP endpoint (host:port).
Address string
// Namespace is the tasks namespace this worker serves.
Namespace string
// TaskQueue is the queue both the workflow and activities sit on.
TaskQueue string
}
Config is the worker's runtime configuration. Mirrors auto's engine.Config.
type Dispatcher ¶ added in v1.6.6
type Dispatcher interface {
// Dispatch enqueues a NotifySendWorkflow execution and returns the
// task id. Implementations must return a non-empty id on success.
Dispatch(ctx context.Context, req SendInput) (string, error)
// Started reports whether the dispatcher is ready to accept work.
// Routes use this to fail-closed (503) when async is requested
// against an unstarted worker.
Started() bool
}
Dispatcher is the narrow surface the HTTP routes need to enqueue an async send. *Worker satisfies it; tests pass an in-memory stub so the send handler can be exercised without dialing tasksd.
type SendInput ¶
type SendInput struct {
MessageID string `json:"message_id"`
TenantSlug string `json:"tenant_slug"`
Channel string `json:"channel"`
Provider string `json:"provider,omitempty"`
To string `json:"to"`
Subject string `json:"subject,omitempty"`
Body string `json:"body"`
TemplateID string `json:"template_id,omitempty"`
Event string `json:"event,omitempty"`
Vars map[string]any `json:"vars,omitempty"`
}
SendInput is the workflow + activity input. One pair per recipient. MessageID is the notifyd-generated message row id the worker updates as the send progresses (queued → sending → sent/failed).
type SendResult ¶
type SendResult struct {
MessageID string `json:"message_id"`
Status string `json:"status"`
Provider string `json:"provider"`
Error string `json:"error,omitempty"`
}
SendResult is the workflow return shape.
func NotifySendWorkflow ¶
func NotifySendWorkflow(ctx workflow.Context, in SendInput) (SendResult, error)
NotifySendWorkflow is the durable-execution entry point for async sends. The workflow body must be deterministic on its inputs — all I/O happens inside the Deliver activity.
Single activity: this is intentional. A "send" is a unit of work, not a DAG; if a provider fails, retry the whole thing rather than trying to split rendering from delivery (rendering against the same vars yields the same body, so re-rendering on retry is free).
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker owns the tasks client + worker lifecycle. Mirrors auto's engine.Worker (we deliberately copy the shape because they're siblings).
func New ¶
func New(cfg Config, acts *Activities) (*Worker, error)
New validates config and returns an un-started worker. The ZAP connection happens in Start so `--help`, `migrate`, and any non-serve command runs without a live tasks server.
func (*Worker) Dispatch ¶
Dispatch enqueues a NotifySendWorkflow execution; returns the task id. The workflow runs asynchronously; clients poll via GET /v1/notify/ messages/{id} (the workflow updates that record).
func (*Worker) Start ¶
Start dials the tasks server, registers the workflow + activities, and starts the poll loop. Idempotent.
Source Files