Webhook Event Ingestion for Real-Time Royalty Events

When a digital service provider settles a stream batch, reverses a chargeback, or flags a takedown, the event that matters to a royalty ledger often lands milliseconds after it happens, not at the end of a nightly export window. Webhook event ingestion inverts the traditional pull model: rather than an ETL job asking a DSP, aggregator, or payment processor “has anything changed?” on a fixed schedule, the source pushes an HTTP POST to a registered endpoint the instant an event occurs. For teams building on the Data Ingestion & Streaming Sync Pipelines reference, that shift moves the hardest reconciliation problems — authenticity, ordering, and duplicate suppression — from the source system’s infrastructure onto the receiving endpoint.

Push Delivery vs. Polling: Concept Context

Polling and webhook delivery solve the same problem — learning that a payee balance, a rights assignment, or a catalog entry has changed — from opposite directions. DSP API Polling Strategies place the burden of freshness on the consumer: a scheduler asks repeatedly and accepts a bounded staleness window between the true event time and detection. Webhook ingestion places that burden on the source, which knows the instant a settlement posts and notifies subscribers directly. The trade is latency for control: a polling consumer decides exactly when to look, while a webhook consumer must always be ready to receive.

That readiness requirement comes with a specific delivery contract. Almost every DSP, aggregator, and payment provider that offers webhooks implements at-least-once delivery: if the receiving endpoint does not return a 2xx response within a short timeout, the sender retries on an exponential schedule for hours or days. This guarantees that a transient outage on the ingestion side does not silently drop a royalty-relevant event, but it also guarantees that the same event will sometimes arrive twice, three times, or more. A pipeline that treats each inbound POST as a unique, one-time fact will double-count settlements; one that treats delivery as untrusted until proven otherwise will not. Two structural elements make untrusted delivery safe: signature verification, confirming the payload originated from the claimed source and was not altered in transit, and idempotency keys, a source-supplied or payload-derived identifier that lets the ingestion layer discard a redelivery before it touches downstream aggregates.

Ordering is the third structural concern. DSPs commonly fan out webhook delivery across multiple workers, so two events for the same payee_id can arrive out of sequence, especially during a retry storm. Rather than trusting arrival order, resilient pipelines carry an explicit sequence number or event timestamp in the payload and reorder before applying business rules.

Python ETL Architecture

A webhook receiver has a fundamentally different shape than a batch job: it must acknowledge quickly and continuously rather than process in scheduled windows. The pattern that works well is a thin, stateless HTTP layer — typically FastAPI or Starlette on an ASGI server — that does the minimum work required to accept custody of an event: capture the raw request body, verify its signature, assign or confirm an idempotency key, and hand it to a durable queue. Heavier work — schema validation with Pydantic, ISRC/ISWC resolution, split-sum recomputation — happens in a separate consumer process reading from that queue, so a slow downstream dependency never causes the DSP-facing endpoint to time out and trigger unnecessary retries.

For the queue itself, a log-structured store (Redis Streams, Kafka, or a managed equivalent) is preferable to a plain task queue because it preserves arrival order per partition and supports replay from an offset. Memory considerations favor streaming: buffer individual events or small micro-batches rather than accumulating an unbounded list, since webhook volume during a settlement run or a DSP’s own backfill can spike well above steady-state traffic. Downstream of the queue, batch-oriented tools such as Polars or DuckDB are appropriate once events are validated and grouped into a batch_id, but the ingestion boundary itself should stay row-at-a-time and non-blocking.

Step-by-Step Implementation

Step 1: Accept and Acknowledge Without Blocking

The receiving endpoint’s only job is to capture the exact bytes delivered, record enough metadata to verify and deduplicate later, and enqueue the envelope before returning a 202. Parsing the JSON body at this stage is a mistake: signature verification must run against the untouched raw bytes.

import hashlib
import json
from typing import Any

from fastapi import FastAPI, Request, Response, status
import redis.asyncio as redis

app = FastAPI()
event_queue = redis.Redis(host="queue.internal", port=6379, db=0)

@app.post("/webhooks/dsp/{source_id}")
async def receive_dsp_webhook(source_id: str, request: Request) -> Response:
    raw_body: bytes = await request.body()
    event_id = request.headers.get("X-Event-Id") or hashlib.sha256(raw_body).hexdigest()

    envelope: dict[str, Any] = {
        "event_id": event_id,
        "source_id": source_id,
        "signature": request.headers.get("X-Signature-256", ""),
        "received_at": request.headers.get("Date", ""),
        "raw_body": raw_body.decode("utf-8", errors="replace"),
    }

    # Ack fast: the raw envelope is durable before any parsing or
    # business-rule validation runs downstream.
    await event_queue.xadd("dsp-events-raw", {"payload": json.dumps(envelope)})
    return Response(status_code=status.HTTP_202_ACCEPTED)

Step 2: Enforce Idempotency Before Any Balance Changes

