amadan
Merge branch 'mail-events-push' into main
Merge branch 'mail-events-push' into main (email events push: tell the app itself, signed, at a path on its own host)
12 files changed,
+2074
−8
cmd/console/main.go+4 −0docs/site/cli.md+33 −2internal/console/apimail.go+177 −0internal/console/apimail_test.go+102 −0internal/console/mailevents.go+21 −6internal/console/mailpush.go+806 −0internal/console/mailpush_test.go+666 −0internal/console/server.go+5 −0internal/console/sesevents.go+6 −0internal/console/store.go+49 −0mail_cli.go+142 −0mail_cli_test.go+63 −0
CI passed — run details
diff --git a/cmd/console/main.go b/cmd/console/main.go| index 0205bf4eb..f70be5037 100644 |
| --- a/cmd/console/main.go |
| +++ b/cmd/console/main.go |
| @@ -351,6 +351,10 @@ func serve() { |
| // above, and it returns immediately when CARLOS_RELAY_REGIONS is empty — |
| // which is every deployment that has not armed a relay. |
| go srv.StartRelay(context.Background()) |
| + // The email-events push (internal/console/mailpush.go): delivers the |
| + // SES outcomes an app asked to be told about. Same placement rule as |
| + // every pass above; with nothing queued, a pass is two indexed queries. |
| + go srv.StartMailEventPush(context.Background()) |
| addr := envOr("CONSOLE_ADDR", "127.0.0.1:8090") |
| ln, err := net.Listen("tcp", addr) |
| if err != nil { |
diff --git a/docs/site/cli.md b/docs/site/cli.md| index 79e2b6185..618e6eab3 100644 |
| --- a/docs/site/cli.md |
| +++ b/docs/site/cli.md |
| @@ -882,8 +882,39 @@ domain, as the day's counts always have. |
| carlos email events --app hello --since 24h |
| carlos email events --app hello --to guest@example.com |
| ``` |
| - `credentials create` mints a |
| -standalone SMTP credential for something not running on CARLOS — a laptop, a |
| + |
| +`events push` tells the app itself. Give it a path and the console POSTs each |
| +delivery, bounce, complaint, rejection and delay SES attributes to the app, as |
| +JSON, to that path on the app's own platform host |
| +(`https://<app>.<account>.<apps domain><path>`), retrying on a widening |
| +schedule for about a day when the app does not answer 2xx. Redirects are not |
| +followed. Only events SES attributed to the app's own sending credential are |
| +sent, so an app is never handed another app's recipients. An app behind |
| +`carlos gate` cannot receive them. |
| + |
| +```sh |
| +carlos email events push --app hello --path /mail/events |
| +carlos restart --app hello |
| +carlos email events push --app hello |
| +``` |
| + |
| +Turning it on delivers `CARLOS_MAIL_EVENTS_KEY` into the app's config — an |
| +Ed25519 public key, standard base64 — which the app reads after a restart. |
| +Every delivery carries a `Carlos-Signature: t=<unix seconds>,ed25519=<base64>` |
| +header over the bytes `carlos-mail-event`, a newline, the `t` value, a newline, |
| +and the request body exactly as received. An app should refuse a signature |
| +that does not verify, a `t` more than five minutes from its own clock, and a |
| +body whose `account` and `app` do not name the host the request arrived on — |
| +`<app>.<account>.` — since the key is the deployment's, not the app's. The |
| +body's `id` is the same on every retry of one event. Bare, `events push` shows |
| +where events go and how delivery has been going; `--off` stops it and takes the |
| +key back out. While an app is in the trash, what was queued for it is given |
| +up on and new events are dropped, not held for a restore. An app that accepts |
| +nothing for a day — every attempt answered with anything but 2xx — has its push |
| +suspended: what is waiting is given up on, nothing new is queued, and `events |
| +push` says so. Running `events push --path` again turns it back on. |
| + |
| +`credentials create` mints a standalone SMTP credential for something not running on CARLOS — a laptop, a |
| cron box — printed once on that command's output and nowhere else. `rotate` |
| mints a fresh delivered credential and leaves the old one live until you run |
| it again with `--finish`. `pause` and `resume` are operator verbs. None of it |
diff --git a/internal/console/apimail.go b/internal/console/apimail.go| index bcc6ed024..773418753 100644 |
| --- a/internal/console/apimail.go |
| +++ b/internal/console/apimail.go |
| @@ -63,6 +63,7 @@ import ( |
| "github.com/carlosframework/platform/internal/dnsrec" |
| "github.com/carlosframework/platform/internal/mail" |
| "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/schedule" |
| "github.com/carlosframework/platform/internal/svcrole" |
| ) |
| @@ -467,6 +468,12 @@ func (s *Server) mailWalkReply(ctx context.Context, w http.ResponseWriter, acct |
| // the people the app wrote to. A credential that may read an app's routes |
| // or deploy it has no need of its recipients' addresses. |
| // |
| +// The three .../email/events/push doors are Admin for the same reason: turning |
| +// the push on is what starts recipients' addresses leaving the console, and |
| +// the status names them in its last failure. The addresses only ever go to |
| +// the app's own platform host (mailpush.go), so this is not a grant to send |
| +// them anywhere — but it is a decision about them, and Admin makes it. |
| +// |
| // The three CLOSED doors (svcrole.Set{} — closed to every role, admin |
| // included) are each closed for their own reason: |
| // |
| @@ -495,6 +502,9 @@ func (s *Server) registerEmailRoutes() { |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/email", svcrole.Allow(svcrole.Admin), s.handleAPIEmailDeclare) |
| s.cliRoute("GET /api/cli/apps/{account}/{app}/email", svcrole.Allow(svcrole.Admin), s.handleAPIEmailStatus) |
| s.cliRoute("GET /api/cli/apps/{account}/{app}/email/events", svcrole.Allow(svcrole.Admin), s.handleAPIEmailEvents) |
| + s.cliRoute("GET /api/cli/apps/{account}/{app}/email/events/push", svcrole.Allow(svcrole.Admin), s.handleAPIEmailPushStatus) |
| + s.cliRoute("PUT /api/cli/apps/{account}/{app}/email/events/push", svcrole.Allow(svcrole.Admin), s.handleAPIEmailPushSet) |
| + s.cliRoute("DELETE /api/cli/apps/{account}/{app}/email/events/push", svcrole.Allow(svcrole.Admin), s.handleAPIEmailPushOff) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/email/domains", svcrole.Allow(svcrole.Admin), s.handleAPIEmailDomainEnsure) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/email/domains/{domain}/verify", svcrole.Allow(svcrole.Admin), s.handleAPIEmailDomainVerify) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/email/enable", svcrole.Allow(svcrole.Admin), s.handleAPIEmailEnable) |
| @@ -715,6 +725,173 @@ func (s *Server) handleAPIEmailEvents(w http.ResponseWriter, r *http.Request) { |
| }) |
| } |
| +// ---- events push ------------------------------------------------------------ |
| + |
| +// emailPushReply is what all three push doors answer: where events go, the |
| +// key an app verifies them with, and how delivery has been going. |
| +func (s *Server) emailPushReply(ctx context.Context, acct *Account, app string) (map[string]any, error) { |
| + hook, err := s.store.MailEventHookFor(ctx, acct.Sqid, app) |
| + if err != nil { |
| + return nil, err |
| + } |
| + var since time.Time |
| + if hook != nil { |
| + since = hook.SetAt |
| + } |
| + st, err := s.store.MailEventPushStatusFor(ctx, acct.Sqid, app, since) |
| + if err != nil { |
| + return nil, err |
| + } |
| + out := map[string]any{"on": hook != nil, "key_env": MailEventsKeyEnv, "status": st} |
| + if hook == nil { |
| + return out, nil |
| + } |
| + if hook.Suspended() { |
| + out["suspended_at"] = hook.SuspendedAt |
| + out["suspended_why"] = hook.SuspendedWhy |
| + } |
| + key, err := s.store.MailEventsSigningKey() |
| + if err != nil { |
| + return nil, err |
| + } |
| + out["path"] = hook.Path |
| + out["set_by"] = hook.SetBy |
| + out["set_at"] = hook.SetAt |
| + out["public_key"] = mailEventsPublicKey(key) |
| + if url, err := s.mailPushURL(acct.Sqid, app, hook.Path); err == nil { |
| + out["url"] = url |
| + } |
| + return out, nil |
| +} |
| + |
| +// handleAPIEmailPushStatus answers `carlos email events push --app <a>` with |
| +// no change asked for. |
| +func (s *Server) handleAPIEmailPushStatus(w http.ResponseWriter, r *http.Request) { |
| + _, acct, app, ok := s.apiAccountApp(w, r) |
| + if !ok { |
| + return |
| + } |
| + if !s.mailProvisionerGate(w) { |
| + return |
| + } |
| + ctx, cancel := context.WithTimeout(r.Context(), mailTimeout) |
| + defer cancel() |
| + out, err := s.emailPushReply(ctx, acct, app) |
| + if err != nil { |
| + log.Printf("console: api email push status %s/%s: %v", acct.Sqid, app, err) |
| + jsonError(w, http.StatusBadGateway, "could not read the email events push — try again") |
| + return |
| + } |
| + writeJSON(w, http.StatusOK, out) |
| +} |
| + |
| +// handleAPIEmailPushSet turns the push on, or moves its path. |
| +// |
| +// The verifying key is delivered BEFORE the hook is recorded, so there is no |
| +// window in which events flow to an app that has nothing to check them |
| +// with. A failure between the two leaves the key delivered and the push off, |
| +// which a retry finishes. |
| +func (s *Server) handleAPIEmailPushSet(w http.ResponseWriter, r *http.Request) { |
| + addr, acct, app, ok := s.apiAccountApp(w, r) |
| + if !ok { |
| + return |
| + } |
| + if !s.mailWriteGates(w, addr) { |
| + return |
| + } |
| + var req struct { |
| + Path string `json:"path"` |
| + } |
| + if !readJSON(w, r, &req) { |
| + return |
| + } |
| + path := strings.TrimSpace(req.Path) |
| + if err := schedule.ValidPath(path); err != nil { |
| + jsonError(w, http.StatusBadRequest, fmt.Sprintf("path %q: %v", req.Path, err)) |
| + return |
| + } |
| + if _, err := s.mailPushURL(acct.Sqid, app, path); err != nil { |
| + jsonError(w, http.StatusServiceUnavailable, err.Error()) |
| + return |
| + } |
| + ctx, cancel := context.WithTimeout(r.Context(), mailTimeout) |
| + defer cancel() |
| + before, err := s.store.MailEventHookFor(ctx, acct.Sqid, app) |
| + if err != nil { |
| + log.Printf("console: api email push read %s/%s: %v", acct.Sqid, app, err) |
| + jsonError(w, http.StatusBadGateway, "could not read the email events push — nothing was changed; try again") |
| + return |
| + } |
| + key, err := s.store.MailEventsSigningKey() |
| + if err != nil { |
| + log.Printf("console: api email push key: %v", err) |
| + jsonError(w, http.StatusBadGateway, "could not read the signing key — nothing was changed; try again") |
| + return |
| + } |
| + sets := map[string]appconfig.Var{MailEventsKeyEnv: {Kind: "env", Value: mailEventsPublicKey(key)}} |
| + if _, err := appconfig.Apply(ctx, s.cfg.Releases, acct.Sqid, app, sets, nil); err != nil { |
| + s.configStoreError(w, fmt.Sprintf("api email push key %s/%s", acct.Sqid, app), err) |
| + return |
| + } |
| + s.bumpSerial(ctx) |
| + if err := s.store.SetMailEventHook(ctx, acct.Sqid, app, path, addr, time.Now()); err != nil { |
| + log.Printf("console: api email push set %s/%s: %v", acct.Sqid, app, err) |
| + jsonError(w, http.StatusBadGateway, "the verifying key was delivered, but the push could not be turned on — try again") |
| + return |
| + } |
| + if before == nil || before.Path != path { |
| + s.audit(acct, app, addr, "email-events-push", "pushing email events to "+path+" (cli)") |
| + } |
| + out, err := s.emailPushReply(ctx, acct, app) |
| + if err != nil { |
| + log.Printf("console: api email push reply %s/%s: %v", acct.Sqid, app, err) |
| + jsonError(w, http.StatusBadGateway, "the push is on, but its status could not be read — try `carlos email events push`") |
| + return |
| + } |
| + writeJSON(w, http.StatusOK, out) |
| +} |
| + |
| +// handleAPIEmailPushOff turns the push off: the hook goes, whatever is still |
| +// queued is given up on, and the key variable is taken back out of the |
| +// app's config. |
| +func (s *Server) handleAPIEmailPushOff(w http.ResponseWriter, r *http.Request) { |
| + addr, acct, app, ok := s.apiAccountApp(w, r) |
| + if !ok { |
| + return |
| + } |
| + if !s.mailWriteGates(w, addr) { |
| + return |
| + } |
| + ctx, cancel := context.WithTimeout(r.Context(), mailTimeout) |
| + defer cancel() |
| + was, err := s.store.DeleteMailEventHook(ctx, acct.Sqid, app, time.Now()) |
| + if err != nil { |
| + log.Printf("console: api email push off %s/%s: %v", acct.Sqid, app, err) |
| + jsonError(w, http.StatusBadGateway, "could not turn the email events push off — try again") |
| + return |
| + } |
| + // The key is taken out whether or not a push was on: turning it on |
| + // delivers the key before recording the hook, so a failure between the |
| + // two leaves a key with no push, and --off is how that is cleaned up. |
| + if removed, err := appconfig.Apply(ctx, s.cfg.Releases, acct.Sqid, app, nil, []string{MailEventsKeyEnv}); err != nil { |
| + // A key left in the environment verifies nothing that will ever |
| + // arrive. Said, not failed. |
| + log.Printf("console: api email push off %s/%s: unsetting %s: %v", acct.Sqid, app, MailEventsKeyEnv, err) |
| + } else if removed > 0 { |
| + s.bumpSerial(ctx) |
| + } |
| + if was { |
| + s.audit(acct, app, addr, "email-events-push-off", "stopped pushing email events (cli)") |
| + } |
| + out, err := s.emailPushReply(ctx, acct, app) |
| + if err != nil { |
| + log.Printf("console: api email push reply %s/%s: %v", acct.Sqid, app, err) |
| + jsonError(w, http.StatusBadGateway, "the push is off, but its status could not be read") |
| + return |
| + } |
| + writeJSON(w, http.StatusOK, out) |
| +} |
| + |
| // ---- domains ---------------------------------------------------------------- |
| // handleAPIEmailDomainEnsure answers `carlos email domains add`: bring a |
diff --git a/internal/console/apimail_test.go b/internal/console/apimail_test.go| index e90bc365f..82f57ed3d 100644 |
| --- a/internal/console/apimail_test.go |
| +++ b/internal/console/apimail_test.go |
| @@ -19,6 +19,7 @@ import ( |
| "testing" |
| "time" |
| + "github.com/carlosframework/platform/internal/appconfig" |
| "github.com/carlosframework/platform/internal/dnsrec" |
| "github.com/carlosframework/platform/internal/mail" |
| "github.com/carlosframework/platform/internal/release" |
| @@ -165,6 +166,14 @@ func TestEmailDoorsAre404ForNonMembers(t *testing.T) { |
| }{ |
| {"GET status", func(base, bearer string) (int, string) { return getRaw(t, base, bearer) }}, |
| {"GET events", func(base, bearer string) (int, string) { return getRaw(t, base+"/events", bearer) }}, |
| + {"GET events push", func(base, bearer string) (int, string) { return getRaw(t, base+"/events/push", bearer) }}, |
| + {"PUT events push", func(base, bearer string) (int, string) { |
| + return putRaw(t, base+"/events/push", map[string]any{"path": "/mail/events"}, bearer) |
| + }}, |
| + {"DELETE events push", func(base, bearer string) (int, string) { |
| + status, _, body := rawReq(t, http.MethodDelete, base+"/events/push", bearer, "") |
| + return status, body |
| + }}, |
| {"POST declare", func(base, bearer string) (int, string) { |
| return postRaw(t, base, map[string]any{"from": "hello@example.com"}, bearer) |
| }}, |
| @@ -227,6 +236,14 @@ func TestEmailDoorsAre404WithNoProvisioner(t *testing.T) { |
| }{ |
| {"GET status", func() (int, string) { return getRaw(t, emailURL(fx, "titogo"), token) }}, |
| {"GET events", func() (int, string) { return getRaw(t, emailURL(fx, "titogo")+"/events", token) }}, |
| + {"GET events push", func() (int, string) { return getRaw(t, emailURL(fx, "titogo")+"/events/push", token) }}, |
| + {"PUT events push", func() (int, string) { |
| + return putRaw(t, emailURL(fx, "titogo")+"/events/push", map[string]any{"path": "/mail/events"}, token) |
| + }}, |
| + {"DELETE events push", func() (int, string) { |
| + status, _, body := rawReq(t, http.MethodDelete, emailURL(fx, "titogo")+"/events/push", token, "") |
| + return status, body |
| + }}, |
| {"POST declare", func() (int, string) { |
| return postRaw(t, emailURL(fx, "titogo"), map[string]any{"from": "hello@example.com"}, token) |
| }}, |
| @@ -1154,3 +1171,88 @@ func TestAPIEmailEventsReadsTheLog(t *testing.T) { |
| } |
| } |
| } |
| + |
| +// TestAPIEmailPushTurnsOnAndOff: turning the push on delivers the key an app |
| +// verifies with into the app's own config, and names where events will go; |
| +// turning it off takes both back. |
| +func TestAPIEmailPushTurnsOnAndOff(t *testing.T) { |
| + fx, _, _, token := emailArmed(t, "code-email-push") |
| + fx.srv.cfg.AppsDomain = "oncarlos.com" |
| + app := "titogo" |
| + push := emailURL(fx, app) + "/events/push" |
| + |
| + status, out := getJSON(t, push, token) |
| + if status != http.StatusOK || out["on"] != false { |
| + t.Fatalf("before asking: %d %v", status, out) |
| + } |
| + |
| + status, body := putRaw(t, push, map[string]any{"path": "/mail/events"}, token) |
| + if status != http.StatusOK { |
| + t.Fatalf("turning it on: %d %s", status, body) |
| + } |
| + var on map[string]any |
| + if err := json.Unmarshal([]byte(body), &on); err != nil { |
| + t.Fatal(err) |
| + } |
| + if on["on"] != true || on["path"] != "/mail/events" || on["url"] != "https://titogo."+fx.sqid+".oncarlos.com/mail/events" { |
| + t.Errorf("reply = %v", on) |
| + } |
| + key, _ := fx.st.MailEventsSigningKey() |
| + vars := appVars(t, fx, app) |
| + if v, ok := vars[MailEventsKeyEnv]; !ok || v.Kind != "env" || v.Value != mailEventsPublicKey(key) { |
| + t.Errorf("%s = %+v, want the verifying key as a plain env var", MailEventsKeyEnv, v) |
| + } |
| + if on["public_key"] != mailEventsPublicKey(key) { |
| + t.Errorf("the reply should name the same key the app was given") |
| + } |
| + |
| + for _, bad := range []string{"", "mail/events", "//elsewhere.example/x", "/a/../b", "/with space"} { |
| + if status, body := putRaw(t, push, map[string]any{"path": bad}, token); status != http.StatusBadRequest { |
| + t.Errorf("path %q = %d %s, want 400", bad, status, body) |
| + } |
| + } |
| + |
| + status, _, body = rawReq(t, http.MethodDelete, push, token, "") |
| + if status != http.StatusOK || !strings.Contains(body, `"on":false`) { |
| + t.Fatalf("turning it off: %d %s", status, body) |
| + } |
| + if _, ok := appVars(t, fx, app)[MailEventsKeyEnv]; ok { |
| + t.Errorf("%s was left in the app's config after the push was turned off", MailEventsKeyEnv) |
| + } |
| +} |
| + |
| +// TestAPIEmailPushNeedsSomewhereToSend: a deployment with no apps domain has |
| +// no platform host to deliver to, and says so rather than recording a push |
| +// that could never arrive. |
| +func TestAPIEmailPushNeedsSomewhereToSend(t *testing.T) { |
| + fx, _, _, token := emailArmed(t, "code-email-push-nodomain") |
| + fx.srv.cfg.AppsDomain = "" |
| + status, body := putRaw(t, emailURL(fx, "titogo")+"/events/push", map[string]any{"path": "/mail/events"}, token) |
| + if status != http.StatusServiceUnavailable { |
| + t.Fatalf("with no apps domain = %d %s, want 503", status, body) |
| + } |
| + if hook, _ := fx.st.MailEventHookFor(context.Background(), fx.sqid, "titogo"); hook != nil { |
| + t.Error("a push was recorded with nowhere to send it") |
| + } |
| +} |
| + |
| +// TestAPIEmailPushOffTakesTheKeyOutEvenWithNoPush: turning the push on |
| +// delivers the key before it records the push, so a failure between the two |
| +// leaves a key and no push — and --off must still take the key out. |
| +func TestAPIEmailPushOffTakesTheKeyOutEvenWithNoPush(t *testing.T) { |
| + fx, _, _, token := emailArmed(t, "code-email-push-halfway") |
| + fx.srv.cfg.AppsDomain = "oncarlos.com" |
| + app := "titogo" |
| + key, _ := fx.st.MailEventsSigningKey() |
| + if _, err := appconfig.Apply(context.Background(), fx.rel, fx.sqid, app, |
| + map[string]appconfig.Var{MailEventsKeyEnv: {Kind: "env", Value: mailEventsPublicKey(key)}}, nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + status, _, body := rawReq(t, http.MethodDelete, emailURL(fx, app)+"/events/push", token, "") |
| + if status != http.StatusOK { |
| + t.Fatalf("off: %d %s", status, body) |
| + } |
| + if _, ok := appVars(t, fx, app)[MailEventsKeyEnv]; ok { |
| + t.Errorf("%s was left behind by --off with no push recorded", MailEventsKeyEnv) |
| + } |
| +} |
diff --git a/internal/console/mailevents.go b/internal/console/mailevents.go| index 9a5777ebe..46874d64f 100644 |
| --- a/internal/console/mailevents.go |
| +++ b/internal/console/mailevents.go |
| @@ -66,10 +66,12 @@ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, |
| return s.pruneMailEvents(ctx, now) |
| } |
| -// mailNoticeKey is the key a notification is kept under: SNS's own |
| -// MessageId, or — for a notification with no id of its own, which is nothing |
| -// SNS sends but the recording path is reachable directly — what it says, so |
| -// a repeat still collapses and two different notifications do not. |
| +// mailNoticeKey is the key a notification is kept and queued under: SNS's |
| +// own MessageId, or — for a notification with no id of its own, which is |
| +// nothing SNS sends but the recording path is reachable directly — what it |
| +// says, so a repeat still collapses and two different notifications do not. |
| +// The log and the push queue share it, so the id an app is handed is the id |
| +// the log has. |
| func mailNoticeKey(notice string, ev mail.Event, now time.Time) string { |
| if notice != "" { |
| return notice |
| @@ -89,7 +91,7 @@ func mailNoticeKey(notice string, ev mail.Event, now time.Time) string { |
| // mailAppTables are the tables keyed by an app's (account sqid, name) that |
| // belong to that app alone: a name that is freed must not hand them on. |
| -var mailAppTables = []string{"mail_events"} |
| +var mailAppTables = []string{"mail_events", "mail_event_hooks", "mail_event_deliveries"} |
| // execer is what carryMailApp and forgetMailApp need from a transaction. |
| type execer interface { |
| @@ -119,6 +121,16 @@ func carryMailApp(ctx context.Context, tx execer, fromSqid, fromApp, toSqid, toA |
| return err |
| } |
| } |
| + if table == "mail_event_deliveries" { |
| + // Not carried: a queued body names the account and app it was |
| + // cut for, so after a move or a rename it names neither, and an |
| + // app checking its host is right to refuse it. The final DELETE |
| + // below drops them. |
| + if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ? AND app = ?`, fromSqid, fromApp); err != nil { |
| + return err |
| + } |
| + continue |
| + } |
| if _, err := tx.ExecContext(ctx, `UPDATE `+table+` SET account = ?, app = ? WHERE account = ? AND app = ? AND `+ |
| mailAppTimeColumn(table)+` >= ?`, toSqid, toApp, fromSqid, fromApp, createdAt); err != nil { |
| return err |
| @@ -156,8 +168,11 @@ func forgetMailApp(ctx context.Context, tx execer, sqid, app string) error { |
| // mailAppTimeColumn is the column that says when a row came to be. |
| func mailAppTimeColumn(table string) string { |
| - if table == "mail_events" { |
| + switch table { |
| + case "mail_events": |
| return "received_at" |
| + case "mail_event_hooks": |
| + return "set_at" |
| } |
| return "created_at" |
| } |
diff --git a/internal/console/mailpush.go b/internal/console/mailpush.go| new file mode 100644 |
| index 000000000..5179f496b |
| --- /dev/null |
| +++ b/internal/console/mailpush.go |
| @@ -0,0 +1,806 @@ |
| +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/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 { |
| + _, err := s.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 { |
| + 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) (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 |
| + } |
| + _, err := s.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) (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 { |
| + _, err := s.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. |
| +func (s *Store) recordMailEventAttempt(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 abandoned_at = 0`, |
| + attempts, now.Unix(), now.Unix(), status, d.ID) |
| + 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 abandoned_at = 0`, |
| + attempts, now.Unix(), now.Unix(), status, failure, d.ID) |
| + 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 abandoned_at = 0`, |
| + attempts, now.Add(mailPushBackoff[attempts-1]).Unix(), now.Unix(), status, failure, d.ID) |
| + 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 { |
| + _, err := s.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 _, 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 |
| + } |
| + 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.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 { |
| + log.Printf("console: mail push: giving up on %s/%s: %v", h.Account, h.App, err) |
| + } |
| + return |
| + } |
| + 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 { |
| + log.Printf("console: mail push: giving up on %s/%s: %v", d.Account, d.App, err) |
| + } |
| + return |
| + } |
| + if err := s.store.recordMailEventAttempt(ctx, d, status, why, clock()); err != nil { |
| + log.Printf("console: mail push: recording delivery %d for %s/%s: %v", d.ID, d.Account, d.App, err) |
| + } |
| + if why != "" { |
| + failure = why |
| + break |
| + } |
| + landed = true |
| + } |
| + suspended, err := s.store.noteMailEventTurn(ctx, h, landed, failure, clock()) |
| + if err != nil { |
| + 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) |
| + } |
| +} |
| + |
| +// 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)) |
| + 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 |
| +} |
diff --git a/internal/console/mailpush_test.go b/internal/console/mailpush_test.go| new file mode 100644 |
| index 000000000..19c965c0a |
| --- /dev/null |
| +++ b/internal/console/mailpush_test.go |
| @@ -0,0 +1,666 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "io" |
| + "net/http" |
| + "net/http/httptest" |
| + "strconv" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/mail" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The scheme has one executable definition, and both ends are held to it: a |
| +// body signed by the console verifies with the key the app is given, and |
| +// nothing about it can be changed on the way. |
| +func TestAMailEventSignatureVerifiesAndNothingElseDoes(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + key, err := fx.st.MailEventsSigningKey() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + again, _ := fx.st.MailEventsSigningKey() |
| + if mailEventsPublicKey(again) != mailEventsPublicKey(key) { |
| + t.Fatal("the signing key changed between reads — every app's delivered key would stop verifying") |
| + } |
| + pub := mailEventsPublicKey(key) |
| + now := time.Date(2026, 9, 15, 9, 0, 0, 0, time.UTC) |
| + body := []byte(`{"id":"n1","account":"beb","app":"jb-book","type":"Bounce"}`) |
| + header := signMailEvent(key, now, body) |
| + |
| + if err := VerifyMailEvent(pub, header, body, now.Add(time.Minute), 5*time.Minute); err != nil { |
| + t.Fatalf("a genuine delivery did not verify: %v", err) |
| + } |
| + other, _ := newConsoleFixture(t).st.MailEventsSigningKey() |
| + for name, c := range map[string]struct { |
| + key, header string |
| + body []byte |
| + at time.Time |
| + }{ |
| + "a changed body": {pub, header, []byte(strings.Replace(string(body), "jb-book", "jb-site", 1)), now}, |
| + "another deployment's key": {mailEventsPublicKey(other), header, body, now}, |
| + "a replay long after": {pub, header, body, now.Add(10 * time.Minute)}, |
| + "a timestamp from the future": {pub, header, body, now.Add(-10 * time.Minute)}, |
| + "a moved timestamp": {pub, strings.Replace(header, "t=", "t=1", 1), body, now}, |
| + "no signature at all": {pub, "t=1726390800", body, now}, |
| + } { |
| + if err := VerifyMailEvent(c.key, c.header, c.body, c.at, 5*time.Minute); err == nil { |
| + t.Errorf("%s verified", name) |
| + } |
| + } |
| +} |
| + |
| +// pushFixture is one app on one domain, asking to be told, with the push |
| +// pointed at a test server instead of the app's platform host. |
| +func pushFixture(t *testing.T, handler http.HandlerFunc) (*consoleFixture, *httptest.Server) { |
| + t.Helper() |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + now := time.Now() |
| + for _, app := range []string{"jb-book", "jb-site"} { |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, app, "bill@tito.io", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + rec := &mail.DomainRecord{ |
| + Domain: "lgc.li", Kind: mail.KindCustom, Apps: []string{"jb-book", "jb-site"}, |
| + Regions: map[string]*mail.RegionState{"eu-west-1": { |
| + IdentityARN: "arn:aws:ses:eu-west-1:1:identity/lgc.li", MailFrom: "mail-eu-west-1.lgc.li", Status: mail.StatusVerified, |
| + }}, |
| + CreatedAt: now, |
| + } |
| + if _, err := mail.PutDomain(ctx, fx.rel, release.OpsAccount, rec, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, app := range []string{"jb-book", "jb-site"} { |
| + if err := fx.st.SetMailEventHook(ctx, release.OpsAccount, app, "/mail/events", "bill@tito.io", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + srv := httptest.NewServer(handler) |
| + t.Cleanup(srv.Close) |
| + fx.srv.mailPushTarget = func(acct, app, path string) string { return srv.URL + path + "?app=" + app } |
| + return fx, srv |
| +} |
| + |
| +func sentBy(app string, ev mail.Event) mail.Event { |
| + ev.Caller = mail.UserName(release.OpsAccount, app) |
| + return ev |
| +} |
| + |
| +func queued(t *testing.T, fx *consoleFixture, app string) int { |
| + t.Helper() |
| + var n int |
| + if err := fx.st.db.QueryRow(`SELECT count(*) FROM mail_event_deliveries WHERE app = ?`, app).Scan(&n); err != nil { |
| + t.Fatal(err) |
| + } |
| + return n |
| +} |
| + |
| +// Only an event SES attributed to the app itself is ever queued for it, and |
| +// only a type worth telling. A notification SNS delivers twice is queued |
| +// once. |
| +func TestOnlyTheSendersOwnOutcomesAreQueued(t *testing.T) { |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {}) |
| + ctx := context.Background() |
| + now := time.Now() |
| + |
| + bounce := sentBy("jb-book", bounceTo("ann@example.com", "m-ann", now)) |
| + fx.srv.recordAndEnforceMail(ctx, bounce, "sns-1") |
| + fx.srv.recordAndEnforceMail(ctx, bounce, "sns-1") |
| + if n := queued(t, fx, "jb-book"); n != 1 { |
| + t.Errorf("jb-book queued %d, want 1 — a redelivered notification is one delivery", n) |
| + } |
| + if n := queued(t, fx, "jb-site"); n != 0 { |
| + t.Errorf("jb-site was queued another app's recipient: %d", n) |
| + } |
| + |
| + // Attributed only by the shared domain: kept in both logs, pushed to |
| + // neither. |
| + fx.srv.recordAndEnforceMail(ctx, bounceTo("bob@example.com", "m-bob", now), "sns-2") |
| + if n := queued(t, fx, "jb-book") + queued(t, fx, "jb-site"); n != 1 { |
| + t.Errorf("a domain-attributed event was queued (%d rows in all)", n-1) |
| + } |
| + |
| + // A Send is not news to the app that sent it. |
| + send := sentBy("jb-book", mail.Event{Type: mail.EventSend, Domain: "lgc.li", MessageID: "m-cat", At: now, Destination: []string{"cat@example.com"}}) |
| + fx.srv.recordAndEnforceMail(ctx, send, "sns-3") |
| + if n := queued(t, fx, "jb-book"); n != 1 { |
| + t.Errorf("a Send was queued") |
| + } |
| + |
| + // And an app that has not asked is not queued anything. |
| + if _, err := fx.st.DeleteMailEventHook(ctx, release.OpsAccount, "jb-book", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("dan@example.com", "m-dan", now)), "sns-4") |
| + if n := queued(t, fx, "jb-book"); n != 1 { |
| + t.Errorf("an app with no hook was queued") |
| + } |
| +} |
| + |
| +// A delivery lands at the app's path, signed with the key the app was |
| +// given, carrying who it was about and what the receiving server said. |
| +func TestADeliveryReachesTheAppSigned(t *testing.T) { |
| + var mu sync.Mutex |
| + var gotPath, gotSig string |
| + var gotBody []byte |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| + mu.Lock() |
| + defer mu.Unlock() |
| + gotPath = r.URL.Path + "?" + r.URL.RawQuery |
| + gotSig = r.Header.Get("Carlos-Signature") |
| + gotBody, _ = io.ReadAll(r.Body) |
| + w.WriteHeader(http.StatusNoContent) |
| + }) |
| + ctx := context.Background() |
| + now := time.Now() |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", now)), "sns-1") |
| + |
| + if err := fx.srv.mailPushPass(ctx, time.Now); err != nil { |
| + t.Fatal(err) |
| + } |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if gotPath != "/mail/events?app=jb-book" { |
| + t.Fatalf("delivered to %q", gotPath) |
| + } |
| + key, _ := fx.st.MailEventsSigningKey() |
| + if err := VerifyMailEvent(mailEventsPublicKey(key), gotSig, gotBody, time.Now(), 5*time.Minute); err != nil { |
| + t.Fatalf("the app could not verify what it was sent: %v", err) |
| + } |
| + var p MailEventPayload |
| + if err := json.Unmarshal(gotBody, &p); err != nil { |
| + t.Fatal(err) |
| + } |
| + if p.ID != "sns-1" || p.Account != release.OpsAccount || p.App != "jb-book" || p.Type != mail.EventBounce || |
| + p.MessageID != "m-ann" || p.Detail != "Permanent/NoEmail" { |
| + t.Errorf("payload = %+v", p) |
| + } |
| + if len(p.Recipients) != 1 || p.Recipients[0].Address != "ann@example.com" || p.Recipients[0].Status != "5.1.1" { |
| + t.Errorf("recipients = %+v", p.Recipients) |
| + } |
| + st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| + if st.Delivered != 1 || st.Pending != 0 || st.LastDelivered.IsZero() { |
| + t.Errorf("status after delivery = %+v", st) |
| + } |
| + |
| + // Delivered is delivered: the next pass sends nothing. |
| + gotPath = "" |
| + mu.Unlock() |
| + fx.srv.mailPushPass(ctx, time.Now) |
| + mu.Lock() |
| + if gotPath != "" { |
| + t.Error("a delivered event was sent again") |
| + } |
| +} |
| + |
| +// A delivery that does not land is tried again on a widening schedule, and |
| +// given up on when the schedule is spent — never retried for ever, and never |
| +// kept past the log it came from. |
| +func TestAFailedDeliveryBacksOffAndIsGivenUpOn(t *testing.T) { |
| + var mu sync.Mutex |
| + hits := 0 |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| + mu.Lock() |
| + hits++ |
| + mu.Unlock() |
| + w.WriteHeader(http.StatusServiceUnavailable) |
| + }) |
| + ctx := context.Background() |
| + now := time.Date(2026, 9, 15, 9, 0, 0, 0, time.UTC) |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", now)), "sns-1") |
| + // Queued by the real clock; start the fake one there. |
| + var created int64 |
| + fx.st.db.QueryRow(`SELECT created_at FROM mail_event_deliveries`).Scan(&created) |
| + clock := time.Unix(created, 0) |
| + tick := func() time.Time { return clock } |
| + |
| + fx.srv.mailPushPass(ctx, tick) |
| + st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| + if st.Pending != 1 || st.LastStatus != 503 || !strings.Contains(st.LastError, "503") || !st.LastFailedAt.Equal(clock) { |
| + t.Fatalf("after one failure: %+v — the failure should be dated when it happened, %s", st, clock.UTC()) |
| + } |
| + // Not due again until the first backoff has passed. |
| + clock = clock.Add(30 * time.Second) |
| + fx.srv.mailPushPass(ctx, tick) |
| + if hits != 1 { |
| + t.Fatalf("retried before its backoff: %d attempts", hits) |
| + } |
| + for _, wait := range mailPushBackoff { |
| + clock = clock.Add(wait) |
| + fx.srv.mailPushPass(ctx, tick) |
| + } |
| + if hits != len(mailPushBackoff)+1 { |
| + t.Errorf("attempts = %d, want %d", hits, len(mailPushBackoff)+1) |
| + } |
| + st, _ = fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| + if st.Abandoned != 1 || st.Pending != 0 { |
| + t.Errorf("after the schedule is spent: %+v", st) |
| + } |
| + clock = clock.Add(24 * time.Hour) |
| + fx.srv.mailPushPass(ctx, tick) |
| + if hits != len(mailPushBackoff)+1 { |
| + t.Error("an abandoned delivery was tried again") |
| + } |
| + |
| + // And the row is gone once the retention window has passed. |
| + clock = time.Unix(created, 0).Add(mailEventsRetention + time.Hour) |
| + fx.srv.mailPushPass(ctx, tick) |
| + if n := queued(t, fx, "jb-book"); n != 0 { |
| + t.Errorf("a queued payload outlived the retention window") |
| + } |
| +} |
| + |
| +// A redirect is where a delivery would become a POST of somebody's address |
| +// at a host the far end chooses. It is not followed. |
| +func TestADeliveryDoesNotFollowARedirect(t *testing.T) { |
| + elsewhere := notReached(t) |
| + defer elsewhere.Close() |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| + http.Redirect(w, r, elsewhere.URL+"/steal", http.StatusTemporaryRedirect) |
| + }) |
| + ctx := context.Background() |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1") |
| + fx.srv.mailPushPass(ctx, time.Now) |
| + st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| + if st.Delivered != 0 || !strings.Contains(st.LastError, "redirect") { |
| + t.Errorf("a redirect counted as delivered, or went unexplained: %+v", st) |
| + } |
| +} |
| + |
| +// Turning the push off stops what is already queued, not only what comes |
| +// next. |
| +func TestTurningThePushOffStopsTheQueue(t *testing.T) { |
| + hits := 0 |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { hits++ }) |
| + ctx := context.Background() |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1") |
| + if _, err := fx.st.DeleteMailEventHook(ctx, release.OpsAccount, "jb-book", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.srv.mailPushPass(ctx, time.Now) |
| + if hits != 0 { |
| + t.Error("a delivery went out after the push was turned off") |
| + } |
| +} |
| + |
| +// The production target is the app's own platform host and nothing else. |
| +func TestThePushGoesOnlyToTheAppsOwnHost(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + fx.srv.cfg.AppsDomain = "OnCarlos.com" |
| + got, err := fx.srv.mailPushURL("beb", "jb-book", "/mail/events") |
| + if err != nil || got != "https://jb-book.beb.oncarlos.com/mail/events" { |
| + t.Errorf("target = %q, %v", got, err) |
| + } |
| + fx.srv.cfg.AppsDomain = "" |
| + if _, err := fx.srv.mailPushURL("beb", "jb-book", "/mail/events"); err == nil { |
| + t.Error("a deployment with no apps domain should have nowhere to send, not a guess") |
| + } |
| +} |
| + |
| +// One app's handler hanging must not be every app's problem. A pass sends to |
| +// apps side by side, and an app that does not answer loses its turn after |
| +// one timeout rather than spending the whole pass. |
| +func TestAHungAppDoesNotHoldUpAnotherApp(t *testing.T) { |
| + defer func(d time.Duration) { mailPushTimeout = d }(mailPushTimeout) |
| + mailPushTimeout = 300 * time.Millisecond |
| + var mu sync.Mutex |
| + hungHits, otherHits := 0, 0 |
| + unblock := make(chan struct{}) |
| + defer close(unblock) |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| + mu.Lock() |
| + if r.URL.Query().Get("app") == "jb-book" { |
| + hungHits++ |
| + mu.Unlock() |
| + select { |
| + case <-unblock: |
| + case <-time.After(5 * time.Second): |
| + } |
| + return |
| + } |
| + otherHits++ |
| + mu.Unlock() |
| + w.WriteHeader(http.StatusNoContent) |
| + }) |
| + ctx := context.Background() |
| + now := time.Now() |
| + for i := 0; i < 5; i++ { |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-hung-"+string(rune('a'+i)), now)), "sns-hung-"+string(rune('a'+i))) |
| + } |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("bob@example.com", "m-site", now)), "sns-site") |
| + |
| + started := time.Now() |
| + if err := fx.srv.mailPushPass(ctx, time.Now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if took := time.Since(started); took > 2*time.Second { |
| + t.Errorf("the pass took %v — a hung app spent it", took) |
| + } |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if otherHits != 1 { |
| + t.Errorf("the other app was sent %d, want its one delivery", otherHits) |
| + } |
| + if hungHits != 1 { |
| + t.Errorf("the hung app was tried %d times in one pass, want 1 — its turn ends at the first silence", hungHits) |
| + } |
| + st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| + if st.Pending != 5 { |
| + t.Errorf("the hung app's deliveries should all still be waiting: %+v", st) |
| + } |
| +} |
| + |
| +// A backlog waits behind itself. An app with far more due than a pass sends |
| +// does not push another app's one delivery out of the pass. |
| +func TestABacklogIsItsOwnAppsToWaitBehind(t *testing.T) { |
| + var mu sync.Mutex |
| + sent := map[string]int{} |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| + mu.Lock() |
| + sent[r.URL.Query().Get("app")]++ |
| + mu.Unlock() |
| + w.WriteHeader(http.StatusNoContent) |
| + }) |
| + ctx := context.Background() |
| + now := time.Now() |
| + for i := 0; i < 3*mailPushPerApp; i++ { |
| + id := "m-busy-" + strconv.Itoa(i) |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, now)), "sns-"+id) |
| + } |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("bob@example.com", "m-quiet", now)), "sns-quiet") |
| + if err := fx.srv.mailPushPass(ctx, time.Now); err != nil { |
| + t.Fatal(err) |
| + } |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if sent["jb-site"] != 1 || sent["jb-book"] != mailPushPerApp { |
| + t.Errorf("sent %v, want jb-site 1 and jb-book %d", sent, mailPushPerApp) |
| + } |
| +} |
| + |
| +// Nothing is sent for an app that is not live, and a trashed app's queue is |
| +// given up on rather than kept waiting for a host it may no longer own. |
| +func TestNothingIsPushedForATrashedApp(t *testing.T) { |
| + hits := 0 |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { hits++ }) |
| + ctx := context.Background() |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1") |
| + if err := fx.st.TrashAppRow(OpsAccountID, "jb-book", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.srv.mailPushPass(ctx, time.Now) |
| + if hits != 0 { |
| + t.Error("a delivery went to a trashed app's host") |
| + } |
| + st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| + if st.Abandoned != 1 || st.Pending != 0 { |
| + t.Errorf("a trashed app's delivery should be given up on: %+v", st) |
| + } |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("bob@example.com", "m-bob", time.Now())), "sns-2") |
| + if n := queued(t, fx, "jb-book"); n != 1 { |
| + t.Errorf("a trashed app was queued something new") |
| + } |
| +} |
| + |
| +// The push is the app's, like its log: a rename takes it along, and a purge |
| +// leaves nothing for the next app of that name to inherit. |
| +func TestThePushFollowsTheAppAndNotItsName(t *testing.T) { |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {}) |
| + ctx := context.Background() |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1") |
| + |
| + if err := fx.st.RenameAppIn(OpsAccountID, "jb-site", "jb-web"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-web"); hook == nil { |
| + t.Error("the renamed app lost its push") |
| + } |
| + if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-site"); hook != nil { |
| + t.Error("the old name kept a push for whoever claims it next") |
| + } |
| + if n := queued(t, fx, "jb-web") + queued(t, fx, "jb-site"); n != 0 { |
| + t.Errorf("a rename carried %d queued deliveries whose bodies name the old app", n) |
| + } |
| + |
| + if err := fx.st.TrashAppRow(OpsAccountID, "jb-web", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.st.PurgeAppRow(OpsAccountID, "jb-web"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-web"); hook != nil { |
| + t.Error("a purged app's push survived for its name") |
| + } |
| + if n := queued(t, fx, "jb-web") + queued(t, fx, "jb-site"); n != 0 { |
| + t.Errorf("a purged app's queue survived: %d rows", n) |
| + } |
| +} |
| + |
| +// An app that answers, but not with a 2xx, ends its turn at the first |
| +// failure: offering it the next row only holds a slot others wait on. |
| +func TestAFailingAppEndsItsTurnAtTheFirstFailure(t *testing.T) { |
| + var mu sync.Mutex |
| + hits := 0 |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| + mu.Lock() |
| + hits++ |
| + mu.Unlock() |
| + w.WriteHeader(http.StatusInternalServerError) |
| + }) |
| + ctx := context.Background() |
| + now := time.Now() |
| + for i := 0; i < mailPushPerApp; i++ { |
| + id := "m-" + strconv.Itoa(i) |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, now)), "sns-"+id) |
| + } |
| + fx.srv.mailPushPass(ctx, time.Now) |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if hits != 1 { |
| + t.Errorf("a failing app was offered %d rows in one pass, want 1", hits) |
| + } |
| +} |
| + |
| +// An app that accepts nothing for a day is suspended: what is waiting is |
| +// given up on, nothing new is queued, and turning the push on again lifts |
| +// it. Without this, a failing app's backlog grows for the thirty days the |
| +// queue is kept. |
| +func TestADayOfFailuresSuspendsThePush(t *testing.T) { |
| + hits := 0 |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| + hits++ |
| + w.WriteHeader(http.StatusNotFound) |
| + }) |
| + ctx := context.Background() |
| + send := func(id string, at time.Time) { |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, at)), "sns-"+id) |
| + } |
| + clock := time.Now() |
| + tick := func() time.Time { return clock } |
| + for i := 0; i < 20; i++ { |
| + send("m-"+strconv.Itoa(i), clock) |
| + } |
| + fx.srv.mailPushPass(ctx, tick) |
| + hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book") |
| + if hook.FailingSince.IsZero() || hook.Suspended() { |
| + t.Fatalf("after the first failure: %+v", hook) |
| + } |
| + |
| + // Most of a day of failing is not a day. |
| + clock = clock.Add(23 * time.Hour) |
| + fx.srv.mailPushPass(ctx, tick) |
| + if hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book"); hook.Suspended() { |
| + t.Fatal("suspended before a day had passed") |
| + } |
| + |
| + clock = clock.Add(2 * time.Hour) |
| + fx.srv.mailPushPass(ctx, tick) |
| + hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book") |
| + if !hook.Suspended() || !strings.Contains(hook.SuspendedWhy, "404") { |
| + t.Fatalf("a day of failures should suspend, saying what the app answered: %+v", hook) |
| + } |
| + st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", hook.SetAt) |
| + if st.Pending != 0 || st.Abandoned != 20 { |
| + t.Errorf("a suspension should give up on the whole queue: %+v", st) |
| + } |
| + |
| + // Suspended: nothing new is queued, and a pass sends nothing. |
| + send("m-late", clock) |
| + if n := queued(t, fx, "jb-book"); n != 20 { |
| + t.Errorf("a suspended push queued more (%d rows)", n) |
| + } |
| + before := hits |
| + clock = clock.Add(time.Hour) |
| + fx.srv.mailPushPass(ctx, tick) |
| + if hits != before { |
| + t.Error("a suspended push was sent to") |
| + } |
| + |
| + // Turning it on again lifts the suspension and starts clean. |
| + if err := fx.st.SetMailEventHook(ctx, release.OpsAccount, "jb-book", "/mail/events", "bill@tito.io", clock); err != nil { |
| + t.Fatal(err) |
| + } |
| + hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book") |
| + if hook.Suspended() || !hook.FailingSince.IsZero() { |
| + t.Errorf("turning it on again should lift the suspension: %+v", hook) |
| + } |
| + send("m-after", clock) |
| + if n := queued(t, fx, "jb-book"); n != 21 { |
| + t.Errorf("after turning it on again, events should queue: %d rows", n) |
| + } |
| +} |
| + |
| +// A run of failures belongs to a backlog, not to the app's history: an app |
| +// whose queue emptied — here by its only delivery running out of retries, |
| +// with nothing ever landing — is not suspended by its next miss days later. |
| +func TestAnOldFailureWithNothingWaitingDoesNotSuspend(t *testing.T) { |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) }) |
| + ctx := context.Background() |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1") |
| + var created int64 |
| + fx.st.db.QueryRow(`SELECT created_at FROM mail_event_deliveries`).Scan(&created) |
| + clock := time.Unix(created, 0) |
| + tick := func() time.Time { return clock } |
| + fx.srv.mailPushPass(ctx, tick) |
| + for _, wait := range mailPushBackoff { |
| + clock = clock.Add(wait) |
| + fx.srv.mailPushPass(ctx, tick) |
| + } |
| + if st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}); st.Pending != 0 || st.Abandoned != 1 { |
| + t.Fatalf("the one delivery should have run out of retries by now: %+v", st) |
| + } |
| + clock = clock.Add(3 * 24 * time.Hour) |
| + fx.srv.mailPushPass(ctx, tick) // nothing waiting: the run ends here |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("bob@example.com", "m-2", clock)), "sns-2") |
| + fx.st.db.Exec(`UPDATE mail_event_deliveries SET next_at = ?, created_at = ? WHERE notice = 'sns-2'`, clock.Unix(), clock.Unix()) |
| + fx.srv.mailPushPass(ctx, tick) |
| + if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book"); hook.Suspended() { |
| + t.Errorf("one miss after days of quiet suspended the push: %+v", hook) |
| + } |
| +} |
| + |
| +// A pass seeks each app's oldest due rows through an index rather than |
| +// sorting everything due: its cost follows the number of apps asking, not |
| +// the size of anyone's backlog. |
| +func TestAPassDoesNotSortAnAppsBacklog(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + rows, err := fx.st.db.Query(`EXPLAIN QUERY PLAN `+mailPushDueSQL, "beb", "jb-book", time.Now().Unix(), mailPushPerApp) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer rows.Close() |
| + var plan []string |
| + for rows.Next() { |
| + var id, parent, notused int |
| + var detail string |
| + rows.Scan(&id, &parent, ¬used, &detail) |
| + plan = append(plan, detail) |
| + } |
| + joined := strings.Join(plan, " | ") |
| + if !strings.Contains(joined, "mail_event_deliveries_app_due") || strings.Contains(joined, "TEMP B-TREE") { |
| + t.Errorf("the due read should seek the per-app index without sorting: %s", joined) |
| + } |
| + |
| + // And a large backlog costs a pass one app's worth of rows. |
| + sent := 0 |
| + fx2, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { sent++; w.WriteHeader(http.StatusNoContent) }) |
| + ctx := context.Background() |
| + now := time.Now() |
| + for i := 0; i < 2000; i++ { |
| + if err := fx2.st.EnqueueMailEventDelivery(ctx, release.OpsAccount, "jb-book", "bulk-"+strconv.Itoa(i), []byte(`{}`), now); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + fx2.srv.mailPushPass(ctx, time.Now) |
| + if sent != mailPushPerApp { |
| + t.Errorf("one pass sent %d, want %d", sent, mailPushPerApp) |
| + } |
| +} |
| + |
| +// Turning the push off and on again, or lifting a suspension, starts the |
| +// status afresh: an old failure is not reported as the last attempt. |
| +func TestTheStatusForgetsFailuresFromBeforeThePushWasTurnedOn(t *testing.T) { |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusServiceUnavailable) }) |
| + ctx := context.Background() |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1") |
| + fx.srv.mailPushPass(ctx, time.Now) |
| + if _, err := fx.st.DeleteMailEventHook(ctx, release.OpsAccount, "jb-book", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + later := time.Now().Add(time.Minute) |
| + if err := fx.st.SetMailEventHook(ctx, release.OpsAccount, "jb-book", "/mail/events", "bill@tito.io", later); err != nil { |
| + t.Fatal(err) |
| + } |
| + st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", later) |
| + if st.LastError != "" || st.LastStatus != 0 { |
| + t.Errorf("a failure from before the push was turned on again was reported: %+v", st) |
| + } |
| + var why, lastErr string |
| + fx.st.db.QueryRow(`SELECT abandoned_why, last_error FROM mail_event_deliveries`).Scan(&why, &lastErr) |
| + if why != "push turned off" || !strings.Contains(lastErr, "503") { |
| + t.Errorf("giving up should record why without overwriting what the app answered: why=%q last_error=%q", why, lastErr) |
| + } |
| +} |
| + |
| +// A move to another account drops the queue, like a rename: the bodies name |
| +// the old account. |
| +func TestAMoveDropsTheQueue(t *testing.T) { |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {}) |
| + ctx := context.Background() |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1") |
| + if err := fx.st.CreateAccount(99, "dest", "Destination", "bill@tito.io", time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.st.MoveApp("jb-site", OpsAccountID, 99); err != nil { |
| + t.Fatal(err) |
| + } |
| + var n int |
| + fx.st.db.QueryRow(`SELECT count(*) FROM mail_event_deliveries WHERE app = 'jb-site'`).Scan(&n) |
| + if n != 0 { |
| + t.Errorf("a move carried %d queued deliveries naming the old account", n) |
| + } |
| + if hook, _ := fx.st.MailEventHookFor(ctx, "dest", "jb-site"); hook == nil { |
| + t.Error("the push itself should move with the app") |
| + } |
| +} |
| + |
| +// A prune that fails is not stamped, and does not cost the pass. |
| +func TestAFailedQueuePruneIsTriedAgain(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + cancel() |
| + if err := fx.st.pruneMailEventDeliveries(ctx, time.Now()); err == nil { |
| + t.Fatal("a prune with a cancelled context should fail") |
| + } |
| + if got := fx.st.mailPushPruned.Load(); got != 0 { |
| + t.Errorf("the stamp was left at %d after a failed prune", got) |
| + } |
| +} |
diff --git a/internal/console/server.go b/internal/console/server.go| index c3afb458c..7efe304fc 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -802,6 +802,11 @@ type Server struct { |
| // ONLY: it never substitutes for the TopicArn or SigningCertURL checks, |
| // which every caller — production and test alike — runs for real. |
| snsVerify func(snsEnvelope) error |
| + // mailPushTarget is the email-events push's test seam (mailpush.go): |
| + // nil means "the app's own platform host, over https", which is the |
| + // only place production ever sends. A test points it at an httptest |
| + // server instead. The path still goes through the same validation. |
| + mailPushTarget func(acct, app, path string) string |
| // snsCerts caches a verified signing certificate's RSA public key by URL |
| // (sesevents.go) — otherwise every request bearing the deployment's |
| // TopicArn, which is not a secret, triggers a fresh outbound HTTPS fetch |
diff --git a/internal/console/sesevents.go b/internal/console/sesevents.go| index c0a4b83a0..d33f8c5ac 100644 |
| --- a/internal/console/sesevents.go |
| +++ b/internal/console/sesevents.go |
| @@ -561,6 +561,12 @@ func (s *Server) recordMailMessage(ctx context.Context, acct, app, key string, e |
| 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 |
diff --git a/internal/console/store.go b/internal/console/store.go| index 726435455..fd47e4e41 100644 |
| --- a/internal/console/store.go |
| +++ b/internal/console/store.go |
| @@ -45,6 +45,8 @@ type Store struct { |
| // this process, which is what makes the first event after a restart |
| // prune. |
| mailEventsPruned atomic.Int64 |
| + // mailPushPruned is the same stamp for the push queue (mailpush.go). |
| + mailPushPruned atomic.Int64 |
| } |
| // Open opens (creating if needed) the console DB at path. |
| @@ -241,6 +243,53 @@ func (s *Store) migrate() error { |
| `CREATE INDEX IF NOT EXISTS mail_events_by_time ON mail_events (account, app, at);`, |
| `CREATE INDEX IF NOT EXISTS mail_events_by_message ON mail_events (account, app, message_id);`, |
| `CREATE INDEX IF NOT EXISTS mail_events_by_age ON mail_events (received_at);`, |
| + // Where an app wants to be told about its email (mailpush.go): one |
| + // path per app, POSTed on the app's own platform host. Absent row = |
| + // the app is not told, which is every app until it asks. |
| + // |
| + // failing_since is when the app's current run of failed deliveries |
| + // began (0: none), and suspended_at when the push was suspended for |
| + // a day of them (0: not suspended) — see mailPushSuspendAfter. Both |
| + // clear when the push is turned on again. |
| + `CREATE TABLE IF NOT EXISTS mail_event_hooks ( |
| + account TEXT NOT NULL, |
| + app TEXT NOT NULL, |
| + path TEXT NOT NULL, |
| + set_by TEXT NOT NULL, |
| + set_at INTEGER NOT NULL, |
| + failing_since INTEGER NOT NULL DEFAULT 0, |
| + suspended_at INTEGER NOT NULL DEFAULT 0, |
| + suspended_why TEXT NOT NULL DEFAULT '', |
| + PRIMARY KEY (account, app) |
| +);`, |
| + // The queue of those notifications, one row per SES notification per |
| + // app, retried until delivered or given up on. The payload carries |
| + // recipients' addresses, so the row lives no longer than the log it |
| + // was cut from (mailEventsRetention), delivered or not. notice is |
| + // UNIQUE with the app so SNS redelivering a notification queues it |
| + // once. |
| + `CREATE TABLE IF NOT EXISTS mail_event_deliveries ( |
| + id INTEGER PRIMARY KEY, |
| + account TEXT NOT NULL, |
| + app TEXT NOT NULL, |
| + notice TEXT NOT NULL, |
| + payload TEXT NOT NULL, |
| + attempts INTEGER NOT NULL DEFAULT 0, |
| + next_at INTEGER NOT NULL, |
| + created_at INTEGER NOT NULL, |
| + delivered_at INTEGER NOT NULL DEFAULT 0, |
| + abandoned_at INTEGER NOT NULL DEFAULT 0, |
| + last_attempt_at INTEGER NOT NULL DEFAULT 0, |
| + last_status INTEGER NOT NULL DEFAULT 0, |
| + last_error TEXT NOT NULL DEFAULT '', |
| + abandoned_why TEXT NOT NULL DEFAULT '', |
| + UNIQUE (account, app, notice) |
| +);`, |
| + // A pass seeks each app's oldest due rows through this index, so its |
| + // cost follows the number of apps asking, not the size of anyone's |
| + // backlog. |
| + `CREATE INDEX IF NOT EXISTS mail_event_deliveries_app_due ON mail_event_deliveries (account, app, delivered_at, abandoned_at, next_at);`, |
| + `CREATE INDEX IF NOT EXISTS mail_event_deliveries_by_age ON mail_event_deliveries (created_at);`, |
| } { |
| if _, err := s.db.Exec(stmt); err != nil { |
| return err |
diff --git a/mail_cli.go b/mail_cli.go| index 1a96f201f..2f9110ff2 100644 |
| --- a/mail_cli.go |
| +++ b/mail_cli.go |
| @@ -84,6 +84,13 @@ const emailUsage = `usage: carlos email <enable|domains|status|test|credentials| |
| newest first, with what the receiving server |
| said. Kept for 30 days. --message takes the id SES |
| gave the message when it was sent. |
| + events push --app <a> [--path <p> | --off] [--json] |
| + tell the app itself: POST each delivery, bounce, |
| + complaint, rejection and delay SES attributes to |
| + it, signed, to <p> on its own platform host. The |
| + verifying key is delivered as CARLOS_MAIL_EVENTS_KEY. |
| + Bare, shows where events go and how delivery is |
| + going; --off stops it. |
| test --app <a> --to <addr> |
| send a REAL message and report the result — the |
| verify half of the no-infra rule: this is the |
| @@ -871,6 +878,10 @@ func sinceParam(raw string, now time.Time) (string, error) { |
| } |
| func runEmailEvents(args []string) { |
| + if len(args) > 0 && args[0] == "push" { |
| + runEmailEventsPush(args[1:]) |
| + return |
| + } |
| fs := flag.NewFlagSet("email events", flag.ExitOnError) |
| app := appFlag(fs, "app whose email events to show") |
| acctRaw, acctIDRaw, _ := accountFlags(fs) |
| @@ -929,6 +940,137 @@ func runEmailEvents(args []string) { |
| printEmailEvents(stdout, name, res) |
| } |
| +// apiEmailPushReply is what the three push doors answer. |
| +type apiEmailPushReply struct { |
| + On bool `json:"on"` |
| + Path string `json:"path,omitempty"` |
| + URL string `json:"url,omitempty"` |
| + KeyEnv string `json:"key_env"` |
| + PublicKey string `json:"public_key,omitempty"` |
| + SetBy string `json:"set_by,omitempty"` |
| + SetAt time.Time `json:"set_at,omitempty"` |
| + // SuspendedAt is set when the push was suspended after a day in which |
| + // the app accepted nothing; SuspendedWhy says what the last attempt got. |
| + SuspendedAt time.Time `json:"suspended_at,omitempty"` |
| + SuspendedWhy string `json:"suspended_why,omitempty"` |
| + Status apiEmailPushDelivery `json:"status"` |
| +} |
| + |
| +type apiEmailPushDelivery 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"` |
| +} |
| + |
| +func (s *apiSession) emailPush(ctx context.Context, a apiAcct, app, path string, off bool) (apiEmailPushReply, error) { |
| + var status int |
| + var body []byte |
| + var err error |
| + endpoint := apiPath(a, app, "/email/events/push") |
| + switch { |
| + case off: |
| + status, body, err = s.deleteJSON(ctx, endpoint, "") |
| + case path != "": |
| + status, body, err = s.putJSON(ctx, endpoint, map[string]string{"path": path}) |
| + default: |
| + status, body, err = s.getJSON(ctx, endpoint) |
| + } |
| + if err != nil { |
| + return apiEmailPushReply{}, err |
| + } |
| + if status != http.StatusOK { |
| + return apiEmailPushReply{}, mailAPIError(status, body) |
| + } |
| + var out apiEmailPushReply |
| + if err := json.Unmarshal(body, &out); err != nil { |
| + return apiEmailPushReply{}, fmt.Errorf("email events push: unreadable response: %w", err) |
| + } |
| + return out, nil |
| +} |
| + |
| +func runEmailEventsPush(args []string) { |
| + fs := flag.NewFlagSet("email events push", flag.ExitOnError) |
| + app := appFlag(fs, "app to tell about its email") |
| + acctRaw, acctIDRaw, _ := accountFlags(fs) |
| + path := fs.String("path", "", "the path on the app's own host to POST each event to, e.g. /mail/events") |
| + off := fs.Bool("off", false, "stop telling the app") |
| + asJSON := fs.Bool("json", false, "print the reply as JSON") |
| + consoleFlag := fs.String("console", "", "console origin (API mode only; default: the only logged-in one)") |
| + fs.Usage = flagUsage(fs, "") |
| + fs.Parse(args) |
| + name := strings.TrimSpace(*app) |
| + usage := "usage: carlos email events push --app <a> [--path <p> | --off] [--json] [--account <sqid-or-name>] [--console <origin>]" |
| + if *off && strings.TrimSpace(*path) != "" { |
| + log.Fatalf("email events push: --path and --off ask for opposite things — give one\n%s", usage) |
| + } |
| + if name != "" { |
| + if err := validAppName(name); err != nil { |
| + log.Fatalf("email events push: %v", err) |
| + } |
| + } |
| + |
| + sess := requireMailAPISession(*consoleFlag) |
| + if !*asJSON { |
| + printUsing(stdout, sess) |
| + } |
| + acct := resolvedAccount(sess, acctRaw, acctIDRaw) |
| + name = resolvedApp(sess, acct, name, usage) |
| + res, err := sess.emailPush(context.Background(), acct, name, strings.TrimSpace(*path), *off) |
| + if err != nil { |
| + log.Fatalf("email events push: %v", err) |
| + } |
| + if *asJSON { |
| + enc := json.NewEncoder(stdout) |
| + enc.SetIndent("", " ") |
| + enc.Encode(res) |
| + return |
| + } |
| + printEmailPush(stdout, name, res, strings.TrimSpace(*path) != "") |
| +} |
| + |
| +// printEmailPush says where events go, what the app needs to check them, |
| +// and how delivery has been going — the last failure in the far end's own |
| +// terms, because "answered 404" means the app has no handler at that path |
| +// yet and "answered 401" means a gate is in the way, and those are two |
| +// different things to go and fix. |
| +func printEmailPush(w io.Writer, app string, res apiEmailPushReply, justSet bool) { |
| + if !res.On { |
| + fmt.Fprintf(w, "%s is not told about its email\n", app) |
| + mutedf(w, "turn it on: carlos email events push --app %s --path /mail/events", app) |
| + return |
| + } |
| + if !res.SuspendedAt.IsZero() { |
| + fmt.Fprintf(w, "%s's email events push is SUSPENDED since %s\n", app, res.SuspendedAt.Local().Format("2006-01-02 15:04")) |
| + fmt.Fprintf(w, " %s\n", res.SuspendedWhy) |
| + fmt.Fprintf(w, " nothing is queued or sent while it is suspended; what was waiting was given up on\n") |
| + mutedf(w, " when %s answers 2xx again, turn it back on: carlos email events push --app %s --path %s", res.URL, app, res.Path) |
| + return |
| + } |
| + fmt.Fprintf(w, "%s is told about its email at %s\n", app, res.URL) |
| + fmt.Fprintf(w, " verify with %s: an Ed25519 public key, standard base64\n", res.KeyEnv) |
| + if justSet { |
| + mutedf(w, " the key has been added to the app's config — restart the app to pick it up: carlos restart --app %s", app) |
| + } |
| + st := res.Status |
| + fmt.Fprintf(w, " %d delivered, %d waiting, %d given up on (last 30 days)\n", st.Delivered, st.Pending, st.Abandoned) |
| + if !st.LastDelivered.IsZero() { |
| + fmt.Fprintf(w, " last delivered %s\n", st.LastDelivered.Local().Format("2006-01-02 15:04:05")) |
| + } |
| + if st.LastError != "" && st.LastFailedAt.After(st.LastDelivered) { |
| + fmt.Fprintf(w, " last attempt failed: %s\n", st.LastError) |
| + switch { |
| + case st.LastStatus == http.StatusNotFound: |
| + mutedf(w, " the app has nothing at %s yet — deploy the handler; waiting deliveries are retried, and a day with none accepted suspends the push", res.Path) |
| + case st.LastStatus == http.StatusUnauthorized: |
| + mutedf(w, " a 401 is usually carlos gate — a gated app cannot receive email events") |
| + } |
| + } |
| +} |
| + |
| // printEmailEvents is one line per recipient per event, newest first: when, |
| // what happened, to whom, and the message it happened to — then, indented |
| // beneath, what the receiving server actually said, which is the part that |
diff --git a/mail_cli_test.go b/mail_cli_test.go| index 8bb049461..7c0dcafa5 100644 |
| --- a/mail_cli_test.go |
| +++ b/mail_cli_test.go |
| @@ -1162,3 +1162,66 @@ func TestSinceParamTakesADurationOrATime(t *testing.T) { |
| t.Error("a word that is neither should be refused, not sent to the console") |
| } |
| } |
| + |
| +// `carlos email events push` says where events go and what the app checks |
| +// them with, and when delivery is failing says so in terms that point at |
| +// the fix: nothing at the path yet, or a gate in the way. |
| +func TestPrintEmailPushPointsAtTheFix(t *testing.T) { |
| + var buf bytes.Buffer |
| + printEmailPush(&buf, "jb-book", apiEmailPushReply{}, false) |
| + if !strings.Contains(buf.String(), "not told") || !strings.Contains(buf.String(), "--path /mail/events") { |
| + t.Errorf("off should say how to turn it on:\n%s", buf.String()) |
| + } |
| + |
| + on := apiEmailPushReply{On: true, Path: "/mail/events", URL: "https://jb-book.beb.oncarlos.com/mail/events", |
| + KeyEnv: "CARLOS_MAIL_EVENTS_KEY", Status: apiEmailPushDelivery{Delivered: 3, Pending: 1}} |
| + buf.Reset() |
| + printEmailPush(&buf, "jb-book", on, true) |
| + out := buf.String() |
| + for _, want := range []string{"https://jb-book.beb.oncarlos.com/mail/events", "CARLOS_MAIL_EVENTS_KEY", "carlos restart --app jb-book", "3 delivered, 1 waiting"} { |
| + if !strings.Contains(out, want) { |
| + t.Errorf("missing %q:\n%s", want, out) |
| + } |
| + } |
| + |
| + delivered := time.Date(2026, 9, 15, 8, 0, 0, 0, time.UTC) |
| + for status, hint := range map[int]string{404: "nothing at /mail/events", 401: "carlos gate"} { |
| + failing := on |
| + failing.Status = apiEmailPushDelivery{Pending: 1, LastDelivered: delivered, LastFailedAt: delivered.Add(time.Hour), |
| + LastStatus: status, LastError: fmt.Sprintf("answered %d", status)} |
| + buf.Reset() |
| + printEmailPush(&buf, "jb-book", failing, false) |
| + if !strings.Contains(buf.String(), fmt.Sprintf("answered %d", status)) || !strings.Contains(buf.String(), hint) { |
| + t.Errorf("a %d should point at %q:\n%s", status, hint, buf.String()) |
| + } |
| + if strings.Contains(buf.String(), "restart") { |
| + t.Errorf("the restart hint belongs to turning it on, not to every look:\n%s", buf.String()) |
| + } |
| + } |
| + |
| + // A failure since recovered from is not news. |
| + recovered := on |
| + recovered.Status = apiEmailPushDelivery{Delivered: 4, LastDelivered: delivered.Add(2 * time.Hour), |
| + LastFailedAt: delivered, LastStatus: 404, LastError: "answered 404"} |
| + buf.Reset() |
| + printEmailPush(&buf, "jb-book", recovered, false) |
| + if strings.Contains(buf.String(), "failed") { |
| + t.Errorf("an old failure was reported after a later delivery landed:\n%s", buf.String()) |
| + } |
| +} |
| + |
| +// A suspended push says so first, says why, and says how to turn it back on. |
| +func TestPrintEmailPushSaysItIsSuspended(t *testing.T) { |
| + var buf bytes.Buffer |
| + printEmailPush(&buf, "jb-book", apiEmailPushReply{ |
| + On: true, Path: "/mail/events", URL: "https://jb-book.beb.oncarlos.com/mail/events", KeyEnv: "CARLOS_MAIL_EVENTS_KEY", |
| + SuspendedAt: time.Date(2026, 9, 15, 9, 0, 0, 0, time.UTC), |
| + SuspendedWhy: "nothing was accepted for a day — the last attempt answered 404", |
| + }, false) |
| + out := buf.String() |
| + for _, want := range []string{"SUSPENDED", "answered 404", "given up on", "carlos email events push --app jb-book --path /mail/events"} { |
| + if !strings.Contains(out, want) { |
| + t.Errorf("missing %q:\n%s", want, out) |
| + } |
| + } |
| +} |