delivery

package
v1.1.0 Latest Latest
Warning

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

Go to latest
Published: Jul 21, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

Package delivery implements outbound delivery feedback (decision 9 / Slice 4b): the async delivery lifecycle SES reports per message and per recipient, plus a per-tenant suppression list. The SES notifications are ingested over SNS (see sns.go / consumer.go); the AWS surface stays at the edge so the transition/suppression logic is testable without AWS.

Index

Constants

View Source
const (
	EventEmailDelivered  = "email.delivered"
	EventEmailBounced    = "email.bounced"
	EventEmailComplained = "email.complained"
	// EventEmailFailed is the canonical stable terminal-failure event
	// (webhookpub.EventEmailFailed — string-duplicated so this package stays a
	// light leaf). The SES Reject path emits it MESSAGE-level, not per
	// recipient; see Process.
	EventEmailFailed      = "email.failed"
	EventSuppressionAdded = "domain.suppression_added" // account-scoped despite the prefix (design vocab)

)

Event push types for delivery outcomes (decision 9 vocabulary).

View Source
const MessageIDHeader = "X-E2A-Message-ID"

MessageIDHeader is the stable e2a correlation marker (async-send-contract §3.1 / async-message-pipeline §4): stamped on the outbound wire message at submit time (internal/outbound.Sender.SubmitOnce) and echoed back by SES in notification payloads (`mail.headers`) when the SES configuration set has "include original headers" enabled. SES overrides supplied Message-ID/Date headers on the wire, so this custom header is the only submit-time value that survives into notifications — it correlates feedback for the SMTP-accept↔mark-sent crash window, where the provider_message_id from the 250 response was never captured.

Variables

This section is empty.

Functions

func ConfirmSubscriptionURL

func ConfirmSubscriptionURL(m *SNSMessage) (string, bool)

ConfirmSubscriptionURL returns the SubscribeURL to GET (out of band) to confirm a subscription, but only for a SubscriptionConfirmation envelope whose SubscribeURL is an allow-listed HTTPS sns.*.amazonaws.com host (SSRF guard). It does not perform the GET — the caller does, after Verify has passed.

func HTTPCertFetcher

func HTTPCertFetcher(ctx context.Context, certURL string) ([]byte, error)

HTTPCertFetcher is the production CertFetcher: an HTTPS GET with a short timeout that returns the PEM cert bytes. It rejects non-2xx responses. The URL must already have passed the host allow-list (see validSigningCertHost).

func Handler

func Handler(v *Verifier, c *Consumer) http.HandlerFunc

Handler returns the HTTP handler for the public SES-over-SNS notifications endpoint. Every request is fail-closed: the SNS signature is verified before anything is acted on (the endpoint is public). It auto-confirms a SubscriptionConfirmation (GET the allow-listed SubscribeURL) and feeds a Notification's SES event to the Consumer.

Responses: 403 on a failed signature; 400 on unparseable JSON; 200 on a handled or safely-ignored message (so SES stops retrying); 500 only when the Consumer hits a real error worth a retry.

Types

type CertFetcher

type CertFetcher func(ctx context.Context, url string) ([]byte, error)

CertFetcher fetches the bytes of an SNS signing certificate (PEM). Injected as a seam so the signature path is testable without network access.

type Consumer

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

Consumer applies a parsed SES Event to the store: per-recipient transitions, delivery events, and auto-suppression. It is idempotent end-to-end (safe to process the same SNS notification twice — at-least-once delivery).

func NewConsumer

func NewConsumer(store Store, fire Firer) *Consumer

NewConsumer builds the consumer. fire may be nil (no events).

func (*Consumer) Process

func (c *Consumer) Process(ctx context.Context, ev *Event) error

Process applies one normalized SES event. Unknown/uncorrelated messages are no-ops (returns nil) — an SNS notification e2a can't act on must still be ACKed so SES stops retrying. A correlated event with no recipient outcomes (e.g. an SES Send) still records provider-accept evidence before returning.

type CorrelatedMessage

