// The SES event ingest endpoint (tenant-email spec §9): SNS delivers send,
// bounce and complaint notifications here, and they become the per-tenant
// counts that enforce a cap and trip an automatic pause (Task 16).
//
// UNAUTHENTICATED, UNLIKE EVERY OTHER DOOR IN THIS PACKAGE. SNS is not a
// signed-in member — it cannot present a session cookie, a CSRF token, or
// prove account ownership — so handleSESEvents has none of those gates.
// What replaces them, in the order they run below, is the door's entire
// security model: bound the body, refuse a topic that isn't ours, validate
// where the verifying certificate may come from, and verify the signature
// before trusting any field the message carries.
//
// THE TRUST MODEL, IN ONE PARAGRAPH (read this before touching anything
// below — four review rounds of history follow it, and the history is
// there so the model doesn't get "corrected" back to something that
// already failed): this door does NOT verify a certificate chain for SNS's
// signing certificate, and that is deliberate. Trust instead comes from
// THREE things together — the request is HTTPS (so the fetch itself rides
// on a TLS connection whose OWN chain the Go http client already validates
// against system roots — that is the chain check, done at the transport
// layer, which is the only layer where it can actually succeed); the host
// is PINNED to this deployment's own sns.<region>.amazonaws.com (or a
// region-shaped fallback), so the TLS-validated connection is going
// somewhere that is provably SNS's own infrastructure and not a stranger's;
// and REDIRECTS are refused, so a 3xx response from that pinned, validated
// host cannot hand control to a different, unvalidated one. On top of that,
// the downloaded certificate's own subject is checked against
// snsCertDNSName as a sanity check (catches a misconfigured or
// wrong-endpoint response), NOT as a trust decision — there is no chain
// verification of the certificate BODY at all, because (see Round 4 below)
// there is no chain a client can build for it.
//
// SECURITY: validSigningCertURL and fetchVerifiedCertKey TOGETHER are the
// entirety of SNS signature security. SNS tells the receiver where to fetch
// the certificate that verifies the message — so an unvalidated URL, or an
// unvalidated certificate, lets anyone who can reach this endpoint host
// their own certificate, sign their own payload, and have it verify. The
// consequence here is not merely bad data: forged bounce events trip the
// automatic pause, so it is a remote off-switch for any tenant's email.
//
// A host allowlist is NOT a signer identity by itself, and a parsed
// certificate is NOT a verified one — but ALSO, a certificate chain is not
// always buildable, and code that insists on building one anyway can fail
// closed on every legitimate message. Four review rounds' worth of fixes
// live in this file because each of those was true in turn:
//
//   - Round 1: the original check accepted any *.amazonaws.com host, which
//     is every AWS customer's identity, not Amazon's.
//   - Round 2 (C1-residual): pinning only the "sns." label and leaving the
//     region label free still accepted "sns.s3.amazonaws.com" — a real,
//     registered S3 bucket — the exact shape of the documented
//     sns-validator bypass. validSigningCertURL now pins to this
//     deployment's OWN region exactly when it is known, and falls back to
//     a region-SHAPED pattern (never a bare service name) only when it is
//     not.
//   - Round 2 (C2): the certificate was parsed and its public key trusted
//     with no chain verification and no subject check at all — a
//     self-signed certificate would have verified.
//   - Round 3 (C2 fixed-closed bug, attempt 1): the FIRST attempt at C2
//     asserted DNSName against the FETCH host (e.g.
//     sns.eu-west-1.amazonaws.com) and decoded only the first PEM block,
//     discarding any intermediate — but a genuine SNS signing certificate
//     names "sns.amazonaws.com" (one identity for the whole service, not
//     one per region), so this failed closed on every real message.
//   - Round 3 (redirect guard): both outbound fetches used
//     http.DefaultClient, which follows redirects — for the SubscribeURL
//     fetch specifically, a redirect target is never re-validated against
//     the host pin, so following one defeats it outright. Both fetches now
//     use snsFetchClient, which refuses to follow a redirect at all. (This
//     fix stands; it is not touched by Round 4.)
//   - Round 4 (C2 fixed-closed bug, attempt 2 — the SECOND reversal on this
//     paragraph, and the one that stuck): attempt 1's fix DID assert the
//     right subject, but it also tried to build an Intermediates pool from
//     the response and verify the chain against system roots. THIS ALSO
//     FAILS CLOSED, for a completely different reason than attempt 1's:
//     AWS serves EXACTLY ONE certificate in a real SigningCertURL response
//     — a bare leaf, no bundled intermediate (verified: `grep -c 'BEGIN
//     CERTIFICATE'` on a real captured SimpleNotificationService-*.pem is
//     1). That leaf's issuer is an intermediate CA (e.g. "Amazon RSA 2048
//     M04") distributed via the leaf's AIA (Authority Information Access)
//     extension — which Go's x509.Verify never fetches; it does not chase
//     AIA. System root stores hold only the top-level Amazon Root CA
//     certificates, several hops above that intermediate. So an
//     Intermediates pool built only from what the response contains is
//     ALWAYS empty against a real message, and cert.Verify against system
//     roots always returns "certificate signed by unknown authority" —
//     proven directly against a real captured certificate, with
//     CurrentTime set inside its validity window so expiry could not mask
//     the result. Every reference implementation agrees there is no chain
//     to build here: aws-sdk-ruby's message_verifier.rb, robbiet480/go.sns,
//     and AWS's own documented `openssl x509 -pubkey -noout` procedure all
//     decode exactly ONE certificate block and verify no chain at all.
//     fetchVerifiedCertKey now does the same: decode one leaf, assert its
//     subject, done — no x509.CertPool, no Intermediates, no cert.Verify.
//     DO NOT reintroduce a chain build here without re-deriving why it was
//     removed twice.
package console

