identity

func (*AgentIdentity) ActualDomain ¶

func (a *AgentIdentity) ActualDomain() string

ActualDomain returns the exact domain present in the agent's email address. It may differ from RegisteredDomain for an inherited subdomain agent.

func (*AgentIdentity) EmailAddress ¶

func (a *AgentIdentity) EmailAddress() string

EmailAddress returns the agent's email address (always the ID).

func (*AgentIdentity) IsSharedDomain ¶

func (a *AgentIdentity) IsSharedDomain(sharedDomain string) bool

IsSharedDomain returns true if the agent's domain matches the configured shared domain (the host that backs slug-based registration). When sharedDomain is empty, the deployment has slug registration disabled and no agent can be on the shared domain.

func (*AgentIdentity) RegisteredDomainName ¶

func (a *AgentIdentity) RegisteredDomainName() string

RegisteredDomainName returns the explicitly registered domain identity that authorizes this agent. It is the DNS/DKIM/sending-state and lifecycle parent. The fallback preserves callers that construct exact-domain identities in memory without populating RegisteredDomain.

func (Message) MarshalJSON ¶

func (m Message) MarshalJSON() ([]byte, error)

MarshalJSON preserves the database-friendly string representation while exposing unavailable inbound identity fields as JSON null. These fields are required-but-nullable in the public export contract.

func (*Message) ThreadMessageID ¶

func (m *Message) ThreadMessageID() string

ThreadMessageID returns the RFC 5322 Message-ID to anchor a reply's In-Reply-To / References on. An inbound message carries the sender's Message-ID in email_message_id; an outbound message the agent sent has no email_message_id (the composer omits Message-ID — see internal/outbound/compose.go) and instead carries the relay/SES-assigned Message-ID, angle-bracketed, in provider_message_id. Threading off the wrong field forks the recipient's mail thread, so callers replying to their own outbound must use this rather than EmailMessageID directly.

func (PendingApprovalEdit) Apply ¶

func (e PendingApprovalEdit) Apply(msg *Message) bool

Apply mutates msg to reflect any fields the reviewer changed. Returns true if any field was actually different from what msg already held (signals the edited flag should be set).

func (*Store) AddAgentSuppression ¶

func (s *Store) AddAgentSuppression(ctx context.Context, userID, agentID, address, reason, source string, onAdded AgentSuppressionTxHook) (sp AgentSuppression, added bool, err error)

AddAgentSuppression idempotently adds an agent-scoped recipient block.

func (*Store) AddAgentSuppressionFromTokenScope ¶

func (s *Store) AddAgentSuppressionFromTokenScope(ctx context.Context, scope UnsubscribeScope, onAdded AgentSuppressionTxHook) (AgentSuppression, bool, error)

AddAgentSuppressionFromTokenScope records consent from an already resolved bearer-token scope. Unlike AddAgentSuppression, it intentionally does not require a live agent: a token issued while the agent existed remains valid after hard deletion. Callers must obtain scope from ResolveUnsubscribeToken; source and reason are fixed so this bypass cannot masquerade as a manual row.

func (*Store) AddSuppression ¶

func (s *Store) AddSuppression(ctx context.Context, userID, address, reason, source, sourceMessageID string) (bool, error)

AddSuppression idempotently inserts a (user, address) suppression. added is false when it already existed, so the caller fires domain.suppression_added at most once per address.

func (*Store) AdoptSharedDomain ¶

func (s *Store) AdoptSharedDomain(ctx context.Context, domain, userID string) (*Domain, error)

AdoptSharedDomain assigns ownership of the server-seeded shared-domain row to userID. The server seeds that row on every boot via EnsureSharedDomain (user_id NULL, verified true, ON CONFLICT DO NOTHING); ClaimOrCreateDomain cannot claim it — it only upserts an unverified, same-user row — yet the probe/system identity that seeds itself lives on the shared domain by design.

The UPDATE is guarded to a VERIFIED, currently-unowned row (or one already owned by userID, so re-seeding is idempotent). Requiring verified=true — the signature EnsureSharedDomain stamps — keeps this method from ever adopting some other ownerless row, so its safety does not rely on the domains schema never producing another NULL-owner row nor on the caller passing only a trusted domain. A row owned by a different account, a nonexistent row, or an ownerless-but-unverified row all match nothing and return ErrDomainTaken. The verified flag and DKIM columns are not modified.

func (*Store) ApproveAndAccept ¶

func (s *Store) ApproveAndAccept(
	ctx context.Context,
	messageID, reviewedByUserID, targetStatus string,
	edited bool,
	acc AcceptedSend,
	enqueue func(ctx context.Context, tx pgx.Tx, messageID string) (int64, error),
	completeIdempotency func(ctx context.Context, tx pgx.Tx, approved *Message) error,
) (*Message, error)

ApproveAndAccept resolves a pending_review outbound hold to an APPROVED, ASYNC-QUEUED state in one transaction, mirroring the API's async accept-tx: it flips status to targetStatus (review_approved for a human approve, review_expired_approved for the TTL sweep) AND delivery_status to 'accepted', persists the composed bytes + envelope, then enqueues the outbound_send job and stamps its id. The existing SendWorker picks the row up by id and performs the actual SMTP submit + email.sent/failed + metering; this method does NOT send and does NOT use the send_attempts gate (async idempotency is the accept-tx atomicity + the worker's delivery_status/alreadyDone guard). reviewedByUserID is "" (→ NULL) for the sweep. When completeIdempotency is non-nil, it runs after the send job is stamped but before commit; an error rolls back the approval, job, and key.

The WHERE status='pending_review' is the compare-and-set guard: RETURNING no row means a human/other worker already resolved the hold → ErrNotPendingApproval (a no-op for the caller). The agent-not-trashed guard sits in the same WHERE so it is atomic with the CAS: a hold whose agent was trashed after the caller's pre-checks (the TTL sweep's candidate SELECT, or a human path's ownership load) must NOT resolve — trashed holds stay pending_review with their clock paused until RestoreAgent shifts approval_expires_at (or the trash purge drops them). Draft body and attachment columns remain retained after approval.

func (*Store) ApproveAndDeliverLocal ¶

func (s *Store) ApproveAndDeliverLocal(
	ctx context.Context,
	messageID, userID string,
	edits PendingApprovalEdit,
	compose func(msg *Message) (SendResult, error),
	beforeCommit LocalDeliveryTxHook,
) (*Message, error)

ApproveAndDeliverLocal atomically resolves a pending outbound review hold whose only recipient is a mailbox owned by this service. Unlike ApproveAndSend, compose must be a local, side-effect-free operation: the outbound update, recipient-side insert, events, and idempotency completion all commit or roll back together, so the SES-oriented send_attempts journal is neither needed nor appropriate.

func (*Store) ApproveAndSend ¶

func (s *Store) ApproveAndSend(
	ctx context.Context,
	messageID, userID string,
	edits PendingApprovalEdit,
	send func(msg *Message) (SendResult, error),
) (*Message, error)

ApproveAndSend finalizes a pending_review message by running it through a caller-supplied send function inside a transaction that holds a row lock on the pending row. If send returns an error the transaction rolls back and the message remains pending. On success the row is updated to 'sent' with the provider-assigned Message-ID while body and attachment columns remain retained.

edits, if any fields are populated, are applied to the in-memory message before send is called and the 'edited' column is set to true when any field differs from what was stored. Approval-via-magic-link callers pass the zero edits value.

Ownership is enforced by the agent -> user join. Messages owned by another user return ErrMessageNotFound. Messages whose status is not 'pending_review' return ErrNotPendingApproval. If another worker is already mid-send for this message (rare; only possible after the approval row lock was released without status changing — e.g. a pool drop mid-send), this returns ErrSendInProgress.

Concurrency / failure mode notes:

• The row-level FOR NO KEY UPDATE lock is held on the messages row for the duration of the send callback. In practice that is bounded by outbound.SMTPRelay's per-attempt deadline (2min) plus its internal retry backoff (1s/5s/15s) — worst case ~6.5min of lock on this single row. Other rows are unaffected; deadlock is not possible because only one row is ever locked per call.

  Why NO KEY UPDATE rather than the stricter FOR UPDATE: the send_attempts INSERT below runs on a SEPARATE pool connection and needs a KEY SHARE lock on this messages row for FK enforcement. FOR UPDATE blocks KEY SHARE; FOR NO KEY UPDATE allows it. The downgrade is safe because nothing in this codebase mutates messages.id (the only key column) after creation — all UPDATEs touch non-key columns, which NO KEY UPDATE serializes against itself exactly like FOR UPDATE.

• The old crash window where send() succeeded at SES but the subsequent UPDATE/Commit failed (DB blip, pool exhaustion) is now closed by the send_attempts table. Around send() we run two small auxiliary transactions that outlive the surrounding approval transaction: ClaimSendAttempt before send(), MarkSendSucceeded (or MarkSendFailed) after. If the approval tx rolls back AFTER send() succeeded, the next retry of ApproveAndSend reads send_attempts.status='sent', reuses the recorded SendResult, and skips the upstream send entirely.

func (*Store) ApproveInboundReview ¶

func (s *Store) ApproveInboundReview(ctx context.Context, messageID, agentID, reviewerID string) error

ApproveInboundReview releases a held inbound message to the agent (status review_approved → visible in the inbox). Scoped to agentID for tenant isolation; reviewerID identifies the human. Held content is retained (the message is now delivered).

func (*Store) AutoDisableFailingWebhooks ¶

func (s *Store) AutoDisableFailingWebhooks(ctx context.Context) (int, error)

AutoDisableFailingWebhooks scans for webhooks whose recent delivery history exceeds the failure threshold and flips them to enabled=false with auto_disabled_at = now(). Returns the count of webhooks newly disabled. Designed to be called periodically (e.g. every 5 minutes) from a janitor goroutine.

"Consecutive failed events" is interpreted as: in the last AutoDisableWindow, at least AutoDisableThreshold rows in webhook_subscriber_deliveries reached status='failed' AND zero rows reached status='delivered'. The zero-delivered guard prevents a noisy webhook that's still mostly working from being disabled.

func (*Store) BootstrapUser ¶

func (s *Store) BootstrapUser(ctx context.Context, email string) (*User, error)

BootstrapUser finds a user by email, or creates one with a synthetic google_subject if none exists. Used by the -bootstrap-email CLI flag for self-host first-run, where there's no Google OAuth flow yet.

func (*Store) ClaimOrCreateDomain ¶

func (s *Store) ClaimOrCreateDomain(ctx context.Context, domain, userID string) (*Domain, error)

