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

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.