import (
	"context"
	"crypto"
	"crypto/rsa"
	"crypto/sha1"
	"crypto/sha256"
	"crypto/x509"
	"encoding/base64"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"regexp"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/mail"
)

// sesEventMaxBody bounds the request body this door reads. SNS messages are
// small (an SES notification is at most a few KB); an unbounded read on an
// unauthenticated endpoint is memory exhaustion anyone can trigger.
const sesEventMaxBody = 256 << 10

// sesCertMaxBody bounds the signing certificate this door fetches (and the
// SubscribeURL response body it discards). AWS's SNS certificates are a few
// KB, even bundled with an intermediate; anything vastly larger than that
// is not a certificate response this code needs to finish reading.
const sesCertMaxBody = 64 << 10

// sesFetchTimeout bounds both outbound calls this door makes — fetching the
// signing certificate and confirming a subscription. Neither should ever
// approach it against a healthy AWS endpoint; it exists so a wedged
// far end costs one request this long, not a goroutine forever.
const sesFetchTimeout = 10 * time.Second

// snsFetchClient is used for both outbound fetches this door makes — the
// signing certificate and the SubscribeURL confirmation.
//
// CheckRedirect refuses to follow. For the SubscribeURL fetch this is not
// optional hardening: validSigningCertURL only ever inspects the ORIGINAL
// URL, so a 3xx response pointing anywhere else would defeat the host pin
// outright — the far end could serve a redirect to any host it likes and
// this client would have followed it there. http.ErrUseLastResponse is the
// documented way to say "hand me the redirect response itself, don't chase
// it" — the caller then sees a non-2xx status and refuses it the same way
// it refuses any other failure. Precedent: internal/edge/prober.go's client
// uses the identical idiom for the identical reason — a redirect is not the
// thing that was asked for.
var snsFetchClient = &http.Client{
	CheckRedirect: func(*http.Request, []*http.Request) error {
		return http.ErrUseLastResponse
	},
}

// snsHostPattern is the FALLBACK shape check, used only when this
// deployment's own region is unknown: "sns." plus a REGION-SHAPED middle
// label — "<2-letter-country>[-gov]-<name>-<digit>", e.g. "eu-west-1",
// "us-gov-west-1", "ap-southeast-4" — plus ".amazonaws.com", and nothing
// looser than that. A bare service name like "s3" or "s3-us-west-2" does
// not have this shape, which is what closes the C1-residual bypass
// described below.
var snsHostPattern = regexp.MustCompile(`^sns\.[a-z]{2}(-gov)?-[a-z]+-\d+\.amazonaws\.com$`)

// validSigningCertURL refuses anything that is not an https URL on an SNS
// host — and, when region is non-empty, anything that is not THIS
// deployment's own SNS regional host exactly. It is also reused, unchanged,
// to validate SubscribeURL before that URL is ever fetched (round-2
// IMPORTANT-1 fix): a verified signature only proves AWS put some string in
// that field, not that the string points somewhere safe to GET, and a real
// SubscribeURL is itself always an SNS regional URL, so the same check is
// the right one for both.
//
// C1-RESIDUAL FIX (round 2 review): the previous pattern,
// ^sns\.[a-z0-9-]+\.amazonaws\.com$, pins the FIRST label ("sns.") but
// leaves the SECOND label completely unconstrained — which is the exact
// shape of the documented sns-validator bypass:
// "https://sns.s3.amazonaws.com/x.pem" matches it, and "sns" is a real,
// registered AWS bucket name (the review confirmed it returns 403
// AccessDenied — owned by someone, reachable, and NOT SNS). When this
// deployment knows its own region (MailDeployment.Region, threaded in by
// the caller), the host must equal "sns.<region>.amazonaws.com" EXACTLY —
// no pattern, no label left free to be a bucket name. Only when the region
// is unknown does this fall back to snsRegionPattern, which is at least
// narrow enough to refuse a customer-chosen name: it requires the middle
// label to have the SHAPE of an actual AWS region code, which "s3" and
// "s3-us-west-2" do not.
//
// The host check is also LABEL-AWARE against the suffix-lookalike class of
// bug: a bare strings.HasSuffix(host, "amazonaws.com") would accept
// "evilamazonaws.com" — it genuinely ends with the string "amazonaws.com",
// with no dot separating the two, so it is a completely different
// registered domain that merely happens to end the right way. Anchoring the
// whole pattern (or the whole exact-match string) is what refuses that, and
// also refuses names like "sns.eu-west-1.amazonaws.com.evil.com" that
// merely CONTAIN the right substring in the middle.
func validSigningCertURL(raw, region string) error {
	u, err := url.Parse(raw)
	if err != nil {
		return fmt.Errorf("sns: unparseable SigningCertURL")
	}
	if u.Scheme != "https" {
		return fmt.Errorf("sns: SigningCertURL is not https")
	}
	host := strings.ToLower(u.Hostname())
	region = strings.ToLower(strings.TrimSpace(region))
	switch {
	case region != "":
		if host != "sns."+region+".amazonaws.com" {
			return fmt.Errorf("sns: SigningCertURL host %q is not this deployment's SNS host (sns.%s.amazonaws.com)", host, region)
		}
	case !snsHostPattern.MatchString(host):
		return fmt.Errorf("sns: SigningCertURL host %q is not an SNS host", host)
	}
	// A traversal in the path can walk off the certificate route on a host
	// that serves more than certificates.
	if strings.Contains(u.Path, "..") {
		return fmt.Errorf("sns: SigningCertURL path contains a traversal")
	}
	return nil
}

