Oban worker for polling AWS SQS queue for email events.

This is the sole SQS poller: an Oban-based approach that allows dynamic enabling/disabling without an application restart (see SQSPollingManager).

Architecture

AWS SES → SNS Topic → SQS Queue → SQSPollingJob (Oban) → SQSProcessor → Database

Multi-account cycle

One Oban chain, N accounts polled inside each cycle — the shape BrevoPollingJob already uses, deliberately copied rather than generalised. An "account" here is an aws_ses Integrations connection referenced by an enabled SendProfile (see PhoenixKit.Modules.Emails.AwsIntegrations), carrying its OWN queue URL and its OWN credentials — an SES account can only publish events to a queue it owns, so polling account A's queue with account B's keys is not a degraded mode, it is silence.

Per-account pipeline settings live in aws_tracking:<integration_uuid> (see Emails.get_aws_tracking/1). Two paths stay alive underneath:

• No active aws_ses SendProfile at all — the legacy single-queue deployment (env-var or plain-Settings credentials, no profile system). The global aws_sqs_queue_url + get_aws_* credentials are polled exactly as before.
• An active account with no aws_tracking: entry of its own — inherits the global aws_sqs_queue_url, but only when the globals can be attributed to it unambiguously: it is the account emails_aws_integration_uuid selects, or it is the only active account. With two-plus unattributed accounts the globals describe one queue and nothing says whose, so those accounts are skipped (and logged) until the operator configures them per-account, rather than having several accounts race on one queue with mismatched keys.

A misconfigured account (credentials that no longer resolve) backs the WHOLE cycle off to @misconfig_backoff_ms, matching BrevoPollingJob — a deliberate simplification: one shared cadence, no per-account state. An account merely missing a queue URL is not "misconfigured" in that sense — it is skipped at full cadence, since a half-configured second account must not slow event delivery for a working first one.

Cycle duration with N accounts

Accounts are polled sequentially and each long-poll waits up to @default_long_poll_timeout seconds, so a quiet cycle over N accounts takes about N x 20s. That does NOT put the self-scheduling chain at risk: this worker does not override Oban.Worker.timeout/1, and Oban's default is :infinity — there is no deadline that could kill the job mid-cycle and hand the chain to Oban's retry machinery. (Adding one would be actively harmful: a killed cycle is retried by Oban WHILE schedule_next_poll/1 has already queued a successor, which is how you get two chains.)

What it does cost is latency and a queue slot: sqs_polling is concurrency 1, so the next tick starts one interval after this cycle ENDS, not after it began. With a handful of accounts that is the difference between a 5s and a ~25s effective cadence — acceptable, and self-correcting in the sense that a busy queue returns immediately rather than waiting out the long-poll. If it ever stops being acceptable, the fix is parallelism or one chain per account (phase 2), not a timeout.

Features

• Dynamic Configuration: Automatically responds to settings changes without restart
• Oban Integration: Uses Oban's job system for reliable background processing
• Self-Scheduling: Each job schedules the next polling cycle
• Batch Processing: Process up to 10 messages at a time
• Error Handling: Retry logic with Dead Letter Queue
• Settings-Based Control: Polling can be enabled/disabled via Settings

Configuration

All settings are retrieved from PhoenixKit Settings:

• sqs_polling_enabled - enable/disable polling (checked before each cycle)
• sqs_polling_interval_ms - interval between polling cycles
• sqs_max_messages_per_poll - maximum messages per batch
• sqs_visibility_timeout - time for message processing
• aws_sqs_queue_url - SQS queue URL
• aws_region - AWS region

Usage

# Enable polling (starts first job)
PhoenixKit.Modules.Emails.SQSPollingManager.enable_polling()

# Disable polling (stops scheduling new jobs)
PhoenixKit.Modules.Emails.SQSPollingManager.disable_polling()

# Trigger immediate polling
PhoenixKit.Modules.Emails.SQSPollingManager.poll_now()

# Check status
PhoenixKit.Modules.Emails.SQSPollingManager.status()

Oban Queue Configuration

Add to your config/config.exs:

config :your_app, Oban,
  repo: YourApp.Repo,
  queues: [
    sqs_polling: 1  # Only one concurrent polling job
  ]

Implementation Notes

• Uses Oban's own unique: (not a manual delete-then-insert) to keep exactly one future job queued — see the unique: option below for the full reasoning; it's the subtle part of this module.
• Schedules next job only if polling is enabled
• Uses SQSProcessor for event processing

Why unique: [period: :infinity, states: [:scheduled]], not more

This looks under-specified at first glance — Oban's own unique-states groups (:incomplete, or the full default) include :executing, :available, :retryable too. Each of those is excluded here for a concrete reason, not an oversight:

• :executing must never be in this worker-level list. The chain works by a currently-:executing job inserting its own successor from inside perform/1. Oban marks a job :executing in the DB before calling perform/1, and its unique-conflict check has no self-exclusion — an insert of the same worker/args while :executing is in states finds that job's own row as the "existing" match and no-ops against it instead of creating a new :scheduled row. The chain would silently stop advancing every single cycle, not just on a crash. (A job orphaned in :executing by a hard kill would make this permanent, on top of merely stalling per-cycle.)
• A wider list (e.g. adding :available) fails the build. Oban's use Oban.Worker, unique: [...] option is checked at compile time (Oban.Worker.__after_compile__/2 → Job.warn_unique/1): any :states list other than the exact literal [:scheduled] that doesn't cover every "incomplete" state (:scheduled, :available, :executing, :retryable, :suspended) emits a compiler warning ("may break uniqueness"), which --warnings-as-errors turns into a build failure. [:scheduled] alone is Oban's own special-cased exception to that check (see Job.warn_unique/1) — it is the only partial list that both compiles clean and excludes :executing.
• [:scheduled] is exactly enough for THIS insert: self-scheduled jobs always land in :scheduled (interval >= 1000ms ⇒ schedule_in >= 1s, never :available), so it dedups the self-reschedule chain against itself with no manual delete step. period: :infinity (not a short window) makes that hold regardless of how long the configured interval is — a short window only caught near-simultaneous double inserts; the old delete-then-insert dance existed specifically to catch the case a short window couldn't (two chains a full interval apart). An unconditional :infinity unique check removes the need for that dance entirely.
• The immediate job from SQSPollingManager.enable_polling/0 / poll_now/0 is a different insert with its own per-call unique:/replace: override (states: [:available, :scheduled]) — see SQSPollingManager's insert_poll_job/0 and insert_forced_poll_job/0. A per-call override on new/2 does NOT go through the compile-time check above (only the worker-level use default does), so it's free to cover :available too. It still excludes :executing for the same self-conflict reason. poll_now/0 additionally carries args: %{"forced" => true}, which puts it in a separate uniqueness namespace (Oban matches on args) so a manual poll never moves the regular chain's next tick.
• Concurrency is capped at 1 by the queue, so parallel execution is already impossible; unique: is what prevents parallel chains.