ClaimOrCreateDomain atomically claims a DNS namespace. A row reserves its exact name plus every ancestor/descendant against other accounts; the same account may explicitly register a child. The verification_token and DKIM keypair are minted on first INSERT and remain stable across re-claims — a caller that has already published the TXT record on DNS (or has mail in flight signed with the DKIM key) isn't silently invalidated by a second call. A different user cannot take over an unverified row; that closes a squatting window where the new owner could verify against a TXT record the original owner already published. The managed bounce.<domain> subtree is reserved once an account owns the parent because SES custom MAIL FROM uses that namespace.

func (*Store) ClaimOutboundForSend ¶

func (s *Store) ClaimOutboundForSend(ctx context.Context, messageID string, jobID int64) (*OutboundSendPayload, error)

ClaimOutboundForSend atomically moves the stamped River job's live outbound message into delivery_status='sending' and returns its provider payload. A retry of the same job may reclaim a row already in sending (River does not run two attempts of one job concurrently); a duplicate job id cannot claim it.

The short transaction is the send/trash linearization point. If trash commits first, the claim misses and the queued send is canceled. If this claim commits first, trash may proceed while provider I/O runs, and MarkOutboundSentTx or MarkOutboundFailedTx still records the terminal result on the trashed row.

func (*Store) ClaimSendAttempt ¶

func (s *Store) ClaimSendAttempt(ctx context.Context, messageID string) (SendAttemptResult, error)

ClaimSendAttempt atomically reserves the send_attempts row for messageID. Concurrency model mirrors internal/idempotency.Claim: one UPSERT with a stale-takeover WHERE clause; the loser does a follow-up SELECT to classify the existing row.

Allowed transitions:

(no row)                              → acquired (fresh INSERT)
status='failed'                       → acquired (UPSERT path takes over)
status='attempting' AND stale         → acquired (UPSERT path takes over)
status='attempting' AND NOT stale     → InFlight (refuse)
status='sent'                         → AlreadySent (reuse recorded result)

Called from ApproveAndSend in a SEPARATE small transaction so the claim row outlives any rollback of the surrounding approval transaction. That's what closes the documented double-send window: once status='sent' is committed here, a retry sees AlreadySent and skips the upstream send even if the messages-row UPDATE inside the approval tx never committed.

func (*Store) ClearExpiredPrevSecrets ¶

func (s *Store) ClearExpiredPrevSecrets(ctx context.Context) (int64, error)

ClearExpiredPrevSecrets nulls signing_secret_prev / signing_secret_prev_expires_at on rows past their grace window. Idempotent. The worker already ignores expired prev secrets at signing time; this janitor is a hygiene pass so GET responses don't carry a meaningless prev_expires_at.

func (*Store) CorrelateByE2AMessageID ¶

func (s *Store) CorrelateByE2AMessageID(ctx context.Context, e2aMessageID string) (*delivery.CorrelatedMessage, bool, error)

CorrelateByE2AMessageID finds the outbound message + owning user/agent by the e2a message id SES echoed back from the X-E2A-Message-ID wire header (delivery.MessageIDHeader) — the §3.1 correlation fallback for the SMTP-accept↔mark-sent crash window, where the provider id from the SMTP 250 was never captured so CorrelateBySESMessageID cannot match. Same return shape as CorrelateBySESMessageID. The id is shape-validated by the caller (internal/delivery) before it reaches this lookup.

func (*Store) CorrelateBySESMessageID ¶

func (s *Store) CorrelateBySESMessageID(ctx context.Context, sesMessageID string) (*delivery.CorrelatedMessage, bool, error)

CorrelateBySESMessageID finds the outbound message + owning user by the SES-assigned provider_message_id captured at send time. found=false when the id is unknown (deleted message, or an event for a different deployment).

The message fields (subject, and the envelope/threading fields the message-level email.failed payload needs on an SES Reject) ride along — they're columns on the very row this query already reads, so including them costs no extra query (contract freeze PR-2: `subject` on delivery events).

The SNS notification carries the BARE SES id (e.g. 010f0193…-000000), but the send path stores it angle-bracketed and sometimes with an @region.amazonses.com suffix (parseMessageIDFromResponse) — same discrepancy LookupConversationID works around. Match all three stored shapes against the bare id: exact, <id>, and <id@…>. SES ids are [A-Za-z0-9-] so they carry no LIKE metacharacters.

func (*Store) CountTemplatesByUser ¶

func (s *Store) CountTemplatesByUser(ctx context.Context, userID string) (int, error)

CountTemplatesByUser returns the number of templates the user owns. Used by CreateTemplate to enforce the per-user cap.

func (*Store) CountWebhooksByUser ¶

func (s *Store) CountWebhooksByUser(ctx context.Context, userID string) (int, error)

CountWebhooksByUser returns the total number of webhooks (enabled + disabled) the user owns. Used by CreateWebhook to enforce the per-user cap from account_limits.max_webhooks.

func (*Store) CreateAPIKey ¶

func (s *Store) CreateAPIKey(ctx context.Context, userID, name string, expiresAt *time.Time) (*APIKey, error)

CreateAPIKey issues a fresh ACCOUNT-scoped API key for the user. expiresAt is the optional hard expiration; pass nil to issue a never-expiring key (the backward-compatible default). This is the account-tier convenience wrapper over CreateScopedAPIKey — the self-host default key.

func (*Store) CreateAgent ¶

func (s *Store) CreateAgent(ctx context.Context, agentEmail, domain, name, webhookURL, agentMode, userID string) (*AgentIdentity, error)

CreateAgent inserts an agent with a domain FK. Does NOT check domain ownership — that's the API handler's responsibility (shared domain skips the check).

webhookURL and agentMode are accepted for signature compatibility but are now IGNORED: the legacy per-agent webhook_url + agent_mode columns were dropped (migration 029). Push is delivered solely via the /v1/webhooks subscriber resource and WebSocket is open to all agents. The params are retained to avoid churning the ~80 call-sites that still pass them; the internal-signature cleanup is a separate follow-up.

func (*Store) CreateAgentTx ¶

func (s *Store) CreateAgentTx(ctx context.Context, tx pgx.Tx, agentEmail, domain, name, webhookURL, agentMode, userID string) (*AgentIdentity, error)

CreateAgentTx inserts an agent inside a caller-owned transaction. Used by the OAuth consent flow so the slug auto-create row and the authorization-code insert (in oauth_auth_codes) commit together — without this, a code-issue failure after the agent commit would leave a phantom inbox the user never authorized. webhookURL and agentMode are accepted but IGNORED — see CreateAgent.

func (*Store) CreateInboundMessage ¶

func (s *Store) CreateInboundMessage(ctx context.Context, id, agentID, senderEmail, recipient, emailMessageID, subject, conversationID, deliveryStatus string, rawMessage []byte, authHeaders map[string]string, authVerdict []byte, flagged bool, flagReason string, toRecipients, cc, replyTo []string, screening InboundScreening) (*Message, error)

CreateInboundMessage stores an inbound message. If id is empty a new one is generated; otherwise the caller's pre-generated ID is used so the upstream signer can bind auth headers to the same ID that gets stored. toRecipients and cc are the parsed To: and Cc: headers from the original RFC 2822 message; recipient is the per-delivery target for this row (may be one of the To: addresses, or absent from the header list when the agent was Bcc'd). replyTo is the parsed Reply-To: header (empty when absent — never silently falls back to sender).

func (*Store) CreateInboundMessageAuthenticated ¶

func (s *Store) CreateInboundMessageAuthenticated(ctx context.Context, id, agentID string, auth InboundAuth, recipient, emailMessageID, subject, conversationID, deliveryStatus string, rawMessage []byte, flagged bool, flagReason string, toRecipients, cc, replyTo []string, screening InboundScreening) (*Message, error)

CreateInboundMessageAuthenticated stores the canonical identity and authentication model. The legacy CreateInboundMessage remains temporarily available while older tests and relay call sites migrate.

func (*Store) CreateInboundMessageAuthenticatedInTx ¶

func (s *Store) CreateInboundMessageAuthenticatedInTx(ctx context.Context, tx pgx.Tx, id, agentID string, auth InboundAuth, recipient, emailMessageID, subject, conversationID, deliveryStatus string, rawMessage []byte, flagged bool, flagReason string, toRecipients, cc, replyTo []string, screening InboundScreening) (*Message, error)

func (*Store) CreateInboundMessageInTx ¶

func (s *Store) CreateInboundMessageInTx(ctx context.Context, tx pgx.Tx, id, agentID, senderEmail, recipient, emailMessageID, subject, conversationID, deliveryStatus string, rawMessage []byte, authHeaders map[string]string, authVerdict []byte, flagged bool, flagReason string, toRecipients, cc, replyTo []string, screening InboundScreening) (*Message, error)

CreateInboundMessageInTx writes the messages row inside the caller's transaction. Used by the slice-3 relay refactor (per design §4.2) so the messages INSERT and the webhook_events outbox INSERT commit together, closing the at-least-once publish-loss window.

Mirrors the CreateAgentTx pattern at store.go:596-607 — same SQL body, executed against either *pgxpool.Pool or pgx.Tx via the messageExecutor interface below.

func (*Store) CreateOrGetUser ¶

func (s *Store) CreateOrGetUser(ctx context.Context, email, name, googleSub string) (*User, error)

func (*Store) CreateOutboundMessage ¶

func (s *Store) CreateOutboundMessage(ctx context.Context, agentID string, toRecipients []string, cc []string, bcc []string, subject, msgType, method, providerMessageID, conversationID string, rawMessage []byte) (*Message, error)

CreateOutboundMessage stores an outbound message with multi-recipient support. The recipient param is kept for backward compat with the singular recipient column; toRecipients, cc, and bcc are the canonical outbound-only multi-recipient fields.

func (*Store) CreateOutboundMessageTx ¶

func (s *Store) CreateOutboundMessageTx(ctx context.Context, tx pgx.Tx, agentID string, toRecipients, cc, bcc []string, subject, msgType, method, providerMessageID, conversationID string, rawMessage []byte, deliveryStatus, envelopeFrom, sentAs string) (*Message, error)

CreateOutboundMessageTx is CreateOutboundMessage on the caller's transaction, for the async accept path (async-message-pipeline.md, slice C). It persists the two-column model: status=MessageStatusSent (the hold/lifecycle column — this row is not held) while delivery_status carries the send progression (the accept-tx passes 'accepted'; the send worker later advances it to 'sent'/'failed'). It also stamps envelope_from + sent_as, decided once at compose time, so the worker can submit the persisted bytes without re-composing. provider_message_id is empty until the worker records the SES id in MarkOutboundSentTx.

func (*Store) CreatePendingOutboundMessage ¶

func (s *Store) CreatePendingOutboundMessage(ctx context.Context, agentID string, toRecipients, cc, bcc []string, subject, bodyText, bodyHTML string, attachmentsJSON []byte, msgType, conversationID, replyToEmailMessageID, replyTo string, ttlSeconds int) (*Message, error)