// snsEnvelope is the outer SNS message shape — the fields this door reads
// off of every delivery, notification or confirmation alike. Field names
// and casing match SNS's own JSON exactly; there is no tolerance for a
// renamed field because these are also exactly the fields the signature
// covers (canonicalString below).
type snsEnvelope struct {
	Type             string `json:"Type"`
	MessageID        string `json:"MessageId"`
	Token            string `json:"Token,omitempty"`
	TopicArn         string `json:"TopicArn"`
	Subject          string `json:"Subject,omitempty"`
	Message          string `json:"Message"`
	SubscribeURL     string `json:"SubscribeURL,omitempty"`
	Timestamp        string `json:"Timestamp"`
	SignatureVersion string `json:"SignatureVersion"`
	Signature        string `json:"Signature"`
	SigningCertURL   string `json:"SigningCertURL"`
}

// canonicalString builds the exact byte string SNS signs, per AWS's
// documented format: alternating field name and value, each followed by a
// newline, fields in ALPHABETICAL order, and only the fields that apply to
// this message's Type. Subject is included only when the message actually
// carries one (SNS omits it from the signed string otherwise);
// SubscribeURL/Token belong to confirmation messages only.
func (e snsEnvelope) canonicalString() string {
	var b strings.Builder
	add := func(k, v string) {
		b.WriteString(k)
		b.WriteByte('\n')
		b.WriteString(v)
		b.WriteByte('\n')
	}
	add("Message", e.Message)
	add("MessageId", e.MessageID)
	switch e.Type {
	case "SubscriptionConfirmation", "UnsubscribeConfirmation":
		add("SubscribeURL", e.SubscribeURL)
		add("Timestamp", e.Timestamp)
		add("Token", e.Token)
		add("TopicArn", e.TopicArn)
		add("Type", e.Type)
	default:
		if e.Subject != "" {
			add("Subject", e.Subject)
		}
		add("Timestamp", e.Timestamp)
		add("TopicArn", e.TopicArn)
		add("Type", e.Type)
	}
	return b.String()
}

