messagelifecycle

package
v1.7.3 Latest Latest
Warning

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

Go to latest
Published: Aug 9, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Overview

Package messagelifecycle defines the canonical vocabulary and validated in-memory representation of message lifecycle observations.

Index

Constants

View Source
const MaxMetricsAgents = 200

MaxMetricsAgents bounds the per-agent breakdown. Plans allow agent counts in the tens of thousands, and a dashboard cannot render that many rows anyway, so the breakdown returns the busiest agents and reports that it truncated rather than silently presenting a partial list as the whole account.

View Source
const MaxMetricsWindow = 92 * 24 * time.Hour

MaxMetricsWindow bounds one metrics read. The aggregate drives off idx_messages_agent_created and then probes the lifecycle ledger once per message in range, so its cost scales with the agent's volume inside the window rather than with the size of either table. A quarter is the widest span that keeps the worst-case probe count bounded for a high-volume agent; wider ranges belong to a pre-aggregated rollup, not to this live read.

Variables

View Source
var ErrDedupeConflict = errors.New("message lifecycle dedupe conflict")

ErrDedupeConflict means a producer reused a message-local dedupe key for a semantically different lifecycle observation.

View Source
var ErrDirectionMismatch = errors.New("message lifecycle direction does not match message")

ErrDirectionMismatch means a producer attempted to append a lifecycle fact whose direction contradicts the owning message.

View Source
var ErrMessageNotFound = errors.New("message lifecycle message not found")

ErrMessageNotFound intentionally covers both absent and foreign messages.

Functions

func Catalog

func Catalog() map[ReasonCode]Definition

Catalog returns a copy of the canonical reason-code catalog.

func IsTerminalSubmissionFailure

func IsTerminalSubmissionFailure(code ReasonCode) bool

IsTerminalSubmissionFailure reports whether code is a catalog-owned submission/failed observation suitable for the durable message failure rollup. Arbitrary database text must never reach AppendTx through this path.

func SafeAuthenticationEvidence

func SafeAuthenticationEvidence(authentication any) (map[string]any, error)

SafeAuthenticationEvidence returns canonical authentication evidence that is safe to append even when diagnostic fields originated in untrusted SMTP/DNS input. In-bounds authentication JSON is preserved exactly. Oversized optional diagnostics are omitted, oversized nullable identifiers become null, and excess DKIM observations are dropped from the tail deterministically until the complete evidence fits the canonical aggregate limit.

The input is intentionally any rather than emailauth.Authentication so this package remains the contract-owning leaf and does not depend on an auth producer package.

func SafeCorrelationIDs

func SafeCorrelationIDs(correlationIDs map[string]string) map[string]string

SafeCorrelationIDs keeps independently valid optional correlations and drops unknown, empty, or oversized values. Producers use it for identifiers derived from remote input so an unsafe optional correlation can never reject the lifecycle fact it merely annotates.

func SafeDiagnostic

func SafeDiagnostic(value string) string

SafeDiagnostic bounds optional untrusted diagnostics to the canonical per-string limit without splitting a UTF-8 encoding. Invalid input bytes are discarded so optional evidence can never reject the lifecycle observation.

func SendSuppressionDedupeKey

func SendSuppressionDedupeKey(jobID int64, attempt int, recipient string) string

SendSuppressionDedupeKey is the stable message-local identity for a recipient suppression observed by one outbound River attempt.

Types

type AccountMetrics added in v1.7.0

type AccountMetrics struct {
	// Totals covers the whole account and is always complete — it is computed
	// across every agent, never from the possibly-truncated Agents slice.
	Totals AgentMetrics

	// Agents is the optional per-agent breakdown, busiest first. Empty when
	// the caller did not ask for it.
	Agents []AgentMetricsGroup

	// AgentsTruncated reports that the account owns more agents with traffic
	// than Agents contains. Totals remain accurate; only the breakdown is cut.
	AgentsTruncated bool
}

AccountMetrics is the aggregate across every agent an account owns.

type AgentMetrics added in v1.7.0

