package console

// Telling an app what happened to its email. An app that asks — `carlos
// email events push --app <a> --path /mail/events` — is POSTed each
// delivery, bounce, complaint, rejection and delay SES reports for a message
// it sent, signed by the console, at that path on its own platform host.
//
// WHAT CROSSES, AND TO WHOM. The payload names the recipient, and a
// recipient's address is the sending app's business and nobody else's. So
// only a notification SES attributed to the app itself (an identity CARLOS
// minted for it, a by_sender row in the log) is ever queued for it: an event that could only
// be attributed by a sending domain several apps share is kept in the log,
// where a member can read it, and is never pushed. A Send is not pushed
// either — the app learnt that one from SES directly, in the reply to the
// message it sent.
//
// WHERE IT GOES. Only ever https://<app>.<account>.<AppsDomain><path>: the
// host is built from names the platform validated, never from anything a
// request supplies, and the path passes schedule.ValidPath, the same check
// that keeps a schedule tick on the instance. Redirects are refused, so the
// far end cannot turn a delivery into a POST of somebody's address at a
// host of its choosing. The request travels the public edge like any other,
// which is why it wakes a hibernating app and why an app behind `carlos
// gate` cannot receive one.
//
// HOW AN APP KNOWS IT IS REAL. Every delivery is signed with an Ed25519 key
// the console keeps for this and nothing else (mailEventsSigningKey), and
// the public half is delivered to the app as CARLOS_MAIL_EVENTS_KEY when it
// asks for events. The signature covers a domain-separation label, the
// timestamp and the exact body:
//
//	Carlos-Signature: t=<unix seconds>,ed25519=<base64 signature>
//	signed bytes:     "carlos-mail-event\n" + t + "\n" + body
//
// An app refuses a bad signature, a timestamp more than a few minutes from
// its own clock, and a body whose account or app is not its own — the key is
// the deployment's, so the last check is what stops one app's event,
// replayed at another, from being believed there.
//
// WHAT ONE APP CAN COST OTHERS. A pass starts from the apps that asked, and
// reads each one's oldest due rows through an index, so its cost follows how
// many apps ask, not how much any of them has queued. Apps are sent to eight
// at a time across the whole deployment, not per account, and an app's turn
// ends at its first failure — so a hung app holds one slot for one timeout a
// pass, and an account with many hung apps costs everyone about ten seconds
// a pass per eight of them. An app that answers 2xx slowly keeps its turn,
// and can hold a slot for up to five timeouts. And an app that has accepted
// nothing for a day is suspended: its queue is given up on and nothing more
// is queued until the push is turned on again (mailPushSuspendAfter), so a
// failing app cannot grow a backlog that a pass has to carry for a month.

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"crypto/x509"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

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

// MailEventsKeyEnv is the variable an app finds the verifying key in.
const MailEventsKeyEnv = "CARLOS_MAIL_EVENTS_KEY"

// mailEventSignatureLabel separates this signature from anything else the
// key could conceivably be asked to sign.
const mailEventSignatureLabel = "carlos-mail-event"

const (
	// mailPushEvery is the queue's cadence. A delivery waits at most this
	// long for its first attempt; notifications themselves arrive seconds
	// after SES hears back from the receiving server.
	mailPushEvery = 15 * time.Second
	// mailPushPerApp is the most one app is sent in a pass, and
	// mailPushApps how many apps are sent to at once. Together they are what
	// keeps one app from being everyone's problem: an app whose handler
	// hangs occupies one of mailPushApps for at most one timeout (see
	// deliverAppMailEvents), and its backlog waits behind itself, not in
	// front of anybody else's.
	mailPushPerApp = 5
	mailPushApps   = 8
	// mailPushSuspendAfter is how long an app may go on failing — every
	// attempt at every row answered with anything but a 2xx — before its
	// push is suspended. A day is past any deploy an app could be halfway
	// through, and it bounds what a failing app can have queued to a day of
	// its own mail.
	mailPushSuspendAfter = 24 * time.Hour
	// mailPushPruneEvery bounds how often the queue is trimmed, as the log is.
	mailPushPruneEvery = time.Hour
	mailPushMaxBody    = 64 << 10
)

// mailPushTimeout bounds one POST: enough for the edge to wake a
// hibernating app and for it to answer. A var so a test can shorten it.
var mailPushTimeout = 10 * time.Second

// mailPushBackoff is when each retry is due after a failed attempt, counted
// from that attempt. After the last, the delivery is given up on — a little
// under a day of trying, which is past any deploy an app could be halfway
// through and short of the retention window the payload is bound by.
var mailPushBackoff = []time.Duration{
	time.Minute, 5 * time.Minute, 15 * time.Minute, time.Hour, 3 * time.Hour, 6 * time.Hour, 12 * time.Hour,
}