// handleSESEvents is where SNS delivers every SES send, bounce and
// complaint notification for this deployment.
//
// The checks below run in this exact order, and the order is the whole
// point:
//
//  1. Bound the body. Nothing about an unauthenticated endpoint should read
//     an unbounded amount of attacker-controlled data before anything else
//     has a chance to refuse it.
//  2. Refuse a foreign TopicArn. This runs before ANY network call this
//     handler might make (the certificate fetch in step 4, the subscribe
//     GET in dispatch) — otherwise a stranger's POST decides who this
//     console makes outbound requests to, which is SSRF wearing an
//     AWS-shaped alibi.
//  3. Validate SigningCertURL's shape — https, an SNS host (this
//     deployment's own region exactly, when known), no path traversal —
//     before it is ever fetched.
//  4. Verify the signature over SNS's canonical string, which includes
//     fetching the certificate and checking ITS chain and subject too.
//     Nothing past this point trusts a single field of the message,
//     including SubscribeURL, until this has succeeded.
//  5. Dispatch: confirm a subscription (after running its SubscribeURL
//     through the SAME host check as step 3), or parse a notification.
//     Anything this deployment deliberately does not act on still answers
//     200 — SNS retries non-2xx, so refusing an unknown case with a
//     failure status would turn one unrecognised field into an unbounded
//     retry flood against the console.
func (s *Server) handleSESEvents(w http.ResponseWriter, r *http.Request) {
	// 1. Bound the body first.
	r.Body = http.MaxBytesReader(w, r.Body, sesEventMaxBody)
	body, err := io.ReadAll(r.Body)
	if err != nil {
		var tooLarge *http.MaxBytesError
		if errors.As(err, &tooLarge) {
			jsonError(w, http.StatusRequestEntityTooLarge, "request body too large")
			return
		}
		jsonError(w, http.StatusBadRequest, "could not read request body")
		return
	}

	var env snsEnvelope
	if err := json.Unmarshal(body, &env); err != nil {
		jsonError(w, http.StatusBadRequest, "malformed SNS envelope")
		return
	}

	// 2. Refuse a foreign TopicArn before fetching anything. An unconfigured
	// SNSTopicARN ("") never matches — a deployment with no topic set
	// accepts nothing, rather than accepting whatever a caller who also
	// sends "" happens to send.
	if s.cfg.MailDeployment.SNSTopicARN == "" || env.TopicArn != s.cfg.MailDeployment.SNSTopicARN {
		s.logRefusedTopic(env.TopicArn)
		jsonError(w, http.StatusForbidden, "unrecognised SNS topic")
		return
	}

	// 3. Validate SigningCertURL before it is ever fetched.
	region := s.cfg.MailDeployment.Region
	if err := validSigningCertURL(env.SigningCertURL, region); err != nil {
		jsonError(w, http.StatusBadRequest, err.Error())
		return
	}

	// 4. Verify the signature before acting on ANY field, including
	// SubscribeURL. snsVerify is the test seam — nil means "do the real
	// verification" (verifySNSSignature, which fetches the certificate over
	// the network). The seam is signature verification ONLY: both checks
	// above already ran unconditionally, for real, regardless of the seam.
	//
	// The real path is wrapped in a closure so it can carry the request's
	// own context (matching confirmSNSSubscription below) rather than
	// context.Background() — without changing the seam's own type, which
	// stays exactly func(snsEnvelope) error so a test's assignment to
	// s.snsVerify is unaffected.
	verify := s.snsVerify
	if verify == nil {
		verify = func(e snsEnvelope) error { return s.verifySNSSignature(r.Context(), e) }
	}
	if err := verify(env); err != nil {
		log.Printf("console: ses events: signature verification failed: %v", err)
		jsonError(w, http.StatusForbidden, "signature verification failed")
		return
	}

	// 5. Dispatch.
	switch env.Type {
	case "SubscriptionConfirmation", "UnsubscribeConfirmation":
		// IMPORTANT-1 fix: SubscribeURL gets the SAME pinned-host check as
		// SigningCertURL before it is fetched. A verified signature only
		// proves AWS put some string in this field — it says nothing about
		// where that string points — so without this an authentic message
		// can still carry a SubscribeURL aimed at an arbitrary host,
		// including link-local metadata endpoints.
		if err := validSigningCertURL(env.SubscribeURL, region); err != nil {
			log.Printf("console: ses events: refusing SubscribeURL: %v", err)
			jsonError(w, http.StatusBadRequest, "invalid SubscribeURL")
			return
		}
		// UnsubscribeConfirmation's SubscribeURL is, by SNS's own naming,
		// actually a RE-subscribe link: fetching it here means an operator
		// who unsubscribes this topic from the SNS console finds the
		// endpoint re-subscribed within seconds, the next time SNS sends an
		// UnsubscribeConfirmation. Defensible — this deployment always
		// wants to be subscribed to its own mail-events topic, so an
		// unsubscribe is far more likely to be SNS housekeeping or an
		// operator mistake than an intentional retirement — but it is not
		// what "Unsubscribe" reads as, so it is named here rather than left
		// to be discovered.
		ctx, cancel := context.WithTimeout(r.Context(), sesFetchTimeout)
		defer cancel()
		if err := s.confirmSNSSubscription(ctx, env.SubscribeURL); err != nil {
			log.Printf("console: ses events: subscription confirm: %v", err)
			jsonError(w, http.StatusBadGateway, "could not confirm the SNS subscription")
			return
		}
		w.WriteHeader(http.StatusOK)
	case "Notification":
		ev, err := mail.ParseEvent([]byte(env.Message))
		if err != nil {
			// Logged and dropped, not retried: an unparseable or
			// unattributable message will never become parseable by SNS
			// resending the identical bytes, so a non-2xx here would only
			// buy a permanent retry loop.
			log.Printf("console: ses events: parse notification: %v", err)
			w.WriteHeader(http.StatusOK)
			return
		}
		if ev.Incomplete != "" {
			// Counted all the same; only the log is missing something.
			log.Printf("console: ses events: %s notification %s read in part: %s", ev.Type, env.MessageID, ev.Incomplete)
		}
		s.recordAndEnforceMail(r.Context(), ev, env.MessageID)
		w.WriteHeader(http.StatusOK)
	default:
		// An SES event type this deployment does not act on (SES has more
		// types than the five this package names). Deliberately ignored,
		// so it is a 200 — see the dispatch comment on the handler above.
		w.WriteHeader(http.StatusOK)
	}
}

// domainMatchesEvent reports whether ev's domain attributes to d — either
// the identity's own domain (the sender-supplied From SES usually reports
// as mail.source) or, for ANY region, d's recorded custom MAIL FROM
// subdomain (I6, review follow-up). SES may report the REWRITTEN envelope
// — mail-<region>.<domain>, the custom MAIL FROM MX target (spec §7) —
// rather than the sender-supplied From, and when it does, a match against
// d.Domain alone attributes the event to nothing: every count for that
// send silently drops, which is a second, independent way the whole of
// spec §9 (caps, the abuse pause) stops working. Matched against RECORDED
// values only, never a prefix-strip heuristic — MailFromDomain's shape
// could change, and this must not assume it.
func domainMatchesEvent(d *mail.DomainRecord, domain string) bool {
	if d.Domain == domain {
		return true
	}
	for _, st := range d.Regions {
		if st != nil && st.MailFrom != "" && st.MailFrom == domain {
			return true
		}
	}
	return false
}