type AgentMetrics struct {
	// Counts holds one entry per reason code observed in the window, ordered
	// by reason code. Codes with no observations are absent rather than zero.
	Counts []ReasonCodeCount

	// MessagesInWindow and MessagesWithLifecycle expose ledger coverage.
	//
	// This aggregate reads only PERSISTED transitions. The per-message read
	// path (ListForMessage) additionally reconstructs observations from
	// durable message state, so a message written before the ledger shipped —
	// or by a path that never appended — shows a lifecycle there while
	// contributing nothing here. Reporting both counts turns that gap into a
	// number the caller can see instead of a silent undercount that reads as
	// a delivery problem.
	MessagesInWindow      int64
	MessagesWithLifecycle int64

	// ReconstructedObservations counts ledger rows in the window that were
	// derived from durable state rather than observed at the boundary. They
	// are included in Counts; this field makes the inferred share visible.
	ReconstructedObservations int64
}

AgentMetrics is the aggregate for one agent over one cohort window.

type AgentMetricsGroup added in v1.7.0

type AgentMetricsGroup struct {
	AgentEmail string
	Metrics    AgentMetrics
}

AgentMetricsGroup is one agent's slice of an account aggregate. AgentEmail is the agent's address, which is also its primary key.

type AppendInput

type AppendInput struct {
	MessageID      string
	DedupeKey      string
	Direction      string
	Recipient      string
	ReasonCode     ReasonCode
	Evidence       map[string]any
	CorrelationIDs map[string]string
	OccurredAt     time.Time
}

AppendInput contains the producer-controlled fields used to construct a transition. DedupeKey is validated here and retained by the persistence operation introduced with the lifecycle store.

type Definition

type Definition struct {
	Stage     Stage
	Outcome   Outcome
	Retryable bool
}

Definition is the fixed meaning of a reason code.

func Lookup

func Lookup(reason ReasonCode) (Definition, bool)

Lookup returns the immutable definition for reason.

type EventSnapshot

type EventSnapshot struct {
	ID        string
	Type      string
	Envelope  json.RawMessage
	CreatedAt time.Time
}

type MessageLifecycleTransition

type MessageLifecycleTransition struct {
	ID                 string            `json:"id"`
	MessageID          string            `json:"message_id"`
	DedupeKey          string            `json:"-"`
	SourceTransitionID string            `json:"-"`
	Direction          string            `json:"direction" enum:"inbound,outbound"`
	Recipient          string            `json:"recipient,omitempty" nullable:"true"`
	Stage              Stage             `json:"stage" enum:"accepted,authentication,review,suppression,queued,submission,delivery,complaint"`
	Outcome            Outcome           `` /* 145-byte string literal not displayed */
	ReasonCode         ReasonCode        `` /* 842-byte string literal not displayed */
	Retryable          bool              `json:"retryable"`
	Evidence           map[string]any    `json:"evidence"`
	CorrelationIDs     map[string]string `json:"correlation_ids"`
	OccurredAt         time.Time         `json:"occurred_at"`
	Reconstructed      bool              `json:"reconstructed"`
}

MessageLifecycleTransition is one validated canonical lifecycle observation.

func AppendTx

func AppendTx(ctx context.Context, tx pgx.Tx, input AppendInput) (MessageLifecycleTransition, error)

AppendTx validates and appends one lifecycle observation in the caller's transaction. Identical retries return the original stored transition.

func MergeTransitions

func MergeTransitions(persisted, reconstructed []MessageLifecycleTransition) []MessageLifecycleTransition

MergeTransitions retains every persisted observation and suppresses only a reconstructed candidate with the same source observation identity.

func NewTransition

func NewTransition(input AppendInput) (MessageLifecycleTransition, error)

NewTransition validates input and derives the semantic tuple from the closed reason catalog. The returned unpersisted transition has an empty ID; the lifecycle store assigns its stable mlt_ identifier when appending it.

func Reconstruct

func Reconstruct(snapshot Snapshot) []MessageLifecycleTransition

Reconstruct derives only facts proven by the supplied durable snapshot.

type Outcome

type Outcome string

Outcome identifies what e2a observed at a lifecycle stage.

