Building a 3-Queue Marketplace Notification Center Backend for Event Audit Polling

Keep each marketplace support queue responsible for its notification templates, freeze the chosen template version when a contact event is accepted, and let email, SMS, audit history, and polling read from the same event identity. That decision prevents a generic notification service from quietly becoming the owner of billing, trust, and seller-support language.

It also gives the backend a clean answer when two channels disagree: the contact event is the durable fact, while each delivery is a separate attempt with its own status. Don't turn a successful email into proof that the SMS succeeded.

Keep those identities separate.

How should a notification center backend expose email SMS delivery history and audit log polling?

Use four records with different jobs: an inbound contact, an immutable notification event, one or more channel attempts, and an append-only audit entry. A polling API returns the event plus its attempts; it does not query an email or SMS gateway on every browser refresh. In this marketplace example, billing, trust, and seller_support own their templates and their routing rules. The central backend owns acceptance, IDs, state transitions, and access control.

That boundary matters more than the transport library. If the trust team changes the wording for a suspected-account-takeover reply, new events can select version 8 while an event already accepted under version 7 remains explainable. Re-rendering an old event from the newest template would make the audit trail look tidy while changing what the system claims it sent.

The data flow is short. POST /contacts validates a form, maps its topic to a queue, records an event with a template snapshot, and creates channel attempts. A worker claims pending attempts and calls a channel adapter. GET /contacts/{contact_id}/notifications polls the local projection with an after cursor. Authentication and queue-level authorization belong at both endpoints in a production service, even though the compact example leaves identity plumbing outside the frame.

A runnable three-queue example

This Python example uses SQLite so the state transitions stay visible. The adapters deliberately return deterministic provider message IDs; replace their bodies at the integration boundary, not the event model. Run the file, submit a contact, call deliver_pending(), and poll the returned contact_id.

import json
import sqlite3
import time
import uuid
from dataclasses import dataclass


DB = sqlite3.connect("notifications.db")
DB.row_factory = sqlite3.Row
DB.executescript(
    """
    CREATE TABLE IF NOT EXISTS contacts (
        id TEXT PRIMARY KEY,
        topic TEXT NOT NULL,
        queue TEXT NOT NULL,
        email TEXT NOT NULL,
        phone TEXT,
        created_at INTEGER NOT NULL
    );
    CREATE TABLE IF NOT EXISTS events (
        id TEXT PRIMARY KEY,
        contact_id TEXT NOT NULL,
        kind TEXT NOT NULL,
        template_owner TEXT NOT NULL,
        template_version INTEGER NOT NULL,
        template_snapshot TEXT NOT NULL,
        created_at INTEGER NOT NULL
    );
    CREATE TABLE IF NOT EXISTS attempts (
        id TEXT PRIMARY KEY,
        event_id TEXT NOT NULL,
        channel TEXT NOT NULL,
        status TEXT NOT NULL,
        provider_message_id TEXT,
        updated_at INTEGER NOT NULL,
        UNIQUE(event_id, channel)
    );
    CREATE TABLE IF NOT EXISTS audit (
        sequence INTEGER PRIMARY KEY AUTOINCREMENT,
        event_id TEXT NOT NULL,
        action TEXT NOT NULL,
        detail TEXT NOT NULL,
        occurred_at INTEGER NOT NULL
    );
    """
)

ROUTES = {
    "payment": "billing",
    "account_safety": "trust",
    "listing": "seller_support",
}

TEMPLATES = {
    "billing": (4, "We received your payment question: {subject}"),
    "trust": (7, "We received your account safety report: {subject}"),
    "seller_support": (3, "We received your listing question: {subject}"),
}


@dataclass(frozen=True)
class ContactCommand:
    topic: str
    subject: str
    email: str
    phone: str | None = None


def record_audit(event_id: str, action: str, detail: dict) -> None:
    DB.execute(
        "INSERT INTO audit(event_id, action, detail, occurred_at) VALUES (?, ?, ?, ?)",
        (event_id, action, json.dumps(detail, sort_keys=True), int(time.time())),
    )


