| 318 | } |
| 319 | |
| 320 | func (a *API) publishSent(ctx context.Context, e webhookpub.Event, outMsg *identity.Message) { |
| 321 | var outboxWrote bool |
| 322 | if a.outbox != nil && outMsg != nil { |
| 323 | // Use deterministic ID so MTA retries (no analog here for |
| 324 | // /send, but rebill of an idempotency key counts as one) |
| 325 | // dedupe through ON CONFLICT (id) DO NOTHING. |
| 326 | e.ID = webhookpub.DeterministicEventID(outMsg.ID, e.Type) |
| 327 | err := a.store.WithTx(ctx, func(tx pgx.Tx) error { |
| 328 | // The messages row is already committed (CreateOutbound- |
| 329 | // Message ran outside this tx because SES already |
| 330 | // accepted the send and the row should be durable even |
| 331 | // if the outbox write fails). So this tx only writes the |
| 332 | // outbox row — best-effort by contract. |
| 333 | outboxWrote = a.outbox.PublishBestEffortTx(ctx, tx, e) |
| 334 | return nil |
| 335 | }) |
| 336 | if err != nil { |
| 337 | log.Printf("[api] outbox tx for email.sent err: %v", err) |
| 338 | outboxWrote = false // tx-level failure also forces fallback |
| 339 | } |
| 340 | } |
| 341 | a.emit().OutboxEventsPublished(e.Type) |
| 342 | if a.shouldFireLegacy(outboxWrote) { |
| 343 | a.publishAsync(e) |
| 344 | } |
| 345 | } |
| 346 | |
| 347 | // publishPendingApproval fires email.pending_review (direction=outbound) |
| 348 | // via the outbox (PublishTx — pre-side-effect: the pending row hasn't been |