const (
	OutcomeAccepted      Outcome = "accepted"
	OutcomePassed        Outcome = "passed"
	OutcomeFailed        Outcome = "failed"
	OutcomeIndeterminate Outcome = "indeterminate"
	OutcomePending       Outcome = "pending"
	OutcomeApproved      Outcome = "approved"
	OutcomeRejected      Outcome = "rejected"
	OutcomeBlocked       Outcome = "blocked"
	OutcomeApplied       Outcome = "applied"
	OutcomeEnqueued      Outcome = "enqueued"
	OutcomeDeferred      Outcome = "deferred"
	OutcomeDelivered     Outcome = "delivered"
	OutcomeBounced       Outcome = "bounced"
	OutcomeReported      Outcome = "reported"
)

type ReasonCode

type ReasonCode string

ReasonCode is a stable machine-readable lifecycle observation.

const (
	ReasonAcceptanceInboundSMTP             ReasonCode = "acceptance.inbound_smtp"
	ReasonAcceptanceOutboundAPI             ReasonCode = "acceptance.outbound_api"
	ReasonAcceptanceLocalLoopback           ReasonCode = "acceptance.local_loopback"
	ReasonAuthenticationDMARCPass           ReasonCode = "authentication.dmarc_pass"
	ReasonAuthenticationDMARCFail           ReasonCode = "authentication.dmarc_fail"
	ReasonAuthenticationDMARCNone           ReasonCode = "authentication.dmarc_none"
	ReasonAuthenticationDMARCTemporaryError ReasonCode = "authentication.dmarc_temporary_error"
	ReasonAuthenticationDMARCPermanentError ReasonCode = "authentication.dmarc_permanent_error"
	ReasonReviewHoldCreated                 ReasonCode = "review.hold_created"
	ReasonReviewApproved                    ReasonCode = "review.approved"
	ReasonReviewRejected                    ReasonCode = "review.rejected"
	ReasonReviewExpiredApproved             ReasonCode = "review.expired_approved"
	ReasonReviewExpiredRejected             ReasonCode = "review.expired_rejected"
	ReasonSuppressionRecipientBlocked       ReasonCode = "suppression.recipient_blocked"
	ReasonSuppressionHardBounceApplied      ReasonCode = "suppression.hard_bounce_applied"
	ReasonSuppressionComplaintApplied       ReasonCode = "suppression.complaint_applied"
	ReasonQueueInboundProcessing            ReasonCode = "queue.inbound_processing"
	ReasonQueueOutboundSubmission           ReasonCode = "queue.outbound_submission"
	ReasonSubmissionUpstreamAccepted        ReasonCode = "submission.upstream_accepted"
	ReasonSubmissionLocalLoopbackAccepted   ReasonCode = "submission.local_loopback_accepted"
	ReasonSubmissionTemporaryFailure        ReasonCode = "submission.temporary_failure"
	ReasonSubmissionProviderRejected        ReasonCode = "submission.provider_rejected"
	ReasonSubmissionLocalRetriesExhausted   ReasonCode = "submission.local_retries_exhausted"
	ReasonSubmissionCancelled               ReasonCode = "submission.cancelled"
	ReasonDeliveryRecipientServerAccepted   ReasonCode = "delivery.recipient_server_accepted"
	ReasonDeliveryTemporaryDelay            ReasonCode = "delivery.temporary_delay"
	ReasonDeliveryPermanentBounce           ReasonCode = "delivery.permanent_bounce"
	ReasonDeliveryTransientBounce           ReasonCode = "delivery.transient_bounce"
	ReasonDeliveryUndeterminedBounce        ReasonCode = "delivery.undetermined_bounce"
	ReasonComplaintRecipientReported        ReasonCode = "complaint.recipient_reported"
)

func AuthenticationReason

func AuthenticationReason(status string) (ReasonCode, error)

AuthenticationReason maps a normalized DMARC status to its canonical reason.

func BounceReason

func BounceReason(bounceType string) ReasonCode

BounceReason maps a normalized provider bounce type to its canonical reason. Unknown values use the provider normalizer's undetermined catch-all.

type ReasonCodeCount added in v1.7.0

type ReasonCodeCount struct {
	ReasonCode   ReasonCode
	Stage        Stage
	Outcome      Outcome
	Retryable    bool
	Observations int64
	Messages     int64
}

ReasonCodeCount is one canonical reason code's tally inside a cohort window.

The two counts answer different questions and routinely differ:

  • Observations counts ledger rows. Retryable codes are appended once per attempt, so a message deferred four times contributes four observations of submission.temporary_failure.
  • Messages counts distinct messages carrying at least one such row. This is the grain funnel arithmetic needs: counting observations there would let a single retried message push a stage above its own denominator.

