senderidentity

type Config ¶

type Config struct {
	// MaxReconcileAttempts bounds the pending→failed TTL. 0 → default.
	MaxReconcileAttempts int
	// ReaperInterval overrides the orphan-sweep cadence. 0 → default.
	ReaperInterval time.Duration
}

Config tunes the Manager. Zero values get sane defaults.

type DNSRecord ¶

type DNSRecord struct {
	Type  string `json:"type"`  // "TXT" | "CNAME" | "MX"
	Name  string `json:"name"`  // record host
	Value string `json:"value"` // record value
}

DNSRecord is a single record the customer must publish for the sending identity. With BYODKIM the customer already published the per-domain DKIM record during register/verify; this now carries the custom MAIL FROM subdomain's MX + SPF records (Return-Path alignment — see ses.go mailFromRecords) and surfaces anything SES reports as still-required.

type DeprovisionArgs ¶

type DeprovisionArgs struct {
	Domain string `json:"domain"`
}

func (DeprovisionArgs) Kind() string

type DeprovisionWorker ¶

type DeprovisionWorker struct {
	river.WorkerDefaults[DeprovisionArgs]
	// contains filtered or unexported fields
}

DeprovisionWorker removes the SES sending identity on domain/account delete. Idempotent: the provider treats a missing identity as success.

func (w *DeprovisionWorker) Work(ctx context.Context, job *river.Job[DeprovisionArgs]) error

type EventFirer ¶

type EventFirer func(ctx context.Context, domain, userID string, status Status, errMsg string)

EventFirer publishes a domain.sending_verified / domain.sending_failed event. Injected as a closure so this package doesn't depend on webhookpub. userID is the domain owner; a nil firer (tests) is a no-op.

type FakeProvider ¶

type FakeProvider struct {
	ProvisionCalls   []string
	StatusCalls      []string
	DeprovisionCalls []string
	// contains filtered or unexported fields
}

FakeProvider is an in-memory Provider for tests. It is concurrency-safe (the workers call it from River goroutines). Configure per-domain behavior with the setters; inspect calls with the recorded slices.

Default behavior with no configuration: Provision returns StatusPending, Status returns StatusPending forever, Deprovision succeeds. Tests that want a domain to verify call SetStatusSequence or SetStatus.

func NewFakeProvider() *FakeProvider

func (f *FakeProvider) Deprovision(ctx context.Context, domain string) error

func (f *FakeProvider) List(ctx context.Context) ([]string, error)

func (f *FakeProvider) Provision(ctx context.Context, domain, dkimSelector string, dkimPrivateKeyDER []byte) (Result, error)

func (f *FakeProvider) SeedIdentity(domain string)

func (f *FakeProvider) SetDeprovisionErr(err error)

func (f *FakeProvider) SetProvisionErr(err error)

func (f *FakeProvider) SetStatus(domain string, r Result)

func (f *FakeProvider) SetStatusErr(domain string, err error)

func (f *FakeProvider) SetStatusNotFound(domain string)

func (f *FakeProvider) SetStatusSequence(domain string, seq ...Result)

func (f *FakeProvider) Status(ctx context.Context, domain string) (Result, error)

type Manager ¶

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

Manager owns the sender-identity job lifecycle on the SHARED River client (internal/jobs), instead of a private client. It is a jobs.Registrar — it contributes the provision/reconcile/deprovision workers + the periodic orphan reaper — and the app's enqueue entry point: EnqueueProvision on domain verify, EnqueueDeprovisionTx in the domain-delete tx. The shared client is injected via SetEnqueuer after jobs.New has built it (which needs this Manager as a Registrar first — the standard two-phase wiring).

func NewManager(store Store, provider Provider, fire EventFirer, cfg Config) *Manager

func (m *Manager) EnqueueDeprovisionTx(ctx context.Context, tx pgx.Tx, domain string) error

func (m *Manager) EnqueueProvision(ctx context.Context, domain string) error

func (m *Manager) RegisterJobs(w *river.Workers) []*river.PeriodicJob

func (m *Manager) SetEnqueuer(e jobs.Enqueuer)

type Provider ¶