Once a consumer pulls an envelope off the queue, the first check is whether this event_id has already been claimed. A unique constraint in the ledger database, rather than an in-memory cache, is what makes this safe across multiple consumer processes and across restarts.

import asyncpg

async def claim_event_once(
    pool: asyncpg.Pool, event_id: str, source_id: str, batch_id: str
) -> bool:
    """Returns True only the first time event_id is seen."""
    async with pool.acquire() as conn:
        result = await conn.execute(
            """
            INSERT INTO webhook_event_log (event_id, source_id, batch_id, received_at)
            VALUES ($1, $2, $3, now())
            ON CONFLICT (event_id) DO NOTHING
            """,
            event_id, source_id, batch_id,
        )
    return result.endswith("1")

Step 3: Reorder by Sequence Number Before Applying Splits

With duplicates suppressed, the consumer validates the payload shape with Pydantic and checks that the event’s sequence number for its payee_id is newer than the last one applied — dropping anything stale rather than assuming arrival order reflects event order.

from pydantic import BaseModel, Field

class RoyaltyWebhookEvent(BaseModel):
    event_id: str
    payee_id: str
    isrc: str
    territory_code: str = Field(pattern=r"^[A-Z]{2}$")
    right_share_pct: float = Field(ge=0.0, le=100.0)
    sequence_number: int
    reporting_period: str

def is_in_order(event: RoyaltyWebhookEvent, last_seen_sequence: dict[str, int]) -> bool:
    watermark = last_seen_sequence.get(event.payee_id, -1)
    if event.sequence_number <= watermark:
        return False  # stale or duplicate delivery — safe to drop
    last_seen_sequence[event.payee_id] = event.sequence_number
    return True

Step 4: Replay a Batch From the Durable Log

Because the raw envelope was persisted before processing, a defect in the consumer logic — or a DSP-initiated backfill of historical events — can be replayed deterministically by reading the log back out, sorting, and deduplicating by event_id.

import polars as pl

def replay_batch(event_log_path: str, batch_id: str) -> pl.DataFrame:
    events = pl.read_ndjson(event_log_path)
    return (
        events
        .filter(pl.col("batch_id") == batch_id)
        .sort(["payee_id", "sequence_number"])
        .unique(subset=["event_id"], keep="first")
    )

Reconciliation Gates & Validation

Every event that survives idempotency and ordering checks still has to clear the same reconciliation gates as any other ingestion path. Pydantic models enforce field shape and range at the boundary; a secondary gate resolves the event’s isrc against the catalog’s ISRC-to-ISWC mapping and quarantines anything that fails to resolve rather than posting against an unknown work. Split-sum assertions run per payee_id and reporting_period: if the right_share_pct values implied by a batch of events do not sum to within a small epsilon of 100%, the batch is held in an exception table for manual adjudication instead of being applied to payee balances. UPC dedup runs at the release level, since some aggregators emit one webhook per territory for what is logically a single sale, and double-counting here is a common source of overpayment.

Failure Modes & Rollback

A webhook receiver that stays healthy under load still needs defined behavior for failure cases specific to push delivery. If the downstream queue or ledger database becomes unavailable, the endpoint should fail closed — return a 5xx rather than a 202 — so the DSP’s retry mechanism keeps the event alive instead of silently dropping it after a false acknowledgment. A circuit breaker in front of the consumer’s downstream dependencies (the catalog registry, the reconciliation store) should trip after a defined error threshold and shed load by pausing consumption rather than accepting events it cannot safely process. Because every state-changing operation is an idempotent UPSERT keyed on event_id, replaying a stalled or partially processed batch after recovery is safe by construction. For a full outage, restoring from the last durable log offset and replaying forward reconstructs ledger state without needing the DSP to resend anything, which matters because some sources cap how far back they will redeliver.

Security & Audit Trail

Raw webhook payloads frequently carry payee identifiers and settlement amounts, so the durable queue and any staging tables holding unprocessed envelopes should be encrypted at rest, with access restricted through role-based controls scoped to the ingestion service account rather than shared broadly across the data platform. Signing secrets and any per-source credentials belong in a secrets manager, never in application config checked into source control, and should be scoped so that a compromised secret for one source_id cannot be used to forge events for another. Every claimed event, whether it passes or fails downstream validation, should leave an append-only audit record — event_id, source_id, batch_id, verification outcome, and processing timestamp — so that a reconciliation dispute months later can be traced back to the exact webhook delivery that produced it.

Related

Webhook ingestion and polling are complementary rather than competing strategies: many production pipelines run DSP API Polling Strategies as a reconciliation backstop for sources whose webhook coverage is incomplete, and lean on shared Error Handling & Retry Mechanisms once an event has been accepted and queued. Sources that support cursor-based backfill in addition to push notifications are best handled through Incremental Sync and Change Data Capture for the historical catch-up, keeping webhooks focused on near-real-time deltas. The mechanics of confirming that an inbound event actually came from the claimed source are covered in depth in Verifying DSP Webhook Signatures, which this reference relies on for every event before it reaches the queue.