def accept_contact(command: ContactCommand) -> str:
    if command.topic not in ROUTES:
        raise ValueError("topic must be payment, account_safety, or listing")

    now = int(time.time())
    contact_id = str(uuid.uuid4())
    event_id = str(uuid.uuid4())
    queue = ROUTES[command.topic]
    version, template = TEMPLATES[queue]
    snapshot = template.format(subject=command.subject)
    channels = ["email"] + (["sms"] if command.phone else [])

    with DB:
        DB.execute(
            "INSERT INTO contacts VALUES (?, ?, ?, ?, ?, ?)",
            (contact_id, command.topic, queue, command.email, command.phone, now),
        )
        DB.execute(
            "INSERT INTO events VALUES (?, ?, ?, ?, ?, ?, ?)",
            (event_id, contact_id, "contact.received", queue, version, snapshot, now),
        )
        for channel in channels:
            DB.execute(
                "INSERT INTO attempts VALUES (?, ?, ?, 'pending', NULL, ?)",
                (str(uuid.uuid4()), event_id, channel, now),
            )
        record_audit(event_id, "event.accepted", {"queue": queue, "channels": channels})
    return contact_id


def send_email(destination: str, body: str) -> str:
    return f"email-{uuid.uuid4()}"


def send_sms(destination: str, body: str) -> str:
    return f"sms-{uuid.uuid4()}"


def deliver_pending() -> None:
    rows = DB.execute(
        """
        SELECT a.id, a.event_id, a.channel, e.template_snapshot,
               c.email, c.phone
        FROM attempts a
        JOIN events e ON e.id = a.event_id
        JOIN contacts c ON c.id = e.contact_id
        WHERE a.status = 'pending'
        ORDER BY e.created_at, a.id
        """
    ).fetchall()

    for row in rows:
        if row["channel"] == "email":
            message_id = send_email(row["email"], row["template_snapshot"])
        else:
            message_id = send_sms(row["phone"], row["template_snapshot"])

        now = int(time.time())
        with DB:
            DB.execute(
                "UPDATE attempts SET status = 'delivered', provider_message_id = ?, updated_at = ? WHERE id = ?",
                (message_id, now, row["id"]),
            )
            record_audit(
                row["event_id"],
                "attempt.delivered",
                {"attempt_id": row["id"], "channel": row["channel"]},
            )


def poll_notifications(contact_id: str, after: int = 0) -> dict:
    event = DB.execute(
        "SELECT * FROM events WHERE contact_id = ?", (contact_id,)
    ).fetchone()
    if event is None:
        raise KeyError("contact not found")

    attempts = DB.execute(
        "SELECT channel, status, updated_at FROM attempts WHERE event_id = ?",
        (event["id"],),
    ).fetchall()
    audit = DB.execute(
        """
        SELECT sequence, action, detail, occurred_at FROM audit
        WHERE event_id = ? AND sequence > ? ORDER BY sequence
        """,
        (event["id"], after),
    ).fetchall()
    return {
        "event_id": event["id"],
        "template_owner": event["template_owner"],
        "template_version": event["template_version"],
        "attempts": [dict(row) for row in attempts],
        "changes": [dict(row) for row in audit],
        "next_after": audit[-1]["sequence"] if audit else after,
    }


if __name__ == "__main__":
    contact_id = accept_contact(
        ContactCommand(
            topic="account_safety",
            subject="Phone number changed without permission",
            email="buyer@example.test",
            phone="+15550100101",
        )
    )
    deliver_pending()
    print(json.dumps(poll_notifications(contact_id), indent=2))

This is intentionally a notebook-sized slice, not a production queue. The useful seam is the attempts table: email and SMS never overwrite each other, and the unique (event_id, channel) constraint stops the same accepted event from creating two initial attempts for one channel. For real retries, add an attempt number or a separate dispatch record rather than weakening that invariant.

Template ownership is an architecture decision