type Provider interface {
	// Provision registers a BYODKIM sending identity for domain, supplying
	// the per-domain DKIM selector + PKCS#1 DER private key that e2a
	// already generated (so DKIM d= aligns with the From domain). Returns
	// the initial Result — typically StatusPending — or an error to retry.
	Provision(ctx context.Context, domain, dkimSelector string, dkimPrivateKeyDER []byte) (Result, error)

	// Status polls the current verification state from the provider.
	// Returns ErrIdentityNotFound if no identity exists for domain.
	Status(ctx context.Context, domain string) (Result, error)

	// Deprovision removes the sending identity. A missing identity MUST be
	// reported as success (idempotent teardown).
	Deprovision(ctx context.Context, domain string) error

	// List returns the domains for which e2a currently has a sending
	// identity at the provider. Used by the orphan reaper to alert on
	// identities with no backing live domain row. SES caps each page; the
	// implementation paginates and returns the full set.
	List(ctx context.Context) ([]string, error)
}

Provider registers, polls, and removes the upstream (SES) sending identity for a domain. Implementations MUST be idempotent: Provision on an already-registered domain is a no-op success, and Deprovision treats a missing identity as success.

type ProvisionArgs ¶

type ProvisionArgs struct {
	Domain string `json:"domain"`
}

func (ProvisionArgs) Kind() string

type ProvisionWorker ¶

type ProvisionWorker struct {
	river.WorkerDefaults[ProvisionArgs]
	// contains filtered or unexported fields
}

ProvisionWorker registers the SES sending identity (BYODKIM) for a domain and, on success, enqueues a reconcile job to poll it to verified.

func (w *ProvisionWorker) Work(ctx context.Context, job *river.Job[ProvisionArgs]) error

type RawStore ¶

type RawStore interface {
	SendingProvisionInputs(ctx context.Context, domain string) (selector string, privateKeyDER []byte, ok bool, err error)
	SetSendingStatus(ctx context.Context, domain, status, dkimStatus, mailFromStatus, errMsg string, recordsJSON []byte) error
	TouchSendingChecked(ctx context.Context, domain string) error
	GetSendingStatus(ctx context.Context, domain string) (string, error)
	DomainOwner(ctx context.Context, domain string) (string, error)
	DomainExists(ctx context.Context, domain string) (bool, error)
}

RawStore is the primitive persistence surface implemented by *identity.Store. It deliberately speaks plain strings / JSON bytes so the core identity package does NOT import senderidentity (and thus does not pull River + the AWS SDK into its dependency graph). NewStoreAdapter wraps it into the typed Store the workers consume.

type ReapArgs ¶

type ReapArgs struct{}

ReapArgs drives the periodic orphan-identity sweep. No fields — it scans the whole provider identity list each run.

func (ReapArgs) Kind() string

type ReapWorker ¶

type ReapWorker struct {
	river.WorkerDefaults[ReapArgs]
	// contains filtered or unexported fields
}

ReapWorker is the teardown BACKSTOP (design decision 4). The primary teardown is the transactional DeprovisionWorker enqueued on domain/account delete; this sweep catches identities orphaned by edge cases (e.g. a delete whose job was somehow lost). It is intentionally ALERT-ONLY: a naïve "list then delete" races a concurrent re-registration of the same domain (stale snapshot deletes a freshly-created identity → silent breakage), so the safe default is to log orphans for an operator to inspect. The TOCTOU-safe conditional delete (SELECT … FOR UPDATE liveness re-confirm) is deferred — see the design doc.

func (w *ReapWorker) Work(ctx context.Context, job *river.Job[ReapArgs]) error

type ReconcileArgs ¶

type ReconcileArgs struct {
	Domain string `json:"domain"`
}

func (ReconcileArgs) Kind() string

type ReconcileWorker ¶

type ReconcileWorker struct {
	river.WorkerDefaults[ReconcileArgs]
	// contains filtered or unexported fields
}

ReconcileWorker polls SES for a pending domain and transitions it to verified/failed. While still pending it returns errStillPending so River retries with backoff; once the attempt budget is exhausted it marks the domain failed (bounded TTL — no infinite poll).

func (w *ReconcileWorker) Work(ctx context.Context, job *river.Job[ReconcileArgs]) error

type Result ¶

type Result struct {
	Status         Status      `json:"status"`
	DkimStatus     Status      `json:"dkim_status,omitempty"`
	MailFromStatus Status      `json:"mail_from_status,omitempty"`
	Error          string      `json:"error,omitempty"`
	DNSRecords     []DNSRecord `json:"dns_records,omitempty"`
}