type CorrelatedMessage struct {
	MessageID string
	UserID    string
	// AgentID is the agent's own address (an agent identity's id IS its
	// email), so it doubles as agent_email on event payloads.
	AgentID        string
	Subject        string
	ConversationID string
	Method         string
	MessageType    string
	From           string
	To             []string
	CC             []string
	BCC            []string
}

CorrelatedMessage is CorrelateBySESMessageID's result: the outbound message a SES provider id maps to, plus the message fields the canonical event payloads need — all columns of the same row/join the correlation SELECT already reads, so widening it costs no extra query.

type Event

type Event struct {
	Kind         EventKind
	SESMessageID string // the mail.messageId — correlates to messages.provider_message_id
	// E2AMessageID is the e2a message id echoed back by SES from the
	// MessageIDHeader stamped at submit time (`mail.headers` — present only
	// when the SES configuration set enables "include original headers").
	// It is the correlation fallback for the SMTP-accept↔mark-sent crash
	// window; empty when headers are absent or the marker isn't among them.
	E2AMessageID string
	Recipients   []RecipientOutcome
	// BounceType / BounceSubType carry the SES bounce classification (Bounce
	// events only; empty otherwise). BounceType is normalized to the stable
	// event vocabulary — permanent | transient | undetermined — the value
	// email.bounced's bounce_type field emits; BounceSubType is the raw SES
	// bounceSubType (e.g. General, NoEmail, MailboxFull).
	BounceType    string
	BounceSubType string
}

Event is a normalized SES notification: which message, which event, and the per-recipient outcomes. The message rollup is the worst status across Recipients by Merge precedence.

func ParseSESNotification

func ParseSESNotification(messageBody []byte) (*Event, error)

ParseSESNotification parses the SES event JSON (the decoded SNS Message body) into a normalized Event. Returns an error for malformed JSON or a missing message id; an unrecognized event kind yields KindOther with no recipient outcomes (caller no-ops).

type EventKind

type EventKind string

EventKind is the normalized SES event category.

const (
	KindDelivery      EventKind = "Delivery"
	KindBounce        EventKind = "Bounce"
	KindComplaint     EventKind = "Complaint"
	KindDeliveryDelay EventKind = "DeliveryDelay"
	KindSend          EventKind = "Send"
	KindReject        EventKind = "Reject"
	KindOther         EventKind = "Other"
)

type FailureSource

type FailureSource string

FailureSource is the provenance of a terminal `failed`: who established it. It is recorded on the message row (messages.delivery_failure_source) when `failed` is written and gates the async-send-contract §3.1 correction rule — only a locally inferred failure may be corrected by later provider evidence.

const (
	// FailureSourceLocal marks a failure e2a inferred without the provider
	// confirming a rejection: retries exhausted on ambiguous errors, the
	// outage retry horizon elapsed, the terminal reconciler swept a dead job,
	// or a queued send was canceled by trash. These are exactly the failures
	// the SMTP-accept↔mark-sent crash window can falsify, so they remain
	// correctable by authoritatively correlated provider-accept evidence.
	FailureSourceLocal FailureSource = "local"
	// FailureSourceProvider marks a failure the provider itself confirmed
	// (permanent SMTP 5xx on submit, SES Reject notification). The provider
	// told us it did not accept this submission — never corrected.
	FailureSourceProvider FailureSource = "provider"
)

func (FailureSource) Correctable

func (fs FailureSource) Correctable() bool

Correctable reports whether a stored `failed` with this provenance may be corrected by authoritatively correlated provider feedback (§3.1). An unknown/empty provenance (rows failed before provenance existed) is treated as locally inferred: those legacy rows are precisely the falsely-failed crash-window rows the correction rule exists for, and correction still requires authoritative per-message correlation plus envelope membership, so unrelated feedback can never revive a genuine failure through this default.

type FiredEvent

type FiredEvent struct {
	UserID         string
	AgentID        string
	ConversationID string
	MessageID      string
	Type           string
	Data           any
	DedupKey       string
}

FiredEvent is the set of fields delivery hands its Firer for one webhook event. Data is the canonical typed payload (eventpayload.EmailDeliveredData / EmailBouncedData / EmailComplainedData / EmailFailedData / DomainSuppressionAddedData).