// mailPushClient refuses redirects: see the file comment. The per-request
// timeout is the context's (deliverMailEvent), so a test can shorten it.
var mailPushClient = &http.Client{
	CheckRedirect: func(*http.Request, []*http.Request) error {
		return http.ErrUseLastResponse
	},
}

// mailPushedTypes are the event types an app is told about.
var mailPushedTypes = map[string]bool{
	mail.EventDelivery: true, mail.EventBounce: true, mail.EventComplaint: true,
	mail.EventReject: true, mail.EventDeliveryDelay: true,
}

// ---- the key ------------------------------------------------------------------

// MailEventsSigningKey returns the Ed25519 key email-event deliveries are
// signed with, generating and persisting it on first use. It is its own key
// rather than SigningKey's: that one signs keymail envelopes, and a key
// that signs two kinds of thing is one whose signatures can be carried from
// one to the other.
func (s *Store) MailEventsSigningKey() (ed25519.PrivateKey, error) {
	load := func() (ed25519.PrivateKey, error) {
		var b64 string
		if err := s.db.QueryRow(`SELECT pkcs8 FROM keys WHERE name = 'mail-events-sign'`).Scan(&b64); err != nil {
			return nil, err
		}
		der, err := base64.StdEncoding.DecodeString(b64)
		if err != nil {
			return nil, fmt.Errorf("console: stored mail events key: %v", err)
		}
		key, err := x509.ParsePKCS8PrivateKey(der)
		if err != nil {
			return nil, fmt.Errorf("console: stored mail events key: %v", err)
		}
		ed, ok := key.(ed25519.PrivateKey)
		if !ok {
			return nil, fmt.Errorf("console: stored mail events key is not Ed25519")
		}
		return ed, nil
	}
	if k, err := load(); err == nil {
		return k, nil
	} else if err != sql.ErrNoRows {
		return nil, err
	}
	_, priv, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		return nil, err
	}
	der, err := x509.MarshalPKCS8PrivateKey(priv)
	if err != nil {
		return nil, err
	}
	// INSERT OR IGNORE + re-load, as SigningKey: a racing opener's row wins
	// and everyone converges on it.
	if _, err := s.db.Exec(`INSERT OR IGNORE INTO keys (name, pkcs8, created_at) VALUES ('mail-events-sign', ?, ?)`,
		base64.StdEncoding.EncodeToString(der), time.Now().Unix()); err != nil {
		return nil, err
	}
	return load()
}

// mailEventsPublicKey is the verifying half, as delivered to an app: the 32
// raw bytes, standard base64.
func mailEventsPublicKey(priv ed25519.PrivateKey) string {
	return base64.StdEncoding.EncodeToString(priv.Public().(ed25519.PublicKey))
}

// signMailEvent is the Carlos-Signature header value for body at t.
func signMailEvent(priv ed25519.PrivateKey, t time.Time, body []byte) string {
	ts := strconv.FormatInt(t.Unix(), 10)
	sig := ed25519.Sign(priv, mailEventSignedBytes(ts, body))
	return "t=" + ts + ",ed25519=" + base64.StdEncoding.EncodeToString(sig)
}

func mailEventSignedBytes(ts string, body []byte) []byte {
	msg := make([]byte, 0, len(mailEventSignatureLabel)+len(ts)+len(body)+2)
	msg = append(msg, mailEventSignatureLabel...)
	msg = append(msg, '\n')
	msg = append(msg, ts...)
	msg = append(msg, '\n')
	return append(msg, body...)
}

// VerifyMailEvent is what an app does with a delivery, in Go: check the
// signature header against the body with the delivered key, and refuse a
// timestamp further than skew from now. It is here so the scheme has one
// executable definition the tests hold both ends to; an app in any
// language does the same four steps.
func VerifyMailEvent(publicKeyB64, header string, body []byte, now time.Time, skew time.Duration) error {
	raw, err := base64.StdEncoding.DecodeString(publicKeyB64)
	if err != nil || len(raw) != ed25519.PublicKeySize {
		return errors.New("mail event: the key is not an Ed25519 public key")
	}
	var ts, sigB64 string
	for _, part := range strings.Split(header, ",") {
		k, v, _ := strings.Cut(strings.TrimSpace(part), "=")
		switch k {
		case "t":
			ts = v
		case "ed25519":
			sigB64 = v
		}
	}
	unix, err := strconv.ParseInt(ts, 10, 64)
	if err != nil {
		return errors.New("mail event: no timestamp in the signature")
	}
	if d := now.Sub(time.Unix(unix, 0)); d > skew || d < -skew {
		return errors.New("mail event: the timestamp is too far from now")
	}
	sig, err := base64.StdEncoding.DecodeString(sigB64)
	if err != nil || len(sig) != ed25519.SignatureSize {
		return errors.New("mail event: no signature")
	}
	if !ed25519.Verify(ed25519.PublicKey(raw), mailEventSignedBytes(ts, body), sig) {
		return errors.New("mail event: the signature does not verify")
	}
	return nil
}

