amadan
email events: Fable review fixes — a move keeps its neighbours' recipients behind
From a Fable review of this branch. A move to another account carried rows that were never the app's. When SES names no CARLOS sender, a notification is logged under every app on the matching domain with by_sender = 0 — it may name a sibling's recipient — and carryMailApp moved those along with the app's own. Across accounts they are now dropped first; a rename within one account still carries both, since the same members could read both before and after. A field only the log reads could cost the count. ParseEvent typed tags, destination and each type's block in the same unmarshal that the counts depend on, so any of them in an unexpected JSON shape refused the whole notification: logged, answered 200, never counted, never enforced — the class this file's header warns about. The counts' fields are now read on their own, as before this branch, and only that read can refuse. The envelope and the event's own block are read separately after it; a part that does not read is left empty, the reason goes in Event.Incomplete, and the ingest logs it and counts the event. A per-recipient event naming nobody was widened to every addressee, so a bounce with an empty bouncedRecipients reported every address on the message as bounced. Only a Send or a Reject — about the whole message — falls back to the destination now. A failed prune kept its stamp and was not retried for an hour; the stamp is put back. ReapAccount now forgets the account's rows in its transaction. And the log's domain fallback requires the matched domain, or MAIL FROM host, to be verified in that region: the domain claim refuses one account another's domain, but not another's MAIL FROM host claimed and never verified. The counts keep matching as they did. Tests for each, each checked to fail with its fix undone, plus the gaps the review named: a notification driven through the SNS door into the log under SNS's MessageId, and the API's since and limit narrowing a real answer.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
7 files changed,
+409
−56
internal/console/apimail_test.go+18 −0internal/console/mailevents.go+32 −2internal/console/mailevents_test.go+168 −0internal/console/sesevents.go+26 −1internal/console/store.go+10 −0internal/mail/events.go+106 −53internal/mail/events_test.go+49 −0
CI passed — run details
diff --git a/internal/console/apimail_test.go b/internal/console/apimail_test.go| index dabb8b2c3..e90bc365f 100644 |
| --- a/internal/console/apimail_test.go |
| +++ b/internal/console/apimail_test.go |
| @@ -14,6 +14,7 @@ import ( |
| "context" |
| "encoding/json" |
| "net/http" |
| + "net/url" |
| "strings" |
| "testing" |
| "time" |
| @@ -1130,6 +1131,23 @@ func TestAPIEmailEventsReadsTheLog(t *testing.T) { |
| t.Errorf("an empty answer = %d %s", status, body) |
| } |
| + // And the narrowing that is not about who: a window and a count. |
| + old := mail.Event{Type: mail.EventDelivery, MessageID: "m-old", At: now.Add(-48 * time.Hour), Outcomes: []mail.Outcome{{Recipient: "cat@example.com"}}} |
| + if err := fx.st.RecordMailMessageEvent(ctx, fx.sqid, app, "sns-m-old", old, true, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + since := url.QueryEscape(now.Add(-24 * time.Hour).UTC().Format(time.RFC3339)) |
| + if status, out := getJSON(t, emailURL(fx, app)+"/events?since="+since, token); status != http.StatusOK { |
| + t.Fatalf("since: %d %v", status, out) |
| + } else if events, _ := out["events"].([]any); len(events) != 2 { |
| + t.Errorf("since the last day = %d events, want the two recent ones: %v", len(events), out["events"]) |
| + } |
| + if status, out := getJSON(t, emailURL(fx, app)+"/events?limit=1", token); status != http.StatusOK { |
| + t.Fatalf("limit: %d %v", status, out) |
| + } else if events, _ := out["events"].([]any); len(events) != 1 { |
| + t.Errorf("limit=1 gave %d events", len(events)) |
| + } |
| + |
| for _, bad := range []string{"since=yesterday", "limit=0", "limit=100000", "limit=many"} { |
| if status, body := getRaw(t, emailURL(fx, app)+"/events?"+bad, token); status != http.StatusBadRequest { |
| t.Errorf("%s = %d %s, want 400", bad, status, body) |
diff --git a/internal/console/mailevents.go b/internal/console/mailevents.go| index d27861e4c..9a5777ebe 100644 |
| --- a/internal/console/mailevents.go |
| +++ b/internal/console/mailevents.go |
| @@ -102,11 +102,23 @@ type execer interface { |
| // and are dropped first, and only rows received since this app's own row was |
| // created travel — the same rule RenameAppIn applies to audit rows, for the |
| // same reason. Whatever is left under the old key is dropped. |
| +// |
| +// Across accounts, a row attributed only by its sending domain (by_sender = |
| +// 0) does not travel. Such a row was given to every app on the domain because |
| +// SES did not say which of them sent the message, so it may name another |
| +// app's recipient — and that app stays behind in the source account. Within |
| +// one account both kinds carry, since the same members could read both |
| +// before and after. |
| func carryMailApp(ctx context.Context, tx execer, fromSqid, fromApp, toSqid, toApp string, createdAt int64) error { |
| for _, table := range mailAppTables { |
| if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ? AND app = ?`, toSqid, toApp); err != nil { |
| return err |
| } |
| + if table == "mail_events" && fromSqid != toSqid { |
| + if _, err := tx.ExecContext(ctx, `DELETE FROM mail_events WHERE account = ? AND app = ? AND by_sender = 0`, fromSqid, fromApp); err != nil { |
| + return err |
| + } |
| + } |
| 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 |
| @@ -118,6 +130,19 @@ func carryMailApp(ctx context.Context, tx execer, fromSqid, fromApp, toSqid, toA |
| return nil |
| } |
| +// forgetMailAccount drops everything an account's apps left in these tables, |
| +// when the account itself is reaped. Nothing can reach the rows afterwards — |
| +// the sqid is tombstoned and never reused — but they are recipients' |
| +// addresses, and a window is not an archive. |
| +func forgetMailAccount(ctx context.Context, tx execer, sqid string) error { |
| + for _, table := range mailAppTables { |
| + if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ?`, sqid); err != nil { |
| + return err |
| + } |
| + } |
| + return nil |
| +} |
| + |
| // forgetMailApp drops everything a purged app left in these tables, so a |
| // later app of the same name starts with none of it. |
| func forgetMailApp(ctx context.Context, tx execer, sqid, app string) error { |
| @@ -148,8 +173,13 @@ func (s *Store) pruneMailEvents(ctx context.Context, now time.Time) error { |
| if !s.mailEventsPruned.CompareAndSwap(last, now.Unix()) { |
| return nil |
| } |
| - _, err := s.db.ExecContext(ctx, `DELETE FROM mail_events WHERE received_at < ?`, now.Add(-mailEventsRetention).Unix()) |
| - return err |
| + if _, err := s.db.ExecContext(ctx, `DELETE FROM mail_events WHERE received_at < ?`, now.Add(-mailEventsRetention).Unix()); err != nil { |
| + // Not pruned, so not stamped: the next notification tries again |
| + // rather than waiting out the hour. |
| + s.mailEventsPruned.CompareAndSwap(now.Unix(), last) |
| + return err |
| + } |
| + return nil |
| } |
| // MailEventQuery narrows a read of the log. The zero value is the newest |
diff --git a/internal/console/mailevents_test.go b/internal/console/mailevents_test.go| index 29a6fa90a..7b807f3a8 100644 |
| --- a/internal/console/mailevents_test.go |
| +++ b/internal/console/mailevents_test.go |
| @@ -2,6 +2,9 @@ package console |
| import ( |
| "context" |
| + "encoding/json" |
| + "net/http" |
| + "strconv" |
| "strings" |
| "testing" |
| "time" |
| @@ -309,3 +312,168 @@ func TestALateBounceCountsAgainstTheDayItWasSent(t *testing.T) { |
| } |
| } |
| } |
| + |
| +// A row attributed only by a shared sending domain may name another app's |
| +// recipient. It stays in the account where that other app lives: a move to |
| +// another account carries only the rows SES said were the app's own. A |
| +// rename within the account carries both, since the same members could read |
| +// both before and after. |
| +func TestAMoveLeavesDomainAttributedRowsBehind(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + now := time.Now() |
| + for _, app := range []string{"moving", "renaming"} { |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, app, "bill@tito.io", now.Add(-time.Hour)); err != nil { |
| + t.Fatal(err) |
| + } |
| + for i, bySender := range []bool{true, false} { |
| + ev := bounceTo("ann@example.com", app+"-"+strconv.FormatBool(bySender), now) |
| + if err := fx.st.RecordMailMessageEvent(ctx, release.OpsAccount, app, "sns-"+app+strconv.Itoa(i), ev, bySender, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + } |
| + if err := fx.st.CreateAccount(99, "dest", "Destination", "bill@tito.io", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.st.MoveApp("moving", OpsAccountID, 99); err != nil { |
| + t.Fatal(err) |
| + } |
| + moved, _ := fx.st.MailEvents(ctx, "dest", "moving", MailEventQuery{}, time.Now()) |
| + if len(moved) != 1 || !moved[0].BySender { |
| + t.Errorf("a move to another account carried %+v, want only the app's own row", moved) |
| + } |
| + if left, _ := fx.st.MailEvents(ctx, release.OpsAccount, "moving", MailEventQuery{}, time.Now()); len(left) != 0 { |
| + t.Errorf("rows were left under the old account's key: %+v", left) |
| + } |
| + |
| + if err := fx.st.RenameAppIn(OpsAccountID, "renaming", "renamed"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if renamed, _ := fx.st.MailEvents(ctx, release.OpsAccount, "renamed", MailEventQuery{}, time.Now()); len(renamed) != 2 { |
| + t.Errorf("a rename within the account carried %d rows, want both", len(renamed)) |
| + } |
| +} |
| + |
| +// A reaped account's email history goes with it. |
| +func TestReapingAnAccountForgetsItsMailLog(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + now := time.Now() |
| + if err := fx.st.CreateAccount(99, "reaped", "Going", "bill@tito.io", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.st.RecordMailMessageEvent(ctx, "reaped", "gone", "sns-1", bounceTo("ann@example.com", "m", now), true, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.st.ReapAccount(99, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + var n int |
| + fx.st.db.QueryRow(`SELECT count(*) FROM mail_events WHERE account = 'reaped'`).Scan(&n) |
| + if n != 0 { |
| + t.Errorf("%d rows outlived the account", n) |
| + } |
| +} |
| + |
| +// A prune that fails is not stamped, so the next notification tries again |
| +// rather than waiting out the hour. |
| +func TestAFailedPruneIsTriedAgain(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + cancel() |
| + if err := fx.st.pruneMailEvents(ctx, time.Now()); err == nil { |
| + t.Fatal("a prune with a cancelled context should fail") |
| + } |
| + if got := fx.st.mailEventsPruned.Load(); got != 0 { |
| + t.Errorf("the stamp was left at %d after a failed prune", got) |
| + } |
| +} |
| + |
| +// The log's domain fallback gives recipients' addresses only to a domain its |
| +// account has verified. The counts keep matching as they always did. |
| +func TestTheDomainFallbackNeedsAVerifiedDomain(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + now := time.Now() |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "squatter", "bill@tito.io", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + rec := &mail.DomainRecord{ |
| + Domain: "mail-eu-west-1.lgc.li", Kind: mail.KindCustom, Apps: []string{"squatter"}, |
| + Regions: map[string]*mail.RegionState{"eu-west-1": { |
| + IdentityARN: "arn:aws:ses:eu-west-1:1:identity/mail-eu-west-1.lgc.li", Status: mail.StatusPending, |
| + }}, |
| + CreatedAt: now, |
| + } |
| + if _, err := mail.PutDomain(ctx, fx.rel, release.OpsAccount, rec, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + ev := bounceTo("ann@example.com", "m-unverified", now) |
| + ev.Domain, ev.SentAt = "mail-eu-west-1.lgc.li", now |
| + fx.srv.recordAndEnforceMail(ctx, ev, "sns-unverified") |
| + if got, _ := fx.st.MailEvents(ctx, release.OpsAccount, "squatter", MailEventQuery{}, time.Now()); len(got) != 0 { |
| + t.Errorf("an unverified domain was given recipients' addresses: %+v", got) |
| + } |
| + if c, _ := fx.st.MailCounts(ctx, release.OpsAccount, "squatter", now.UTC().Format("2006-01-02")); c.Bounces != 1 { |
| + t.Errorf("bounces = %d — the counts' matching is not this change's to narrow", c.Bounces) |
| + } |
| +} |
| + |
| +// End to end through the SNS door: a signed notification lands in the log |
| +// under SNS's own MessageId, with what the receiving server said, and a |
| +// redelivery of the same notification is still one row. |
| +func TestANotificationThroughTheDoorLandsInTheLog(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + fx.srv.cfg.MailDeployment.SNSTopicARN = "arn:aws:sns:eu-west-1:1:carlos-mail" |
| + fx.srv.snsVerify = func(_ snsEnvelope) error { return nil } |
| + ctx := context.Background() |
| + now := time.Now() |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "jb-book", "bill@tito.io", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + rec := &mail.DomainRecord{ |
| + Domain: "lgc.li", Kind: mail.KindCustom, Apps: []string{"jb-book"}, |
| + Regions: map[string]*mail.RegionState{"eu-west-1": { |
| + IdentityARN: "arn:aws:ses:eu-west-1:1:identity/lgc.li", Status: mail.StatusVerified, |
| + }}, |
| + CreatedAt: now, |
| + } |
| + if _, err := mail.PutDomain(ctx, fx.rel, release.OpsAccount, rec, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + message, _ := json.Marshal(map[string]any{ |
| + "eventType": "Bounce", |
| + "mail": map[string]any{"source": "bookings@lgc.li", "messageId": "0102-door", "timestamp": now.UTC().Format(time.RFC3339), |
| + "tags": map[string][]string{"ses:caller-identity": {mail.UserName(release.OpsAccount, "jb-book")}}}, |
| + "bounce": map[string]any{"bounceType": "Permanent", "bounceSubType": "NoEmail", "timestamp": now.UTC().Format("2006-01-02T15:04:05.000Z"), |
| + "bouncedRecipients": []map[string]string{{"emailAddress": "ann@example.com", "status": "5.1.1", "diagnosticCode": "550 no such user"}}}, |
| + }) |
| + envelope, _ := json.Marshal(map[string]string{ |
| + "Type": "Notification", "MessageId": "sns-door-1", "TopicArn": "arn:aws:sns:eu-west-1:1:carlos-mail", |
| + "Message": string(message), "SigningCertURL": "https://sns.eu-west-1.amazonaws.com/x.pem", "Signature": "x", |
| + }) |
| + for i := 0; i < 2; i++ { |
| + resp := sesPost(t, fx, string(envelope)) |
| + resp.Body.Close() |
| + if resp.StatusCode != http.StatusOK { |
| + t.Fatalf("status = %d", resp.StatusCode) |
| + } |
| + } |
| + got, err := fx.st.MailEvents(ctx, release.OpsAccount, "jb-book", MailEventQuery{}, time.Now()) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(got) != 1 { |
| + t.Fatalf("rows = %d, want 1: %+v", len(got), got) |
| + } |
| + e := got[0] |
| + if e.MessageID != "0102-door" || e.Recipient != "ann@example.com" || e.Status != "5.1.1" || e.Detail != "Permanent/NoEmail" || !e.BySender { |
| + t.Errorf("row = %+v", e) |
| + } |
| + var notice string |
| + fx.st.db.QueryRow(`SELECT notice FROM mail_events WHERE message_id = '0102-door'`).Scan(¬ice) |
| + if notice != "sns-door-1" { |
| + t.Errorf("notice = %q, want SNS's own MessageId", notice) |
| + } |
| +} |
diff --git a/internal/console/sesevents.go b/internal/console/sesevents.go| index c4a90683c..c0a4b83a0 100644 |
| --- a/internal/console/sesevents.go |
| +++ b/internal/console/sesevents.go |
| @@ -404,6 +404,10 @@ func (s *Server) handleSESEvents(w http.ResponseWriter, r *http.Request) { |
| w.WriteHeader(http.StatusOK) |
| return |
| } |
| + if ev.Incomplete != "" { |
| + // Counted all the same; only the log is missing something. |
| + log.Printf("console: ses events: %s notification %s read in part: %s", ev.Type, env.MessageID, ev.Incomplete) |
| + } |
| s.recordAndEnforceMail(r.Context(), ev, env.MessageID) |
| w.WriteHeader(http.StatusOK) |
| default: |
| @@ -506,8 +510,14 @@ func (s *Server) recordAndEnforceMail(ctx context.Context, ev mail.Event, notice |
| if !domainMatchesEvent(d, ev.Domain) { |
| continue |
| } |
| + // The log's fallback asks more of the match than the counts do: |
| + // only a domain this account actually verified gets recipients' |
| + // addresses. The domain claim refuses a second account the same |
| + // name, but not another account's MAIL FROM host claimed as a |
| + // domain of its own and never verified. |
| + logHere := !named && domainVerifiedForEvent(d, ev.Domain) |
| for _, app := range d.Apps { |
| - if !named { |
| + if logHere { |
| s.recordMailMessage(ctx, a.Sqid, app, key, ev, false, now) |
| } |
| if err := s.store.RecordMailEvent(ctx, a.Sqid, app, day, ev.Type); err != nil { |
| @@ -524,6 +534,21 @@ func (s *Server) recordAndEnforceMail(ctx context.Context, ev mail.Event, notice |
| } |
| } |
| +// domainVerifiedForEvent is domainMatchesEvent restricted to what this |
| +// record has proved: the identity domain in a region SES verified it in, or |
| +// the MAIL FROM host of a verified region. |
| +func domainVerifiedForEvent(d *mail.DomainRecord, domain string) bool { |
| + for _, st := range d.Regions { |
| + if st == nil || st.Status != mail.StatusVerified { |
| + continue |
| + } |
| + if d.Domain == domain || (st.MailFrom != "" && st.MailFrom == domain) { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| // recordMailMessage keeps one notification in one app's log, if that app is |
| // live. A trashed app keeps nothing new: its name is not a member-facing app |
| // any more, and what it is sent is not somebody else's to read after a |
diff --git a/internal/console/store.go b/internal/console/store.go| index 5ffbc7e9e..726435455 100644 |
| --- a/internal/console/store.go |
| +++ b/internal/console/store.go |
| @@ -1190,6 +1190,16 @@ func (s *Store) ReapAccount(accountID int64, now time.Time) error { |
| return err |
| } |
| } |
| + // Email history and settings are keyed by sqid (mailevents.go). |
| + var sqid string |
| + switch err := tx.QueryRow(`SELECT sqid FROM accounts WHERE id = ?`, accountID).Scan(&sqid); { |
| + case err == nil: |
| + if err := forgetMailAccount(context.Background(), tx, sqid); err != nil { |
| + return err |
| + } |
| + case err != sql.ErrNoRows: |
| + return err |
| + } |
| if _, err := tx.Exec(`UPDATE users SET last_account = 0 WHERE last_account = ?`, accountID); err != nil { |
| return err |
| } |
diff --git a/internal/mail/events.go b/internal/mail/events.go| index e830e02eb..4fb9b62a7 100644 |
| --- a/internal/mail/events.go |
| +++ b/internal/mail/events.go |
| @@ -48,6 +48,10 @@ type Event struct { |
| Outcomes []Outcome |
| // Detail is the event-wide explanation, empty where the type has none. |
| Detail EventDetail |
| + // Incomplete says why part of what only the log uses could not be read — |
| + // a field in a shape this code did not expect. The event is still good |
| + // for the counts; the caller logs this. |
| + Incomplete string |
| } |
| // Outcome is one recipient's part in an event. |
| @@ -143,44 +147,17 @@ func DomainOf(address string) string { |
| // is worse than no event, because it would be silently dropped from the |
| // counts that enforce a cap. |
| func ParseEvent(body []byte) (Event, error) { |
| - type recipient struct { |
| - EmailAddress string `json:"emailAddress"` |
| - Status string `json:"status"` |
| - DiagnosticCode string `json:"diagnosticCode"` |
| - } |
| + // Two reads of the same body, deliberately. This one is everything the |
| + // counts and the cap depend on, typed exactly as it was before the log |
| + // existed, and it is the only one allowed to refuse the event: a refused |
| + // event is a count silently dropped. |
| var raw struct { |
| EventType string `json:"eventType"` |
| Mail struct { |
| - Timestamp string `json:"timestamp"` |
| - MessageID string `json:"messageId"` |
| - Source string `json:"source"` |
| - Destination []string `json:"destination"` |
| - Tags map[string][]string `json:"tags"` |
| + Timestamp string `json:"timestamp"` |
| + MessageID string `json:"messageId"` |
| + Source string `json:"source"` |
| } `json:"mail"` |
| - Bounce struct { |
| - Timestamp string `json:"timestamp"` |
| - BounceType string `json:"bounceType"` |
| - BounceSubType string `json:"bounceSubType"` |
| - BouncedRecipients []recipient `json:"bouncedRecipients"` |
| - } `json:"bounce"` |
| - Complaint struct { |
| - Timestamp string `json:"timestamp"` |
| - ComplaintFeedbackType string `json:"complaintFeedbackType"` |
| - ComplainedRecipients []recipient `json:"complainedRecipients"` |
| - } `json:"complaint"` |
| - Delivery struct { |
| - Timestamp string `json:"timestamp"` |
| - Recipients []string `json:"recipients"` |
| - SMTPResponse string `json:"smtpResponse"` |
| - } `json:"delivery"` |
| - Reject struct { |
| - Reason string `json:"reason"` |
| - } `json:"reject"` |
| - DeliveryDelay struct { |
| - Timestamp string `json:"timestamp"` |
| - DelayType string `json:"delayType"` |
| - DelayedRecipients []recipient `json:"delayedRecipients"` |
| - } `json:"deliveryDelay"` |
| } |
| if err := json.Unmarshal(body, &raw); err != nil { |
| return Event{}, fmt.Errorf("mail: parse ses event: %w", err) |
| @@ -232,27 +209,104 @@ func ParseEvent(body []byte) (Event, error) { |
| ev.At = at |
| } |
| } |
| + ev.MessageID = cleanField(ev.MessageID, maxDiagnosticLen) |
| + readLogFields(body, raw.EventType, &ev) |
| + return ev, nil |
| +} |
| + |
| +// readLogFields is the second read: what only the per-message log uses — |
| +// who the message went to, who sent it, and what the receiving server said. |
| +// A field here in a shape this code does not expect is not a reason to lose |
| +// the event, so nothing here refuses it: the part that does not read is left |
| +// empty, the reason is kept in ev.Incomplete for the caller to log, and the |
| +// count goes ahead. The envelope and the event's own block are read |
| +// separately, so a malformed bounce block does not also cost the sender. |
| +func readLogFields(body []byte, eventType string, ev *Event) { |
| + type recipient struct { |
| + EmailAddress string `json:"emailAddress"` |
| + Status string `json:"status"` |
| + DiagnosticCode string `json:"diagnosticCode"` |
| + } |
| + var problems []string |
| + var envelope struct { |
| + Mail struct { |
| + Destination []string `json:"destination"` |
| + Tags map[string][]string `json:"tags"` |
| + } `json:"mail"` |
| + } |
| + if err := json.Unmarshal(body, &envelope); err != nil { |
| + problems = append(problems, "mail: "+err.Error()) |
| + } |
| + if ids := envelope.Mail.Tags["ses:caller-identity"]; len(ids) > 0 { |
| + ev.Caller = cleanField(ids[0], maxDiagnosticLen) |
| + } |
| + for _, to := range envelope.Mail.Destination { |
| + if to = cleanField(to, maxAddressLen); to != "" { |
| + ev.Destination = append(ev.Destination, to) |
| + } |
| + } |
| + |
| + // Each type carries its own block; the others are simply empty. Read |
| + // only the one this event is, so a malformed message that carries two |
| + // cannot have one type's recipients reported under another's name. |
| + var raw struct { |
| + Bounce struct { |
| + Timestamp string `json:"timestamp"` |
| + BounceType string `json:"bounceType"` |
| + BounceSubType string `json:"bounceSubType"` |
| + BouncedRecipients []recipient `json:"bouncedRecipients"` |
| + } `json:"bounce"` |
| + Complaint struct { |
| + Timestamp string `json:"timestamp"` |
| + ComplaintFeedbackType string `json:"complaintFeedbackType"` |
| + ComplainedRecipients []recipient `json:"complainedRecipients"` |
| + } `json:"complaint"` |
| + Delivery struct { |
| + Timestamp string `json:"timestamp"` |
| + Recipients []string `json:"recipients"` |
| + SMTPResponse string `json:"smtpResponse"` |
| + } `json:"delivery"` |
| + Reject struct { |
| + Reason string `json:"reason"` |
| + } `json:"reject"` |
| + DeliveryDelay struct { |
| + Timestamp string `json:"timestamp"` |
| + DelayType string `json:"delayType"` |
| + DelayedRecipients []recipient `json:"delayedRecipients"` |
| + } `json:"deliveryDelay"` |
| + } |
| + block := map[string]string{ |
| + EventBounce: "bounce", EventComplaint: "complaint", EventDelivery: "delivery", |
| + EventReject: "reject", EventDeliveryDelay: "deliveryDelay", |
| + }[eventType] |
| + if block != "" { |
| + var blocks map[string]json.RawMessage |
| + if err := json.Unmarshal(body, &blocks); err == nil { |
| + // Wrap the one block back under its own key and read only that. |
| + if one, ok := blocks[block]; ok { |
| + if err := json.Unmarshal([]byte(`{"`+block+`":`+string(one)+`}`), &raw); err != nil { |
| + problems = append(problems, block+": "+err.Error()) |
| + } |
| + } |
| + } |
| + } |
| + if len(problems) > 0 { |
| + ev.Incomplete = cleanField(strings.Join(problems, "; "), maxDiagnosticLen) |
| + } |
| + |
| // When the thing this event reports happened, where SES says: a bounce |
| // or a delay can arrive a day after the send, and an event dated by the |
| // send is both wrong on a screen and invisible to "since yesterday". |
| eventTime := map[string]string{ |
| EventBounce: raw.Bounce.Timestamp, EventComplaint: raw.Complaint.Timestamp, |
| EventDelivery: raw.Delivery.Timestamp, EventDeliveryDelay: raw.DeliveryDelay.Timestamp, |
| - }[raw.EventType] |
| + }[eventType] |
| if eventTime != "" { |
| if at, err := time.Parse(time.RFC3339, eventTime); err == nil { |
| ev.At = at |
| } |
| } |
| - ev.MessageID = cleanField(ev.MessageID, maxDiagnosticLen) |
| - if ids := raw.Mail.Tags["ses:caller-identity"]; len(ids) > 0 { |
| - ev.Caller = cleanField(ids[0], maxDiagnosticLen) |
| - } |
| - for _, to := range raw.Mail.Destination { |
| - if to = cleanField(to, maxAddressLen); to != "" { |
| - ev.Destination = append(ev.Destination, to) |
| - } |
| - } |
| + |
| outcomes := func(rs []recipient) { |
| for _, r := range rs { |
| to := cleanField(r.EmailAddress, maxAddressLen) |
| @@ -263,10 +317,7 @@ func ParseEvent(body []byte) (Event, error) { |
| Status: cleanField(r.Status, maxDetailLen), Diagnostic: cleanField(r.DiagnosticCode, maxDiagnosticLen)}) |
| } |
| } |
| - // Each type carries its own block; the others are simply empty. Read |
| - // only the one this event is, so a malformed message that carries two |
| - // cannot have one type's recipients reported under another's name. |
| - switch raw.EventType { |
| + switch eventType { |
| case EventBounce: |
| ev.Detail.BounceType = cleanField(raw.Bounce.BounceType, maxDetailLen) |
| ev.Detail.BounceSubType = cleanField(raw.Bounce.BounceSubType, maxDetailLen) |
| @@ -287,19 +338,21 @@ func ParseEvent(body []byte) (Event, error) { |
| ev.Detail.DelayType = cleanField(raw.DeliveryDelay.DelayType, maxDetailLen) |
| outcomes(raw.DeliveryDelay.DelayedRecipients) |
| } |
| - return ev, nil |
| } |
| // Recipients is who this event is about, one entry per address: the |
| // recipients it reports an outcome for, or — for an event about the whole |
| -// message, a Send or a Reject — everyone it was addressed to. An event that |
| -// names nobody at all yields a single empty entry, so it is still recorded |
| -// once rather than not at all. |
| +// message, a Send or a Reject — everyone it was addressed to. A per-recipient |
| +// event (a bounce, a delivery, a complaint, a delay) that names nobody is NOT |
| +// widened to every addressee: that would report a bounce for addresses that |
| +// did not bounce. It, like any event that names nobody at all, yields a |
| +// single empty entry, so it is still recorded once rather than not at all. |
| func (ev Event) Recipients() []Outcome { |
| if len(ev.Outcomes) > 0 { |
| return ev.Outcomes |
| } |
| - if len(ev.Destination) == 0 { |
| + wholeMessage := ev.Type == EventSend || ev.Type == EventReject |
| + if len(ev.Destination) == 0 || !wholeMessage { |
| return []Outcome{{}} |
| } |
| out := make([]Outcome, 0, len(ev.Destination)) |
diff --git a/internal/mail/events_test.go b/internal/mail/events_test.go| index 33b58e96e..34143db3e 100644 |
| --- a/internal/mail/events_test.go |
| +++ b/internal/mail/events_test.go |
| @@ -362,3 +362,52 @@ func TestAnEventIsDatedByWhatItReports(t *testing.T) { |
| t.Errorf("a Send is dated by the send: %v / %v", send.At, send.SentAt) |
| } |
| } |
| + |
| +// What only the log reads must never cost the count. A notification whose |
| +// tags, destination or type block arrive in a shape this code does not |
| +// expect still parses for everything the counts and the cap depend on, and |
| +// says what it could not read. |
| +func TestAMisshapedLogFieldDoesNotCostTheEvent(t *testing.T) { |
| + for name, body := range map[string]string{ |
| + "tags as a string": `{"eventType":"Send","mail":{"source":"a@lgc.li","timestamp":"2026-09-15T08:00:00.000Z","tags":{"ses:caller-identity":"x"}}}`, |
| + "destination as a string": `{"eventType":"Send","mail":{"source":"a@lgc.li","timestamp":"2026-09-15T08:00:00.000Z","destination":"ann@example.com"}}`, |
| + "recipients as objects": `{"eventType":"Delivery","mail":{"source":"a@lgc.li","timestamp":"2026-09-15T08:00:00.000Z"},"delivery":{"recipients":[{"a":1}]}}`, |
| + "reason as an object": `{"eventType":"Reject","mail":{"source":"a@lgc.li","timestamp":"2026-09-15T08:00:00.000Z"},"reject":{"reason":{"x":1}}}`, |
| + "bounce type as a number": `{"eventType":"Bounce","mail":{"source":"a@lgc.li","timestamp":"2026-09-15T08:00:00.000Z","tags":{"ses:caller-identity":["carlos-mail-beb-jb-book"]}},"bounce":{"bounceType":7}}`, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + ev, err := ParseEvent([]byte(body)) |
| + if err != nil { |
| + t.Fatalf("refused — a count would be dropped for a field only the log reads: %v", err) |
| + } |
| + if ev.Type == "" || ev.Domain != "lgc.li" || ev.SentAt.IsZero() { |
| + t.Errorf("the fields the counts need did not survive: %+v", ev) |
| + } |
| + if ev.Incomplete == "" { |
| + t.Error("what could not be read should be said, for the log") |
| + } |
| + }) |
| + } |
| + // A malformed type block does not also cost the envelope beside it. |
| + ev, _ := ParseEvent([]byte(`{"eventType":"Bounce","mail":{"source":"a@lgc.li","tags":{"ses:caller-identity":["carlos-mail-beb-jb-book"]}},"bounce":{"bounceType":7}}`)) |
| + if ev.Caller != "carlos-mail-beb-jb-book" { |
| + t.Errorf("a malformed bounce block cost the sender: %q", ev.Caller) |
| + } |
| + good, _ := ParseEvent([]byte(bounceEvent)) |
| + if good.Incomplete != "" { |
| + t.Errorf("a well-formed event claims to be incomplete: %q", good.Incomplete) |
| + } |
| +} |
| + |
| +// A bounce is about the recipients it names. One that names nobody is not a |
| +// bounce for everybody the message was addressed to. |
| +func TestAPerRecipientEventThatNamesNobodyIsNotWidened(t *testing.T) { |
| + ev, err := ParseEvent([]byte(`{"eventType":"Bounce","mail":{"source":"a@lgc.li","messageId":"m", |
| + "destination":["one@example.com","two@example.com","three@example.com"]},"bounce":{"bounceType":"Permanent","bouncedRecipients":[]}}`)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if rs := ev.Recipients(); len(rs) != 1 || rs[0].Recipient != "" { |
| + t.Errorf("a bounce naming nobody was widened to %+v", rs) |
| + } |
| +} |