type RecipientSnapshot

type RecipientSnapshot struct {
	ID        string
	Address   string
	Status    string
	Detail    string
	UpdatedAt time.Time
}

type Snapshot

type Snapshot struct {
	MessageID             string
	AgentID               string
	UserID                string
	Direction             string
	Method                string
	CreatedAt             time.Time
	Authentication        json.RawMessage
	Status                string
	ApprovalExpiresAt     *time.Time
	ReviewedAt            *time.Time
	SendJobID             *int64
	JobCreatedAt          *time.Time
	ProviderAcceptedAt    *time.Time
	ProviderMessageID     string
	EmailMessageID        string
	DeliveryStatus        string
	DeliveryFailureSource string
	Recipients            []RecipientSnapshot
	Suppressions          []SuppressionSnapshot
	Events                []EventSnapshot
}

Snapshot is the bounded durable state used to reconstruct one message.

type Stage

type Stage string

Stage identifies the boundary at which e2a made an observation.

const (
	StageAccepted       Stage = "accepted"
	StageAuthentication Stage = "authentication"
	StageReview         Stage = "review"
	StageSuppression    Stage = "suppression"
	StageQueued         Stage = "queued"
	StageSubmission     Stage = "submission"
	StageDelivery       Stage = "delivery"
	StageComplaint      Stage = "complaint"
)

type Store

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

Store reads canonical and conservatively reconstructed message lifecycle facts.

func NewStore

func NewStore(pool *pgxpool.Pool) *Store

func (*Store) CountByReasonCode added in v1.7.0

func (s *Store) CountByReasonCode(ctx context.Context, agentID string, start, end time.Time) (AgentMetrics, error)

CountByReasonCode aggregates persisted lifecycle observations for one agent's messages, anchored on the message's own created_at.

The window is anchored on the SEND cohort, not on observation time, and is half-open: [start, end). Anchoring on occurred_at would let a bounce that lands six hours after midnight pollute the following day's rate and would leave every ratio permanently unable to converge, because the numerator and denominator would be drawn from different populations. The cost of cohort anchoring is that a recent window keeps moving as late feedback arrives — callers must treat roughly the last 72 hours as still settling.

Trashed-but-unpurged messages are deliberately included: the mail was still accepted and still sent, and excluding it would silently restate history whenever a customer cleans up an inbox.

func (*Store) CountByReasonCodeForAccount added in v1.7.0

func (s *Store) CountByReasonCodeForAccount(ctx context.Context, userID string, start, end time.Time, groupByAgent bool) (AccountMetrics, error)

CountByReasonCodeForAccount aggregates persisted lifecycle observations across every agent owned by userID, on the same cohort-window contract as CountByReasonCode (see its doc comment: half-open, anchored on the message's own created_at, feedback still settling for ~72h).

Trashed-but-unpurged MESSAGES are included: that mail was still sent, and dropping it would let inbox housekeeping restate past delivery rates.

Trashed AGENTS are excluded. That mail was also still sent, so this does restate history slightly — the tradeoff is deliberate. A deleted agent is gone from every other account-scoped surface (quota counts, list, get without an explicit opt-in), so counting it only here was the odd one out; and because agent deletion is soft, an account that churns agents keeps every tombstone for the whole 30-day retention window. Past a few thousand of them the planner abandons idx_messages_agent_created and sequentially scans the entire messages/transitions tables: measured on a test account with ~7k trashed agents, the same aggregate went 22ms → 1015ms. Scoping to live agents keeps the cost proportional to what the account currently owns rather than to everything it has ever owned.

When groupByAgent is set, the busiest MaxMetricsAgents agents also come back individually. Totals are computed independently of that cap.

func (*Store) ListForMessage

func (s *Store) ListForMessage(ctx context.Context, messageID, agentID string) ([]MessageLifecycleTransition, error)

ListForMessage returns the lifecycle for one message owned by agentID. Reconstruction is read-only and is never persisted by this method.

type SuppressionSnapshot

type SuppressionSnapshot struct {
	ID              string
	Address         string
	Source          string
	SourceMessageID string
	CreatedAt       time.Time
}

Jump to

Keyboard shortcuts

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