AgentID, ConversationID, and MessageID are the ENVELOPE ROUTING KEYS: they let subscribers' agent_ids/labels filters match AND they populate the webhook_events row's agent_id/conversation_id/message_id columns, which is what makes the persisted event findable via GET /v1/events?message_id= / ?conversation_id= — the reconciliation query an integrator uses to learn a specific message's fate. A message-backed delivery-feedback event must set all three (they mirror the send-worker path's envelope, giving full payload+envelope parity); account-scoped events (suppression) leave them empty. DedupKey makes redeliveries idempotent (the publisher derives a stable event id from it, independent of the routing keys).

type Firer

type Firer func(ctx context.Context, e FiredEvent)

Firer publishes one delivery/suppression webhook event to the owning user's subscribers. Injected as a closure so this package does not depend on webhookpub.

type RecipientOutcome

type RecipientOutcome struct {
	Address  string
	Status   Status
	Detail   string
	Suppress bool // hard bounce or complaint → add to the suppression list
}

RecipientOutcome is the delivery result for one address from one SES event.

type SNSMessage

type SNSMessage struct {
	Type             string `json:"Type"` // Notification | SubscriptionConfirmation | UnsubscribeConfirmation
	MessageId        string `json:"MessageId"`
	TopicArn         string `json:"TopicArn"`
	Subject          string `json:"Subject,omitempty"` // Notifications only, optional
	Message          string `json:"Message"`
	Timestamp        string `json:"Timestamp"`
	SignatureVersion string `json:"SignatureVersion"`
	Signature        string `json:"Signature"`
	SigningCertURL   string `json:"SigningCertURL"`
	// SubscribeURL and Token are present only on SubscriptionConfirmation and
	// UnsubscribeConfirmation envelopes.
	SubscribeURL string `json:"SubscribeURL,omitempty"`
	Token        string `json:"Token,omitempty"`
}

SNSMessage is the standard Amazon SNS HTTP/S POST body. The SES delivery notifications arrive wrapped in this envelope; Message carries the SES event JSON (see ParseSESNotification). Field tags match the SNS wire format.

type Status

type Status string

Status is the outbound delivery lifecycle of a message (or a single recipient of it). It maps 1:1 onto messages.delivery_status / message_recipients.status.

const (
	// StatusAccepted is the async-pipeline entry state (async-send-contract.md
	// §3.1): durably persisted + queued for submission, no network I/O yet.
	// Replaces the legacy StatusQueued, which was never emitted in production.
	StatusAccepted Status = "accepted"
	// StatusSending means a worker holds the lease and is submitting to SES.
	StatusSending    Status = "sending"
	StatusQueued     Status = "queued"     // DEPRECATED legacy alias for accepted; kept so any historical row stays Valid()
	StatusSent       Status = "sent"       // accepted by the relay (SES) — NON-terminal
	StatusDeferred   Status = "deferred"   // transient delay (SES deliveryDelay); poll, no event
	StatusDelivered  Status = "delivered"  // SES confirmed delivery to the recipient MTA
	StatusBounced    Status = "bounced"    // hard/soft bounce
	StatusComplained Status = "complained" // recipient marked spam (FBL complaint)
	StatusFailed     Status = "failed"     // terminal send failure (retries exhausted / permanent reject)
)

func Merge

func Merge(current, incoming Status) Status

Merge returns the status that should result from observing `incoming` while currently at `current`, applying the monotonic precedence: the higher-ranked status wins, so a transition only ever moves "up". An unknown current (e.g. empty) is treated as the lowest rank so any valid incoming status applies.

func ResolveMessageRollup

func ResolveMessageRollup(current Status, source FailureSource, rollup Status) Status

ResolveMessageRollup decides the messages.delivery_status that results from re-rolling up per-recipient provider feedback while the row currently holds `current` with failure provenance `source`. It implements the async-send-contract §3.1 correction exception:

  • a non-failed current keeps today's semantics — the recipient rollup is authoritative;
  • a locally inferred (or legacy unknown-provenance) `failed` is CORRECTED by a rollup that proves the provider accepted the submission (the falsely-declared terminal failure from a final-attempt crash);
  • a provider-confirmed `failed` is never revived.

