| 1 | package console |
| 2 | |
| 3 | // The per-message email log: what happened to each message an app sent, to |
| 4 | // whom, and what the receiving server said. The counts in mailcounts.go are |
| 5 | // the abuse signal and stay exactly as they were; this sits beside them and |
| 6 | // is fed from the same verified SES notifications, one row per recipient |
| 7 | // per notification. |
| 8 | |
| 9 | import ( |
| 10 | "context" |
| 11 | "database/sql" |
| 12 | "strings" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/mail" |
| 16 | ) |
| 17 | |
| 18 | // mailEventsRetention is how long a row is kept. The log carries recipient |
| 19 | // addresses — somebody's guests, customers, users — so it is a window, not |
| 20 | // an archive: long enough to answer "did that email last week arrive", no |
| 21 | // longer. |
| 22 | const mailEventsRetention = 30 * 24 * time.Hour |
| 23 | |
| 24 | // mailEventsPruneEvery bounds how often the prune runs. It is one indexed |
| 25 | // DELETE, but it would otherwise run once per notification, and an hour's |
| 26 | // slack on a thirty-day window changes nothing anybody can see. |
| 27 | const mailEventsPruneEvery = time.Hour |
| 28 | |
| 29 | // MailEvent is one row of the log. |
| 30 | type MailEvent struct { |
| 31 | MessageID string `json:"message_id"` |
| 32 | Type string `json:"type"` |
| 33 | Recipient string `json:"recipient"` |
| 34 | Status string `json:"status,omitempty"` |
| 35 | Diagnostic string `json:"diagnostic,omitempty"` |
| 36 | Detail string `json:"detail,omitempty"` |
| 37 | BySender bool `json:"by_sender"` |
| 38 | At time.Time `json:"at"` |
| 39 | } |
| 40 | |
| 41 | // RecordMailMessageEvent keeps one notification for one app: a row per |
| 42 | // recipient it concerns. notice is SNS's MessageId for the notification, |
| 43 | // and a notification SNS delivers twice writes nothing the second time. |
| 44 | // bySender says whether SES itself named this app as the sender, rather |
| 45 | // than the app being one of several that share the sending domain. |
| 46 | func (s *Store) RecordMailMessageEvent(ctx context.Context, acct, app, notice string, ev mail.Event, bySender bool, now time.Time) error { |
| 47 | at := ev.At |
| 48 | if at.IsZero() { |
| 49 | at = now |
| 50 | } |
| 51 | notice = mailNoticeKey(notice, ev, now) |
| 52 | sender := 0 |
| 53 | if bySender { |
| 54 | sender = 1 |
| 55 | } |
| 56 | detail := ev.Detail.Summary() |
| 57 | if recipients := ev.Recipients(); len(recipients) > 0 { |
| 58 | // acct/app's rows only, under that app's permit (claim_mail_sql.go). |
| 59 | err := s.withMailAppSQL(ctx, acct, app, "sql.RecordMailMessageEvent", func(v *Store) error { |
| 60 | for _, o := range recipients { |
| 61 | if _, err := v.db.ExecContext(ctx, `INSERT OR IGNORE INTO mail_events |
| 62 | (account, app, notice, recipient, message_id, type, status, diagnostic, detail, by_sender, at, received_at) |
| 63 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, |
| 64 | acct, app, notice, o.Recipient, ev.MessageID, ev.Type, o.Status, o.Diagnostic, detail, sender, |
| 65 | at.Unix(), now.Unix()); err != nil { |
| 66 | return err |
| 67 | } |
| 68 | } |
| 69 | return nil |
| 70 | }) |
| 71 | if err != nil { |
| 72 | return err |
| 73 | } |
| 74 | } |
| 75 | // After the app's own write and outside it: the prune touches other |
| 76 | // apps' rows, each under its own permit. |
| 77 | return s.pruneMailEvents(ctx, now) |
| 78 | } |
| 79 | |
| 80 | // mailNoticeKey is the key a notification is kept and queued under: SNS's |
| 81 | // own MessageId, or — for a notification with no id of its own, which is |
| 82 | // nothing SNS sends but the recording path is reachable directly — what it |
| 83 | // says, so a repeat still collapses and two different notifications do not. |
| 84 | // The log and the push queue share it, so the id an app is handed is the id |
| 85 | // the log has. |
| 86 | func mailNoticeKey(notice string, ev mail.Event, now time.Time) string { |
| 87 | if notice != "" { |
| 88 | return notice |
| 89 | } |
| 90 | at := ev.At |
| 91 | if at.IsZero() { |
| 92 | at = now |
| 93 | } |
| 94 | var to []string |
| 95 | for _, o := range ev.Recipients() { |
| 96 | to = append(to, o.Recipient) |
| 97 | } |
| 98 | // Two bounces for one message — one per recipient, or a delay reported |
| 99 | // twice — are two notifications, so what they are about is in the key. |
| 100 | return ev.MessageID + "/" + ev.Type + "/" + at.UTC().Format(time.RFC3339Nano) + "/" + strings.Join(to, ",") |
| 101 | } |
| 102 | |
| 103 | // mailAppTables are the tables keyed by an app's (account sqid, name) that |
| 104 | // belong to that app alone: a name that is freed must not hand them on. |
| 105 | // |
| 106 | // mail_sender_counts joins the list (senders design §8), which also closes |
| 107 | // the gap where a rename left an app's counts behind under its old name. |
| 108 | // It cannot join blindly, though: every sweep here keys on a TIMESTAMP |
| 109 | // column and this table has a day string instead, so it gets the explicit |
| 110 | // case in mailAppTimeColumn and carryMailApp below. mail_counts is |
| 111 | // deliberately not here: nothing writes it any more and |
| 112 | // DropLegacyMailCounts removes it whole. |
| 113 | var mailAppTables = []string{"mail_events", "mail_event_hooks", "mail_event_deliveries", "mail_sender_counts"} |
| 114 | |
| 115 | // execer is what carryMailApp and forgetMailApp need from a transaction. |
| 116 | type execer interface { |
| 117 | ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) |
| 118 | } |
| 119 | |
| 120 | // carryMailApp moves an app's email history with the app, when a rename or |
| 121 | // a move to another account changes the (sqid, name) it is keyed by. Rows |
| 122 | // already at the destination belong to a purged predecessor of that name |
| 123 | // and are dropped first, and only rows received since this app's own row was |
| 124 | // created travel — the same rule RenameAppIn applies to audit rows, for the |
| 125 | // same reason. Whatever is left under the old key is dropped. |
| 126 | // |
| 127 | // Across accounts, a row attributed only by its sending domain (by_sender = |
| 128 | // 0) does not travel. Such a row was given to every app on the domain because |
| 129 | // SES did not say which of them sent the message, so it may name another |
| 130 | // app's recipient — and that app stays behind in the source account. Within |
| 131 | // one account both kinds carry, since the same members could read both |
| 132 | // before and after. |
| 133 | func carryMailApp(ctx context.Context, tx execer, fromSqid, fromApp, toSqid, toApp string, createdAt int64) error { |
| 134 | for _, table := range mailAppTables { |
| 135 | if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ? AND app = ?`, toSqid, toApp); err != nil { |
| 136 | return err |
| 137 | } |
| 138 | if table == "mail_events" && fromSqid != toSqid { |
| 139 | if _, err := tx.ExecContext(ctx, `DELETE FROM mail_events WHERE account = ? AND app = ? AND by_sender = 0`, fromSqid, fromApp); err != nil { |
| 140 | return err |
| 141 | } |
| 142 | } |
| 143 | if table == "mail_event_deliveries" { |
| 144 | // Not carried: a queued body names the account and app it was |
| 145 | // cut for, so after a move or a rename it names neither, and an |
| 146 | // app checking its host is right to refuse it. The final DELETE |
| 147 | // below drops them. |
| 148 | if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ? AND app = ?`, fromSqid, fromApp); err != nil { |
| 149 | return err |
| 150 | } |
| 151 | continue |
| 152 | } |
| 153 | if _, err := tx.ExecContext(ctx, `UPDATE `+table+` SET account = ?, app = ? WHERE account = ? AND app = ? AND `+ |
| 154 | mailAppTimeColumn(table)+` >= ?`, toSqid, toApp, fromSqid, fromApp, mailAppTimeSince(table, createdAt)); err != nil { |
| 155 | return err |
| 156 | } |
| 157 | if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ? AND app = ?`, fromSqid, fromApp); err != nil { |
| 158 | return err |
| 159 | } |
| 160 | } |
| 161 | return nil |
| 162 | } |
| 163 | |
| 164 | // forgetMailAccount drops everything an account's apps left in these tables, |
| 165 | // when the account itself is reaped. Nothing can reach the rows afterwards — |
| 166 | // the sqid is tombstoned and never reused — but they are recipients' |
| 167 | // addresses, and a window is not an archive. |
| 168 | func forgetMailAccount(ctx context.Context, tx execer, sqid string) error { |
| 169 | for _, table := range mailAppTables { |
| 170 | if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ?`, sqid); err != nil { |
| 171 | return err |
| 172 | } |
| 173 | } |
| 174 | return nil |
| 175 | } |
| 176 | |
| 177 | // forgetMailApp drops everything a purged app left in these tables, so a |
| 178 | // later app of the same name starts with none of it. |
| 179 | func forgetMailApp(ctx context.Context, tx execer, sqid, app string) error { |
| 180 | for _, table := range mailAppTables { |
| 181 | if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ? AND app = ?`, sqid, app); err != nil { |
| 182 | return err |
| 183 | } |
| 184 | } |
| 185 | return nil |
| 186 | } |
| 187 | |
| 188 | // mailAppTimeColumn is the column that says when a row came to be. Most are |
| 189 | // unix seconds; mail_sender_counts keys on a UTC date string, which compares |
| 190 | // correctly against another ISO date and against nothing else. |
| 191 | func mailAppTimeColumn(table string) string { |
| 192 | switch table { |
| 193 | case "mail_events": |
| 194 | return "received_at" |
| 195 | case "mail_event_hooks": |
| 196 | return "set_at" |
| 197 | case "mail_sender_counts": |
| 198 | return "day" |
| 199 | } |
| 200 | return "created_at" |
| 201 | } |
| 202 | |
| 203 | // mailAppTimeSince renders createdAt in whatever mailAppTimeColumn(table) |
| 204 | // compares against: a unix stamp for a timestamp column, a UTC date string |
| 205 | // for the one day column. Comparing a unix integer against a 'YYYY-MM-DD' |
| 206 | // string in SQLite is not an error, it is simply always false, so getting |
| 207 | // this wrong would silently carry nothing. |
| 208 | func mailAppTimeSince(table string, createdAt int64) any { |
| 209 | if mailAppTimeColumn(table) == "day" { |
| 210 | return time.Unix(createdAt, 0).UTC().Format("2006-01-02") |
| 211 | } |
| 212 | return createdAt |
| 213 | } |
| 214 | |
| 215 | // pruneMailEvents trims the log to its retention window, at most once per |
| 216 | // mailEventsPruneEvery. The stamp is taken before the DELETE runs, so two |
| 217 | // notifications arriving together do not both prune. |
| 218 | func (s *Store) pruneMailEvents(ctx context.Context, now time.Time) error { |
| 219 | last := s.mailEventsPruned.Load() |
| 220 | if last != 0 && now.Unix()-last < int64(mailEventsPruneEvery/time.Second) { |
| 221 | return nil |
| 222 | } |
| 223 | if !s.mailEventsPruned.CompareAndSwap(last, now.Unix()) { |
| 224 | return nil |
| 225 | } |
| 226 | if s.gate != nil && s.bound == nil { |
| 227 | // Gated: one app at a time, each under its own permit. The stamp |
| 228 | // stays on a partial failure: an app that refuses (an operation |
| 229 | // holds it) is retried next window, not on every notification, |
| 230 | // which would open an operation per app per event. |
| 231 | return s.pruneMailAppRows(ctx, "mail_events", "received_at", now.Add(-mailEventsRetention).Unix(), "sql.PruneMailEvents") |
| 232 | } |
| 233 | if _, err := s.db.ExecContext(ctx, `DELETE FROM mail_events WHERE received_at < ?`, now.Add(-mailEventsRetention).Unix()); err != nil { |
| 234 | // Not pruned, so not stamped: the next notification tries again |
| 235 | // rather than waiting out the hour. |
| 236 | s.mailEventsPruned.CompareAndSwap(now.Unix(), last) |
| 237 | return err |
| 238 | } |
| 239 | return nil |
| 240 | } |
| 241 | |
| 242 | // MailEventQuery narrows a read of the log. The zero value is the newest |
| 243 | // mailEventsDefaultLimit rows of the whole window. |
| 244 | type MailEventQuery struct { |
| 245 | Since time.Time // only rows SES timestamped at or after this |
| 246 | MessageID string // only this message |
| 247 | Recipient string // only this address, case-insensitively |
| 248 | Limit int |
| 249 | } |
| 250 | |
| 251 | const ( |
| 252 | mailEventsDefaultLimit = 100 |
| 253 | mailEventsMaxLimit = 1000 |
| 254 | ) |
| 255 | |
| 256 | // MailEvents reads one app's log, newest first. A row older than the |
| 257 | // retention window is never returned, whether or not the prune has reached |
| 258 | // it yet — what a member is told is kept for thirty days has to be what |
| 259 | // they can see. |
| 260 | func (s *Store) MailEvents(ctx context.Context, acct, app string, q MailEventQuery, now time.Time) ([]MailEvent, error) { |
| 261 | limit := q.Limit |
| 262 | if limit <= 0 { |
| 263 | limit = mailEventsDefaultLimit |
| 264 | } |
| 265 | if limit > mailEventsMaxLimit { |
| 266 | limit = mailEventsMaxLimit |
| 267 | } |
| 268 | where := []string{"account = ?", "app = ?", "received_at >= ?"} |
| 269 | args := []any{acct, app, now.Add(-mailEventsRetention).Unix()} |
| 270 | if !q.Since.IsZero() { |
| 271 | where = append(where, "at >= ?") |
| 272 | args = append(args, q.Since.Unix()) |
| 273 | } |
| 274 | if q.MessageID != "" { |
| 275 | where = append(where, "message_id = ?") |
| 276 | args = append(args, q.MessageID) |
| 277 | } |
| 278 | if q.Recipient != "" { |
| 279 | where = append(where, "recipient = ? COLLATE NOCASE") |
| 280 | args = append(args, q.Recipient) |
| 281 | } |
| 282 | args = append(args, limit) |
| 283 | rows, err := s.db.QueryContext(ctx, `SELECT message_id, type, recipient, status, diagnostic, detail, by_sender, at |
| 284 | FROM mail_events WHERE `+strings.Join(where, " AND ")+` |
| 285 | ORDER BY at DESC, received_at DESC, recipient LIMIT ?`, args...) |
| 286 | if err != nil { |
| 287 | return nil, err |
| 288 | } |
| 289 | defer rows.Close() |
| 290 | var out []MailEvent |
| 291 | for rows.Next() { |
| 292 | var e MailEvent |
| 293 | var at int64 |
| 294 | var sender int |
| 295 | if err := rows.Scan(&e.MessageID, &e.Type, &e.Recipient, &e.Status, &e.Diagnostic, &e.Detail, &sender, &at); err != nil { |
| 296 | return nil, err |
| 297 | } |
| 298 | e.BySender = sender == 1 |
| 299 | e.At = time.Unix(at, 0).UTC() |
| 300 | out = append(out, e) |
| 301 | } |
| 302 | if err := rows.Err(); err != nil && err != sql.ErrNoRows { |
| 303 | return nil, err |
| 304 | } |
| 305 | return out, nil |
| 306 | } |
| 307 | |
amadan