| name | landing_page-oban |
| description | Patterns for creating Oban background workers in landing_page_backend — queue configuration, job arguments, retry strategies, and dynamic query processing. |
| metadata | {"author":"Vũ Lưu","version":"2026.03.30","source":"Source code landing_page_backend"} |
Landing Page — Oban Workers
Oban v2.10 background job patterns for async processing.
1. Creating a New Worker
defmodule LandingPage.Oban.SyncPosWorker do
use Oban.Worker,
queue: :sync_pos, # Queue name (atom)
priority: 0, # 0 = highest priority
max_attempts: 1000, # Retry count
tags: ["sync_pos"] # Tags for monitoring
import Ecto.Query, warn: false
alias LandingPage.{Repo, FormData, Tools}
alias LandingPage.Services.PosService
alias LandingPage.Syncs.SyncInfo
alias LandingPage.FormData.PageFormData
@impl Oban.Worker
def perform(%{args: args} = _job) do
%{"page_id" => page_id, "order_id" => order_id} = args
# 1. Fetch required data
sync_info = SyncInfo
|> where([s], s.page_id == ^page_id and not s.is_deleted)
|> Repo.one()
order = case FormData.get_form_data(order_id) do
{:ok, v} -> v
_ -> nil
end
# 2. Process
cond do
sync_info && order ->
sync_info = PosService.get_sync_info_stable(sync_info)
settings = sync_info.settings
# 3. Check for duplicates with dynamic query
pending = if order.phone_number && settings["allow_merge_duplicate_form_data"] do
back_day = NaiveDateTime.utc_now() |> Tools.shift_from_day_to_day(-7)
dyn_query = dynamic([pfd],
pfd.inserted_at > ^back_day and
pfd.phone_number == ^order.phone_number and
pfd.sync_pos_status != ^Enum.SyncPosStatus.value(:synced)
)
count = PageFormData
|> where([pfd], ^dyn_query)
|> select([pfd], count("*"))
|> Repo.one()
count > 0
else
false
end
# 4. Execute or defer
if !pending do
case PosService.update_pos_form_data(page_id, order_id) do
:ok -> {:ok, :success}
{:ok, _} -> {:ok, :success}
e -> {:error, e}
end
else
{:error, :pending}
end
true ->
:ok # Nothing to process
end
end
end
2. Enqueueing Jobs
# Basic enqueue
%{page_id: page_id, order_id: order_id}
|> LandingPage.Oban.SyncPosWorker.new()
|> Oban.insert()
# With schedule (delay)
%{page_id: page_id, order_id: order_id}
|> LandingPage.Oban.SyncPosWorker.new(schedule_in: 60) # 60 seconds
|> Oban.insert()
# With priority override
%{page_id: page_id}
|> LandingPage.Oban.SyncPosWorker.new(priority: 3)
|> Oban.insert()
# Inside Ecto.Multi (transactional)
Multi.new()
|> Multi.run(:resource, fn _, _ -> create_resource(attrs) end)
|> Multi.run(:oban_job, fn _, %{resource: resource} ->
%{resource_id: resource.id}
|> MyWorker.new()
|> Oban.insert()
end)
|> Repo.transaction()
3. Worker Return Values
| Return | Behavior |
|---|
:ok | Job completed successfully |
{:ok, value} | Job completed successfully |
{:error, reason} | Job failed — sẽ retry theo max_attempts |
{:cancel, reason} | Job cancelled — KHÔNG retry |
{:snooze, seconds} | Delay rồi retry sau seconds giây |
4. Existing Queues
| Queue | Purpose |
|---|
:sync_pos | Sync form data to POS systems |
:sync_sheet | Sync to Google Sheets |
:sync_crm | Sync to CRM services |
:email | Send emails |
:image_processing | Image optimization/background removal |
:page_publish | Page publishing pipeline |
5. Important Patterns
- Args phải serializable: Chỉ dùng strings, numbers, maps, lists trong args. KHÔNG dùng structs hay atoms.
- Idempotent: Worker phải handle được việc chạy lại (duplicate-safe).
- Dynamic queries trong worker: Dùng
dynamic() macro cho conditional filtering (xem ví dụ trên).
- Hostname-based distribution: Một số workers chỉ chạy trên specific servers dựa vào hostname.
# Trong application.ex — conditional worker start
{:ok, host} = :inet.gethostname()
strhost = :erlang.iolist_to_binary(host)
children = if strhost == "lp-backend-01" do
children ++ [LandingPage.Scheduler]
else
children
end