Oban worker for polling the Brevo transactional-email events API.

Mirrors SQSPollingJob's self-scheduling chain (see that module's @moduledoc for the general shape/rationale — the self-reschedule dedup, unique window, and misconfig back-off all apply identically here) with two differences dictated by Brevo's API instead of SQS's:

• Sender-aware gate: SQS polls unconditionally once enabled; Brevo only fetches events while at least one enabled PhoenixKit.Email.SendProfile actually points at a "brevo_api" integration — an idle Brevo integration with no active profile has nothing to correlate events against, so polling it is pure waste (and, unlike SQS's long-poll, would burn Brevo API quota on a timer). The chain still keeps re-scheduling itself while brevo_events_enabled stays on even when there are zero active profiles right now — a profile added later must not require a manual re-trigger to be picked up.
• No message ack step: SQS deletes each message after processing to avoid redelivery. Brevo's events endpoint is a plain paginated report with no consumption side-effect — this job tracks its own read position instead (see "Watermark cursor" below). Re-processing an already-seen event is still safe and cheap regardless: Event.create_event/1's partial unique indexes (see Event.changeset/2) make it a no-op, not a duplicate row — the watermark exists for performance and completeness on top of that existing guarantee, not to provide it.

Watermark cursor

Brevo's startDate/endDate query params only accept whole calendar dates (YYYY-MM-DD) — there is no "give me events after this exact timestamp/offset" primitive. An early version of this job re-fetched the fixed [yesterday, today] window from offset: 0 every cycle, which has two problems at volume: every already-processed event pays the full re-fetch + dedup-lookup cost again, and a sender producing more than @max_pages_per_integration * page_limit() events in that window (25,000 by default) can never reach its newest events — the page cap is always hit before the tail of an ever-re-fetched window is reached.

Each integration's progress is now tracked as a persisted %{date: Date.t(), offset: integer} watermark (Emails.get_brevo_watermark/1 / set_brevo_watermark/3, stored via PhoenixKit.Settings' JSON-by-prefix helpers — no dedicated table). Every request queries a single day (startDate == endDate == watermark.date), never a growing window — this is the key property that makes offset mean the same thing across cycles. (A naive "just remember the offset" fix without this breaks at every day boundary: if startDate is recomputed as yesterday() each cycle the way the old code did, the window silently slides forward by one day every day, and a stale offset from window [D-1, D] gets replayed against the unrelated window [D, D+1] the next cycle — same numeric offset, different result set entirely.)

A day only advances (offset resets to 0, date moves to the next day) once a short page (fewer than page_limit() events) is seen for it and it's strictly before today. Today's date never auto-closes from a short page alone — more events (opens/clicks/bounces on earlier sends) can land on it any time before midnight, and closing it early would permanently skip them (nothing ever re-visits a closed day by design — see the trailing re-check below for the one exception). The watermark is persisted after every page (not once at the end of a cycle), so a crash mid-cycle replays at most one page of already-processed (dedup-absorbed) events, not the whole cycle.

Trailing safety re-check

Whenever the watermark's date equals today — meaning the forward walk has already closed yesterday and moved on, and (unlike a still- lagging backlog) will never fetch yesterday again on its own — this job additionally re-scans yesterday relative to right now (Date.add(today, -1)) from offset: 0 before advancing the watermark: a fixed one-day lookback with no cursor of its own (it never persists a position; there's nothing to remember, it always starts over at 0). This guards two distinct risks that both surface as "an event that belongs to a day the watermark has already closed and will never revisit":