Result is what a Provider reports for a domain.

Status is the all-or-nothing rollup (mapSESStatus): `verified` only when EVERY sending axis is good. DkimStatus and MailFromStatus are the per-axis breakdown SES reports independently (DkimAttributes.Status and MailFromAttributes.MailFromDomainStatus), so a domain with good DKIM but a broken custom MAIL FROM surfaces as DkimStatus=verified + MailFromStatus=failed while the rollup Status stays `failed`. They are empty ("") when the provider has no per-axis signal (e.g. Provision, which only registers the identity); consumers fall back to the rollup in that case.

type SESProvider ¶

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

SESProvider is the real Provider backed by AWS SES v2. It registers domain sending identities with BYODKIM, reusing e2a's per-domain DKIM key so the DKIM d= aligns with the From domain (DMARC passes on DKIM alignment), and configures a custom MAIL FROM subdomain (bounce.<domain>) so the Return-Path aligns too (SPF passes on the From org-domain → no "via e2a"). NOTE: only exercised against live AWS — CI/tests use FakeProvider; the status-mapping helpers below are unit-tested with a stub.

func NewSESProvider(api sesAPI, region string) *SESProvider

func NewSESProviderFromConfig(ctx context.Context, region string) (*SESProvider, error)

func (p *SESProvider) Deprovision(ctx context.Context, domain string) error

func (p *SESProvider) List(ctx context.Context) ([]string, error)

func (p *SESProvider) Provision(ctx context.Context, domain, dkimSelector string, dkimPrivateKeyDER []byte) (Result, error)

func (p *SESProvider) Status(ctx context.Context, domain string) (Result, error)

type Status ¶

type Status string

Status is the verification state of a domain's sending identity. It maps 1:1 onto the domains.sending_status column.

const (
	// StatusNone — no sending identity registered. Default for every
	// domain; self-host / SES-not-configured deployments stay here, which
	// keeps outbound on the relay From (fail-closed).
	StatusNone Status = "none"
	// StatusPending — identity registered with SES (BYODKIM), awaiting
	// asynchronous verification. The reconciler polls until it resolves.
	StatusPending Status = "pending"
	// StatusVerified — SES confirmed the identity; own-address From is now
	// used for this domain's agents.
	StatusVerified Status = "verified"
	// StatusFailed — verification failed, or `pending` exceeded its TTL.
	// Carries an actionable reason; outbound stays on the relay From.
	StatusFailed Status = "failed"
)

func (s Status) Valid() bool

type Store ¶

type Store interface {
	// SendingProvisionInputs returns the per-domain DKIM selector + PKCS#1
	// DER private key used for BYODKIM. ok=false means the domain has no key
	// material (can't provision). err is for real DB failures (retryable).
	SendingProvisionInputs(ctx context.Context, domain string) (selector string, privateKeyDER []byte, ok bool, err error)
	// SetSendingStatus writes a terminal/transition status (+ the per-axis
	// dkim/mailFrom breakdown + error + DNS records) and stamps
	// sending_last_checked_at. dkimStatus/mailFromStatus may be empty ("")
	// when the caller has no per-axis signal (e.g. provision, or a terminal
	// failure with no SES poll); persisting empty lets the read path fall back
	// to the all-or-nothing rollup.
	SetSendingStatus(ctx context.Context, domain string, status, dkimStatus, mailFromStatus Status, errMsg string, records []DNSRecord) error
	// TouchSendingChecked stamps sending_last_checked_at without changing the
	// status — used on a still-pending poll.
	TouchSendingChecked(ctx context.Context, domain string) error
	// GetSendingStatus reads the current status. Returns pgx.ErrNoRows when
	// the domain row is gone (deleted mid-flight) — workers treat that as
	// "nothing to do".
	GetSendingStatus(ctx context.Context, domain string) (Status, error)
	// DomainOwner returns the user_id owning the domain, for event routing.
	// Empty string (e.g. the system shared domain) means "no owner" → no event.
	DomainOwner(ctx context.Context, domain string) (string, error)
	// DomainExists reports whether a live domain row exists. Used by the
	// orphan reaper to decide if a provider identity is backed.
	DomainExists(ctx context.Context, domain string) (bool, error)
}

Store is the narrow persistence surface the workers need. *identity.Store satisfies it. Kept minimal so the workers don't depend on the whole store.

func NewStoreAdapter(raw RawStore) Store