CreatePendingOutboundMessage stores a fully composed outbound email in pending_review status, including body_text, body_html, and attachments so that approval can reconstruct the original SendRequest (or accept edits) without the caller needing to retain it. ttlSeconds sets how long the message remains pending before the expiration worker resolves it.

replyToEmailMessageID is the RFC 5322 Message-ID of the inbound being replied to (e.g. "<abc@gmail.com>"), or "" for fresh sends and test emails. It reuses the email_message_id column, which is unused for outbound rows in every other path — the column semantically carries "the Message-ID this row references" in both directions.

attachmentsJSON must be a JSON array matching the public Attachment shape ([{filename, content_type, data}, ...]) or nil. Callers that already have an []outbound.Attachment slice should json.Marshal it before passing in.

func (*Store) CreatePendingOutboundMessageManaged ¶

func (s *Store) CreatePendingOutboundMessageManaged(ctx context.Context, agentID string, toRecipients, cc, bcc []string, subject, bodyText, bodyHTML string, attachmentsJSON []byte, msgType, conversationID, replyToEmailMessageID, replyTo string, ttlSeconds int, managedUnsubscribe bool) (*Message, error)

func (*Store) CreatePendingOutboundMessageManagedTx ¶

func (s *Store) CreatePendingOutboundMessageManagedTx(ctx context.Context, tx pgx.Tx, agentID string, toRecipients, cc, bcc []string, subject, bodyText, bodyHTML string, attachmentsJSON []byte, msgType, conversationID, replyToEmailMessageID, replyTo string, ttlSeconds int, managedUnsubscribe bool) (*Message, error)

func (*Store) CreatePendingOutboundMessageTx ¶

func (s *Store) CreatePendingOutboundMessageTx(ctx context.Context, tx pgx.Tx, agentID string, toRecipients, cc, bcc []string, subject, bodyText, bodyHTML string, attachmentsJSON []byte, msgType, conversationID, replyToEmailMessageID, replyTo string, ttlSeconds int) (*Message, error)

CreatePendingOutboundMessageTx is the in-tx sibling of CreatePendingOutboundMessage, letting the HITL hold write the pending_review row and enqueue its approval-notification job (QueueNotify) in ONE transaction so neither can exist without the other (docs/design/hitl-notify-river.md). Mirrors CreateOutboundMessageTx / the send accept-tx.

func (*Store) CreateProtectionEvent ¶

func (s *Store) CreateProtectionEvent(ctx context.Context, ev ProtectionEvent) error

CreateProtectionEvent appends a screening event. The insert is idempotent on the primary key: callers that set a DeterministicProtectionEventID get exactly-once recording across retries. If ev.ID is empty a random id is assigned. A conflicting id is a no-op (returns nil).

func (*Store) CreateScopedAPIKey ¶

func (s *Store) CreateScopedAPIKey(ctx context.Context, userID, name, scope, agentID string, expiresAt *time.Time) (*APIKey, error)

CreateScopedAPIKey issues a fresh API key with an explicit scope (Slice 5a).

• ScopeAccount: account-wide admin; agentID must be empty; prefix e2a_acct_.
• ScopeAgent: bound to agentID (which must be a non-empty agent owned by the user); prefix e2a_agt_. The key can only act as that one agent.

The visible prefix makes a key's blast radius obvious at a glance, and the DB CHECK (scope='agent') == (agent_id IS NOT NULL) backstops the binding.

func (*Store) CreateTemplate ¶

func (s *Store) CreateTemplate(ctx context.Context, userID string, in TemplateCreate) (*Template, error)

CreateTemplate inserts a new template. Returns ErrTemplateLimitReached at the per-user cap and ErrTemplateAliasTaken on a per-user alias collision.

Syntax validation (Parse of all three parts) is the handler's job; the storage layer only enforces the count cap and alias uniqueness. The cap check races like the webhooks one (bounded overshoot of one row under concurrent creates) — acceptable at a cap of 10.

func (*Store) CreateUserSession ¶

func (s *Store) CreateUserSession(ctx context.Context, userID string) (string, error)

func (*Store) CreateWebhook ¶

func (s *Store) CreateWebhook(ctx context.Context, userID, url, description string, events []string, filters WebhookFilters) (*Webhook, error)

CreateWebhook inserts a new row and returns it with the plaintext signing secret populated. The plaintext is only available on this response — subsequent GET/list calls scrub it.

Filters validation (charset, count caps, agent ownership) is the handler's job in slice 2; the storage layer only verifies the per-user count cap from account_limits.max_webhooks.

func (*Store) CreateWebhookIdem ¶

func (s *Store) CreateWebhookIdem(ctx context.Context, userID, url, description string, events []string, filters WebhookFilters, idemCompleteTx WebhookIdemCompleter) (*Webhook, error)

CreateWebhookIdem is CreateWebhook with an optional idempotency completer. When idemCompleteTx is non-nil the insert runs in a transaction and the completer is invoked inside it (mirroring the send/approve same-tx pattern — idempotency.Store.CompleteTx): the webhook row and the completed idempotency key commit together, so there is no window in which a webhook exists without a replayable cached response. A completer error aborts the create entirely. With a nil completer the insert stays the original single statement.

func (*Store) DeferOutboundTerminalFailure ¶

func (s *Store) DeferOutboundTerminalFailure(ctx context.Context, messageID string, jobID int64, detail string) error

DeferOutboundTerminalFailure records a final attempt's diagnostic and releases the I/O claim WITHOUT declaring the message failed. The worker calls it when its last attempt errored ambiguously (the provider may have accepted the submission before the failure — the SMTP-accept↔mark-sent crash window): the terminal reconciler then declares the outcome after the provider-evidence grace window, preferring this stored detail over its generic sweep message, or settles the row as sent if evidence arrives. Job-scoped so a stale worker can't scribble on a re-enqueued message.

func (*Store) DeleteAPIKey ¶

func (s *Store) DeleteAPIKey(ctx context.Context, keyID, userID string) error

func (*Store) DeleteAgent ¶

func (s *Store) DeleteAgent(ctx context.Context, agentID, userID string) (messagesDeleted int64, err error)

DeleteAgent removes an agent and everything bound to it. It returns the number of message rows removed with the agent — the messages are deleted explicitly rather than left to the agent_identities → messages FK cascade. Besides providing the API's deletion receipt, this ensures the storage metering trigger can resolve the owning user while the agent row exists.

func (*Store) DeleteDomain ¶

func (s *Store) DeleteDomain(ctx context.Context, domain, userID string) error

DeleteDomain deletes a domain only if owned by the user. The handler should check for existing agents first.

func (*Store) DeleteDomainTx ¶

func (s *Store) DeleteDomainTx(ctx context.Context, domain, userID string, inTx func(ctx context.Context, tx pgx.Tx) error) error

DeleteDomainTx deletes a domain and, before committing, runs inTx within the SAME transaction. The hook is how sender-identity teardown is enqueued transactionally (decision 4): the River deprovision job is committed atomically with the domain-row delete, so it can never be lost if SES is unreachable at delete time. A nil hook is a plain delete (dev / no SES).

inTx runs only after the DELETE affected a row (the domain existed and was owned by userID); it never runs for a not-found / FK-blocked delete.

func (*Store) DeleteExpiredMessages ¶

func (s *Store) DeleteExpiredMessages(ctx context.Context) (int64, error)

DeleteExpiredMessages purges message rows whose TrashRetention window has lapsed. Live rows are retained indefinitely. Messages belonging to an agent in trash are removed by PurgeDeletedAgents when that agent reaches the same retention boundary.

func (*Store) DeleteExpiredUserSessions ¶

func (s *Store) DeleteExpiredUserSessions(ctx context.Context) (int64, error)

func (*Store) DeleteTemplate ¶

func (s *Store) DeleteTemplate(ctx context.Context, templateID, userID string) error

DeleteTemplate removes a template owned by the user. Deleting a missing or cross-user template returns ErrTemplateNotFound, never silent success.

func (*Store) DeleteUserData ¶

func (s *Store) DeleteUserData(ctx context.Context, userID string) (*DeleteUserDataResult, error)

DeleteUserData wipes everything tied to a user in a single transaction.

Schema cascades cover most of it (user_sessions, domains, agent_identities, api_keys, usage_summaries all `ON DELETE CASCADE` from users; messages cascade through agent_identities; webhook_deliveries cascade through messages; agent_suppressions and agent_unsubscribe_tokens cascade directly from users). The one row that doesn't is usage_events: its FK is `ON DELETE SET NULL` so analytics survives, which we explicitly override here for full deletion.

Per-table counts are returned to the caller so an operator can attest to what was removed in audit / compliance contexts.

func (*Store) DeleteUserDataTx ¶

func (s *Store) DeleteUserDataTx(ctx context.Context, userID string, perDomainInTx func(ctx context.Context, tx pgx.Tx, domain string) error) (*DeleteUserDataResult, error)

DeleteUserDataTx is DeleteUserData with an optional per-domain in-tx hook run for every domain the user owns, BEFORE the cascade deletes them. It is how sender-identity teardown is enqueued for an account delete (decision 4): each owned domain's SES deprovision job commits atomically with the user delete. A nil hook is a plain account delete (dev / no SES). The DB FK cascade still removes the domain rows; the hook only schedules the remote SES cleanup that the cascade cannot do.

func (*Store) DeleteUserSession ¶

func (s *Store) DeleteUserSession(ctx context.Context, token string) error

func (*Store) DeleteWebhook ¶

func (s *Store) DeleteWebhook(ctx context.Context, webhookID, userID string) error

DeleteWebhook removes a webhook owned by the user. Idempotent: deleting a non-existent or cross-user webhook returns ErrWebhookNotFound, never silently succeeds. The ON DELETE CASCADE on webhook_subscriber_deliveries.webhook_id drops pending delivery rows automatically — no separate cleanup needed.