// recordAndEnforceMail resolves ev.Domain to the account(s)/app(s) that own
// it and, for each, records the count and re-evaluates the cap and rate
// thresholds (Task 16, mailcounts.go). Best-effort throughout: a lookup or
// storage failure here is logged and swallowed, never turned into a non-2xx
// — the caller has already committed to answering 200 for every recognised
// notification, retried or not, and losing one count must not become an
// unbounded SNS retry flood.
//
// Domain identity (mail.DomainRecord) is ACCOUNT-scoped, and there is no
// reverse index from a bare domain to the account that owns it — so every
// live console account is walked. Console deployments run a small,
// bounded number of accounts; this is the same shape ListDomains' own
// callers (mail.Teardown, mail.grantedDomains) already accept for a single
// account, one level up.
//
// Each notification is also kept in the per-message log (mailevents.go),
// under notice — SNS's own id for it, and, since senders (design §8), the
// counts follow the same rule: where the caller identity resolves to a
// sender CARLOS minted, that sender alone is counted, once, and the
// domain-wide loop below is skipped entirely for it. Only when the caller
// resolves to nothing does the count fall back to every app on the matching
// domain, against its app sender, which is the whole of what this endpoint
// did before senders existed.
func (s *Server) recordAndEnforceMail(ctx context.Context, ev mail.Event, notice string) {
	accts, err := s.store.Accounts()
	if err != nil {
		log.Printf("console: ses events: listing accounts for %s: %v", ev.Domain, err)
		return
	}
	// The SEND decides which day's counters an event lands in — never the
	// bounce, the complaint or the delivery, which can arrive hours or days
	// later. The rate limiter divides a day's bounces and complaints by that
	// same day's sends, so an event must count against the day its message
	// was sent: a bounce at 00:05 for a message sent at 23:50 counted on the
	// new day reads as a bounce rate over a handful of sends, and pauses a
	// healthy tenant. (ev.At is the event's own time, for the log; ev.SentAt
	// is mail.timestamp.) A missing send time (mail.ParseEvent leaves it
	// zero when SES omits or mis-shapes it) falls back to now rather than
	// bucketing into 0001-01-01. The SAME instant is then reused as
	// enforceMailLimits' "now", so the cap/rate check reads the same day the
	// count just landed in — using time.Now() there instead could evaluate a
	// different UTC day than the one just written, right at midnight.
	at := ev.SentAt
	if at.IsZero() {
		at = time.Now()
	}
	day := at.UTC().Format("2006-01-02")
	now := time.Now()
	key := mailNoticeKey(notice, ev, now)

	// ATTRIBUTION BY CALLER (senders design §8). Every sender has its own
	// IAM user and SES stamps its name on Send, Bounce, Complaint, Delivery
	// and Reject, so where the sending domain can only name every app that
	// shares it, the caller names the one sender that actually sent. The
	// resolver parses before it reads, so an app enabled before senders
	// existed still resolves with no index entry of its own.
	callerAcct, callerApp, callerSender, named := mail.SenderOfCaller(ctx, s.cfg.Releases, ev.Caller)
	if named {
		// The COUNT is gated on the app being live, exactly as the log row
		// already is (recordMailMessage, sesevents.go:558). A caller name
		// can parse into an account and an app that is not a live app —
		// mail.UserName(acct, "not-an-app") is a well-formed name for an
		// app that was trashed, or never existed — and counting it would
		// insert a mail_sender_counts row keyed on a name no app answers
		// to, which nothing would ever read or reap. Returning rather than
		// falling through is deliberate: SES named a sender, so the
		// domain-wide loop is not the right answer for it either.
		//
		// But that only holds for a CONFIRMED absence. A store ERROR here
		// is not proof the app is gone, and treating it as one would drop
		// the event outright: no count, no cap check, and no domain
		// fallback either, with SNS still seeing a 200 so nothing ever
		// retries it. The fallback loop below applies exactly the opposite
		// rule to its own read errors ("A read failure is not a reason to
		// drop the event: fall through and charge") for the same reason, so
		// a transient error here falls through to it too rather than being
		// swallowed.
		live, err := s.mailAppLiveErr(callerAcct, callerApp)
		if err != nil {
			log.Printf("console: ses events: checking %s/%s live: %v; falling back to domain attribution", callerAcct, callerApp, err)
		} else if !live {
			return
		} else {
			s.recordMailMessage(ctx, callerAcct, callerApp, key, ev, true, now)
			// Counted ONCE, against the sender SES named, and the domain loop
			// below is skipped entirely. This is the visible change in the roll
			// note: a sibling app on the same domain is no longer charged for
			// this send, which is the intent, but a tenant watching its own
			// numbers will see them move.
			if err := s.store.RecordMailEvent(ctx, callerAcct, callerApp, callerSender, day, ev.Type); err != nil {
				log.Printf("console: ses events: recording %s for %s/%s/%s -> %v", ev.Type, callerAcct, callerApp, callerSender, err)
				return
			}
			if _, reason, err := s.enforceMailLimits(ctx, callerAcct, callerApp, callerSender, at); err != nil {
				log.Printf("console: ses events: enforcing limits for %s/%s/%s -> %v", callerAcct, callerApp, callerSender, err)
			} else if reason != "" {
				log.Printf("console: ses events: paused %s/%s/%s -> %s", callerAcct, callerApp, callerSender, reason)
			}
			return
		}
	}

	// The fallback, unchanged: SES named nobody CARLOS minted, so the
	// sending domain is all there is to go on and every app on it is
	// charged, against its app sender.
	for _, a := range accts {
		domains, err := mail.ListDomains(ctx, s.cfg.Releases, a.Sqid)
		if err != nil {
			log.Printf("console: ses events: listing domains for %s: %v", a.Sqid, err)
			continue
		}
		for _, d := range domains {
			if !domainMatchesEvent(d, ev.Domain) {
				continue
			}
			// The log's fallback asks more of the match than the counts do:
			// only a domain this account actually verified gets recipients'
			// addresses. The domain claim refuses a second account the same
			// name, but not another account's MAIL FROM host claimed as a
			// domain of its own and never verified.
			logHere := domainVerifiedForEvent(d, ev.Domain)
			for _, app := range d.Apps {
				// An app in per-instance mode is skipped (design §8). Its
				// app sender is deliberately dead — no instance sends
				// through it — so charging it for a domain-wide event it
				// cannot have caused would pause a credential nobody holds
				// and hide the instance that actually sent. A read failure
				// is not a reason to drop the event: fall through and
				// charge, which is today's behaviour.
				if sn, _, err := mail.GetSender(ctx, s.cfg.Releases, a.Sqid, app, mail.SenderAppID); err != nil {
					log.Printf("console: ses events: reading the app sender for %s/%s -> %v", a.Sqid, app, err)
				} else if sn != nil && sn.PerInstance {
					continue
				}
				if logHere {
					s.recordMailMessage(ctx, a.Sqid, app, key, ev, false, now)
				}
				if err := s.store.RecordMailEvent(ctx, a.Sqid, app, mail.SenderAppID, day, ev.Type); err != nil {
					log.Printf("console: ses events: recording %s for %s/%s: %v", ev.Type, a.Sqid, app, err)
					continue
				}
				if _, reason, err := s.enforceMailLimits(ctx, a.Sqid, app, mail.SenderAppID, at); err != nil {
					log.Printf("console: ses events: enforcing limits for %s/%s: %v", a.Sqid, app, err)
				} else if reason != "" {
					log.Printf("console: ses events: paused %s/%s: %s", a.Sqid, app, reason)
				}
			}
		}
	}
}