// ---- what an app is sent --------------------------------------------------------

// MailEventPayload is the body of one delivery.
type MailEventPayload struct {
	// ID is stable across retries of the same notification, so an app that
	// receives one twice — a retry after an answer that never arrived — can
	// tell.
	ID         string               `json:"id"`
	Account    string               `json:"account"`
	App        string               `json:"app"`
	Type       string               `json:"type"`
	MessageID  string               `json:"message_id"`
	At         time.Time            `json:"at"`
	Detail     string               `json:"detail,omitempty"`
	Recipients []MailEventRecipient `json:"recipients"`
}

// MailEventRecipient is one recipient's part in a delivery.
type MailEventRecipient struct {
	Address    string `json:"address"`
	Status     string `json:"status,omitempty"`
	Diagnostic string `json:"diagnostic,omitempty"`
}

func mailEventPayload(acct, app, notice string, ev mail.Event, now time.Time) MailEventPayload {
	at := ev.At
	if at.IsZero() {
		at = now
	}
	p := MailEventPayload{ID: notice, Account: acct, App: app, Type: ev.Type, MessageID: ev.MessageID,
		At: at.UTC(), Detail: ev.Detail.Summary(), Recipients: []MailEventRecipient{}}
	for _, o := range ev.Recipients() {
		if o.Recipient == "" {
			continue
		}
		p.Recipients = append(p.Recipients, MailEventRecipient{Address: o.Recipient, Status: o.Status, Diagnostic: o.Diagnostic})
	}
	return p
}

// ---- hooks ------------------------------------------------------------------------

// MailEventHook is where an app has asked to be told.
type MailEventHook struct {
	Path  string
	SetBy string
	SetAt time.Time
	// FailingSince is when the app's current run of failed deliveries began;
	// zero when it has none. SuspendedAt is when the push was suspended for
	// failing for a day, and SuspendedWhy says what the last attempt got.
	FailingSince time.Time
	SuspendedAt  time.Time
	SuspendedWhy string
}

// Suspended reports whether the push is suspended.
func (h *MailEventHook) Suspended() bool { return h != nil && !h.SuspendedAt.IsZero() }

// SetMailEventHook turns the push on, or moves its path — and, either way,
// lifts a suspension and forgets the run of failures behind it: turning it
// on again is how a member says the handler is there now.
func (s *Store) SetMailEventHook(ctx context.Context, acct, app, path, by string, now time.Time) error {
	return s.withMailAppSQL(ctx, acct, app, "sql.SetMailEventHook", func(v *Store) error {
		_, err := v.db.ExecContext(ctx, `INSERT INTO mail_event_hooks (account, app, path, set_by, set_at) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(account, app) DO UPDATE SET path = excluded.path, set_by = excluded.set_by, set_at = excluded.set_at,
  failing_since = 0, suspended_at = 0, suspended_why = ''`,
			acct, app, path, by, now.Unix())
		return err
	})
}

const mailEventHookColumns = `account, app, path, set_by, set_at, failing_since, suspended_at, suspended_why`

type mailEventHookRow struct {
	claimAppID   string // Captured before reading queued payloads; never rebound by a delayed worker.
	Account, App string
	MailEventHook
}

func scanMailEventHook(scan func(...any) error) (mailEventHookRow, error) {
	var h mailEventHookRow
	var setAt, failing, suspended int64
	if err := scan(&h.Account, &h.App, &h.Path, &h.SetBy, &setAt, &failing, &suspended, &h.SuspendedWhy); err != nil {
		return h, err
	}
	h.SetAt = time.Unix(setAt, 0).UTC()
	if failing > 0 {
		h.FailingSince = time.Unix(failing, 0).UTC()
	}
	if suspended > 0 {
		h.SuspendedAt = time.Unix(suspended, 0).UTC()
	}
	return h, nil
}

// MailEventHookFor answers nil when the app has not asked.
func (s *Store) MailEventHookFor(ctx context.Context, acct, app string) (*MailEventHook, error) {
	h, err := scanMailEventHook(s.db.QueryRowContext(ctx,
		`SELECT `+mailEventHookColumns+` FROM mail_event_hooks WHERE account = ? AND app = ?`, acct, app).Scan)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &h.MailEventHook, nil
}