There are three plausible owners. A central communications team can own every template, each support queue can own its own, or queues can own content while a central team owns a constrained layout and policy layer. For marketplace contact routing, the hybrid model is usually the sharper starting point: billing knows which payment context is safe to mention, trust controls security wording, and seller support understands listing terminology; the platform still enforces required metadata, channel eligibility, and rendering tests.

The catch is coordination. Queue ownership is not suitable when teams cannot review copy, maintain translations, or respond to policy changes. In that environment, keep templates central and make queue-specific fields explicit. At the other extreme, a centralized template team becomes a release bottleneck when domain language changes weekly. There isn't a universal winner — ownership should follow who can approve meaning, while the platform approves mechanics.

Treat template selection like code selection. Store a stable template key and version, validate required variables before accepting the event, and test rendered output with representative subjects including empty, long, and non-ASCII input. Snapshotting the final body gives the strongest audit explanation, but it stores more personal data; storing only version plus variables minimizes duplication but requires long-term access to the exact renderer. Pick one policy deliberately, set retention around it, and document who may read the result.

Failure modes worth testing before deployment

The first test should submit the same logical command twice. The sample's random contact ID makes those two distinct contacts, which is correct only if the caller intended two submissions. A public endpoint needs an idempotency token tied to the authenticated actor and request body so a browser retry can return the original result. Keep that token separate from a provider message ID because it protects a different boundary.

Next, make the channels disagree. Email may be delivered while SMS remains pending, and the API must report both states without collapsing them into sent: true. Then reorder worker completion, poll with the same cursor twice, and confirm that audit entries are stable. A cursor should identify an ordered local record, not wall-clock time; two updates can share a one-second timestamp.

Security messages need a separate review. OWASP's forgot-password guidance says reset flows should return consistent messages and timing, use a side channel for reset tokens, rate-limit requests, and avoid changing the account until a valid token is presented. A marketplace contact acknowledgement is not automatically a password-reset channel. Don't place credentials or reset tokens into a generic support template just because email and SMS adapters already exist.

Marketing is another boundary. The FTC's CAN-SPAM guide describes requirements for commercial email, including accurate headers and subject lines, identification as an advertisement, a physical postal address, and a working opt-out mechanism. A transactional reply to a contact form and a promotional follow-up should therefore enter different policy paths. I'm not sure a message with mixed transactional and promotional content will be classified the way your product team expects; legal review of the actual template and audience is what resolves that uncertainty.

Tiny tests catch expensive ambiguity.

Operate the polling API as a projection

Polling should be boring: authorize the contact, read local state, return a monotonic cursor, and cap page size. Use conditional responses or a modest client interval to reduce empty reads, but do not make clients infer completion from silence. Define terminal states per channel and return an event-level summary only as a derived value.

Poll local state.

Operationally, watch acceptance latency, the age of the oldest pending attempt, attempts by channel and status, template-version distribution by queue, and polling volume per active contact. Alert on age and backlog rather than raw failure counts alone, because ten delayed attempts during quiet traffic can matter more than a larger burst that clears immediately. Logs should carry contact_id, event_id, and attempt_id, while message bodies and destination addresses stay out unless an approved diagnostic policy requires them. Before shipping, walk one event from form submission through queue selection, template version, each channel attempt, audit cursor, and retention deletion. Verify authorization with a contact owned by another account. Exercise long subjects, missing phone numbers, duplicate submissions, delayed workers, and out-of-order completions. Finally, run the same cases in the eval harness whenever routing or templates change; a prompt-assisted classifier can propose a queue, but a fixed labeled set and a deterministic fallback should decide whether that proposal is safe enough for production.

The durable design rule is simple: domain teams own message meaning, the notification backend owns delivery facts, and the polling API exposes those facts without pretending that two channels share one outcome.

References

• https://cheatsheetseries.owasp.org/cheatsheets/Forgot_Password_Cheat_Sheet.html
• https://www.ftc.gov/business-guidance/resources/can-spam-act-compliance-guide-business