// domainVerifiedForEvent is domainMatchesEvent restricted to what this
// record has proved: the identity domain in a region SES verified it in, or
// the MAIL FROM host of a verified region.
func domainVerifiedForEvent(d *mail.DomainRecord, domain string) bool {
	for _, st := range d.Regions {
		if st == nil || st.Status != mail.StatusVerified {
			continue
		}
		if d.Domain == domain || (st.MailFrom != "" && st.MailFrom == domain) {
			return true
		}
	}
	return false
}

// recordMailMessage keeps one notification in one app's log, if that app is
// live. A trashed app keeps nothing new: its name is not a member-facing app
// any more, and what it is sent is not somebody else's to read after a
// restore or a claim. Best effort, like the rest of the ingest: a failure
// is logged and costs neither the count nor the 200.
func (s *Server) recordMailMessage(ctx context.Context, acct, app, key string, ev mail.Event, bySender bool, now time.Time) {
	if !s.mailAppLive(acct, app) {
		return
	}
	if err := s.store.RecordMailMessageEvent(ctx, acct, app, key, ev, bySender, now); err != nil {
		log.Printf("console: ses events: logging %s for %s/%s: %v", ev.Type, acct, app, err)
	}
	// Pushed to the app only when SES itself named it as the sender: a
	// domain-attributed event names a recipient who may be another app's
	// (mailpush.go).
	if bySender {
		s.queueMailEventPush(ctx, acct, app, key, ev, now)
	}
}

// mailAppLive reports whether acct (a sqid) holds app as a live, untrashed
// app right now. It collapses a store error into false, like an absent app,
// which is fine for its one caller (recordMailMessage's best-effort log
// guard) but wrong for anything that treats "not live" as a reason to stop
// processing an event entirely — use mailAppLiveErr there instead.
func (s *Server) mailAppLive(acct, app string) bool {
	live, err := s.mailAppLiveErr(acct, app)
	return err == nil && live
}

// mailAppLiveErr is mailAppLive with the error kept apart from a confirmed
// absence: err != nil means the store read itself failed, not that the app
// is gone. A caller deciding whether to drop an event needs that
// distinction — an errored read is not evidence of anything.
func (s *Server) mailAppLiveErr(acct, app string) (bool, error) {
	a, err := s.store.AccountBySqid(acct)
	if err != nil {
		return false, err
	}
	if a == nil {
		return false, nil
	}
	return s.store.HasAppIn(a.ID, app)
}

// confirmSNSSubscription GETs the verified, host-checked SubscribeURL. It is
// reached only after the enclosing message's signature has verified AND
// SubscribeURL has passed validSigningCertURL — the signature proves AWS
// put the string there, the host check proves the string points at SNS
// itself rather than somewhere this console should never fetch on a
// stranger's behalf. It uses snsFetchClient (never http.DefaultClient) so a
// redirect from that already-verified host cannot hand control to a
// DIFFERENT, unverified one — see snsFetchClient's doc comment.
func (s *Server) confirmSNSSubscription(ctx context.Context, subscribeURL string) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, subscribeURL, nil)
	if err != nil {
		return fmt.Errorf("sns: building subscribe request: %w", err)
	}
	resp, err := snsFetchClient.Do(req)
	if err != nil {
		return fmt.Errorf("sns: fetching SubscribeURL: %w", err)
	}
	defer resp.Body.Close()
	io.Copy(io.Discard, io.LimitReader(resp.Body, sesCertMaxBody))
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("sns: SubscribeURL confirm status %d", resp.StatusCode)
	}
	return nil
}