Slice 1 includes this method (rather than deferring to slice 2's handler work) because tests need it for setup teardown and the implementation is trivial.

func (*Store) DomainExists ¶

func (s *Store) DomainExists(ctx context.Context, domain string) (bool, error)

DomainExists reports whether a live domain row exists (orphan reaper).

func (*Store) DomainOwner ¶

func (s *Store) DomainOwner(ctx context.Context, domain string) (string, error)

DomainOwner returns the user_id owning a domain, or "" for an unowned (system) domain. pgx.ErrNoRows → ("", nil) so the caller treats a missing domain as "no owner, no event".

func (*Store) EffectiveSuppressions ¶

func (s *Store) EffectiveSuppressions(ctx context.Context, userID, agentID string, addresses []string) ([]string, error)

EffectiveSuppressions returns the deduplicated union of account-wide and exact-agent recipient blocks present in addresses.

func (*Store) EncryptLegacyDKIMKeys ¶

func (s *Store) EncryptLegacyDKIMKeys(ctx context.Context) (int, error)

EncryptLegacyDKIMKeys re-encrypts any DKIM private keys still stored as plaintext DER (rows written before encryption-at-rest, #144). It is idempotent and self-terminating: only untagged rows (first byte != dkimBlobV1) are selected, so a second run is a no-op. No-op when no cipher is configured. The domains table is small, so it is safe to run at every startup. Returns the number of rows encrypted.

func (*Store) EnsureSharedDomain ¶

func (s *Store) EnsureSharedDomain(ctx context.Context, domain string) error

EnsureSharedDomain inserts a system row for the configured shared mail domain so slug-based agent registration can satisfy the agent_identities.registered_domain → domains.domain foreign key. The row is owned by no user (user_id = NULL) and pre-verified — it represents infrastructure the operator runs, not user-claimed identity.

Called once at server startup. Idempotent via ON CONFLICT DO NOTHING, and a no-op when the operator has not configured a shared domain. Without this, any deployment whose shared_domain differs from the hardcoded migration seed (`agents.e2a.dev`) gets an FK violation the first time a user tries to register a slug-based agent.

func (*Store) ExpireAndDeliverLocal ¶

func (s *Store) ExpireAndDeliverLocal(
	ctx context.Context,
	messageID string,
	compose func(msg *Message) (SendResult, error),
	beforeCommit LocalDeliveryTxHook,
) (*Message, error)

ExpireAndDeliverLocal is the TTL-worker counterpart of ApproveAndDeliverLocal. It requires an expired pending row, records the review_expired_approved lifecycle, and atomically creates the Inbox copy and outcome events without using the external-provider send journal.

func (*Store) ExpireApproveAndSend ¶

func (s *Store) ExpireApproveAndSend(
	ctx context.Context,
	messageID string,
	send func(msg *Message) (SendResult, error),
) (*Message, error)

ExpireApproveAndSend is the worker-side counterpart to ApproveAndSend: no user ownership check (the caller is the expiration worker, which is system-scoped), SELECT ... FOR NO KEY UPDATE SKIP LOCKED so concurrent workers don't race for the same row, and the terminal status is 'review_expired_approved' instead of 'sent'. On send failure the transaction rolls back; the worker should then call ExpireReject to move the row to a final state so the row doesn't get picked up on every sweep.

Exactly-once guarantee: like ApproveAndSend, this method runs the send() callback under a send_attempts gate so a crash between SES acceptance and the surrounding tx commit does NOT cause the next worker poll to re-send. ClaimSendAttempt / MarkSendSucceeded / MarkSendFailed run in separate small transactions that outlive the approval tx; on retry, an AlreadySent verdict reuses the cached SendResult and skips the upstream send entirely. Without this, the polling-loop nature of the worker would guarantee a re-send on any commit failure — strictly worse than the human-approval path, where a re-send needs an explicit click.

SKIP LOCKED means multiple app instances can run the worker without contending on the same row. The row-level FOR NO KEY UPDATE lock on messages is held for the duration of the send callback (bounded by SMTPRelay timeouts); FOR NO KEY UPDATE rather than FOR UPDATE so the send_attempts INSERT in a separate connection can acquire its KEY SHARE lock for FK enforcement — see ApproveAndSend's docstring for the full rationale.

If a concurrent worker is mid-send for the same row (the send_attempts row is 'attempting' and not yet stale), returns ErrSendInProgress. The worker loop should treat this like ErrNotPendingApproval — skip silently and let the next poll handle it.

func (*Store) ExpireApproveReview ¶

func (s *Store) ExpireApproveReview(ctx context.Context, messageID string) error

ExpireApproveReview is the worker-side TTL auto-approve: releases the message (status review_expired_approved) with no human reviewer. System-scoped.

func (*Store) ExpireReject ¶

func (s *Store) ExpireReject(ctx context.Context, messageID, reason string) (*Message, error)

ExpireReject transitions a pending_review message to review_expired_rejected while retaining its content. No user ownership check — this is the worker path. If the row is no longer pending (racing worker, already handled), or the agent was moved to the trash after the sweep listed the row (the hold must survive intact, and RestoreAgent shifts approval_expires_at so the clock resumes on restore), returns ErrNotPendingApproval; caller can treat as a no-op. The trash guard lives INSIDE the CAS so it is atomic with the transition — a separate read would reopen the sweep's TOCTOU window.

func (*Store) ExpireRejectReview ¶

func (s *Store) ExpireRejectReview(ctx context.Context, messageID, reason string) error

ExpireRejectReview is the worker-side TTL auto-reject: drops the message (status review_expired_rejected) with no human reviewer. System-scoped.

func (*Store) ExportUserData ¶

func (s *Store) ExportUserData(ctx context.Context, userID string) (*UserExport, error)

ExportUserData gathers everything a user owns into a single struct for the right-of-access flow. Reads run inside a REPEATABLE READ transaction so the snapshot is internally consistent even if writes arrive while the export is being assembled.

func (*Store) GetAgentByEmail ¶

func (s *Store) GetAgentByEmail(ctx context.Context, email string) (*AgentIdentity, error)

GetAgentByEmail looks up an agent by email address (same as GetAgentByID since ID = email).

func (*Store) GetAgentByID ¶

func (s *Store) GetAgentByID(ctx context.Context, id string) (*AgentIdentity, error)

GetAgentByID looks up an agent by its ID (full email) with domain verification status.

func (*Store) GetAgentByIDAnyState ¶

func (s *Store) GetAgentByIDAnyState(ctx context.Context, id string) (*AgentIdentity, error)

GetAgentByIDAnyState loads an agent regardless of trash state (deleted_at populated when trashed). For the trash surfaces only — restore, permanent delete, trash listing detail. Every live path uses GetAgentByID, which treats a trashed agent as nonexistent.

func (*Store) GetConversationByID ¶

func (s *Store) GetConversationByID(ctx context.Context, agentID, conversationID string) (*ConversationDetail, error)

GetConversationByID returns the aggregate summary fields plus every member message, ordered oldest-first (chronological reading order). Returns ErrMessageNotFound when no live messages exist for the given (agentID, conversationID) — mirrors the "looks-like-not-found-on-cross-agent" convention used by single- message reads. The same code path handles "wrong agent" and "real non-existent": either way the agent has no business seeing it.

Participants are computed as the union of sender + recipient + each row's to_recipients / cc / bcc (when populated). Empty strings are dropped. Labels are the union of all members' labels[]; both are sorted lexicographically for stable output.

func (*Store) GetDKIMKeyInternal ¶

func (s *Store) GetDKIMKeyInternal(ctx context.Context, domain string) (string, []byte, error)

GetDKIMKeyInternal returns the stored selector + private key bytes for a domain. The "Internal" suffix is load-bearing: this function does NOT scope by user — it takes a domain name and returns whoever owns that domain's signing key. ONLY call from server-internal codepaths where the domain has already been resolved from a trusted source (e.g. an outbound message's sender field, after the agent layer has authenticated the owner). A handler that ever takes a user-supplied domain string and feeds it to this function becomes a "sign as anyone" primitive: don't.

Returns ("", nil, nil) when the domain has no key — callers MUST treat this as "skip signing" and fall back to whatever the relay-level fallback does.

func (*Store) GetDashboardStats ¶

func (s *Store) GetDashboardStats(ctx context.Context, userID string, windowDays int) (*DashboardStats, error)

GetDashboardStats returns workspace-level aggregates for the authenticated user, with a configurable lookback window. windowDays controls Inbound/Outbound totals AND the delivery-success ratio's sample period — passing 0 falls back to DashboardDefaultWindowDays (7), values above DashboardMaxWindowDays (90) are clamped.

Three independent reads — kept separate because the source tables have different indexes and one slow read shouldn't slow the others. All reads are O(rows-for-this-user-only) thanks to the existing per-user indexes.

Robust to missing data: deployments without usage tracking enabled (E2A_USAGE_TRACKING=false — the default for self-hosters) return zero counts rather than erroring. Same for users who have no messages yet. The UI renders zero values as "—".

Delta percentages: today vs yesterday on usage_summaries. Avoids divide-by-zero when yesterday was zero by returning 0. 100% in/de- crease maps to ±100; values clipped at ±999 for integer width.

func (*Store) GetEventEnvelope ¶

func (s *Store) GetEventEnvelope(ctx context.Context, messageID, eventType string) ([]byte, error)

GetEventEnvelope returns the exact durable event envelope for a message. WebSocket reconnect drain uses this instead of rebuilding an event whose timestamp or attachment metadata could differ under the same event id.

func (*Store) GetInboundByEmailMessageID ¶

func (s *Store) GetInboundByEmailMessageID(ctx context.Context, agentID, emailMessageID string) (*Message, error)

GetInboundByEmailMessageID looks up an inbound message by its RFC 5322 Message-ID for the given agent. Used by HITL flows to reach the parent inbound at approval time so the References chain can be rebuilt — the pending-outbound row only stores the parent's Message-ID, not its raw message. Scoped to agent_id to prevent any cross-agent reach across shared infra. Returns sql.ErrNoRows when the inbound has expired or was never persisted; callers must tolerate that and fall back to legacy single-id threading.

This reply-threading lookup selects the stored header context needed to rebuild References, including the raw message and legacy auth-header map. It omits structured authentication evidence, which message/review detail paths load separately.

func (*Store) GetInboundMessage ¶

func (s *Store) GetInboundMessage(ctx context.Context, id string) (*Message, error)

func (*Store) GetMessageByEmailMessageID ¶

func (s *Store) GetMessageByEmailMessageID(ctx context.Context, agentID, messageID string) (*Message, error)

GetMessageByEmailMessageID looks up a message by its RFC 5322 Message-ID for the given agent, regardless of direction. It is the direction-agnostic sibling of GetInboundByEmailMessageID: the HITL approve path uses it to rebuild the References chain of a held reply, and the reply's parent can be an outbound the agent sent (reply-to-own-message), not just a received inbound.

The id is matched against email_message_id (where inbound rows carry the sender's Message-ID) OR provider_message_id (where outbound rows carry the relay/SES-assigned Message-ID — outbound rows have no email_message_id, see ThreadMessageID), so a held reply threaded onto either kind of parent resolves. Same expiry/held exclusions apply. Returns sql.ErrNoRows when the parent has expired or was never persisted; callers must tolerate that and fall back to legacy single-id threading.

func (*Store) GetMessageWithContent ¶

func (s *Store) GetMessageWithContent(ctx context.Context, messageID, agentID string) (*Message, error)

GetMessageWithContent returns a full message including raw MIME and canonical authentication evidence. Marks the message as 'read' if it was 'unread'.

Unlike the list/reply/threading paths, this deliberately returns TRASHED rows too (DeletedAt set) so the dashboard trash can open a deleted message — Gmail's "view message in trash". Callers branch on DeletedAt when trash rows must not qualify.

func (*Store) GetMessagesByAgent ¶

func (s *Store) GetMessagesByAgent(ctx context.Context, f MessageListFilter) ([]Message, error)

GetMessagesByAgent returns messages for an agent, filtered by status, direction, and the optional search filters on the MessageListFilter struct.

• direction: "inbound" (default for SDK polling), "outbound", or "all" (used by the dashboard inbox).
• status: "unread" | "read" | "all" — only applies when direction selects inbound rows; ignored on pure outbound queries.
• descending: cursor walks newest→oldest when true; oldest→newest when false (FIFO polling).
• From / SubjectContains: case-insensitive substring (ILIKE).
• ConversationID: exact match.
• Since / Until: time-range bracket on created_at.

The SELECT includes columns both consumers need: the inbox needs `status` (outbound HITL lifecycle), `webhook_status`/`last_error` (outbound delivery), and `octet_length(raw_message)` (size column); the polling SDK ignores these fields and reads only the existing inbound-relevant ones from the Message struct. Canonical identity and authentication evidence is selected so the WS drain fallback matches live delivery; raw_message stays deliberately unselected (the blob). Legacy auth columns remain selected only for fail-closed migration fallback.

func (*Store) GetOutboundMessageForUser ¶

func (s *Store) GetOutboundMessageForUser(ctx context.Context, messageID, userID string) (*Message, error)

GetOutboundMessageForUser returns a full message row (including body, HITL fields, and attachments) if it exists and is owned by userID (via the agent the row belongs to). Inbound messages and cross-user access both return ErrMessageNotFound — the caller should not be able to distinguish "does not exist" from "belongs to someone else".

func (*Store) GetPrincipalByAPIKey ¶

func (s *Store) GetPrincipalByAPIKey(ctx context.Context, apiKey string) (*Principal, error)

GetPrincipalByAPIKey authenticates a bearer token and returns the full principal — the owning user PLUS the key's scope and bound agent (Slice 5a). Same validation/last_used semantics as GetUserByAPIKey (it delegates here). A legacy key with a NULL scope column resolves to ScopeAccount, preserving pre-redesign authority.

func (*Store) GetRepliableMessage ¶

func (s *Store) GetRepliableMessage(ctx context.Context, id string) (*Message, error)

GetRepliableMessage loads a message that can be the target of a reply or forward, regardless of direction: an inbound the agent received or an outbound the agent sent. It is the direction-agnostic sibling of GetInboundMessage — same columns (plus provider_message_id, which carries the outbound Message-ID for threading; see ThreadMessageID) — but without the `direction = 'inbound'` predicate, so an agent can continue a thread off its own sent message (mirrors how mail clients let you reply to a message in your Sent folder).

The held-status exclusion is kept for BOTH directions: a message still in review (pending/rejected/expired) has not actually been delivered, so it is not a legitimate reply/forward anchor. Callers still scope the result to the owning agent (id-only lookup here does not).

method + delivery_status are loaded (beyond GetInboundMessage's column list) because `status` is the review/hold axis, not the delivery axis: an outbound row reads status='sent' the instant it is accepted, long before the send worker submits it. The reply/forward handlers need the delivery axis to tell an outbound that is still queued for external submission (no provider_message_id yet → ThreadMessageID() would return "" and silently drop In-Reply-To/References) from one that is genuinely terminal. See httpapi.parentNotYetSubmitted.

func (*Store) GetReviewMessage ¶

func (s *Store) GetReviewMessage(ctx context.Context, messageID, agentID string) (*ReviewMessageMeta, error)

GetReviewMessage is the review-queue single-item getter: it loads a message scoped to agentID REGARDLESS of held status (unlike every agent-facing read path, which excludes heldInboundStatuses). It exists so the account-scoped /approve+/reject endpoints can resolve a held message's direction and dispatch the correct release path.

SECURITY: this deliberately bypasses the held-inbound read boundary, so it MUST only ever be reachable from an account-scoped reviewer flow that has already proven ownership of agentID (resolveOwnedAgent). It is NOT an agent read path — do not wire it onto any agent-scoped surface. The agentID filter is the tenant-isolation guard: a reviewer can only resolve a message belonging to an agent they own. Returns sql.ErrNoRows when no such message exists for the agent.

func (*Store) GetReviewWithContent ¶

func (s *Store) GetReviewWithContent(ctx context.Context, userID, messageID string) (*Message, error)

GetReviewWithContent loads one held message (either direction) with its body, scoped to userID — the detail view behind GET /v1/reviews/{id}. Like GetReviewMessage it deliberately bypasses the held-inbound exclusion, so it is account-scoped reviewer flow ONLY (the user join is the tenant guard). Unlike GetMessageWithContent it does NOT mark the message read (reviewing isn't reading) and does NOT exclude held rows. Returns sql.ErrNoRows when no such message exists for the user.

func (*Store) GetSendOutcome ¶

func (s *Store) GetSendOutcome(ctx context.Context, messageID string) (SendOutcome, error)

GetSendOutcome reads the current delivery_status + provider id + sent_as + detail for an outbound message — the wait=sent poll. Returns a zero-value outcome (empty DeliveryStatus) if the row is gone, never an error for a missing row.

func (*Store) GetSendingStatus ¶

func (s *Store) GetSendingStatus(ctx context.Context, domain string) (string, error)

GetSendingStatus returns the domain's sending_status. Propagates pgx.ErrNoRows when the domain row is gone.

func (*Store) GetTemplateByAlias ¶

func (s *Store) GetTemplateByAlias(ctx context.Context, alias, userID string) (*Template, error)

GetTemplateByAlias resolves a template by its per-user alias. Missing or cross-user aliases return ErrTemplateNotFound.

func (*Store) GetTemplateByID ¶

func (s *Store) GetTemplateByID(ctx context.Context, templateID, userID string) (*Template, error)

GetTemplateByID returns the template iff it's owned by userID. Cross-user reads (or missing rows) return ErrTemplateNotFound.

func (*Store) GetUserByAPIKey ¶

func (s *Store) GetUserByAPIKey(ctx context.Context, apiKey string) (*User, error)

GetUserByAPIKey authenticates a bearer token and returns the owning user. Rejects revoked keys and time-expired keys; touches last_used_at only on the success path so the column stays a real "last successful authentication" signal (rather than "last attempt").

Expiration semantics: expires_at IS NULL means the key never expires (preserves the pre-migration default). A non-null expires_at must be in the future, evaluated against now() in the same query so there's no clock skew between row read and check.

func (*Store) GetUserByID ¶

func (s *Store) GetUserByID(ctx context.Context, id string) (*User, error)

func (*Store) GetUserSession ¶

func (s *Store) GetUserSession(ctx context.Context, token string) (*User, error)

func (*Store) GetWebhookByID ¶

func (s *Store) GetWebhookByID(ctx context.Context, webhookID, userID string) (*Webhook, error)

GetWebhookByID returns the webhook iff it's owned by userID. Cross- user reads (or missing rows) return ErrWebhookNotFound — same not-found-on-cross-user convention used elsewhere in the codebase (conversation reads, message reads).

The returned Webhook has SigningSecret populated for the delivery worker's benefit; the public API layer scrubs this field before responding to GETs.

func (*Store) GetWebhookByIDInternal ¶

func (s *Store) GetWebhookByIDInternal(ctx context.Context, webhookID string) (*Webhook, error)

GetWebhookByIDInternal returns the webhook by ID with no ownership check. INTERNAL USE ONLY — handler code MUST use GetWebhookByID which scopes by user_id. The retry worker uses this to look up the URL + signing secret for a delivery row whose ownership was already established when the publisher inserted it.

The suffix Internal mirrors the convention in dkim.GetDKIMKeyInternal: a method name that calls out "skipping the standard authorization check" so a reviewer doesn't have to read the body to know why.

func (*Store) HasAgentsOnDomain ¶

func (s *Store) HasAgentsOnDomain(ctx context.Context, domain, userID string) (bool, error)

HasAgentsOnDomain checks whether the owned domain still has agents.

func (*Store) InsertInboundIntakeTx ¶

func (s *Store) InsertInboundIntakeTx(ctx context.Context, tx pgx.Tx, id, recipient, envelopeFrom, heloDomain, remoteIP, messageID, contentHash string, raw []byte) (inserted bool, err error)

InsertInboundIntakeTx writes an accepted intake row inside the accept-tx. It returns inserted=false (no error) when the row is a duplicate — the dedup unique index (recipient, message_id, content_hash) suppressed it — so the caller knows NOT to enqueue a second job and to still answer 250 (idempotent accept).

func (*Store) ListAPIKeys ¶

func (s *Store) ListAPIKeys(ctx context.Context, userID string, limit int, afterCreatedAt time.Time, afterID string) ([]APIKey, error)

ListAPIKeys returns one page of the user's live (non-revoked) API keys, newest-first, keyset-paginated on (created_at, id). limit<=0 returns every key unpaginated (the all-consumers: auth dashboard + prober seed); a positive limit fetches that many (pass limit+1 to detect a further page) starting after the (afterCreatedAt, afterID) key from the previous page's last row (zero afterCreatedAt = first page).

func (*Store) ListActivityByAgent ¶

func (s *Store) ListActivityByAgent(ctx context.Context, agentID string, limit int) ([]Message, error)

func (*Store) ListAgentSuppressions ¶

func (s *Store) ListAgentSuppressions(ctx context.Context, userID, agentID string, limit int, after time.Time, afterAddress string) ([]AgentSuppression, error)

ListAgentSuppressions returns one newest-first keyset page for an exact account and agent.

func (*Store) ListAgentsByUser ¶

func (s *Store) ListAgentsByUser(ctx context.Context, userID string, limit int, afterCreatedAt time.Time, afterID string) ([]AgentIdentity, error)

Agents are joined with domain verification AND enriched with per-agent stats for the dashboard. Correlated subqueries compute inbound/outbound 7-day counts, pending approvals, last delivery, and webhook status in a single round-trip. Other load paths (GetAgentByID, GetAgentByEmail) intentionally don't compute these — only the dashboard and the account export need them.

ListAgentsByUser returns one page of the user's agents, newest-first, keyset-paginated on (created_at, id). limit<=0 returns every agent unpaginated (the all-consumers: auth dashboard views + webhook filter ownership validation); a positive limit fetches that many (pass limit+1 to detect a further page) starting after the (afterCreatedAt, afterID) key from the previous page's last row (zero afterCreatedAt = first page).

func (*Store) ListConversationsByAgent ¶

func (s *Store) ListConversationsByAgent(ctx context.Context, f ConversationListFilter) ([]ConversationSummary, error)

ListConversationsByAgent groups the agent's live messages by conversation_id and returns one row per conversation sorted by most-recent activity. Messages without a conversation_id are not included in any conversation — they remain individually visible via GetMessagesByAgent.

func (*Store) ListDeletedAgentsByUser ¶

func (s *Store) ListDeletedAgentsByUser(ctx context.Context, userID string, limit int, afterCreatedAt time.Time, afterID string) ([]AgentIdentity, error)

ListDeletedAgentsByUser is the trash listing: agents the user soft-deleted, newest-first, same keyset pagination as the live list. DeletedAt is populated on every row.

func (*Store) ListDomainsByUser ¶

func (s *Store) ListDomainsByUser(ctx context.Context, userID string, limit int, afterCreatedAt time.Time, afterDomain string) ([]Domain, error)

AgentCount is computed inline via a correlated subquery — one round-trip regardless of how many domains the page has, and the per-row count is supported by the existing user and registered-domain indexes. Excludes system rows.

ListDomainsByUser returns one page of the user's domains, newest-first, keyset-paginated on (created_at, domain) — domain is the table's unique key, so it is the deterministic tiebreak. limit<=0 returns every domain unpaginated; a positive limit fetches that many (pass limit+1 to detect a further page) starting after the (afterCreatedAt, afterDomain) key from the previous page's last row (zero afterCreatedAt = first page).

func (*Store) ListEnabledWebhooksForRouting ¶

func (s *Store) ListEnabledWebhooksForRouting(ctx context.Context, userID, eventType string) ([]Webhook, error)

ListEnabledWebhooksForRouting is the hot-path query used by the event publisher. Returns enabled webhooks for the user that subscribe to the given event type. The in-process publisher then applies filter matching in Go (cheaper than encoding the AND-across-types + OR-within-type rule in SQL at slice-1 scale).

The partial index idx_webhooks_user_enabled WHERE enabled = true keeps this O(log n) on the common case (a user has a small number of enabled webhooks).

func (*Store) ListExpiredPending ¶

func (s *Store) ListExpiredPending(ctx context.Context, limit int) ([]ExpirationCandidate, error)

ListExpiredPending returns pending_review messages whose approval_expires_at is in the past, joined with their agent's hitl_expiration_action. Ordered by approval_expires_at ASC so earliest-expired are handled first.

func (*Store) ListExpiredReviews ¶

func (s *Store) ListExpiredReviews(ctx context.Context, limit int) ([]ExpirationCandidate, error)

ListExpiredReviews returns inbound pending_review messages whose approval_expires_at has passed, joined with their agent's hitl_expiration_action — the inbound analogue of ListExpiredPending. The expiry worker uses these to auto-resolve held messages per the agent's policy.

func (*Store) ListPendingOutboundForUser ¶

func (s *Store) ListPendingOutboundForUser(ctx context.Context, userID string, limit int) ([]Message, error)

ListPendingOutboundForUser returns pending-approval messages across all of the user's agents, sorted by approval_expires_at ASC (expiring-soonest first). Body columns are not returned from this path — callers should use GetOutboundMessageForUser for detail.

func (*Store) ListProtectionEventsByAgent ¶

func (s *Store) ListProtectionEventsByAgent(ctx context.Context, agentID string, limit int) ([]ProtectionEvent, error)

ListProtectionEventsByAgent returns an agent's most recent screening events (newest first, capped by limit) for the security/analytics view.

func (*Store) ListProtectionEventsByMessage ¶

func (s *Store) ListProtectionEventsByMessage(ctx context.Context, messageID string) ([]ProtectionEvent, error)

ListProtectionEventsByMessage returns every screening event recorded against a message, newest first — the breakdown a reviewer sees for a held item.

func (*Store) ListReviews ¶

func (s *Store) ListReviews(ctx context.Context, userID string, limit int, afterCreatedAt time.Time, afterID string) ([]ReviewListItem, error)

ListReviews returns one page of held (pending_review) messages — BOTH directions — across all of userID's agents, newest-first, keyset-paginated on (created_at, id). The caller passes limit (fetch limit+1 to detect a further page; limit<=0 returns every row unpaginated) and the after-key from the previous page's last row (zero afterCreatedAt = first page). The review queue grows unbounded with the pending-review backlog, so it needs real pagination rather than returning the whole set. This is the operator review queue: it intentionally includes held inbound (which every agent-facing read path excludes). SECURITY: account-scoped reviewer flow only; the user join is the tenant-isolation guard.

func (*Store) ListSuppressions ¶

func (s *Store) ListSuppressions(ctx context.Context, userID string, limit int, afterCreatedAt time.Time, afterAddress string) ([]Suppression, error)

ListSuppressions returns the user's suppression list, newest first. ListSuppressions returns one page of the user's suppressed addresses, newest first, keyset-paginated on (created_at, address). The caller passes limit (fetch limit+1 to detect a further page) and the after-key from the previous page's last row (zero afterCreatedAt = first page). (A-5: the suppression list auto-grows on every bounce/complaint, so it needs real pagination, not a single page.)

func (*Store) ListTemplatesByUser ¶

func (s *Store) ListTemplatesByUser(ctx context.Context, userID string, limit int, afterCreatedAt time.Time, afterID string) ([]TemplateSummary, error)

Summaries only (no body/html_body).

ListTemplatesByUser returns one page of the user's templates, newest-first, keyset-paginated on (created_at, id). limit<=0 returns every template unpaginated; a positive limit fetches that many (pass limit+1 to detect a further page) starting after the (afterCreatedAt, afterID) key from the previous page's last row (zero afterCreatedAt = first page).

func (*Store) ListWebhooksByUser ¶

func (s *Store) ListWebhooksByUser(ctx context.Context, userID string, limit int, afterCreatedAt time.Time, afterID string) ([]Webhook, error)

Storage layer surfaces enabled and disabled rows alike; filter at the handler if needed.

ListWebhooksByUser returns one page of the user's webhooks, newest-first, keyset-paginated on (created_at, id). limit<=0 returns every webhook unpaginated (the all-consumers: prober seed); a positive limit fetches that many (pass limit+1 to detect a further page) starting after the (afterCreatedAt, afterID) key from the previous page's last row (zero afterCreatedAt = first page).

func (*Store) LoadInboundIntake ¶

func (s *Store) LoadInboundIntake(ctx context.Context, intakeID string) (*InboundIntake, error)

LoadInboundIntake reads an intake row for the worker. Returns (nil, nil) when the row is gone (pruned) so the worker treats it as a no-op.

func (*Store) LoadOutboundDraft ¶

func (s *Store) LoadOutboundDraft(ctx context.Context, messageID string) (*Message, error)

LoadOutboundDraft loads a pending_review outbound message's full draft content (recipients, subject, body, attachments, reply-to) by id, system-scoped (no user filter, no side effects) — the TTL sweep uses it to reconstruct the SendRequest and compose before ApproveAndAccept. Returns ErrMessageNotFound if the row is gone or not an outbound message. The caller must still handle the pending_review CAS in ApproveAndAccept (a human may resolve the hold before the transition).

func (*Store) LoadOutboundForSend ¶

func (s *Store) LoadOutboundForSend(ctx context.Context, messageID string) (*OutboundSendPayload, error)

LoadOutboundForSend returns the payload the async send worker submits, or nil if the row is gone (agent delete / TTL) — the worker treats that as a no-op. A message in the trash, or one whose AGENT is in the trash, counts as gone: deleting is the user's one lever to stop a queued-but-unsent message (e.g. snoozed behind a provider-outage retry horizon), and a trashed inbox must never emit mail. Reads the envelope (to+cc+bcc) and the persisted wire bytes; does not touch message_recipients (those are written at MarkSent).

func (*Store) LoadPendingNotify ¶

func (s *Store) LoadPendingNotify(ctx context.Context, messageID string) (*PendingNotify, error)

LoadPendingNotify returns the row the notification worker needs, or (nil, nil) when there is nothing to notify about — the message was deleted/pruned, or its owning agent no longer exists (an orphaned hold; other paths finalize it). The worker treats a nil return as a no-op.

func (*Store) LookupConversationID ¶

func (s *Store) LookupConversationID(ctx context.Context, agentID string, messageIDs []string) (string, error)

LookupConversationID finds a conversation_id by matching In-Reply-To / References message IDs against stored messages. Checks both email_message_id (inbound) and provider_message_id (outbound). Uses prefix matching because SES bare IDs stored in provider_message_id (e.g. <010f...>) may lack the @region.amazonses.com suffix that appears in the actual email headers sent to recipients.

func (*Store) LookupCoveringDomain ¶

func (s *Store) LookupCoveringDomain(ctx context.Context, sub, userID string) (*Domain, error)

LookupCoveringDomain returns the MOST-SPECIFIC registered domain owned by userID that is a proper label-boundary parent of sub, or an error if none covers it. This backs subdomain-agent creation: a verified parent SES identity (e.g. team.mnexa.ai) DKIM-signs and DMARC-aligns mail from any subdomain (acme.team.mnexa.ai), so e2a lets an agent be created on the subdomain without a separate registration. The caller must inspect Verified: an explicitly registered pending child is authoritative for its subtree and must not be masked by a verified grandparent. The parent must be an exact ancestor on a DNS label boundary — see coveringParentCandidates. The returned Domain.Domain is what the caller stores in agent_identities.registered_domain so the FK, quota JOIN, DKIM signer, and sending-status lookup all resolve to the parent.

func (*Store) LookupDomain ¶

func (s *Store) LookupDomain(ctx context.Context, domain, userID string) (*Domain, error)

LookupDomain returns a domain if it exists and is owned by the given user.

func (*Store) MarkInboundIntakeFailed ¶

func (s *Store) MarkInboundIntakeFailed(ctx context.Context, intakeID, detail string) error

MarkInboundIntakeFailed marks an intake terminally failed (unparseable body / exhausted retries) with a diagnostic detail. Own transaction — a terminal record for ops visibility; the message is dropped (we already 250'd).

func (*Store) MarkInboundIntakeProcessedTx ¶

func (s *Store) MarkInboundIntakeProcessedTx(ctx context.Context, tx pgx.Tx, intakeID, messageFK string) error

MarkInboundIntakeProcessedTx flips the intake to 'processed' and links the created messages row, in the worker's terminal tx (same tx as the messages insert + outbox publish). This is the worker's idempotency gate: a re-drive that finds 'processed' no-ops instead of re-creating the message.

func (*Store) MarkMessageNotified ¶

func (s *Store) MarkMessageNotified(ctx context.Context, messageID string) error

MarkMessageNotified stamps notified_at after the approval-notification email is sent. Set only AFTER a successful send, so it is the send-dedup marker that makes a crash-after-send River re-drive a no-op without ever risking a lost notification (loss would require setting it before the send).

func (*Store) MarkMessageSent ¶

func (s *Store) MarkMessageSent(ctx context.Context, messageID, sentAs string, to, cc, bcc []string) error

MarkMessageSent records that an outbound message was accepted by the relay: delivery_status='sent', the From identity actually used, and one message_recipients row per recipient (to/cc/bcc) at 'sent'. Called after a successful relay accept. Idempotent on the recipient rows.

func (*Store) MarkOutboundFailedTx ¶

func (s *Store) MarkOutboundFailedTx(ctx context.Context, tx pgx.Tx, messageID, detail string, source delivery.FailureSource) (*OutboundSentInfo, error)

MarkOutboundFailedTx records, within the caller's transaction, a terminal outbound send failure: delivery_status='failed' + delivery_detail + the failure provenance (delivery_failure_source — 'provider' for an explicit provider rejection, 'local' for an inferred failure; §3.1 correction only ever applies to 'local'). Returns the row + owning user for the caller to emit email.failed, or nil if the row is gone, already left accepted/sending, or carries provider-accept evidence — the CAS's `provider_accepted_at IS NULL` arm is the terminal-failure guard's last line: a row the provider demonstrably accepted can never be declared failed, no matter which caller races in (the caller settles it via ResolveOutboundProviderAcceptedTx instead). A stored delivery_detail (a deferred final attempt's diagnostic) is preferred over the caller's generic detail. A post-accept trash does not suppress terminal bookkeeping on the hidden row.

func (*Store) MarkOutboundSentTx ¶

func (s *Store) MarkOutboundSentTx(ctx context.Context, tx pgx.Tx, messageID, providerMessageID string) (*OutboundSentInfo, error)

MarkOutboundSentTx records, within the caller's transaction, that a claimed outbound message was submitted to the provider: delivery_status='sent', provider_message_id, and one message_recipients row per recipient at 'sent' (mirrors MarkMessageSent's recipient shape). sent_as is left as the accept-time value. Returns the row + owning user/domain for the caller to emit email.sent + meter usage, or nil if the row is gone or no longer in `sending`. Trashed rows remain eligible because trash may commit after the durable send claim.

func (*Store) MarkSendFailed ¶

func (s *Store) MarkSendFailed(ctx context.Context, messageID, errMsg string) error

MarkSendFailed records that the upstream send failed for messageID, so the next ClaimSendAttempt is allowed to take over and retry. Idempotent against double-call: only updates rows still 'attempting'.

func (*Store) MarkSendSucceeded ¶

func (s *Store) MarkSendSucceeded(ctx context.Context, messageID string, r SendResult) error

MarkSendSucceeded records the result of a successful upstream send. Idempotent against double-call: only updates rows still 'attempting' so a stray re-Mark from a buggy caller cannot overwrite a previous success or revive a failed row.

func (*Store) MarkSendSucceededWithRetry ¶

func (s *Store) MarkSendSucceededWithRetry(messageID string, r SendResult) error

MarkSendSucceededWithRetry calls MarkSendSucceeded with bounded exponential backoff. The retry runs on a fresh detached context (15s budget) so that once SES has accepted the send, request-level cancellation (client disconnect, request timeout) cannot prevent us from durably recording that fact.

Closes the C4 audit finding: if MarkSendSucceeded's single attempt failed (transient DB blip — pool exhaustion, statement_timeout, momentary leader failover), the row stayed `attempting` and the 10-minute stale takeover would later re-invoke the upstream send, duplicating the email at the recipient. Most transient DB failures resolve well within the retry budget; the residual risk window shrinks from "any single blip" to "persistent DB failure for >7s during a specific 10-minute window."

Returns nil if any attempt succeeded; the last error otherwise. Callers should treat a returned error as a loud signal: the upstream send happened, but we couldn't record it — manual reconciliation against the SES Configuration Set events log may be needed.

func (*Store) MaxTemplatesForUser ¶

func (s *Store) MaxTemplatesForUser(ctx context.Context, userID string) (int, error)

MaxTemplatesForUser returns the per-user cap from account_limits, or DefaultMaxTemplates when the user has no row.

func (*Store) MaxWebhooksForUser ¶

func (s *Store) MaxWebhooksForUser(ctx context.Context, userID string) (int, error)

func (*Store) ModifyMessageLabels ¶

func (s *Store) ModifyMessageLabels(ctx context.Context, messageID, agentID string, add, remove []string) ([]string, error)

ModifyMessageLabels applies a delta — add then remove — to a message's labels[] in a single atomic statement. Returns the updated labels (deduplicated, sorted) so the caller can echo them back in the response without a second round-trip.

Inputs are assumed already normalized (lowercased, charset-validated, dedup'd within each list, e2a:* gated). The store layer:

• applies adds first, then removes (so a label in both lists ends up removed)
• rejects if the post-add total would exceed MaxLabelsPerMessage
• returns ErrMessageNotFound if the row is missing / trashed / cross-agent

The whole thing runs as one UPDATE so a concurrent PATCH from a second client can't observe a partial state.

func (*Store) PruneProcessedIntake ¶

func (s *Store) PruneProcessedIntake(ctx context.Context, olderThan time.Duration) (int64, error)

PruneProcessedIntake deletes processed intake rows whose terminal time is older than olderThan. Safe because the raw MIME also lives in messages.raw_message once processed; failed rows are deliberately retained for inspection. Returns the count pruned. Called by the inbound retention periodic (QueueMaintenance).

func (*Store) PurgeDeletedAgents ¶

func (s *Store) PurgeDeletedAgents(ctx context.Context) (int64, error)

PurgeDeletedAgents hard-deletes agents whose trash retention has lapsed (deleted_at older than TrashRetention). One agent per transaction, its messages deleted explicitly BEFORE the agent row (not via ON DELETE CASCADE) so the storage-metering trigger — which resolves the owning user through the agent row — still reconciles account_usage; see DeleteAgent. Idempotent and interruption-safe: each agent commits independently, so a janitor timeout mid-backlog resumes next tick.

func (*Store) PurgeMessage ¶

func (s *Store) PurgeMessage(ctx context.Context, messageID, agentID string) error

PurgeMessage permanently deletes a message that is already in the trash ("delete forever" — the Gmail journey is delete → trash → delete forever, so a live message must be trashed first). Returns ErrNotInTrash for a live message, ErrMessageNotFound otherwise.

func (*Store) PutUnsubscribeToken ¶

func (s *Store) PutUnsubscribeToken(ctx context.Context, tokenHash []byte, userID, agentID, address string) error

PutUnsubscribeToken idempotently records a token hash's exact scope.

func (*Store) RecordDeliveryOutcome ¶

func (s *Store) RecordDeliveryOutcome(ctx context.Context, messageID, address string, status delivery.Status, detail string) error

RecordDeliveryOutcome upserts one recipient's status monotonically (by the delivery precedence) and recomputes the message's rollup delivery_status as the worst status across its recipients. Runs in a tx with FOR UPDATE so concurrent SNS events can't race the merge. Idempotent: a duplicate or older event is a no-op for the status (detail still refreshes on an equal/higher).

Correction rule (async-send-contract §3.1): when the message-level status is a locally inferred `failed` (delivery_failure_source 'local' or legacy NULL), a recomputed rollup that proves provider acceptance CORRECTS the row — the falsely-declared terminal failure from a final-attempt crash. A provider-confirmed `failed` is never revived, and feedback for an address outside the message's own envelope is ignored on a failed row so unrelated feedback can never resurrect a genuine failure.

func (*Store) RecordProviderAcceptEvidence ¶

func (s *Store) RecordProviderAcceptEvidence(ctx context.Context, messageID, sesMessageID string) error

RecordProviderAcceptEvidence durably notes that SES reported having accepted this message's submission (the SNS consumer calls it for every correlated post-acceptance notification) and repairs a provider_message_id lost to the SMTP-accept↔mark-sent crash window. Idempotent: the first-seen timestamp and a previously captured provider id are never overwritten. The terminal-failure guards (send worker, terminal reconciler, MarkOutboundFailedTx's CAS) read this evidence before declaring an accepted/sending row failed.

func (*Store) RejectInboundReview ¶

func (s *Store) RejectInboundReview(ctx context.Context, messageID, agentID, reviewerID, reason string) error

RejectInboundReview drops a held inbound message (status review_rejected → stays hidden from the agent). Scoped to agentID for tenant isolation; reviewerID identifies the human; reason is operator-facing. The raw payload is retained (hidden) indefinitely for security forensics unless the message is deleted.

func (*Store) RejectPending ¶

func (s *Store) RejectPending(ctx context.Context, messageID, userID, reason string) (*Message, error)

RejectPending transitions a pending_review message to rejected, records the reviewer's reason (empty string allowed), and retains outbound content. Ownership checked; missing rows return ErrMessageNotFound. Non-pending rows return ErrNotPendingApproval.

func (*Store) ReleaseOutboundSendClaim ¶

func (s *Store) ReleaseOutboundSendClaim(ctx context.Context, messageID string, jobID int64) error

ReleaseOutboundSendClaim moves a side-effect-free provider failure back to accepted so River backoff is not mistaken for an active provider call.

func (*Store) RemoveAgentSuppression ¶

func (s *Store) RemoveAgentSuppression(ctx context.Context, userID, agentID, address string) (bool, error)

RemoveAgentSuppression deletes only the exact account/agent/address row.

func (*Store) RemoveSuppression ¶

func (s *Store) RemoveSuppression(ctx context.Context, userID, address string) (bool, error)

RemoveSuppression deletes a (user, address) suppression. found=false when no such entry existed.

func (*Store) ResolveOutboundOwner ¶

func (s *Store) ResolveOutboundOwner(ctx context.Context, messageID string) (userID, agentID string, err error)

ResolveOutboundOwner looks up the user_id and agent_id for an outbound message without requiring the caller to know the user_id up-front. It exists for token-authenticated paths (magic-link approve/reject) where the HMAC token itself is the authorization and the handler just needs enough context to dispatch into the existing user-scoped store methods.

Returns ErrMessageNotFound if the message doesn't exist or isn't outbound. The returned user_id is guaranteed to own the message's agent (via the agent_identities.user_id join).

func (*Store) ResolveOutboundProviderAcceptedTx ¶

func (s *Store) ResolveOutboundProviderAcceptedTx(ctx context.Context, tx pgx.Tx, messageID string) (*OutboundSentInfo, string, error)

ResolveOutboundProviderAcceptedTx settles, within the caller's transaction, a pre-terminal outbound row for which authoritative provider-accept evidence has been recorded (provider_accepted_at, written by the SNS consumer): delivery_status='sent' with the evidence-repaired provider id + the standard per-recipient 'sent' rows. This is the terminal-failure guard's positive branch (async-send-contract §3.1 / pipeline §7): a final attempt or the terminal reconciler about to declare `failed` first settles the row as sent when the provider demonstrably has the message. Returns (nil, "", nil) when the row is gone, already sent/terminal, or carries no evidence — the caller proceeds to the failure write, whose own CAS re-checks the evidence.

func (*Store) ResolveUnsubscribeToken ¶

func (s *Store) ResolveUnsubscribeToken(ctx context.Context, tokenHash []byte) (*UnsubscribeScope, error)

ResolveUnsubscribeToken resolves a token hash, returning nil for an unknown token without exposing the stored hash.

func (*Store) RestoreAgent ¶

func (s *Store) RestoreAgent(ctx context.Context, agentID, userID string) error

RestoreAgent brings a trashed agent back to life, messages and config intact. Live messages never expire; for still-held drafts only approval_expires_at is shifted forward by the time spent in the trash so a hold cannot silently lapse while the inbox is trashed. Returns ErrNotInTrash when the agent exists but is live, and ErrAgentNotFound when it doesn't exist (or isn't the caller's).

func (*Store) RestoreMessage ¶

func (s *Store) RestoreMessage(ctx context.Context, messageID, agentID string) error

RestoreMessage brings an indefinitely retained trashed message back to the inbox. Returns ErrNotInTrash when the message exists but is live, ErrMessageNotFound otherwise.

func (*Store) RotateSecret ¶

func (s *Store) RotateSecret(ctx context.Context, webhookID, userID string) (newPlaintext string, prevExpiresAt time.Time, err error)

RotateSecret generates a new signing secret, moves the current secret into signing_secret_prev with a 24h expiry, and returns the new plaintext (shown once). During the 24h grace window the delivery worker dual-signs each request so receivers can verify with either secret while they update their handler.

func (*Store) SendingProvisionInputs ¶

func (s *Store) SendingProvisionInputs(ctx context.Context, domain string) (selector string, privateKeyDER []byte, ok bool, err error)

SendingProvisionInputs returns the per-domain DKIM selector + private key for BYODKIM provisioning. ok=false means no usable key material. Like GetDKIMKeyInternal this is unscoped — call only with a server-resolved domain.

func (*Store) SetAccountClass ¶

func (s *Store) SetAccountClass(ctx context.Context, userID, class string) error

SetAccountClass sets a user's account_class (standard|internal|system|demo). Used by the prober's seed to mark the synthetic probe account as system so its traffic is never metered (see usage.PolicyFor). The CHECK constraint in migration 037 rejects any other value.

func (*Store) SetDKIMCipher ¶

func (s *Store) SetDKIMCipher(c *DKIMCipher)

SetDKIMCipher enables envelope encryption of DKIM private keys at rest (#144). Optional-setter (matches SetEnforcer) so NewStore's signature — and the many tests that call NewStore(pool) — stay unchanged. When unset, keys are stored as plaintext DER. cmd/e2a always sets it in production, where Signing.HMACSecret is enforced ≥32 bytes.

func (*Store) SetMessageScreening ¶

func (s *Store) SetMessageScreening(ctx context.Context, messageID, agentID, reviewReason string, scanScore *float64, scanAction string) error

SetMessageScreening denormalizes a screening verdict onto an already-created message row (review_reason / scan_score / scan_action). The inbound path sets these at INSERT time; the outbound path creates the row first (send or HITL hold) and annotates after, so it needs this UPDATE. Agent-scoped to keep the write within the owning tenant. scanScore may be nil (gate-only verdicts).

func (*Store) SetSendingStatus ¶

func (s *Store) SetSendingStatus(ctx context.Context, domain, status, dkimStatus, mailFromStatus, errMsg string, recordsJSON []byte) error

SetSendingStatus writes the sending lifecycle state for a domain and stamps sending_last_checked_at. recordsJSON may be nil (cleared). dkimStatus and mailFromStatus are the per-axis SES breakdown (migration 049); an empty string for either is written as SQL NULL so the read path falls back to the all-or-nothing sending_status rollup (and the CHECK constraint, which allows NULL but not ”, is satisfied).

func (*Store) SoftDeleteAgent ¶

func (s *Store) SoftDeleteAgent(ctx context.Context, agentID, userID string) error

SoftDeleteAgent moves a live agent to the trash (docs/design/ trash-soft-delete.md): it disappears from every live lookup — inbound mail bounces, per-agent API calls 404, its held messages leave the review queue — until restored or purged by the janitor after TrashRetention. The email address stays reserved while in the trash.

func (*Store) SoftDeleteMessage ¶

func (s *Store) SoftDeleteMessage(ctx context.Context, messageID, agentID string) error

SoftDeleteMessage moves a live message to the trash. Idempotent on an already-trashed message (nil). A held message (status pending_review, either direction) cannot be trashed — the review queue is its resolution surface — and returns ErrMessageHeld. A missing message returns ErrMessageNotFound.

func (*Store) StampInboundIntakeJobIDTx ¶

func (s *Store) StampInboundIntakeJobIDTx(ctx context.Context, tx pgx.Tx, intakeID string, jobID int64) error

StampInboundIntakeJobIDTx records the River job id on the intake row, in the same accept-tx as the insert + enqueue (so a committed accepted row always has its job).

func (*Store) StampNotifyJobIDTx ¶

func (s *Store) StampNotifyJobIDTx(ctx context.Context, tx pgx.Tx, messageID string, jobID int64) error

StampNotifyJobIDTx records the River QueueNotify job id on the pending_review message, within the hold accept-tx, so the notification reconciler can find rows stranded without a job (pending_review AND notify_job_id IS NULL). Mirrors StampSendJobIDTx.

func (*Store) StampSendJobIDTx ¶

func (s *Store) StampSendJobIDTx(ctx context.Context, tx pgx.Tx, messageID string, jobID int64) error

StampSendJobIDTx records the River outbound_send job id on the accepted message, within the accept-tx, so the async-send reconciler can find stranded rows ('accepted' with send_job_id IS NULL). Mirrors the webhook_subscriber_deliveries .job_id stamp.

func (*Store) SuppressedAddresses ¶

func (s *Store) SuppressedAddresses(ctx context.Context, userID string, addrs []string) ([]string, error)

SuppressedAddresses returns the subset of addrs that are suppressed for the user — the send-time enforcement read. Empty input → empty result.

func (*Store) TouchDomainLastChecked ¶

func (s *Store) TouchDomainLastChecked(ctx context.Context, domain, userID string) error

TouchDomainLastChecked records that the verification probe ran. Call this from POST /v1/domains/{domain}/verify whether the probe succeeded or not — the LastCheckedAt column is "when did we last try", not "when did we last succeed" (the latter is verified_at).

func (*Store) TouchSendingChecked ¶

func (s *Store) TouchSendingChecked(ctx context.Context, domain string) error

TouchSendingChecked stamps sending_last_checked_at without changing status (a still-pending poll).

func (*Store) UpdateAgentHITL ¶

func (s *Store) UpdateAgentHITL(ctx context.Context, agentID, userID string, ttlSeconds int, expirationAction string) error

UpdateAgentHITL updates all three HITL settings on an agent owned by userID. The TTL and expiration action are validated against the same rules as the DB CHECK constraints so callers get a clean error rather than a raw SQL error.

func (*Store) UpdateAgentInboundPolicy ¶

func (s *Store) UpdateAgentInboundPolicy(ctx context.Context, agentID, userID, policy string, allowlist []string) error

func (*Store) UpdateAgentName ¶

func (s *Store) UpdateAgentName(ctx context.Context, agentID, userID, name string) error

UpdateAgentName sets an agent's display name for an agent owned by userID. The name is a UI label only — the agent's identity is its email. Returns an error if the agent isn't found or not owned.

func (*Store) UpdateAgentProtection ¶

func (s *Store) UpdateAgentProtection(ctx context.Context, agentID, userID string, c ProtectionConfig) error

UpdateAgentProtection writes the full protection posture for an agent owned by userID in a single statement, validating first. It writes the sensitivity columns (the API source of truth) AND the derived scan toggle + float thresholds the piguard engine reads, so the two never diverge and the engine needs no change. Returns an error if the agent isn't found or not owned.

func (*Store) UpdateAgentScanConfig ¶

func (s *Store) UpdateAgentScanConfig(ctx context.Context, agentID, userID string, c ScanConfig) error

UpdateAgentScanConfig writes the full screening config for an agent owned by userID, validating the effective posture first. Returns an error if the agent isn't found or isn't owned by the user.

func (*Store) UpdateMessageDeliveryStatus ¶

func (s *Store) UpdateMessageDeliveryStatus(ctx context.Context, messageID, agentID, status string) error

UpdateMessageDeliveryStatus sets the inbox_status on a message.

func (*Store) UpdateTemplate ¶

func (s *Store) UpdateTemplate(ctx context.Context, templateID, userID string, u TemplateUpdate) (*Template, error)

UpdateTemplate applies a partial update to a template owned by the user. Only non-nil fields are touched; updated_at is always bumped. Returns the updated row, ErrTemplateNotFound for missing/cross-user rows, and ErrTemplateAliasTaken on an alias collision.

func (*Store) UpdateUserName ¶

func (s *Store) UpdateUserName(ctx context.Context, userID, name string) (*User, error)

UpdateUserName persists a new display name on the user row and returns the updated User. Input validation (length, whitespace) is the caller's responsibility — this layer only enforces that the row exists.

func (*Store) UpdateWebhook ¶

func (s *Store) UpdateWebhook(ctx context.Context, webhookID, userID string, u WebhookUpdate) (*Webhook, error)

UpdateWebhook applies a partial update to a webhook. Only fields with a non-nil pointer in WebhookUpdate are touched. Returns the updated row.

Validation (charset, count caps, agent ownership) is the handler's job; the storage layer enforces only the re-enable cooldown and the per-row CHECK constraints (events non-empty, url non-empty).

func (*Store) VerifyDomain ¶

func (s *Store) VerifyDomain(ctx context.Context, domain, userID string) error

VerifyDomain marks a domain as verified, only if owned by the given user.

func (*Store) WithTx ¶

func (s *Store) WithTx(ctx context.Context, fn func(tx pgx.Tx) error) error

WithTx opens a transaction, runs fn inside it, and commits if fn returns nil (or rolls back if fn returns an error). Used by the slice-3 relay refactor so the messages INSERT and the webhook_events outbox INSERT commit together, closing the at-least-once publish-loss window.

The relay handler is the primary v1 caller; future trigger sites (slice 4 outbound + HITL) reuse the same helper. Keeps callers from having to import pgxpool directly.