// activeMailEventHooks is every push that is on and not suspended: where a
// pass starts.
func (s *Store) activeMailEventHooks(ctx context.Context) ([]mailEventHookRow, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT `+mailEventHookColumns+` FROM mail_event_hooks WHERE suspended_at = 0 ORDER BY account, app`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []mailEventHookRow
	for rows.Next() {
		h, err := scanMailEventHook(rows.Scan)
		if err != nil {
			return nil, err
		}
		out = append(out, h)
	}
	return out, rows.Err()
}

// noteMailEventTurn records how an app's turn in a pass went: a delivery
// that landed ends any run of failures; a failure starts one if none is
// running; and a run that has lasted mailPushSuspendAfter suspends the push
// and gives up on its queue, in one transaction. It answers whether the push
// is now suspended.
func (s *Store) noteMailEventTurn(ctx context.Context, h mailEventHookRow, landed bool, failure string, now time.Time) (suspended bool, err error) {
	err = s.withMailAppSQL(ctx, h.Account, h.App, "sql.NoteMailEventTurn", func(v *Store) error {
		suspended, err = v.claimRawNoteMailEventTurn(ctx, h, landed, failure, now)
		return err
	})
	return suspended, err
}

func (s *Store) claimRawNoteMailEventTurn(ctx context.Context, h mailEventHookRow, landed bool, failure string, now time.Time) (bool, error) {
	switch {
	case landed:
		if h.FailingSince.IsZero() {
			return false, nil
		}
		_, err := s.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = 0 WHERE account = ? AND app = ?`, h.Account, h.App)
		return false, err
	case failure == "":
		return false, nil
	case h.FailingSince.IsZero():
		_, err := s.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = ? WHERE account = ? AND app = ? AND failing_since = 0`,
			now.Unix(), h.Account, h.App)
		return false, err
	case now.Sub(h.FailingSince) < mailPushSuspendAfter:
		return false, nil
	}
	why := "nothing was accepted for a day — the last attempt " + strings.TrimPrefix(failure, "no answer: ")
	if strings.HasPrefix(failure, "no answer") {
		why = "nothing was accepted for a day — the last attempt got no answer"
	}
	if len(why) > 300 {
		why = why[:300]
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return false, err
	}
	defer tx.Rollback()
	if _, err := tx.ExecContext(ctx, `UPDATE mail_event_hooks SET suspended_at = ?, suspended_why = ? WHERE account = ? AND app = ? AND set_at = ?`,
		now.Unix(), why, h.Account, h.App, h.SetAt.Unix()); err != nil {
		return false, err
	}
	if _, err := tx.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = 'push suspended'
WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), h.Account, h.App); err != nil {
		return false, err
	}
	return true, tx.Commit()
}

// clearMailEventFailing ends a run of failures that has no queue left behind
// it: failing_since measures one backlog's failing, not the app's history,
// so an app that failed once last week is not suspended by its next miss.
func (s *Store) clearMailEventFailing(ctx context.Context, h mailEventHookRow) error {
	if h.FailingSince.IsZero() {
		return nil
	}
	return s.withMailAppSQL(ctx, h.Account, h.App, "sql.ClearMailEventFailing", func(v *Store) error {
		_, err := v.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = 0
WHERE account = ? AND app = ? AND NOT EXISTS (
  SELECT 1 FROM mail_event_deliveries WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0)`,
			h.Account, h.App, h.Account, h.App)
		return err
	})
}

// DeleteMailEventHook turns the push off, and gives up on everything still
// queued for it: an app that has stopped asking must not keep being sent
// addresses from a queue it has forgotten about. One transaction, so the
// push is never off with its queue still live.
func (s *Store) DeleteMailEventHook(ctx context.Context, acct, app string, now time.Time) (was bool, err error) {
	err = s.withMailAppSQL(ctx, acct, app, "sql.DeleteMailEventHook", func(v *Store) error {
		was, err = v.claimRawDeleteMailEventHook(ctx, acct, app, now)
		return err
	})
	return was, err
}

func (s *Store) claimRawDeleteMailEventHook(ctx context.Context, acct, app string, now time.Time) (bool, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return false, err
	}
	defer tx.Rollback()
	res, err := tx.ExecContext(ctx, `DELETE FROM mail_event_hooks WHERE account = ? AND app = ?`, acct, app)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	if _, err := tx.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = 'push turned off'
WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), acct, app); err != nil {
		return false, err
	}
	return n > 0, tx.Commit()
}

// ---- the queue --------------------------------------------------------------------

// EnqueueMailEventDelivery queues one notification for one app. A notice
// already queued for the app queues nothing.
func (s *Store) EnqueueMailEventDelivery(ctx context.Context, acct, app, notice string, payload []byte, now time.Time) error {
	return s.withMailAppSQL(ctx, acct, app, "sql.EnqueueMailEventDelivery", func(v *Store) error {
		_, err := v.db.ExecContext(ctx, `INSERT OR IGNORE INTO mail_event_deliveries (account, app, notice, payload, next_at, created_at)
VALUES (?, ?, ?, ?, ?, ?)`, acct, app, notice, string(payload), now.Unix(), now.Unix())
		return err
	})
}