Callers must only feed it rollups computed from authoritatively correlated feedback (provider_message_id match or the SNS-verified X-E2A-Message-ID header echo) for the message's own recipients — correlation strength is the caller's gate, provenance is this function's.

func (Status) ProvesProviderAcceptance

func (s Status) ProvesProviderAcceptance() bool

ProvesProviderAcceptance reports whether s is provider feedback that implies the provider accepted the submission: `sent` and every per-recipient outcome that can only follow an accepted submission. `failed` is excluded (an SES Reject explicitly means the submission was NOT accepted), as are the local pre-send states.

func (Status) Terminal

func (s Status) Terminal() bool

Terminal reports whether s is a final state that should never transition further. (`sent`/`queued`/`deferred` are non-terminal; the rest are final.)

func (Status) Valid

func (s Status) Valid() bool

Valid reports whether s is a known status.

type Store

type Store interface {
	// CorrelateBySESMessageID finds the outbound message + owning user + agent
	// (plus the message fields the event payloads need — same single SELECT,
	// no extra query) by the SES-assigned provider_message_id captured at send
	// time. found=false when the id is unknown (message deleted, or an event
	// for another deployment).
	CorrelateBySESMessageID(ctx context.Context, sesMessageID string) (m *CorrelatedMessage, found bool, err error)
	// CorrelateByE2AMessageID finds the outbound message by the e2a message id
	// SES echoed back from the MessageIDHeader stamped at submit time — the
	// correlation fallback for the SMTP-accept↔mark-sent crash window, where
	// the provider id from the 250 response was never captured. Same return
	// shape as CorrelateBySESMessageID; found=false for unknown/non-outbound.
	CorrelateByE2AMessageID(ctx context.Context, e2aMessageID string) (m *CorrelatedMessage, found bool, err error)
	// RecordProviderAcceptEvidence durably notes that SES reported having
	// accepted this message's submission (any correlated post-acceptance
	// notification kind) and repairs a provider_message_id lost to the
	// crash window. Idempotent. The worker/terminal-reconciler guards read
	// this evidence before declaring a terminal failure.
	RecordProviderAcceptEvidence(ctx context.Context, messageID, sesMessageID string) error
	// RecordDeliveryOutcome upserts the per-recipient status monotonically and
	// recomputes the message rollup (worst status by precedence). Idempotent:
	// a duplicate/older event is a no-op.
	RecordDeliveryOutcome(ctx context.Context, messageID, address string, status Status, detail string) error
	// AddSuppression idempotently inserts a (user, address) suppression.
	// added=false when it already existed (so the event fires at most once).
	AddSuppression(ctx context.Context, userID, address, reason, source, sourceMessageID string) (added bool, err error)
}

Store is the narrow persistence surface the consumer needs. *identity.Store satisfies it.

type Verifier

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

Verifier validates the authenticity of an SNS message. Because the SES event ingress is a public, unauthenticated HTTP endpoint, signature verification is a fail-closed security boundary: any failure rejects the message.

func NewVerifier

func NewVerifier(allowedTopicARNs []string, fetch CertFetcher) *Verifier

NewVerifier builds a Verifier. allowedTopicARNs is the set of SNS topics e2a trusts; an empty set rejects every message (fail closed). fetch retrieves the signing certificate (use HTTPCertFetcher in production).

func (*Verifier) IsSubscriptionConfirmation

func (v *Verifier) IsSubscriptionConfirmation(m *SNSMessage) bool

IsSubscriptionConfirmation reports whether m is a topic subscription handshake that the HTTP handler should confirm.

func (*Verifier) Verify

func (v *Verifier) Verify(ctx context.Context, m *SNSMessage) error

Verify checks an SNS message fail-closed and returns nil only if it is authentic and from a trusted topic. The checks run in order so the cheap, no-network guards (topic allow-list, URL host) reject before any fetch.

Jump to

Keyboard shortcuts

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