// snsCertCache caches a signing certificate's verified RSA public key,
// keyed on its (already pinned-host-validated) URL — IMPORTANT-2 fix.
// Without it, every request bearing the deployment's TopicArn — which is
// not a secret, it appears in IaC and logs — triggers a fresh outbound
// HTTPS fetch and a full certificate-chain verification: unauthenticated
// outbound amplification, one connection out per request in. The AWS SDKs
// cache verification certs by URL for the same reason.
//
// A value type with a nil-safe zero value on purpose: Server embeds one
// directly (no constructor call needed), the same way ceClientOnce
// (server.go) needs no init.
type snsCertCache struct {
	mu   sync.Mutex
	keys map[string]*rsa.PublicKey
}

func (c *snsCertCache) get(certURL string) (*rsa.PublicKey, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	pub, ok := c.keys[certURL]
	return pub, ok
}

func (c *snsCertCache) put(certURL string, pub *rsa.PublicKey) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.keys == nil {
		c.keys = map[string]*rsa.PublicKey{}
	}
	c.keys[certURL] = pub
}

// verifiedCertKey returns the RSA public key for a signing certificate URL
// — already pinned to an SNS host by the caller (validSigningCertURL) —
// using the cache on a hit, and fetching plus fully verifying the
// certificate on a miss.
func (s *Server) verifiedCertKey(ctx context.Context, certURL string) (*rsa.PublicKey, error) {
	if pub, ok := s.snsCerts.get(certURL); ok {
		return pub, nil
	}
	pub, err := fetchVerifiedCertKey(ctx, certURL)
	if err != nil {
		return nil, err
	}
	s.snsCerts.put(certURL, pub)
	return pub, nil
}

// snsCertDNSName is the hostname a genuine SNS message-signing certificate
// actually carries in its subject/SAN — NOT the regional fetch host (e.g.
// sns.eu-west-1.amazonaws.com). SNS mints one signing identity for the
// whole service, not one per region. Verified against a genuine SNS
// signing certificate the round-2 review pulled from AWS's own SDK test
// fixture (aws-sdk-ruby, gems/aws-sdk-sns/spec/message_verifier_spec.rb):
// subject=CN=sns.amazonaws.com, SAN=[sns.amazonaws.com],
// issuer=CN=Amazon,OU=Server CA 1B,O=Amazon,C=US. That fixture is old and
// is not treated as sufficient evidence that today's live certificate is
// identical — see sesevents.go's package comment and the task report for
// what remains unproven until a real message arrives.
const snsCertDNSName = "sns.amazonaws.com"

// fetchVerifiedCertKey downloads the certificate at certURL and returns its
// RSA public key. certURL is assumed already pinned (validSigningCertURL)
// and fetched with redirects refused (snsFetchClient) — those two, plus the
// TLS chain validation the http client performs automatically as part of
// making an HTTPS request, are what make the BYTES this function receives
// trustworthy. See the package comment's "TRUST MODEL" paragraph for the
// full picture.
//
// NO CHAIN IS BUILT OR VERIFIED FOR THE CERTIFICATE ITSELF, DELIBERATELY.
// This reverses an EARLIER fix in this same file — twice (see the package
// comment, "Round 4") — and the removal is deliberate, not an oversight to
// "fix" back:
//
//   - AWS serves EXACTLY ONE certificate in a real SigningCertURL response
//     — a bare leaf, with no bundled intermediate.
//   - That leaf's issuer is an intermediate CA distributed via the leaf's
//     AIA extension, which Go's x509.Verify never fetches (no AIA
//     chasing). System root stores hold only the top-level Amazon Root CA
//     certificates, several hops above that intermediate.
//   - So a chain built ONLY from what the response contains is ALWAYS
//     empty against a real message, and verifying it against system roots
//     ALWAYS fails with "certificate signed by unknown authority" —
//     proven directly against a real captured certificate. Every
//     reference implementation (aws-sdk-ruby's message_verifier.rb,
//     robbiet480/go.sns, AWS's own documented `openssl x509 -pubkey
//     -noout` procedure) agrees: decode one leaf, verify no chain.
//
// What remains is the ONE check that both means something and can actually
// succeed: the certificate's own subject must be snsCertDNSName — SNS's
// fixed identity, not the regional host it happened to be fetched from.
// This is a SANITY check, not a trust decision on its own — it catches a
// misconfigured or wrong-endpoint response, and it is deliberately paired
// with the host pin and redirect guard elsewhere, which are what make the
// FETCH itself trustworthy in the first place.
func fetchVerifiedCertKey(ctx context.Context, certURL string) (*rsa.PublicKey, error) {
	fctx, cancel := context.WithTimeout(ctx, sesFetchTimeout)
	defer cancel()
	req, err := http.NewRequestWithContext(fctx, http.MethodGet, certURL, nil)
	if err != nil {
		return nil, fmt.Errorf("sns: building cert request: %w", err)
	}
	resp, err := snsFetchClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("sns: fetching signing cert: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("sns: signing cert fetch status %d", resp.StatusCode)
	}
	certPEM, err := io.ReadAll(io.LimitReader(resp.Body, sesCertMaxBody))
	if err != nil {
		return nil, fmt.Errorf("sns: reading signing cert: %w", err)
	}
	block, _ := pem.Decode(certPEM)
	if block == nil {
		return nil, fmt.Errorf("sns: signing cert is not PEM")
	}
	leaf, err := x509.ParseCertificate(block.Bytes)
	if err != nil {
		return nil, fmt.Errorf("sns: parsing signing cert: %w", err)
	}
	if err := leaf.VerifyHostname(snsCertDNSName); err != nil {
		return nil, fmt.Errorf("sns: signing cert subject is not %s: %w", snsCertDNSName, err)
	}
	pub, ok := leaf.PublicKey.(*rsa.PublicKey)
	if !ok {
		return nil, fmt.Errorf("sns: signing cert public key is not RSA")
	}
	return pub, nil
}

