| 1 | package console |
| 2 | |
| 3 | // Telling an app what happened to its email. An app that asks — `carlos |
| 4 | // email events push --app <a> --path /mail/events` — is POSTed each |
| 5 | // delivery, bounce, complaint, rejection and delay SES reports for a message |
| 6 | // it sent, signed by the console, at that path on its own platform host. |
| 7 | // |
| 8 | // WHAT CROSSES, AND TO WHOM. The payload names the recipient, and a |
| 9 | // recipient's address is the sending app's business and nobody else's. So |
| 10 | // only a notification SES attributed to the app itself (an identity CARLOS |
| 11 | // minted for it, a by_sender row in the log) is ever queued for it: an event that could only |
| 12 | // be attributed by a sending domain several apps share is kept in the log, |
| 13 | // where a member can read it, and is never pushed. A Send is not pushed |
| 14 | // either — the app learnt that one from SES directly, in the reply to the |
| 15 | // message it sent. |
| 16 | // |
| 17 | // WHERE IT GOES. Only ever https://<app>.<account>.<AppsDomain><path>: the |
| 18 | // host is built from names the platform validated, never from anything a |
| 19 | // request supplies, and the path passes schedule.ValidPath, the same check |
| 20 | // that keeps a schedule tick on the instance. Redirects are refused, so the |
| 21 | // far end cannot turn a delivery into a POST of somebody's address at a |
| 22 | // host of its choosing. The request travels the public edge like any other, |
| 23 | // which is why it wakes a hibernating app and why an app behind `carlos |
| 24 | // gate` cannot receive one. |
| 25 | // |
| 26 | // HOW AN APP KNOWS IT IS REAL. Every delivery is signed with an Ed25519 key |
| 27 | // the console keeps for this and nothing else (mailEventsSigningKey), and |
| 28 | // the public half is delivered to the app as CARLOS_MAIL_EVENTS_KEY when it |
| 29 | // asks for events. The signature covers a domain-separation label, the |
| 30 | // timestamp and the exact body: |
| 31 | // |
| 32 | // Carlos-Signature: t=<unix seconds>,ed25519=<base64 signature> |
| 33 | // signed bytes: "carlos-mail-event\n" + t + "\n" + body |
| 34 | // |
| 35 | // An app refuses a bad signature, a timestamp more than a few minutes from |
| 36 | // its own clock, and a body whose account or app is not its own — the key is |
| 37 | // the deployment's, so the last check is what stops one app's event, |
| 38 | // replayed at another, from being believed there. |
| 39 | // |
| 40 | // WHAT ONE APP CAN COST OTHERS. A pass starts from the apps that asked, and |
| 41 | // reads each one's oldest due rows through an index, so its cost follows how |
| 42 | // many apps ask, not how much any of them has queued. Apps are sent to eight |
| 43 | // at a time across the whole deployment, not per account, and an app's turn |
| 44 | // ends at its first failure — so a hung app holds one slot for one timeout a |
| 45 | // pass, and an account with many hung apps costs everyone about ten seconds |
| 46 | // a pass per eight of them. An app that answers 2xx slowly keeps its turn, |
| 47 | // and can hold a slot for up to five timeouts. And an app that has accepted |
| 48 | // nothing for a day is suspended: its queue is given up on and nothing more |
| 49 | // is queued until the push is turned on again (mailPushSuspendAfter), so a |
| 50 | // failing app cannot grow a backlog that a pass has to carry for a month. |
| 51 | |
| 52 | import ( |
| 53 | "bytes" |
| 54 | "context" |
| 55 | "crypto/ed25519" |
| 56 | "crypto/rand" |
| 57 | "crypto/x509" |
| 58 | "database/sql" |
| 59 | "encoding/base64" |
| 60 | "encoding/json" |
| 61 | "errors" |
| 62 | "fmt" |
| 63 | "io" |
| 64 | "log" |
| 65 | "net/http" |
| 66 | "strconv" |
| 67 | "strings" |
| 68 | "sync" |
| 69 | "time" |
| 70 | |
| 71 | "github.com/carlosframework/platform/internal/hostops" |
| 72 | "github.com/carlosframework/platform/internal/mail" |
| 73 | ) |
| 74 | |
| 75 | // MailEventsKeyEnv is the variable an app finds the verifying key in. |
| 76 | const MailEventsKeyEnv = "CARLOS_MAIL_EVENTS_KEY" |
| 77 | |
| 78 | // mailEventSignatureLabel separates this signature from anything else the |
| 79 | // key could conceivably be asked to sign. |
| 80 | const mailEventSignatureLabel = "carlos-mail-event" |
| 81 | |
| 82 | const ( |
| 83 | // mailPushEvery is the queue's cadence. A delivery waits at most this |
| 84 | // long for its first attempt; notifications themselves arrive seconds |
| 85 | // after SES hears back from the receiving server. |
| 86 | mailPushEvery = 15 * time.Second |
| 87 | // mailPushPerApp is the most one app is sent in a pass, and |
| 88 | // mailPushApps how many apps are sent to at once. Together they are what |
| 89 | // keeps one app from being everyone's problem: an app whose handler |
| 90 | // hangs occupies one of mailPushApps for at most one timeout (see |
| 91 | // deliverAppMailEvents), and its backlog waits behind itself, not in |
| 92 | // front of anybody else's. |
| 93 | mailPushPerApp = 5 |
| 94 | mailPushApps = 8 |
| 95 | // mailPushSuspendAfter is how long an app may go on failing — every |
| 96 | // attempt at every row answered with anything but a 2xx — before its |
| 97 | // push is suspended. A day is past any deploy an app could be halfway |
| 98 | // through, and it bounds what a failing app can have queued to a day of |
| 99 | // its own mail. |
| 100 | mailPushSuspendAfter = 24 * time.Hour |
| 101 | // mailPushPruneEvery bounds how often the queue is trimmed, as the log is. |
| 102 | mailPushPruneEvery = time.Hour |
| 103 | mailPushMaxBody = 64 << 10 |
| 104 | ) |
| 105 | |
| 106 | // mailPushTimeout bounds one POST: enough for the edge to wake a |
| 107 | // hibernating app and for it to answer. A var so a test can shorten it. |
| 108 | var mailPushTimeout = 10 * time.Second |
| 109 | |
| 110 | // mailPushBackoff is when each retry is due after a failed attempt, counted |
| 111 | // from that attempt. After the last, the delivery is given up on — a little |
| 112 | // under a day of trying, which is past any deploy an app could be halfway |
| 113 | // through and short of the retention window the payload is bound by. |
| 114 | var mailPushBackoff = []time.Duration{ |
| 115 | time.Minute, 5 * time.Minute, 15 * time.Minute, time.Hour, 3 * time.Hour, 6 * time.Hour, 12 * time.Hour, |
| 116 | } |
| 117 | |
| 118 | // mailPushClient refuses redirects: see the file comment. The per-request |
| 119 | // timeout is the context's (deliverMailEvent), so a test can shorten it. |
| 120 | var mailPushClient = &http.Client{ |
| 121 | CheckRedirect: func(*http.Request, []*http.Request) error { |
| 122 | return http.ErrUseLastResponse |
| 123 | }, |
| 124 | } |
| 125 | |
| 126 | // mailPushedTypes are the event types an app is told about. |
| 127 | var mailPushedTypes = map[string]bool{ |
| 128 | mail.EventDelivery: true, mail.EventBounce: true, mail.EventComplaint: true, |
| 129 | mail.EventReject: true, mail.EventDeliveryDelay: true, |
| 130 | } |
| 131 | |
| 132 | // ---- the key ------------------------------------------------------------------ |
| 133 | |
| 134 | // MailEventsSigningKey returns the Ed25519 key email-event deliveries are |
| 135 | // signed with, generating and persisting it on first use. It is its own key |
| 136 | // rather than SigningKey's: that one signs keymail envelopes, and a key |
| 137 | // that signs two kinds of thing is one whose signatures can be carried from |
| 138 | // one to the other. |
| 139 | func (s *Store) MailEventsSigningKey() (ed25519.PrivateKey, error) { |
| 140 | load := func() (ed25519.PrivateKey, error) { |
| 141 | var b64 string |
| 142 | if err := s.db.QueryRow(`SELECT pkcs8 FROM keys WHERE name = 'mail-events-sign'`).Scan(&b64); err != nil { |
| 143 | return nil, err |
| 144 | } |
| 145 | der, err := base64.StdEncoding.DecodeString(b64) |
| 146 | if err != nil { |
| 147 | return nil, fmt.Errorf("console: stored mail events key: %v", err) |
| 148 | } |
| 149 | key, err := x509.ParsePKCS8PrivateKey(der) |
| 150 | if err != nil { |
| 151 | return nil, fmt.Errorf("console: stored mail events key: %v", err) |
| 152 | } |
| 153 | ed, ok := key.(ed25519.PrivateKey) |
| 154 | if !ok { |
| 155 | return nil, fmt.Errorf("console: stored mail events key is not Ed25519") |
| 156 | } |
| 157 | return ed, nil |
| 158 | } |
| 159 | if k, err := load(); err == nil { |
| 160 | return k, nil |
| 161 | } else if err != sql.ErrNoRows { |
| 162 | return nil, err |
| 163 | } |
| 164 | _, priv, err := ed25519.GenerateKey(rand.Reader) |
| 165 | if err != nil { |
| 166 | return nil, err |
| 167 | } |
| 168 | der, err := x509.MarshalPKCS8PrivateKey(priv) |
| 169 | if err != nil { |
| 170 | return nil, err |
| 171 | } |
| 172 | // INSERT OR IGNORE + re-load, as SigningKey: a racing opener's row wins |
| 173 | // and everyone converges on it. |
| 174 | if _, err := s.db.Exec(`INSERT OR IGNORE INTO keys (name, pkcs8, created_at) VALUES ('mail-events-sign', ?, ?)`, |
| 175 | base64.StdEncoding.EncodeToString(der), time.Now().Unix()); err != nil { |
| 176 | return nil, err |
| 177 | } |
| 178 | return load() |
| 179 | } |
| 180 | |
| 181 | // mailEventsPublicKey is the verifying half, as delivered to an app: the 32 |
| 182 | // raw bytes, standard base64. |
| 183 | func mailEventsPublicKey(priv ed25519.PrivateKey) string { |
| 184 | return base64.StdEncoding.EncodeToString(priv.Public().(ed25519.PublicKey)) |
| 185 | } |
| 186 | |
| 187 | // signMailEvent is the Carlos-Signature header value for body at t. |
| 188 | func signMailEvent(priv ed25519.PrivateKey, t time.Time, body []byte) string { |
| 189 | ts := strconv.FormatInt(t.Unix(), 10) |
| 190 | sig := ed25519.Sign(priv, mailEventSignedBytes(ts, body)) |
| 191 | return "t=" + ts + ",ed25519=" + base64.StdEncoding.EncodeToString(sig) |
| 192 | } |
| 193 | |
| 194 | func mailEventSignedBytes(ts string, body []byte) []byte { |
| 195 | msg := make([]byte, 0, len(mailEventSignatureLabel)+len(ts)+len(body)+2) |
| 196 | msg = append(msg, mailEventSignatureLabel...) |
| 197 | msg = append(msg, '\n') |
| 198 | msg = append(msg, ts...) |
| 199 | msg = append(msg, '\n') |
| 200 | return append(msg, body...) |
| 201 | } |
| 202 | |
| 203 | // VerifyMailEvent is what an app does with a delivery, in Go: check the |
| 204 | // signature header against the body with the delivered key, and refuse a |
| 205 | // timestamp further than skew from now. It is here so the scheme has one |
| 206 | // executable definition the tests hold both ends to; an app in any |
| 207 | // language does the same four steps. |
| 208 | func VerifyMailEvent(publicKeyB64, header string, body []byte, now time.Time, skew time.Duration) error { |
| 209 | raw, err := base64.StdEncoding.DecodeString(publicKeyB64) |
| 210 | if err != nil || len(raw) != ed25519.PublicKeySize { |
| 211 | return errors.New("mail event: the key is not an Ed25519 public key") |
| 212 | } |
| 213 | var ts, sigB64 string |
| 214 | for _, part := range strings.Split(header, ",") { |
| 215 | k, v, _ := strings.Cut(strings.TrimSpace(part), "=") |
| 216 | switch k { |
| 217 | case "t": |
| 218 | ts = v |
| 219 | case "ed25519": |
| 220 | sigB64 = v |
| 221 | } |
| 222 | } |
| 223 | unix, err := strconv.ParseInt(ts, 10, 64) |
| 224 | if err != nil { |
| 225 | return errors.New("mail event: no timestamp in the signature") |
| 226 | } |
| 227 | if d := now.Sub(time.Unix(unix, 0)); d > skew || d < -skew { |
| 228 | return errors.New("mail event: the timestamp is too far from now") |
| 229 | } |
| 230 | sig, err := base64.StdEncoding.DecodeString(sigB64) |
| 231 | if err != nil || len(sig) != ed25519.SignatureSize { |
| 232 | return errors.New("mail event: no signature") |
| 233 | } |
| 234 | if !ed25519.Verify(ed25519.PublicKey(raw), mailEventSignedBytes(ts, body), sig) { |
| 235 | return errors.New("mail event: the signature does not verify") |
| 236 | } |
| 237 | return nil |
| 238 | } |
| 239 | |
| 240 | // ---- what an app is sent -------------------------------------------------------- |
| 241 | |
| 242 | // MailEventPayload is the body of one delivery. |
| 243 | type MailEventPayload struct { |
| 244 | // ID is stable across retries of the same notification, so an app that |
| 245 | // receives one twice — a retry after an answer that never arrived — can |
| 246 | // tell. |
| 247 | ID string `json:"id"` |
| 248 | Account string `json:"account"` |
| 249 | App string `json:"app"` |
| 250 | Type string `json:"type"` |
| 251 | MessageID string `json:"message_id"` |
| 252 | At time.Time `json:"at"` |
| 253 | Detail string `json:"detail,omitempty"` |
| 254 | Recipients []MailEventRecipient `json:"recipients"` |
| 255 | } |
| 256 | |
| 257 | // MailEventRecipient is one recipient's part in a delivery. |
| 258 | type MailEventRecipient struct { |
| 259 | Address string `json:"address"` |
| 260 | Status string `json:"status,omitempty"` |
| 261 | Diagnostic string `json:"diagnostic,omitempty"` |
| 262 | } |
| 263 | |
| 264 | func mailEventPayload(acct, app, notice string, ev mail.Event, now time.Time) MailEventPayload { |
| 265 | at := ev.At |
| 266 | if at.IsZero() { |
| 267 | at = now |
| 268 | } |
| 269 | p := MailEventPayload{ID: notice, Account: acct, App: app, Type: ev.Type, MessageID: ev.MessageID, |
| 270 | At: at.UTC(), Detail: ev.Detail.Summary(), Recipients: []MailEventRecipient{}} |
| 271 | for _, o := range ev.Recipients() { |
| 272 | if o.Recipient == "" { |
| 273 | continue |
| 274 | } |
| 275 | p.Recipients = append(p.Recipients, MailEventRecipient{Address: o.Recipient, Status: o.Status, Diagnostic: o.Diagnostic}) |
| 276 | } |
| 277 | return p |
| 278 | } |
| 279 | |
| 280 | // ---- hooks ------------------------------------------------------------------------ |
| 281 | |
| 282 | // MailEventHook is where an app has asked to be told. |
| 283 | type MailEventHook struct { |
| 284 | Path string |
| 285 | SetBy string |
| 286 | SetAt time.Time |
| 287 | // FailingSince is when the app's current run of failed deliveries began; |
| 288 | // zero when it has none. SuspendedAt is when the push was suspended for |
| 289 | // failing for a day, and SuspendedWhy says what the last attempt got. |
| 290 | FailingSince time.Time |
| 291 | SuspendedAt time.Time |
| 292 | SuspendedWhy string |
| 293 | } |
| 294 | |
| 295 | // Suspended reports whether the push is suspended. |
| 296 | func (h *MailEventHook) Suspended() bool { return h != nil && !h.SuspendedAt.IsZero() } |
| 297 | |
| 298 | // SetMailEventHook turns the push on, or moves its path — and, either way, |
| 299 | // lifts a suspension and forgets the run of failures behind it: turning it |
| 300 | // on again is how a member says the handler is there now. |
| 301 | func (s *Store) SetMailEventHook(ctx context.Context, acct, app, path, by string, now time.Time) error { |
| 302 | return s.withMailAppSQL(ctx, acct, app, "sql.SetMailEventHook", func(v *Store) error { |
| 303 | _, err := v.db.ExecContext(ctx, `INSERT INTO mail_event_hooks (account, app, path, set_by, set_at) VALUES (?, ?, ?, ?, ?) |
| 304 | ON CONFLICT(account, app) DO UPDATE SET path = excluded.path, set_by = excluded.set_by, set_at = excluded.set_at, |
| 305 | failing_since = 0, suspended_at = 0, suspended_why = ''`, |
| 306 | acct, app, path, by, now.Unix()) |
| 307 | return err |
| 308 | }) |
| 309 | } |
| 310 | |
| 311 | const mailEventHookColumns = `account, app, path, set_by, set_at, failing_since, suspended_at, suspended_why` |
| 312 | |
| 313 | type mailEventHookRow struct { |
| 314 | claimAppID string // Captured before reading queued payloads; never rebound by a delayed worker. |
| 315 | Account, App string |
| 316 | MailEventHook |
| 317 | } |
| 318 | |
| 319 | func scanMailEventHook(scan func(...any) error) (mailEventHookRow, error) { |
| 320 | var h mailEventHookRow |
| 321 | var setAt, failing, suspended int64 |
| 322 | if err := scan(&h.Account, &h.App, &h.Path, &h.SetBy, &setAt, &failing, &suspended, &h.SuspendedWhy); err != nil { |
| 323 | return h, err |
| 324 | } |
| 325 | h.SetAt = time.Unix(setAt, 0).UTC() |
| 326 | if failing > 0 { |
| 327 | h.FailingSince = time.Unix(failing, 0).UTC() |
| 328 | } |
| 329 | if suspended > 0 { |
| 330 | h.SuspendedAt = time.Unix(suspended, 0).UTC() |
| 331 | } |
| 332 | return h, nil |
| 333 | } |
| 334 | |
| 335 | // MailEventHookFor answers nil when the app has not asked. |
| 336 | func (s *Store) MailEventHookFor(ctx context.Context, acct, app string) (*MailEventHook, error) { |
| 337 | h, err := scanMailEventHook(s.db.QueryRowContext(ctx, |
| 338 | `SELECT `+mailEventHookColumns+` FROM mail_event_hooks WHERE account = ? AND app = ?`, acct, app).Scan) |
| 339 | if err == sql.ErrNoRows { |
| 340 | return nil, nil |
| 341 | } |
| 342 | if err != nil { |
| 343 | return nil, err |
| 344 | } |
| 345 | return &h.MailEventHook, nil |
| 346 | } |
| 347 | |
| 348 | // activeMailEventHooks is every push that is on and not suspended: where a |
| 349 | // pass starts. |
| 350 | func (s *Store) activeMailEventHooks(ctx context.Context) ([]mailEventHookRow, error) { |
| 351 | rows, err := s.db.QueryContext(ctx, `SELECT `+mailEventHookColumns+` FROM mail_event_hooks WHERE suspended_at = 0 ORDER BY account, app`) |
| 352 | if err != nil { |
| 353 | return nil, err |
| 354 | } |
| 355 | defer rows.Close() |
| 356 | var out []mailEventHookRow |
| 357 | for rows.Next() { |
| 358 | h, err := scanMailEventHook(rows.Scan) |
| 359 | if err != nil { |
| 360 | return nil, err |
| 361 | } |
| 362 | out = append(out, h) |
| 363 | } |
| 364 | return out, rows.Err() |
| 365 | } |
| 366 | |
| 367 | // noteMailEventTurn records how an app's turn in a pass went: a delivery |
| 368 | // that landed ends any run of failures; a failure starts one if none is |
| 369 | // running; and a run that has lasted mailPushSuspendAfter suspends the push |
| 370 | // and gives up on its queue, in one transaction. It answers whether the push |
| 371 | // is now suspended. |
| 372 | func (s *Store) noteMailEventTurn(ctx context.Context, h mailEventHookRow, landed bool, failure string, now time.Time) (suspended bool, err error) { |
| 373 | err = s.withMailAppSQL(ctx, h.Account, h.App, "sql.NoteMailEventTurn", func(v *Store) error { |
| 374 | suspended, err = v.claimRawNoteMailEventTurn(ctx, h, landed, failure, now) |
| 375 | return err |
| 376 | }) |
| 377 | return suspended, err |
| 378 | } |
| 379 | |
| 380 | func (s *Store) claimRawNoteMailEventTurn(ctx context.Context, h mailEventHookRow, landed bool, failure string, now time.Time) (bool, error) { |
| 381 | switch { |
| 382 | case landed: |
| 383 | if h.FailingSince.IsZero() { |
| 384 | return false, nil |
| 385 | } |
| 386 | _, err := s.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = 0 WHERE account = ? AND app = ?`, h.Account, h.App) |
| 387 | return false, err |
| 388 | case failure == "": |
| 389 | return false, nil |
| 390 | case h.FailingSince.IsZero(): |
| 391 | _, err := s.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = ? WHERE account = ? AND app = ? AND failing_since = 0`, |
| 392 | now.Unix(), h.Account, h.App) |
| 393 | return false, err |
| 394 | case now.Sub(h.FailingSince) < mailPushSuspendAfter: |
| 395 | return false, nil |
| 396 | } |
| 397 | why := "nothing was accepted for a day — the last attempt " + strings.TrimPrefix(failure, "no answer: ") |
| 398 | if strings.HasPrefix(failure, "no answer") { |
| 399 | why = "nothing was accepted for a day — the last attempt got no answer" |
| 400 | } |
| 401 | if len(why) > 300 { |
| 402 | why = why[:300] |
| 403 | } |
| 404 | tx, err := s.db.BeginTx(ctx, nil) |
| 405 | if err != nil { |
| 406 | return false, err |
| 407 | } |
| 408 | defer tx.Rollback() |
| 409 | if _, err := tx.ExecContext(ctx, `UPDATE mail_event_hooks SET suspended_at = ?, suspended_why = ? WHERE account = ? AND app = ? AND set_at = ?`, |
| 410 | now.Unix(), why, h.Account, h.App, h.SetAt.Unix()); err != nil { |
| 411 | return false, err |
| 412 | } |
| 413 | if _, err := tx.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = 'push suspended' |
| 414 | WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), h.Account, h.App); err != nil { |
| 415 | return false, err |
| 416 | } |
| 417 | return true, tx.Commit() |
| 418 | } |
| 419 | |
| 420 | // clearMailEventFailing ends a run of failures that has no queue left behind |
| 421 | // it: failing_since measures one backlog's failing, not the app's history, |
| 422 | // so an app that failed once last week is not suspended by its next miss. |
| 423 | func (s *Store) clearMailEventFailing(ctx context.Context, h mailEventHookRow) error { |
| 424 | if h.FailingSince.IsZero() { |
| 425 | return nil |
| 426 | } |
| 427 | return s.withMailAppSQL(ctx, h.Account, h.App, "sql.ClearMailEventFailing", func(v *Store) error { |
| 428 | _, err := v.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = 0 |
| 429 | WHERE account = ? AND app = ? AND NOT EXISTS ( |
| 430 | SELECT 1 FROM mail_event_deliveries WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0)`, |
| 431 | h.Account, h.App, h.Account, h.App) |
| 432 | return err |
| 433 | }) |
| 434 | } |
| 435 | |
| 436 | // DeleteMailEventHook turns the push off, and gives up on everything still |
| 437 | // queued for it: an app that has stopped asking must not keep being sent |
| 438 | // addresses from a queue it has forgotten about. One transaction, so the |
| 439 | // push is never off with its queue still live. |
| 440 | func (s *Store) DeleteMailEventHook(ctx context.Context, acct, app string, now time.Time) (was bool, err error) { |
| 441 | err = s.withMailAppSQL(ctx, acct, app, "sql.DeleteMailEventHook", func(v *Store) error { |
| 442 | was, err = v.claimRawDeleteMailEventHook(ctx, acct, app, now) |
| 443 | return err |
| 444 | }) |
| 445 | return was, err |
| 446 | } |
| 447 | |
| 448 | func (s *Store) claimRawDeleteMailEventHook(ctx context.Context, acct, app string, now time.Time) (bool, error) { |
| 449 | tx, err := s.db.BeginTx(ctx, nil) |
| 450 | if err != nil { |
| 451 | return false, err |
| 452 | } |
| 453 | defer tx.Rollback() |
| 454 | res, err := tx.ExecContext(ctx, `DELETE FROM mail_event_hooks WHERE account = ? AND app = ?`, acct, app) |
| 455 | if err != nil { |
| 456 | return false, err |
| 457 | } |
| 458 | n, _ := res.RowsAffected() |
| 459 | if _, err := tx.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = 'push turned off' |
| 460 | WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), acct, app); err != nil { |
| 461 | return false, err |
| 462 | } |
| 463 | return n > 0, tx.Commit() |
| 464 | } |
| 465 | |
| 466 | // ---- the queue -------------------------------------------------------------------- |
| 467 | |
| 468 | // EnqueueMailEventDelivery queues one notification for one app. A notice |
| 469 | // already queued for the app queues nothing. |
| 470 | func (s *Store) EnqueueMailEventDelivery(ctx context.Context, acct, app, notice string, payload []byte, now time.Time) error { |
| 471 | return s.withMailAppSQL(ctx, acct, app, "sql.EnqueueMailEventDelivery", func(v *Store) error { |
| 472 | _, err := v.db.ExecContext(ctx, `INSERT OR IGNORE INTO mail_event_deliveries (account, app, notice, payload, next_at, created_at) |
| 473 | VALUES (?, ?, ?, ?, ?, ?)`, acct, app, notice, string(payload), now.Unix(), now.Unix()) |
| 474 | return err |
| 475 | }) |
| 476 | } |
| 477 | |
| 478 | type mailEventDelivery struct { |
| 479 | ID int64 |
| 480 | Account string |
| 481 | App string |
| 482 | Payload []byte |
| 483 | Attempts int |
| 484 | } |
| 485 | |
| 486 | // mailPushDueSQL is the read a pass makes per app. A test holds its query |
| 487 | // plan to the index. |
| 488 | const mailPushDueSQL = `SELECT id, account, app, payload, attempts FROM mail_event_deliveries |
| 489 | INDEXED BY mail_event_deliveries_app_due |
| 490 | WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0 AND next_at <= ? |
| 491 | ORDER BY next_at LIMIT ?` |
| 492 | |
| 493 | // dueMailEventDeliveriesFor is one app's oldest due rows, at most limit of |
| 494 | // them, sought through mail_event_deliveries_app_due: the four equalities |
| 495 | // and the range on next_at are all in the index, in its order, so the rows |
| 496 | // come back without sorting the app's backlog. |
| 497 | func (s *Store) dueMailEventDeliveriesFor(ctx context.Context, acct, app string, now time.Time, limit int) ([]mailEventDelivery, error) { |
| 498 | rows, err := s.db.QueryContext(ctx, mailPushDueSQL, acct, app, now.Unix(), limit) |
| 499 | if err != nil { |
| 500 | return nil, err |
| 501 | } |
| 502 | defer rows.Close() |
| 503 | var out []mailEventDelivery |
| 504 | for rows.Next() { |
| 505 | var d mailEventDelivery |
| 506 | var payload string |
| 507 | if err := rows.Scan(&d.ID, &d.Account, &d.App, &payload, &d.Attempts); err != nil { |
| 508 | return nil, err |
| 509 | } |
| 510 | d.Payload = []byte(payload) |
| 511 | out = append(out, d) |
| 512 | } |
| 513 | return out, rows.Err() |
| 514 | } |
| 515 | |
| 516 | // recordMailEventAttempt writes one attempt's outcome: delivered, due again |
| 517 | // after the backoff, or given up on once the backoff is spent. A row already |
| 518 | // given up on (the push was turned off while this attempt was in flight) |
| 519 | // stays given up on: it is not counted twice, and it is not revived. |
| 520 | // |
| 521 | // The row is d's own: d.Account and d.App were read from it, the write runs |
| 522 | // under that app's permit, and every statement also matches them, so an id |
| 523 | // naming another app's row updates nothing. |
| 524 | func (s *Store) recordMailEventAttempt(ctx context.Context, d mailEventDelivery, status int, failure string, now time.Time) error { |
| 525 | return s.withMailAppSQL(ctx, d.Account, d.App, "sql.RecordMailEventAttempt", func(v *Store) error { |
| 526 | return v.claimRawRecordMailEventAttempt(ctx, d, status, failure, now) |
| 527 | }) |
| 528 | } |
| 529 | |
| 530 | func (s *Store) claimRawRecordMailEventAttempt(ctx context.Context, d mailEventDelivery, status int, failure string, now time.Time) error { |
| 531 | attempts := d.Attempts + 1 |
| 532 | if failure == "" { |
| 533 | _, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries |
| 534 | SET attempts = ?, delivered_at = ?, last_attempt_at = ?, last_status = ?, last_error = '' WHERE id = ? AND account = ? AND app = ? AND abandoned_at = 0`, |
| 535 | attempts, now.Unix(), now.Unix(), status, d.ID, d.Account, d.App) |
| 536 | return err |
| 537 | } |
| 538 | if len(failure) > 300 { |
| 539 | failure = failure[:300] |
| 540 | } |
| 541 | if attempts > len(mailPushBackoff) { |
| 542 | _, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries |
| 543 | SET attempts = ?, abandoned_at = ?, abandoned_why = 'retries spent', last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ? AND account = ? AND app = ? AND abandoned_at = 0`, |
| 544 | attempts, now.Unix(), now.Unix(), status, failure, d.ID, d.Account, d.App) |
| 545 | return err |
| 546 | } |
| 547 | _, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries |
| 548 | SET attempts = ?, next_at = ?, last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ? AND account = ? AND app = ? AND abandoned_at = 0`, |
| 549 | attempts, now.Add(mailPushBackoff[attempts-1]).Unix(), now.Unix(), status, failure, d.ID, d.Account, d.App) |
| 550 | return err |
| 551 | } |
| 552 | |
| 553 | // abandonMailEventDeliveries gives up on everything still queued for one |
| 554 | // app, in one statement. The reason goes in abandoned_why, never over |
| 555 | // last_error: what the app last answered stays what it last answered. |
| 556 | func (s *Store) abandonMailEventDeliveries(ctx context.Context, acct, app, reason string, now time.Time) error { |
| 557 | return s.withMailAppSQL(ctx, acct, app, "sql.AbandonMailEventDeliveries", func(v *Store) error { |
| 558 | _, err := v.db.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = ? |
| 559 | WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), reason, acct, app) |
| 560 | return err |
| 561 | }) |
| 562 | } |
| 563 | |
| 564 | // pruneMailEventDeliveries drops every queued row past the retention window, |
| 565 | // delivered or not: the payload is a copy of the log's addresses and lives |
| 566 | // no longer than they do. |
| 567 | func (s *Store) pruneMailEventDeliveries(ctx context.Context, now time.Time) error { |
| 568 | last := s.mailPushPruned.Load() |
| 569 | if last != 0 && now.Unix()-last < int64(mailPushPruneEvery/time.Second) { |
| 570 | return nil |
| 571 | } |
| 572 | if !s.mailPushPruned.CompareAndSwap(last, now.Unix()) { |
| 573 | return nil |
| 574 | } |
| 575 | if s.gate != nil && s.bound == nil { |
| 576 | // Gated: per app, each under its own permit; the stamp stays on a |
| 577 | // partial failure (pruneMailEvents says why). |
| 578 | return s.pruneMailAppRows(ctx, "mail_event_deliveries", "created_at", now.Add(-mailEventsRetention).Unix(), "sql.PruneMailEventDeliveries") |
| 579 | } |
| 580 | if _, err := s.db.ExecContext(ctx, `DELETE FROM mail_event_deliveries WHERE created_at < ?`, now.Add(-mailEventsRetention).Unix()); err != nil { |
| 581 | // Not pruned, so not stamped: the next pass tries again. |
| 582 | s.mailPushPruned.CompareAndSwap(now.Unix(), last) |
| 583 | return err |
| 584 | } |
| 585 | return nil |
| 586 | } |
| 587 | |
| 588 | // MailEventPushStatus is what `carlos email events push` shows. |
| 589 | type MailEventPushStatus struct { |
| 590 | Pending int `json:"pending"` |
| 591 | Delivered int `json:"delivered"` |
| 592 | Abandoned int `json:"abandoned"` |
| 593 | LastDelivered time.Time `json:"last_delivered,omitempty"` |
| 594 | LastFailedAt time.Time `json:"last_failed_at,omitempty"` |
| 595 | LastStatus int `json:"last_status,omitempty"` |
| 596 | LastError string `json:"last_error,omitempty"` |
| 597 | } |
| 598 | |
| 599 | // MailEventPushStatusFor reads the counts over the retention window, and the |
| 600 | // last failure since `since` — the moment the push was last turned on — so a |
| 601 | // failure from before a member turned it off and on again, or before a |
| 602 | // suspension was lifted, is not reported as today's news. |
| 603 | func (s *Store) MailEventPushStatusFor(ctx context.Context, acct, app string, since time.Time) (MailEventPushStatus, error) { |
| 604 | var st MailEventPushStatus |
| 605 | var lastDelivered sql.NullInt64 |
| 606 | if err := s.db.QueryRowContext(ctx, `SELECT |
| 607 | COALESCE(SUM(delivered_at = 0 AND abandoned_at = 0), 0), |
| 608 | COALESCE(SUM(delivered_at > 0), 0), |
| 609 | COALESCE(SUM(abandoned_at > 0), 0), |
| 610 | MAX(NULLIF(delivered_at, 0)) |
| 611 | FROM mail_event_deliveries WHERE account = ? AND app = ?`, acct, app).Scan(&st.Pending, &st.Delivered, &st.Abandoned, &lastDelivered); err != nil { |
| 612 | return st, err |
| 613 | } |
| 614 | if lastDelivered.Valid { |
| 615 | st.LastDelivered = time.Unix(lastDelivered.Int64, 0).UTC() |
| 616 | } |
| 617 | // The most recent attempt that did not land, since the push was turned |
| 618 | // on. The reader decides whether it is news: one older than the last |
| 619 | // delivery has been recovered from. Giving up on a row is not an attempt |
| 620 | // and never writes these columns. |
| 621 | var failedAt int64 |
| 622 | err := s.db.QueryRowContext(ctx, `SELECT last_attempt_at, last_status, last_error FROM mail_event_deliveries |
| 623 | WHERE account = ? AND app = ? AND delivered_at = 0 AND last_attempt_at > 0 AND last_attempt_at >= ? |
| 624 | ORDER BY last_attempt_at DESC, id DESC LIMIT 1`, acct, app, since.Unix()). |
| 625 | Scan(&failedAt, &st.LastStatus, &st.LastError) |
| 626 | switch { |
| 627 | case err == sql.ErrNoRows: |
| 628 | case err != nil: |
| 629 | return st, err |
| 630 | default: |
| 631 | st.LastFailedAt = time.Unix(failedAt, 0).UTC() |
| 632 | } |
| 633 | return st, nil |
| 634 | } |
| 635 | |
| 636 | // ---- producing and sending ---------------------------------------------------------- |
| 637 | |
| 638 | // queueMailEventPush is the ingest's half: queue ev for app if the app has |
| 639 | // asked and the event is one an app is told about. The caller has already |
| 640 | // established that SES named this app as the sender — see the file comment |
| 641 | // for why nothing else is ever queued. Best effort, like the rest of the |
| 642 | // ingest: a failure is logged and the notification still answers 200. |
| 643 | func (s *Server) queueMailEventPush(ctx context.Context, acct, app, notice string, ev mail.Event, now time.Time) { |
| 644 | if !mailPushedTypes[ev.Type] { |
| 645 | return |
| 646 | } |
| 647 | hook, err := s.store.MailEventHookFor(ctx, acct, app) |
| 648 | if err != nil { |
| 649 | log.Printf("console: mail push: reading the hook for %s/%s: %v", acct, app, err) |
| 650 | return |
| 651 | } |
| 652 | if hook == nil || hook.Suspended() { |
| 653 | return |
| 654 | } |
| 655 | body, err := json.Marshal(mailEventPayload(acct, app, notice, ev, now)) |
| 656 | if err != nil { |
| 657 | log.Printf("console: mail push: encoding %s for %s/%s: %v", ev.Type, acct, app, err) |
| 658 | return |
| 659 | } |
| 660 | if err := s.store.EnqueueMailEventDelivery(ctx, acct, app, notice, body, now); err != nil { |
| 661 | log.Printf("console: mail push: queueing %s for %s/%s: %v", ev.Type, acct, app, err) |
| 662 | } |
| 663 | } |
| 664 | |
| 665 | // mailPushURL is where a delivery for app goes. |
| 666 | func (s *Server) mailPushURL(acct, app, path string) (string, error) { |
| 667 | if s.mailPushTarget != nil { |
| 668 | return s.mailPushTarget(acct, app, path), nil |
| 669 | } |
| 670 | domain := strings.ToLower(strings.TrimSpace(s.cfg.AppsDomain)) |
| 671 | if domain == "" { |
| 672 | return "", errors.New("this deployment has no apps domain to deliver to") |
| 673 | } |
| 674 | return "https://" + app + "." + acct + "." + domain + path, nil |
| 675 | } |
| 676 | |
| 677 | // StartMailEventPush runs the queue every mailPushEvery until ctx is done. |
| 678 | // Started from cmd/console like every other pass — never from NewServer, |
| 679 | // which tests construct by the dozen. |
| 680 | func (s *Server) StartMailEventPush(ctx context.Context) { |
| 681 | t := time.NewTicker(mailPushEvery) |
| 682 | defer t.Stop() |
| 683 | for { |
| 684 | if err := s.mailPushPass(ctx, time.Now); err != nil { |
| 685 | log.Printf("console: mail push: %v", err) |
| 686 | } |
| 687 | select { |
| 688 | case <-ctx.Done(): |
| 689 | return |
| 690 | case <-t.C: |
| 691 | } |
| 692 | } |
| 693 | } |
| 694 | |
| 695 | // mailPushPass sends what is due and trims the queue. It starts from the |
| 696 | // apps that asked — a short list — and seeks each one's oldest due rows, so |
| 697 | // its cost does not grow with anyone's backlog. Apps are sent to in |
| 698 | // parallel, mailPushApps at a time, and each app's own rows in order; see |
| 699 | // deliverAppMailEvents for why one app cannot hold up another. clock is read |
| 700 | // per delivery so a slow pass records honest times. |
| 701 | func (s *Server) mailPushPass(ctx context.Context, clock func() time.Time) error { |
| 702 | start := clock() |
| 703 | if err := s.store.pruneMailEventDeliveries(ctx, start); err != nil { |
| 704 | // A prune that fails costs an hour's slack on a thirty-day window, |
| 705 | // not the pass: what is due still goes out. |
| 706 | log.Printf("console: mail push: pruning the queue: %v", err) |
| 707 | } |
| 708 | hooks, err := s.store.activeMailEventHooks(ctx) |
| 709 | if err != nil { |
| 710 | return fmt.Errorf("reading the hooks: %w", err) |
| 711 | } |
| 712 | if len(hooks) == 0 { |
| 713 | return nil |
| 714 | } |
| 715 | var key ed25519.PrivateKey |
| 716 | sem := make(chan struct{}, mailPushApps) |
| 717 | var wg sync.WaitGroup |
| 718 | for _, h := range hooks { |
| 719 | if ctx.Err() != nil { |
| 720 | break |
| 721 | } |
| 722 | if s.cfg.HostClaim { |
| 723 | h.claimAppID, err = s.mutationAppID(ctx, h.Account, h.App) |
| 724 | if err != nil { |
| 725 | continue |
| 726 | } |
| 727 | } |
| 728 | rows, err := s.store.dueMailEventDeliveriesFor(ctx, h.Account, h.App, start, mailPushPerApp) |
| 729 | if err != nil { |
| 730 | log.Printf("console: mail push: reading the queue for %s/%s: %v", h.Account, h.App, err) |
| 731 | continue |
| 732 | } |
| 733 | if len(rows) == 0 { |
| 734 | if err := s.store.clearMailEventFailing(ctx, h); err != nil { |
| 735 | log.Printf("console: mail push: %s/%s: %v", h.Account, h.App, err) |
| 736 | } |
| 737 | continue |
| 738 | } |
| 739 | if key == nil { |
| 740 | if key, err = s.store.MailEventsSigningKey(); err != nil { |
| 741 | wg.Wait() |
| 742 | return fmt.Errorf("reading the signing key: %w", err) |
| 743 | } |
| 744 | } |
| 745 | wg.Add(1) |
| 746 | sem <- struct{}{} |
| 747 | go func() { |
| 748 | defer func() { <-sem; wg.Done() }() |
| 749 | s.deliverAppMailEvents(ctx, h, rows, key, clock) |
| 750 | }() |
| 751 | } |
| 752 | wg.Wait() |
| 753 | return nil |
| 754 | } |
| 755 | |
| 756 | // deliverAppMailEvents sends one app's due rows in order. Any attempt that |
| 757 | // does not land ends the app's turn for this pass: an app that answers 500 |
| 758 | // after nine seconds is no more likely to take the next row than the last, |
| 759 | // and each row it is offered holds one of mailPushApps for a timeout that |
| 760 | // every other app waits on. The rest keep their place for the next pass, and |
| 761 | // how the turn went is noted on the hook — which is where a day of nothing |
| 762 | // but failures suspends the push. |
| 763 | // |
| 764 | // An app that is no longer live, or whose push was turned off since the pass |
| 765 | // began, has all of its rows given up on at once — there is nothing to wait |
| 766 | // for. |
| 767 | func (s *Server) deliverAppMailEvents(ctx context.Context, h mailEventHookRow, rows []mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) { |
| 768 | if !s.cfg.HostClaim { |
| 769 | _ = s.deliverAppMailEventsHeld(ctx, h, rows, key, clock) |
| 770 | return |
| 771 | } |
| 772 | acct, err := s.store.AccountBySqid(h.Account) |
| 773 | if err == nil && acct == nil { |
| 774 | err = hostops.ErrIdentity |
| 775 | } |
| 776 | if err == nil { |
| 777 | currentID, readErr := s.mutationAppID(ctx, h.Account, h.App) |
| 778 | err = readErr |
| 779 | if err == nil && (h.claimAppID == "" || currentID != h.claimAppID) { |
| 780 | err = hostops.ErrIdentity |
| 781 | } |
| 782 | } |
| 783 | if err == nil { |
| 784 | err = s.claimAppMutation(ctx, acct, h.App, "mail-event-delivery", func(held context.Context) error { |
| 785 | if h.claimAppID == "" { |
| 786 | return hostops.ErrIdentity |
| 787 | } |
| 788 | if err := hostops.RequireAppMutation(held, s.cfg.Releases, h.Account, h.App, h.claimAppID); err != nil { |
| 789 | return err |
| 790 | } |
| 791 | // The queue may have changed while this goroutine waited for a slot. |
| 792 | hook, err := s.store.MailEventHookFor(held, h.Account, h.App) |
| 793 | if err != nil { |
| 794 | return err |
| 795 | } |
| 796 | if hook == nil || hook.Suspended() { |
| 797 | return nil |
| 798 | } |
| 799 | h.MailEventHook = *hook |
| 800 | rows, err = s.store.dueMailEventDeliveriesFor(held, h.Account, h.App, clock(), mailPushPerApp) |
| 801 | if err != nil { |
| 802 | return err |
| 803 | } |
| 804 | return s.deliverAppMailEventsHeld(held, h, rows, key, clock) |
| 805 | }) |
| 806 | } |
| 807 | if err != nil { |
| 808 | log.Printf("console: mail push: %s/%s: %v", h.Account, h.App, err) |
| 809 | } |
| 810 | } |
| 811 | |
| 812 | // The admitted context covers delivery and all SQL settlement. A transport |
| 813 | // failure can have reached the app, so its issued effect remains owned. |
| 814 | func (s *Server) deliverAppMailEventsHeld(ctx context.Context, h mailEventHookRow, rows []mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) error { |
| 815 | var result error |
| 816 | if !s.mailAppLive(h.Account, h.App) { |
| 817 | // A trashed app's name is no longer a member's live app, and while it |
| 818 | // sits in the trash its platform host is not reliably its own (the |
| 819 | // instances door treats a trashed sibling's apex as free), so nothing |
| 820 | // is sent to it; a purged, renamed or moved app's rows have already |
| 821 | // been forgotten, carried or dropped. |
| 822 | if err := s.store.abandonMailEventDeliveries(ctx, h.Account, h.App, "the app is not live", clock()); err != nil { |
| 823 | result = errors.Join(result, err) |
| 824 | log.Printf("console: mail push: giving up on %s/%s: %v", h.Account, h.App, err) |
| 825 | } |
| 826 | return result |
| 827 | } |
| 828 | landed, failure := false, "" |
| 829 | for _, d := range rows { |
| 830 | if ctx.Err() != nil { |
| 831 | break |
| 832 | } |
| 833 | status, why, giveUp := s.deliverMailEvent(ctx, d, h.Path, key, clock) |
| 834 | if giveUp { |
| 835 | if err := s.store.abandonMailEventDeliveries(ctx, d.Account, d.App, why, clock()); err != nil { |
| 836 | result = errors.Join(result, err) |
| 837 | log.Printf("console: mail push: giving up on %s/%s: %v", d.Account, d.App, err) |
| 838 | } |
| 839 | return result |
| 840 | } |
| 841 | if err := s.store.recordMailEventAttempt(ctx, d, status, why, clock()); err != nil { |
| 842 | result = errors.Join(result, err) |
| 843 | log.Printf("console: mail push: recording delivery %d for %s/%s: %v", d.ID, d.Account, d.App, err) |
| 844 | } |
| 845 | if s.cfg.HostClaim && status == 0 && why != "" { |
| 846 | result = errors.Join(result, errors.New(why)) |
| 847 | } |
| 848 | if why != "" { |
| 849 | failure = why |
| 850 | break |
| 851 | } |
| 852 | landed = true |
| 853 | } |
| 854 | suspended, err := s.store.noteMailEventTurn(ctx, h, landed, failure, clock()) |
| 855 | if err != nil { |
| 856 | result = errors.Join(result, err) |
| 857 | log.Printf("console: mail push: noting the turn for %s/%s: %v", h.Account, h.App, err) |
| 858 | } |
| 859 | if suspended { |
| 860 | log.Printf("console: mail push: suspended %s/%s — nothing accepted for %s", h.Account, h.App, mailPushSuspendAfter) |
| 861 | } |
| 862 | return result |
| 863 | } |
| 864 | |
| 865 | // deliverMailEvent makes one attempt, and answers the HTTP status, what went |
| 866 | // wrong ("" when it landed), and whether to give up on the row now rather |
| 867 | // than retry it. |
| 868 | func (s *Server) deliverMailEvent(ctx context.Context, d mailEventDelivery, path string, key ed25519.PrivateKey, clock func() time.Time) (int, string, bool) { |
| 869 | hook, err := s.store.MailEventHookFor(ctx, d.Account, d.App) |
| 870 | if err != nil { |
| 871 | return 0, "could not read the hook: " + err.Error(), false |
| 872 | } |
| 873 | if hook == nil || hook.Suspended() { |
| 874 | // Turned off, or suspended, since the pass began. Both give up on the |
| 875 | // queue themselves; this covers the race. |
| 876 | return 0, "push turned off", true |
| 877 | } |
| 878 | if hook.Path != path { |
| 879 | path = hook.Path // moved since the pass began; the new path is the one asked for |
| 880 | } |
| 881 | target, err := s.mailPushURL(d.Account, d.App, path) |
| 882 | if err != nil { |
| 883 | return 0, err.Error(), false |
| 884 | } |
| 885 | rctx, cancel := context.WithTimeout(ctx, mailPushTimeout) |
| 886 | defer cancel() |
| 887 | req, err := http.NewRequestWithContext(rctx, http.MethodPost, target, bytes.NewReader(d.Payload)) |
| 888 | if err != nil { |
| 889 | return 0, "could not build the request: " + err.Error(), false |
| 890 | } |
| 891 | req.Header.Set("Content-Type", "application/json") |
| 892 | req.Header.Set("User-Agent", "carlos-mail-events/1") |
| 893 | req.Header.Set("Carlos-Signature", signMailEvent(key, clock(), d.Payload)) |
| 894 | if s.cfg.HostClaim { |
| 895 | appID, err := s.mutationAppID(ctx, d.Account, d.App) |
| 896 | if err == nil { |
| 897 | err = hostops.RequireAppMutation(ctx, s.cfg.Releases, d.Account, d.App, appID) |
| 898 | } |
| 899 | if err != nil { |
| 900 | return 0, err.Error(), false |
| 901 | } |
| 902 | hostops.MarkAppMutationIssued(ctx) |
| 903 | } |
| 904 | resp, err := mailPushClient.Do(req) |
| 905 | if err != nil { |
| 906 | // The URL is the app's own and not a secret, but the error can be |
| 907 | // long; the reason is what a member needs. |
| 908 | return 0, "no answer: " + strings.TrimPrefix(err.Error(), "Post \""+target+"\": "), false |
| 909 | } |
| 910 | defer resp.Body.Close() |
| 911 | io.Copy(io.Discard, io.LimitReader(resp.Body, mailPushMaxBody)) |
| 912 | if resp.StatusCode >= 200 && resp.StatusCode < 300 { |
| 913 | return resp.StatusCode, "", false |
| 914 | } |
| 915 | if resp.StatusCode >= 300 && resp.StatusCode < 400 { |
| 916 | return resp.StatusCode, "answered with a redirect, which is not followed — POST to the path itself", false |
| 917 | } |
| 918 | return resp.StatusCode, fmt.Sprintf("answered %d", resp.StatusCode), false |
| 919 | } |
| 920 | |
amadan