1. Indexing lag. Brevo's event log has been observed to lag real time by roughly 30–60 seconds. An event that occurs at 23:59:50 can be indexed on Brevo's side a moment after midnight — by which point the watermark may have already closed yesterday with a short page that didn't yet include it.
2. startDate/endDate timezone ambiguity. Brevo's API documentation states the returned date field is UTC, but does not document which timezone startDate/endDate are interpreted in — plausibly the receiving Brevo account's configured timezone rather than UTC. If so, this job's UTC-computed today() and Brevo's own day boundary can disagree by up to (but never more than) one calendar day — at any given instant, at most two calendar dates are "current" anywhere on Earth, so a one-day lookback is a sufficient, not merely heuristic, margin here. Chosen deliberately: extend the re-checked window rather than guess at a specific timezone (nothing in Brevo's docs settles it either way).

A one-day lookback is a margin against both risks above as observed — sub-minute indexing lag, and at most a one-day timezone offset — not a general guarantee against arbitrarily late indexing. An event indexed more than 24h after it occurred would land on a day this re-check no longer reaches (that day has since closed too, and closed days are never revisited by design — see above). This is an accepted trade-off given the lag actually observed, not a gap this job tries to close.

The re-check itself is capped at its own fixed @trailing_recheck_page_budget (2 pages), independent of the forward walk's budget — see "Per-cycle event cap" for why, and for the resulting trade-off: on a day whose total event count exceeds what that budget can page through (starting over at offset 0 every cycle — it keeps no cursor of its own), a late-indexed event sorted past that point won't be reached by the re-check either, on this or any later cycle. Widening this budget (or giving the re-check a remembered starting offset instead of always restarting at 0) would close that gap at the cost of the "own small budget" property this section exists to have — not done here; see the "Per-cycle event cap" section for the reasoning that led to a small fixed budget over an unbounded one.

The watermark.date == today gate isn't a cost-cutting shortcut — skipping it in every other state is required for correctness of a different kind: cold start's forward walk also begins at {Date.add(today, -1), 0} (see below), and a still-lagging backlog's forward walk will reach yesterday itself in due course either way — re-scanning it here too would be the exact same request twice in one cycle, not extra safety. Once the gate condition is true it stays true, and gets re-checked, every cycle for as long as the forward walk sits on today (i.e. once a day, renewed daily as "today" itself advances) — not a one-time check right after the boundary, so there's no grace-window length to tune. It runs first when it runs at all, ahead of the forward watermark walk (see "Per-cycle event cap") — but on its own small fixed budget, not a share of the forward walk's, so it can never starve the forward walk regardless of how large yesterday turned out to be.

Stale watermark cleanup

A watermark is keyed by integration uuid and outlives the integration itself unless cleaned up — perform/1 prunes any stored watermark whose integration uuid is no longer in the current cycle's active set (active_brevo_integrations/0 — deleted, or explicitly excluded via Emails.brevo_polling_excluded_integrations/0) before running the cycle. Excluding an integration and later re-including it starts it over from cold-start rather than resuming a possibly very stale position — the same trade-off a first-time poll already makes, not a new one.

Multi-integration

More than one Brevo Integrations connection can exist (e.g. two distinct Brevo accounts, each with its own SendProfile). Each cycle polls every distinct integration referenced by an enabled Brevo profile — not once per profile, since profiles may share an integration and its API key. If any integration in the cycle turns out to be misconfigured (missing/invalid credentials), the whole cycle's next poll backs off to @misconfig_backoff_ms rather than the configured interval — same rationale as SQSPollingJob, and a deliberate choice, not an oversight: a per-integration backoff would need per-integration scheduling state this self-scheduling chain doesn't have, and a broken integration is expected to be rare/ transient enough that briefly slowing the whole cycle is an acceptable trade for the simplicity of one shared interval.

Per-cycle event cap

Each integration is capped at @max_pages_per_integration (10) pages of @default_page_limit (2500, overridable — see page_limit/0) events per cycle — at most 25,000 events per integration per poll. A sender busy enough to exceed that in one polling_interval_ms window will lag (the cap logs a warning and picks up the remainder next cycle, from the correct watermark position rather than from the start) rather than the poll cycle growing unbounded. A backlog spanning several low-volume days can close out more than one of them in a single cycle — the cap is on total pages fetched, not on how many distinct days get walked.

This budget is not shared with the trailing safety re-check — that has its own separate, much smaller @trailing_recheck_page_budget (2 pages). It used to draw from this same cap, which meant a single very high-volume yesterday could exhaust the entire cap on the re-check alone (it restarts from offset 0 every cycle, re-reading pages it already re-confirmed last cycle) and leave nothing for the forward walk to advance today with — stale today data every cycle for as long as that backlog lasted, not just a one-off. Giving the re-check its own small budget instead means it can only ever cost a couple of pages, at the trade-off described in "Trailing safety re-check" above (it may not reach the tail of an unusually large day).

Oban queue configuration

Requires a :brevo_polling Oban queue in the host app's config, same as :sqs_polling already does:

config :your_app, Oban,
  queues: [
    brevo_polling: 1  # concurrency MUST stay 1 — the self-scheduling
                       # chain assumes only one cycle runs at a time;
                       # see SQSPollingJob's queue config docs for why
  ]