This guide begins where the Architecture guide stops. Hold one route in your head: a host asks for a direct subscription, Accrue persists the resulting local projection, and a later webhook converges that projection before entitlements read it.
The excerpts below are shortened from current source. A # ... marks every deliberate cut.
Internal modules and private functions are shown to explain the design; they are not promised public
API. Integrations should enter through the documented host facade and Accrue.Billing surface.
1. The host owns the front door
The generated facade begins as a delegation point. In a real application, authorization and product policy accumulate here while processor normalization stays in core.
# ...
alias Accrue.Billing
# Host policy hook: authorize who may subscribe this billable.
def subscribe(billable, price_id, opts \\ []) do
Billing.subscribe(billable, price_id, opts)
end
# ...Notice that the billable is still a host value. A controller or LiveView need not know how it maps to a processor customer.
2. The library validates, then supervises nothing
Accrue's OTP application performs safety checks before any billing work. Its supervisor deliberately has no children because the host owns Repo, Oban, HTTP clients, and optional PDF processes.
# ...
@impl true
def start(_type, _args) do
:ok = Accrue.Config.validate_at_boot!()
:ok = Accrue.Auth.Default.boot_check!()
:ok = warn_on_secret_collision()
:ok = warn_pdf_adapter_unavailable()
:ok = warn_oban_queue_vs_pdf_pool()
:ok = warn_company_address_locale_mismatch()
children = []
Supervisor.start_link(children, strategy: :one_for_one, name: Accrue.Supervisor)
end
# ...Boot failure is preferable to discovering an invalid adapter or missing safety boundary after a money-moving request begins.
3. Configuration fails before billing starts
Accrue.Config validates only keys it owns. Adapter-specific keys can coexist in the application
environment without being mistaken for core options.
# ...
@spec validate_at_boot!() :: :ok
def validate_at_boot! do
known_keys = Keyword.keys(@schema)
opts =
:accrue
|> Application.get_all_env()
|> Keyword.take(known_keys)
_ = NimbleOptions.validate!(opts, @schema)
_ = maybe_validate_boot_setup!(opts)
:ok
end
# ...This is the configuration half of the ownership boundary: Accrue defines what must be true; the host chooses values and starts the required infrastructure.
4. The public operation adds observability
The public context is intentionally small at the point of entry. Its added value here is a named billing span and a stable facade; the detailed state transition lives elsewhere.
# ...
def subscribe(user, price_id_or_opts \\ [], opts \\ []) do
span_billing(:subscription, :create, user, opts, fn ->
SubscriptionActions.subscribe(user, price_id_or_opts, opts)
end)
end
def subscribe!(user, price_id_or_opts \\ [], opts \\ []) do
span_billing(:subscription, :create, user, opts, fn ->
SubscriptionActions.subscribe!(user, price_id_or_opts, opts)
end)
end
# ...The non-raising and raising variants share the same telemetry boundary.
5. A billable becomes a customer
Subscription actions accept either an existing local customer or a host billable. The latter takes the lazy customer path before the direct-create operation continues.
# ...
def subscribe(billable, price_spec, opts \\ [])
def subscribe(%Customer{} = customer, price_spec, opts) do
do_subscribe(customer, price_spec, opts)
end
def subscribe(billable, price_spec, opts) do
with {:ok, customer} <- Accrue.Billing.customer(billable) do
do_subscribe(customer, price_spec, opts)
end
end
# ...That split keeps host identity concerns out of adapter request assembly.
6. One function establishes support, identity, and atomicity
This is the outbound journey's center. It checks the declared support slice, derives a deterministic
key, calls the adapter, and turns the response into one transactional local outcome. The source still
uses the historical variable name stripe_sub; the configured adapter is the actual authority.
# ...
defp do_subscribe(%Customer{} = customer, price_spec, opts) do
with :ok <- ensure_subscribe_support() do
do_subscribe_supported(customer, price_spec, opts)
end
end
defp do_subscribe_supported(%Customer{} = customer, price_spec, opts) do
{price_id, quantity} = normalize_price_spec(price_spec)
op_id = resolve_operation_id(opts)
idem_key =
Idempotency.key(
:create_subscription,
customer.id,
op_id,
subscribe_sequence(price_id, quantity, opts)
)
{item_params, trial_end} = build_subscribe_params({price_id, quantity}, opts)
result =
with {:ok, processor_params} <-
build_subscription_request(customer, item_params, trial_end, opts),
_result <- :ok do
case ensure_customer_tax_location(customer, opts) do
:ok ->
case Processor.__impl__().create_subscription(
processor_params,
[idempotency_key: idem_key] ++ sanitize_opts(opts)
) do
{:ok, stripe_sub} ->
Repo.transact(fn ->
with :ok <- ensure_valid_tax_location(stripe_sub, opts),
{:ok, attrs} <- SubscriptionProjection.decompose(stripe_sub),
{:ok, sub} <- insert_subscription(customer.id, attrs),
{:ok, _items} <- upsert_items(sub, stripe_sub),
{:ok, _} <-
record_event("subscription.created", sub, %{price_id: price_id}) do
sub = Repo.preload(sub, :subscription_items, force: true)
{:ok, sub}
end
end)
# ...
end
# ...
end
end
IntentResult.wrap(result)
end
# ...The omitted branches release reserved discount state and return errors. They do not create an
alternate success path. IntentResult.wrap/1 promotes payment authentication into the public
action-required shape.
7. The processor behaviour is a narrow waist
The behaviour defines the value crossing the boundary. Runtime resolution and capability labels let the same core choose Stripe, Braintree, Fake, or a custom adapter without claiming they support the same slices.
# ...
@callback create_subscription(params(), opts()) :: result()
@callback retrieve_subscription(id(), opts()) :: result()
@callback update_subscription(id(), params(), opts()) :: result()
@callback cancel_subscription(id(), opts()) :: result()
# ...
@doc false
@spec __impl__() :: module()
def __impl__, do: Application.get_env(:accrue, :processor, Accrue.Processor.Fake)
# ...
@spec first_party_supported?(atom() | [atom()]) :: boolean()
def first_party_supported?(path) when is_atom(path), do: first_party_supported?([path])
def first_party_supported?(path) when is_list(path),
do: Accrue.Processor.Capabilities.first_party_supported?(capabilities(), path)
# ...Fake is an executable contract for deterministic tests and demos. It is not evidence that remote authentication, vaulting, webhook naming, or provider failures are interchangeable.
8. Stripe implements the direct-create contract
The Stripe adapter adds the expansion needed for SCA inspection, derives processor options, calls LatticeStripe, and translates its response into Accrue's map-shaped adapter result.
# ...
@impl Accrue.Processor
def create_subscription(params, opts) when is_map(params) and is_list(opts) do
client = build_client!(opts)
params = ensure_expand(params, ["latest_invoice.payment_intent"])
stripe_opts = stripe_opts(:create_subscription, subject_of(params, "sub"), opts)
client
|> LatticeStripe.Subscription.create(stringify_keys(params), stripe_opts)
|> translate_resource()
end
# ...Braintree's direct-create implementation is intentionally not excerpted here. Its supported path requires a host-acquired vault reference, while Stripe accepts the Stripe-shaped request assembled by subscription actions.
9. Processor data becomes a local projection
Projection code is where provider response shape stops leaking. It selects a provider-specific normalizer and produces attributes accepted by the local subscription schema.
# ...
@spec decompose(map(), keyword()) :: {:ok, map()}
def decompose(subscription, opts \\ [])
def decompose(subscription, opts) when is_map(subscription) and is_list(opts) do
processor = Keyword.get(opts, :processor, processor_atom())
case processor do
:paddle -> {:ok, paddle_attrs(subscription)}
:braintree -> {:ok, braintree_attrs(subscription)}
_ -> {:ok, stripe_attrs(subscription)}
end
end
defp stripe_attrs(stripe_sub) do
automatic_tax = automatic_tax_fields(get(stripe_sub, :automatic_tax))
%{
processor_id: get(stripe_sub, :id),
status: parse_status(get(stripe_sub, :status)),
cancel_at_period_end: get(stripe_sub, :cancel_at_period_end) || false,
pause_collection: parse_pause_collection(get(stripe_sub, :pause_collection)),
automatic_tax: automatic_tax.enabled,
automatic_tax_status: automatic_tax.status,
automatic_tax_disabled_reason: automatic_tax.disabled_reason,
current_period_start: unix_to_dt(get(stripe_sub, :current_period_start)),
current_period_end: unix_to_dt(get(stripe_sub, :current_period_end)),
# ...
data: normalize_data(stripe_sub),
metadata: get(stripe_sub, :metadata) || %{}
}
end
# ...The full raw map remains normalized into data for round-trip safety, but named fields give local
queries stable lifecycle semantics.
10. Lifecycle is more than a status atom
The schema stores remote identity, lifecycle overlays, and ordering watermarks. The entitlement predicate composes the meaningful conditions instead of asking product code to repeat them.
# ...
schema "accrue_subscriptions" do
belongs_to(:customer, Accrue.Billing.Customer)
field(:processor, :string)
field(:processor_id, :string)
field(:status, Ecto.Enum, values: @statuses)
field(:cancel_at_period_end, :boolean, default: false)
field(:pause_collection, :map)
field(:ended_at, :utc_datetime_usec)
field(:last_stripe_event_ts, :utc_datetime_usec)
field(:last_stripe_event_id, :string)
# ...
end
# ...
@spec entitling?(%__MODULE__{} | map()) :: boolean()
def entitling?(sub), do: active?(sub) and not paused?(sub) and not canceled?(sub)
# ...The last_stripe_event_* names are legacy Stripe-shaped column names even when another adapter owns
the row. They are local ordering metadata, not proof that Stripe is always configured.
11. Webhook receipt verifies raw bytes
The inbound journey starts from bytes retained before body parsing. A missing signature is an error, not a development-mode bypass.
# ...
defp do_call(conn, :stripe, endpoint) do
raw_body = flatten_raw_body(conn)
sig_header = get_req_header(conn, "stripe-signature") |> List.first()
unless sig_header do
raise Accrue.SignatureError, reason: "missing stripe-signature header"
end
secrets = resolve_secrets!(endpoint, :stripe)
stripe_event = Signature.verify!(raw_body, sig_header, secrets)
# Transactional persist + Oban enqueue.
Accrue.Webhook.Ingest.run(conn, :stripe, stripe_event, raw_body, endpoint)
end
# ...Only a verified event reaches ingest. Endpoint-aware secret lookup also fails closed when the named endpoint is not configured.
12. Receipt becomes three durable facts
Ingest commits the evidence row, dispatch job, and received ledger event together. The duplicate branch returns the already stored row and skips both secondary writes.
# ...
result =
Accrue.Repo.transact(fn repo ->
case persist_event(repo, processor_str, stripe_event, raw_body, endpoint_atom) do
{:ok, {:duplicate, _} = duplicate} ->
{:ok, duplicate}
{:ok, {:new, row} = persisted} ->
with {:ok, _job} <- repo.insert(DispatchWorker.new(%{webhook_event_id: row.id})),
{:ok, _event} <- record_received_event(processor_str, stripe_event, row) do
{:ok, persisted}
else
{:error, reason} -> {:error, reason}
end
{:error, reason} ->
{:error, reason}
end
end)
case result do
{:ok, {:new, _}} ->
conn |> send_resp(200, Jason.encode!(%{ok: true})) |> halt()
{:ok, {:duplicate, _}} ->
conn |> send_resp(200, Jason.encode!(%{ok: true})) |> halt()
{:error, reason} ->
Logger.error("Webhook ingest failed: #{inspect(reason, limit: 200)}")
conn |> send_resp(500, Jason.encode!(%{ok: false})) |> halt()
end
# ...The HTTP response can be retried safely because uniqueness is enforced at persistence, not inferred from timing in the Plug process.
13. The queued event is intentionally lean
The worker projection extracts only the identity required to route and refetch. This excerpt also
exposes two current constraints: queued created_at is receipt time, and the bounded processor map
does not include Braintree.
# ...
@processor_atoms %{
"stripe" => :stripe,
"stripe_connect" => :stripe_connect,
"fake" => :fake
}
# ...
@spec from_webhook_event(Accrue.Webhook.WebhookEvent.t()) :: t()
def from_webhook_event(%Accrue.Webhook.WebhookEvent{} = row) do
object_id =
case row.data do
%{"data" => %{"object" => %{"id" => id}}} -> id
_ -> nil
end
%__MODULE__{
type: row.type,
object_id: object_id,
livemode: row.livemode,
created_at: row.received_at,
processor_event_id: row.processor_event_id,
processor: processor_to_atom(row.processor)
}
end
# ...Consequently, canonical refetch is the queued path's primary convergence guarantee. Braintree receipt persists, but queued conversion currently raises before its dedicated handler can run. Direct Braintree subscription creation does not use this dispatch conversion and is unaffected.
14. Dispatch separates core retries from user extensions
The worker records status, establishes webhook actor context, runs the non-disableable reducer first, then isolates user handlers. Only a built-in handler failure drives the 25-attempt retry and eventual dead-letter lifecycle.
# ...
use Oban.Worker,
queue: :accrue_webhooks,
max_attempts: 25
# ...
event = Event.from_webhook_event(row)
ctx = %{
attempt: attempt,
max_attempts: max_attempts,
webhook_event_id: id,
endpoint: row.endpoint,
meter_error_object: meter_error_object
}
Accrue.Actor.put_current(%{type: :webhook, id: row.processor_event_id})
default_handler =
case row.endpoint do
:connect -> ConnectHandler
_ -> DefaultHandler
end
default_result = safe_handle(default_handler, event, ctx)
_user_results =
Accrue.Config.webhook_handlers()
|> Enum.map(fn handler -> safe_handle(handler, event, ctx) end)
case default_result do
:ok ->
mark_succeeded(repo, row)
:ok
{:error, reason} ->
mark_failed_or_dead(repo, row, attempt, max_attempts)
{:error, reason}
end
# ...
def safe_handle(handler, event, ctx) do
handler.handle_event(event.type, event, ctx)
rescue
e ->
Logger.error(
"Webhook handler #{inspect(handler)} crashed: #{Exception.format(:error, e, __STACKTRACE__)}"
)
:telemetry.execute(
[:accrue, :webhook, :handler, :exception],
%{},
%{module: handler, error: e}
)
{:error, e}
end
# ...The rescue contains the failure locally, logs it, and emits telemetry before returning the error tuple to the caller that decides whether the failure is retryable.
15. Reduction checks order, then refetches truth
Subscription reduction runs inside the shared transactional wrapper. A strictly older watermark returns without a processor call. Otherwise the reducer refetches canonical state, projects it, updates items, and appends its domain event before the transaction completes.
# ...
defp reduce_subscription(action, evt_id, evt_ts, obj) do
stripe_id = get(obj, :id)
reduce_row(:subscription, stripe_id, evt_ts, evt_id, fn row ->
with {:ok, canonical} <- Processor.__impl__().fetch(:subscription, stripe_id),
{:ok, attrs} <- SubscriptionProjection.decompose(canonical),
attrs <- stamp_watermark(attrs, evt_ts, evt_id),
{:ok, upsert_result} <- upsert_subscription(row, canonical, attrs) do
case upsert_result do
:deferred ->
{:ok, :deferred}
%Subscription{} = updated ->
with {:ok, _} <- upsert_subscription_items(updated, canonical),
:ok <- maybe_emit_dunning_exhaustion(row, updated, canonical),
:ok <- maybe_finalize_dunning_campaign(row, updated, canonical),
{:ok, _} <-
record_event(
subscription_event_type(action),
"Subscription",
updated.id,
evt_id
) do
{:ok, updated}
end
end
end
end)
end
# ...
defp reduce_row(object_type, stripe_id, evt_ts, evt_id, fun) do
Repo.transact(fn ->
row = load_row(object_type, stripe_id)
case check_stale(row, evt_ts) do
:stale ->
:telemetry.execute(
[:accrue, :webhooks, :stale_event],
%{},
%{object_type: object_type, stripe_id: stripe_id, event_id: evt_id}
)
{:ok, :stale}
:ok ->
fun.(row)
end
end)
end
# ...For queued events, evt_ts is receipt time because of the previous projection. For the raw/Fake
entry point, it can be processor-created time. Either way, the reducer never treats the event's
object snapshot as the canonical current subscription.
16. Entitlements read the projection and fail closed
The default resolver performs a local query only. Query.entitling/1 is the SQL twin of the
subscription predicate; joining items yields the processor price IDs and quantities that the
configured plan catalog can fold.
# ...
defp none_lane_items(customer_id) do
Subscription
|> Query.entitling()
|> where([s], s.customer_id == ^customer_id)
|> join(:inner, [s], i in SubscriptionItem, on: i.subscription_id == s.id)
|> select([_s, i], {i.price_id, i.quantity})
|> Accrue.Repo.all()
|> Enum.map(fn {price_id, quantity} -> {price_id, quantity, false} end)
end
# ...No customer, no qualifying subscription, no catalog mapping, or a resolver failure all produce a
deny result at the public Accrue.Entitlements boundary. Admin may diagnose the same local fold, but
it does not become authoritative for access.
17. The duplicate test states the ingest contract
The most useful tests read like architecture. This one proves that HTTP success on a retry does not mean duplicated durable work.
# ...
test "duplicate POST returns 200 with no second row or Oban job" do
{body, _sig} = signed_event()
stripe_event = build_lattice_event(body)
conn1 = Plug.Test.conn(:post, "/webhook/stripe")
result1 = Ingest.run(conn1, @processor, stripe_event, body)
assert result1.status == 200
# Second call with same event
conn2 = Plug.Test.conn(:post, "/webhook/stripe")
result2 = Ingest.run(conn2, @processor, stripe_event, body)
assert result2.status == 200
# Still only one row
events = Accrue.TestRepo.all(WebhookEvent)
assert length(events) == 1
# Only one Oban job (from first call)
jobs = Accrue.TestRepo.all(Oban.Job)
assert length(jobs) == 1
end
# ...The database uniqueness contract, not a mocked call count, is the behavior under test.
18. The ordering test states the reducer contract
The raw/Fake reducer test pins the strict comparison: an older processor-created timestamp emits stale telemetry and cannot move the watermark backward.
# ...
test "older event is skipped with :stale_event telemetry and no refetch", %{sub: sub} do
newer_ts = DateTime.add(Accrue.Clock.utc_now(), 3600, :second)
{:ok, _} =
sub
|> Subscription.changeset(%{
last_stripe_event_ts: newer_ts,
last_stripe_event_id: "evt_new"
})
|> Repo.update()
test_pid = self()
:telemetry.attach(
"test-stale-#{System.unique_integer([:positive])}",
[:accrue, :webhooks, :stale_event],
fn evt, meas, meta, _ -> send(test_pid, {:stale, evt, meas, meta}) end,
nil
)
older_event =
StripeFixtures.webhook_event(
"customer.subscription.updated",
StripeFixtures.subscription_created(%{"id" => sub.processor_id}),
%{
"id" => "evt_older",
"created" => DateTime.to_unix(DateTime.add(newer_ts, -1800, :second))
}
)
DefaultHandler.handle(older_event)
assert_received {:stale, _, _, %{event_id: "evt_older"}}
unchanged = Repo.get!(Subscription, sub.id)
assert unchanged.last_stripe_event_id == "evt_new"
end
# ...Read this beside the queued timestamp caveat rather than generalizing it to provider-created ordering on every entry path.
Your next source-reading sessions
- Subscribe:
Accrue.Billing→Accrue.Billing.SubscriptionActions→Accrue.Processor→Accrue.Billing.SubscriptionProjection. Where do host policy, remote authority, and local atomicity change hands? - Webhook:
Accrue.Webhook.Plug→Accrue.Webhook.Ingest→Accrue.Webhook.DispatchWorker→Accrue.Webhook.DefaultHandler. What survives a duplicate, a crash, and a final retry? - Entitlements:
Accrue.Entitlements→Accrue.Entitlements.Resolver.LocalMap→Accrue.Billing.Query. Which exact local states can grant access, and how does uncertainty deny? - Dunning:
Accrue.Jobs.DunningSweeper→Accrue.Workers.DunningStep→ subscription predicates. What anchors a campaign, and what makes recovery cancel outstanding work? - Invoice and PDF:
Accrue.Billing.InvoiceActions→Accrue.InvoiceRenderer→ the configured renderer. Which work is synchronous, queued, or host-supervised? - Audit:
Accrue.Events→Accrue.Events.Event. Which writes share a transaction with their audit evidence, and how is tampering blocked? - Admin:
AccrueAdmin.Router→ an Admin LiveView → core queries or facade. Does each operator mutation preserve host auth and reuse core semantics? - Portal:
AccruePortal.Router→AccruePortal.BillingReadModel→Accrue.Billing. Which actions are local UI concerns, and which cross the processor boundary?
Return to the Architecture guide when you need the package and data-authority map. For operational depth, continue with Webhooks, Lifecycle semantics, Entitlements, Testing, Dunning, PDF rendering, and Email.