type mailEventDelivery struct {
	ID       int64
	Account  string
	App      string
	Payload  []byte
	Attempts int
}

// mailPushDueSQL is the read a pass makes per app. A test holds its query
// plan to the index.
const mailPushDueSQL = `SELECT id, account, app, payload, attempts FROM mail_event_deliveries
INDEXED BY mail_event_deliveries_app_due
WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0 AND next_at <= ?
ORDER BY next_at LIMIT ?`

// dueMailEventDeliveriesFor is one app's oldest due rows, at most limit of
// them, sought through mail_event_deliveries_app_due: the four equalities
// and the range on next_at are all in the index, in its order, so the rows
// come back without sorting the app's backlog.
func (s *Store) dueMailEventDeliveriesFor(ctx context.Context, acct, app string, now time.Time, limit int) ([]mailEventDelivery, error) {
	rows, err := s.db.QueryContext(ctx, mailPushDueSQL, acct, app, now.Unix(), limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []mailEventDelivery
	for rows.Next() {
		var d mailEventDelivery
		var payload string
		if err := rows.Scan(&d.ID, &d.Account, &d.App, &payload, &d.Attempts); err != nil {
			return nil, err
		}
		d.Payload = []byte(payload)
		out = append(out, d)
	}
	return out, rows.Err()
}

// recordMailEventAttempt writes one attempt's outcome: delivered, due again
// after the backoff, or given up on once the backoff is spent. A row already
// given up on (the push was turned off while this attempt was in flight)
// stays given up on: it is not counted twice, and it is not revived.
//
// The row is d's own: d.Account and d.App were read from it, the write runs
// under that app's permit, and every statement also matches them, so an id
// naming another app's row updates nothing.
func (s *Store) recordMailEventAttempt(ctx context.Context, d mailEventDelivery, status int, failure string, now time.Time) error {
	return s.withMailAppSQL(ctx, d.Account, d.App, "sql.RecordMailEventAttempt", func(v *Store) error {
		return v.claimRawRecordMailEventAttempt(ctx, d, status, failure, now)
	})
}

func (s *Store) claimRawRecordMailEventAttempt(ctx context.Context, d mailEventDelivery, status int, failure string, now time.Time) error {
	attempts := d.Attempts + 1
	if failure == "" {
		_, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries
SET attempts = ?, delivered_at = ?, last_attempt_at = ?, last_status = ?, last_error = '' WHERE id = ? AND account = ? AND app = ? AND abandoned_at = 0`,
			attempts, now.Unix(), now.Unix(), status, d.ID, d.Account, d.App)
		return err
	}
	if len(failure) > 300 {
		failure = failure[:300]
	}
	if attempts > len(mailPushBackoff) {
		_, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries
SET attempts = ?, abandoned_at = ?, abandoned_why = 'retries spent', last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ? AND account = ? AND app = ? AND abandoned_at = 0`,
			attempts, now.Unix(), now.Unix(), status, failure, d.ID, d.Account, d.App)
		return err
	}
	_, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries
SET attempts = ?, next_at = ?, last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ? AND account = ? AND app = ? AND abandoned_at = 0`,
		attempts, now.Add(mailPushBackoff[attempts-1]).Unix(), now.Unix(), status, failure, d.ID, d.Account, d.App)
	return err
}

// abandonMailEventDeliveries gives up on everything still queued for one
// app, in one statement. The reason goes in abandoned_why, never over
// last_error: what the app last answered stays what it last answered.
func (s *Store) abandonMailEventDeliveries(ctx context.Context, acct, app, reason string, now time.Time) error {
	return s.withMailAppSQL(ctx, acct, app, "sql.AbandonMailEventDeliveries", func(v *Store) error {
		_, err := v.db.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = ?
WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), reason, acct, app)
		return err
	})
}

// pruneMailEventDeliveries drops every queued row past the retention window,
// delivered or not: the payload is a copy of the log's addresses and lives
// no longer than they do.
func (s *Store) pruneMailEventDeliveries(ctx context.Context, now time.Time) error {
	last := s.mailPushPruned.Load()
	if last != 0 && now.Unix()-last < int64(mailPushPruneEvery/time.Second) {
		return nil
	}
	if !s.mailPushPruned.CompareAndSwap(last, now.Unix()) {
		return nil
	}
	if s.gate != nil && s.bound == nil {
		// Gated: per app, each under its own permit; the stamp stays on a
		// partial failure (pruneMailEvents says why).
		return s.pruneMailAppRows(ctx, "mail_event_deliveries", "created_at", now.Add(-mailEventsRetention).Unix(), "sql.PruneMailEventDeliveries")
	}
	if _, err := s.db.ExecContext(ctx, `DELETE FROM mail_event_deliveries WHERE created_at < ?`, now.Add(-mailEventsRetention).Unix()); err != nil {
		// Not pruned, so not stamped: the next pass tries again.
		s.mailPushPruned.CompareAndSwap(now.Unix(), last)
		return err
	}
	return nil
}

// MailEventPushStatus is what `carlos email events push` shows.
type MailEventPushStatus struct {
	Pending       int       `json:"pending"`
	Delivered     int       `json:"delivered"`
	Abandoned     int       `json:"abandoned"`
	LastDelivered time.Time `json:"last_delivered,omitempty"`
	LastFailedAt  time.Time `json:"last_failed_at,omitempty"`
	LastStatus    int       `json:"last_status,omitempty"`
	LastError     string    `json:"last_error,omitempty"`
}

// MailEventPushStatusFor reads the counts over the retention window, and the
// last failure since `since` — the moment the push was last turned on — so a
// failure from before a member turned it off and on again, or before a
// suspension was lifted, is not reported as today's news.
func (s *Store) MailEventPushStatusFor(ctx context.Context, acct, app string, since time.Time) (MailEventPushStatus, error) {
	var st MailEventPushStatus
	var lastDelivered sql.NullInt64
	if err := s.db.QueryRowContext(ctx, `SELECT
  COALESCE(SUM(delivered_at = 0 AND abandoned_at = 0), 0),
  COALESCE(SUM(delivered_at > 0), 0),
  COALESCE(SUM(abandoned_at > 0), 0),
  MAX(NULLIF(delivered_at, 0))
FROM mail_event_deliveries WHERE account = ? AND app = ?`, acct, app).Scan(&st.Pending, &st.Delivered, &st.Abandoned, &lastDelivered); err != nil {
		return st, err
	}
	if lastDelivered.Valid {
		st.LastDelivered = time.Unix(lastDelivered.Int64, 0).UTC()
	}
	// The most recent attempt that did not land, since the push was turned
	// on. The reader decides whether it is news: one older than the last
	// delivery has been recovered from. Giving up on a row is not an attempt
	// and never writes these columns.
	var failedAt int64
	err := s.db.QueryRowContext(ctx, `SELECT last_attempt_at, last_status, last_error FROM mail_event_deliveries
WHERE account = ? AND app = ? AND delivered_at = 0 AND last_attempt_at > 0 AND last_attempt_at >= ?
ORDER BY last_attempt_at DESC, id DESC LIMIT 1`, acct, app, since.Unix()).
		Scan(&failedAt, &st.LastStatus, &st.LastError)
	switch {
	case err == sql.ErrNoRows:
	case err != nil:
		return st, err
	default:
		st.LastFailedAt = time.Unix(failedAt, 0).UTC()
	}
	return st, nil
}

// ---- producing and sending ----------------------------------------------------------

// queueMailEventPush is the ingest's half: queue ev for app if the app has
// asked and the event is one an app is told about. The caller has already
// established that SES named this app as the sender — see the file comment
// for why nothing else is ever queued. Best effort, like the rest of the
// ingest: a failure is logged and the notification still answers 200.
func (s *Server) queueMailEventPush(ctx context.Context, acct, app, notice string, ev mail.Event, now time.Time) {
	if !mailPushedTypes[ev.Type] {
		return
	}
	hook, err := s.store.MailEventHookFor(ctx, acct, app)
	if err != nil {
		log.Printf("console: mail push: reading the hook for %s/%s: %v", acct, app, err)
		return
	}
	if hook == nil || hook.Suspended() {
		return
	}
	body, err := json.Marshal(mailEventPayload(acct, app, notice, ev, now))
	if err != nil {
		log.Printf("console: mail push: encoding %s for %s/%s: %v", ev.Type, acct, app, err)
		return
	}
	if err := s.store.EnqueueMailEventDelivery(ctx, acct, app, notice, body, now); err != nil {
		log.Printf("console: mail push: queueing %s for %s/%s: %v", ev.Type, acct, app, err)
	}
}

// mailPushURL is where a delivery for app goes.
func (s *Server) mailPushURL(acct, app, path string) (string, error) {
	if s.mailPushTarget != nil {
		return s.mailPushTarget(acct, app, path), nil
	}
	domain := strings.ToLower(strings.TrimSpace(s.cfg.AppsDomain))
	if domain == "" {
		return "", errors.New("this deployment has no apps domain to deliver to")
	}
	return "https://" + app + "." + acct + "." + domain + path, nil
}

// StartMailEventPush runs the queue every mailPushEvery until ctx is done.
// Started from cmd/console like every other pass — never from NewServer,
// which tests construct by the dozen.
func (s *Server) StartMailEventPush(ctx context.Context) {
	t := time.NewTicker(mailPushEvery)
	defer t.Stop()
	for {
		if err := s.mailPushPass(ctx, time.Now); err != nil {
			log.Printf("console: mail push: %v", err)
		}
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// mailPushPass sends what is due and trims the queue. It starts from the
// apps that asked — a short list — and seeks each one's oldest due rows, so
// its cost does not grow with anyone's backlog. Apps are sent to in
// parallel, mailPushApps at a time, and each app's own rows in order; see
// deliverAppMailEvents for why one app cannot hold up another. clock is read
// per delivery so a slow pass records honest times.
func (s *Server) mailPushPass(ctx context.Context, clock func() time.Time) error {
	start := clock()
	if err := s.store.pruneMailEventDeliveries(ctx, start); err != nil {
		// A prune that fails costs an hour's slack on a thirty-day window,
		// not the pass: what is due still goes out.
		log.Printf("console: mail push: pruning the queue: %v", err)
	}
	hooks, err := s.store.activeMailEventHooks(ctx)
	if err != nil {
		return fmt.Errorf("reading the hooks: %w", err)
	}
	if len(hooks) == 0 {
		return nil
	}
	var key ed25519.PrivateKey
	sem := make(chan struct{}, mailPushApps)
	var wg sync.WaitGroup
	for _, h := range hooks {
		if ctx.Err() != nil {
			break
		}
		if s.cfg.HostClaim {
			h.claimAppID, err = s.mutationAppID(ctx, h.Account, h.App)
			if err != nil {
				continue
			}
		}
		rows, err := s.store.dueMailEventDeliveriesFor(ctx, h.Account, h.App, start, mailPushPerApp)
		if err != nil {
			log.Printf("console: mail push: reading the queue for %s/%s: %v", h.Account, h.App, err)
			continue
		}
		if len(rows) == 0 {
			if err := s.store.clearMailEventFailing(ctx, h); err != nil {
				log.Printf("console: mail push: %s/%s: %v", h.Account, h.App, err)
			}
			continue
		}
		if key == nil {
			if key, err = s.store.MailEventsSigningKey(); err != nil {
				wg.Wait()
				return fmt.Errorf("reading the signing key: %w", err)
			}
		}
		wg.Add(1)
		sem <- struct{}{}
		go func() {
			defer func() { <-sem; wg.Done() }()
			s.deliverAppMailEvents(ctx, h, rows, key, clock)
		}()
	}
	wg.Wait()
	return nil
}

// deliverAppMailEvents sends one app's due rows in order. Any attempt that
// does not land ends the app's turn for this pass: an app that answers 500
// after nine seconds is no more likely to take the next row than the last,
// and each row it is offered holds one of mailPushApps for a timeout that
// every other app waits on. The rest keep their place for the next pass, and
// how the turn went is noted on the hook — which is where a day of nothing
// but failures suspends the push.
//
// An app that is no longer live, or whose push was turned off since the pass
// began, has all of its rows given up on at once — there is nothing to wait
// for.
func (s *Server) deliverAppMailEvents(ctx context.Context, h mailEventHookRow, rows []mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) {
	if !s.cfg.HostClaim {
		_ = s.deliverAppMailEventsHeld(ctx, h, rows, key, clock)
		return
	}
	acct, err := s.store.AccountBySqid(h.Account)
	if err == nil && acct == nil {
		err = hostops.ErrIdentity
	}
	if err == nil {
		currentID, readErr := s.mutationAppID(ctx, h.Account, h.App)
		err = readErr
		if err == nil && (h.claimAppID == "" || currentID != h.claimAppID) {
			err = hostops.ErrIdentity
		}
	}
	if err == nil {
		err = s.claimAppMutation(ctx, acct, h.App, "mail-event-delivery", func(held context.Context) error {
			if h.claimAppID == "" {
				return hostops.ErrIdentity
			}
			if err := hostops.RequireAppMutation(held, s.cfg.Releases, h.Account, h.App, h.claimAppID); err != nil {
				return err
			}
			// The queue may have changed while this goroutine waited for a slot.
			hook, err := s.store.MailEventHookFor(held, h.Account, h.App)
			if err != nil {
				return err
			}
			if hook == nil || hook.Suspended() {
				return nil
			}
			h.MailEventHook = *hook
			rows, err = s.store.dueMailEventDeliveriesFor(held, h.Account, h.App, clock(), mailPushPerApp)
			if err != nil {
				return err
			}
			return s.deliverAppMailEventsHeld(held, h, rows, key, clock)
		})
	}
	if err != nil {
		log.Printf("console: mail push: %s/%s: %v", h.Account, h.App, err)
	}
}

// The admitted context covers delivery and all SQL settlement. A transport
// failure can have reached the app, so its issued effect remains owned.
func (s *Server) deliverAppMailEventsHeld(ctx context.Context, h mailEventHookRow, rows []mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) error {
	var result error
	if !s.mailAppLive(h.Account, h.App) {
		// A trashed app's name is no longer a member's live app, and while it
		// sits in the trash its platform host is not reliably its own (the
		// instances door treats a trashed sibling's apex as free), so nothing
		// is sent to it; a purged, renamed or moved app's rows have already
		// been forgotten, carried or dropped.
		if err := s.store.abandonMailEventDeliveries(ctx, h.Account, h.App, "the app is not live", clock()); err != nil {
			result = errors.Join(result, err)
			log.Printf("console: mail push: giving up on %s/%s: %v", h.Account, h.App, err)
		}
		return result
	}
	landed, failure := false, ""
	for _, d := range rows {
		if ctx.Err() != nil {
			break
		}
		status, why, giveUp := s.deliverMailEvent(ctx, d, h.Path, key, clock)
		if giveUp {
			if err := s.store.abandonMailEventDeliveries(ctx, d.Account, d.App, why, clock()); err != nil {
				result = errors.Join(result, err)
				log.Printf("console: mail push: giving up on %s/%s: %v", d.Account, d.App, err)
			}
			return result
		}
		if err := s.store.recordMailEventAttempt(ctx, d, status, why, clock()); err != nil {
			result = errors.Join(result, err)
			log.Printf("console: mail push: recording delivery %d for %s/%s: %v", d.ID, d.Account, d.App, err)
		}
		if s.cfg.HostClaim && status == 0 && why != "" {
			result = errors.Join(result, errors.New(why))
		}
		if why != "" {
			failure = why
			break
		}
		landed = true
	}
	suspended, err := s.store.noteMailEventTurn(ctx, h, landed, failure, clock())
	if err != nil {
		result = errors.Join(result, err)
		log.Printf("console: mail push: noting the turn for %s/%s: %v", h.Account, h.App, err)
	}
	if suspended {
		log.Printf("console: mail push: suspended %s/%s — nothing accepted for %s", h.Account, h.App, mailPushSuspendAfter)
	}
	return result
}

// deliverMailEvent makes one attempt, and answers the HTTP status, what went
// wrong ("" when it landed), and whether to give up on the row now rather
// than retry it.
func (s *Server) deliverMailEvent(ctx context.Context, d mailEventDelivery, path string, key ed25519.PrivateKey, clock func() time.Time) (int, string, bool) {
	hook, err := s.store.MailEventHookFor(ctx, d.Account, d.App)
	if err != nil {
		return 0, "could not read the hook: " + err.Error(), false
	}
	if hook == nil || hook.Suspended() {
		// Turned off, or suspended, since the pass began. Both give up on the
		// queue themselves; this covers the race.
		return 0, "push turned off", true
	}
	if hook.Path != path {
		path = hook.Path // moved since the pass began; the new path is the one asked for
	}
	target, err := s.mailPushURL(d.Account, d.App, path)
	if err != nil {
		return 0, err.Error(), false
	}
	rctx, cancel := context.WithTimeout(ctx, mailPushTimeout)
	defer cancel()
	req, err := http.NewRequestWithContext(rctx, http.MethodPost, target, bytes.NewReader(d.Payload))
	if err != nil {
		return 0, "could not build the request: " + err.Error(), false
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("User-Agent", "carlos-mail-events/1")
	req.Header.Set("Carlos-Signature", signMailEvent(key, clock(), d.Payload))
	if s.cfg.HostClaim {
		appID, err := s.mutationAppID(ctx, d.Account, d.App)
		if err == nil {
			err = hostops.RequireAppMutation(ctx, s.cfg.Releases, d.Account, d.App, appID)
		}
		if err != nil {
			return 0, err.Error(), false
		}
		hostops.MarkAppMutationIssued(ctx)
	}
	resp, err := mailPushClient.Do(req)
	if err != nil {
		// The URL is the app's own and not a secret, but the error can be
		// long; the reason is what a member needs.
		return 0, "no answer: " + strings.TrimPrefix(err.Error(), "Post \""+target+"\": "), false
	}
	defer resp.Body.Close()
	io.Copy(io.Discard, io.LimitReader(resp.Body, mailPushMaxBody))
	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
		return resp.StatusCode, "", false
	}
	if resp.StatusCode >= 300 && resp.StatusCode < 400 {
		return resp.StatusCode, "answered with a redirect, which is not followed — POST to the path itself", false
	}
	return resp.StatusCode, fmt.Sprintf("answered %d", resp.StatusCode), false
}