// verifySNSSignature is the real (non-test-seam) verification path: resolve
// the certificate's verified public key (cached; fetched and chain-checked
// on a miss) and check the signature over the message's own canonical
// string.
func (s *Server) verifySNSSignature(ctx context.Context, env snsEnvelope) error {
	pub, err := s.verifiedCertKey(ctx, env.SigningCertURL)
	if err != nil {
		return err
	}
	return verifySNSSignatureBytes(pub, []byte(env.canonicalString()), env.SignatureVersion, env.Signature)
}

// verifySNSSignatureBytes checks a base64 signature against canon using pub,
// per SignatureVersion. Split out from verifySNSSignature so this — the
// actual cryptographic check — can be tested with a plain in-memory RSA
// keypair: no network, no certificate, no trust store involved.
//
// SNS signs with SHA1withRSA under SignatureVersion "1" (the historical
// default, and what SNS still sends unless a topic opts into version "2",
// SHA256withRSA). Both are verified here; an unrecognised version is
// refused rather than guessed at.
func verifySNSSignatureBytes(pub *rsa.PublicKey, canon []byte, sigVersion, sigB64 string) error {
	sig, err := base64.StdEncoding.DecodeString(sigB64)
	if err != nil {
		return fmt.Errorf("sns: signature is not base64: %w", err)
	}
	switch sigVersion {
	case "", "1":
		sum := sha1.Sum(canon)
		if err := rsa.VerifyPKCS1v15(pub, crypto.SHA1, sum[:], sig); err != nil {
			return fmt.Errorf("sns: signature verification failed: %w", err)
		}
	case "2":
		sum := sha256.Sum256(canon)
		if err := rsa.VerifyPKCS1v15(pub, crypto.SHA256, sum[:], sig); err != nil {
			return fmt.Errorf("sns: signature verification failed: %w", err)
		}
	default:
		return fmt.Errorf("sns: unsupported SignatureVersion %q", sigVersion)
	}
	return nil
}

// refusedTopicWindow is how long one distinct refused ARN stays quiet after
// being logged, and refusedTopicMax how many distinct ARNs are remembered.
const (
	refusedTopicWindow = time.Hour
	refusedTopicMax    = 8
)

// logRefusedTopic names a rejected TopicArn in the log, at most once per
// hour per distinct ARN.
//
// WHY LOG AT ALL. The TopicArn allowlist is this door's first real gate, and
// a stale or mistyped CARLOS_MAIL_SNS_TOPIC_ARN is the single most likely
// misconfiguration in the whole event path. Before this it left NO trace:
// 403s at a door nobody watches, and counts that never move. "Events stopped
// arriving and nothing said so" is precisely the silence this is for. The
// ARN is not a secret — it is published in terraform and read off the box —
// so naming it costs nothing.
//
// WHY BOUNDED. This door is UNAUTHENTICATED and internet-reachable by
// design (SNS has no credential to present), so an unbounded log line here
// is a stranger's write access to the disk. Bounding answers that without
// giving up the diagnostic: the misconfiguration this exists to catch
// produces exactly ONE distinct ARN, so one line an hour says everything,
// while a flood of fabricated ARNs costs at most refusedTopicMax lines an
// hour and a map of the same size.
//
// The 403 body is deliberately unchanged. The external answer stays
// constant and uninformative; the operator's answer goes to the log. They
// are different readers and only one of them is trusted.
func (s *Server) logRefusedTopic(got string) {
	// %q escapes control bytes, so a caller cannot forge log lines; the
	// truncation bounds the damage a very long value can do to the line. A
	// real ARN is about 100 bytes.
	if len(got) > 256 {
		got = got[:256]
	}
	want := s.cfg.MailDeployment.SNSTopicARN
	wantStr := fmt.Sprintf("%q", want)
	if want == "" {
		// Its own misconfiguration signal, and a different one: the
		// deployment believes email events are off while SNS is actively
		// delivering to it.
		wantStr = "(none configured)"
	}

	s.refusedTopicsMu.Lock()
	now := time.Now()
	if s.refusedTopics == nil {
		s.refusedTopics = map[string]time.Time{}
	}
	if last, seen := s.refusedTopics[got]; seen && now.Sub(last) < refusedTopicWindow {
		s.refusedTopicsMu.Unlock()
		return
	}
	if _, seen := s.refusedTopics[got]; !seen && len(s.refusedTopics) >= refusedTopicMax {
		// Evict the oldest rather than refusing to record: a genuine new
		// misconfiguration arriving after a flood must still be logged.
		oldestK, oldestT := "", now
		for k, t := range s.refusedTopics {
			if t.Before(oldestT) || oldestK == "" {
				oldestK, oldestT = k, t
			}
		}
		delete(s.refusedTopics, oldestK)
	}
	s.refusedTopics[got] = now
	s.refusedTopicsMu.Unlock()

	log.Printf("console: ses events: refused topic %q — this deployment expects %s", got, wantStr)
}