Consume transaction events

Transaction POS Log (TPLA) moves transaction data into the event-driven space. Rather than long-polling a REST endpoint for new receipts, FiftyTwo pushes events to you as they arrive.

Why events

From the API’s own documentation, the trade this makes:

• No long polling. Less network traffic, no “have you got anything yet” loop.
• Real time. You react as the transaction happens, not on your next poll.
• Resilient. If your consumer goes down, the data is ordered and waiting when it comes back.
• Horizontally scalable. Add consumers to share a high-volume workload, round-robin.
• No status-code business logic. An event arriving is the data.

The event payload is identical to the one the REST endpoints return, so you can develop against REST and switch to events without reshaping your parser.

Transport: Apache Pulsar

FiftyTwo uses Apache Pulsar. Topics are structured:

persistent://customer-name/chain-name/topic-name

Pulsar’s multi-tenancy is why the topic path carries the customer and chain: your data is namespaced from every other tenant’s.

Consuming

import json
import pulsar
client = pulsar.Client("pulsar+ssl://<broker>:6651", authentication=...)
consumer = client.subscribe(
    "persistent://your-customer/your-chain/transaction-pos-log",
    subscription_name="analytics-loader",
    # Shared lets you scale out: add consumers on the same subscription and
    # Pulsar round-robins between them.
    consumer_type=pulsar.ConsumerType.Shared,
)
while True:
    msg = consumer.receive()
    try:
        event = json.loads(msg.data())
        handle(event)                 # must be idempotent -- see below
        consumer.acknowledge(msg)
    except Exception:
        # Negative-ack so Pulsar redelivers rather than dropping the event.
        consumer.negative_acknowledge(msg)

Design for redelivery

Make the handler idempotent on the transaction’s own identity:

def handle(event):
    tx_id = event["transactionPosLogId"]
    if store.already_processed(tx_id):
        return                        # duplicate delivery; nothing to do
    store.record(tx_id, event)

Ordering is per-topic-partition, not global. If your logic depends on sequence, key on the store’s session and sequence numbers rather than assuming arrival order.

The REST endpoints

Three endpoints remain, and they are the right tool for backfill and reconciliation rather than steady-state ingestion:

The lastTransactionPosLogId cursor makes these useful for catching up after an extended outage, or for verifying that your event pipeline has not silently dropped anything.

Migrating off the Receipt API

If you currently poll the Receipt API, this replaces it. See Migrating off the Receipt API for the endpoint mapping.

Related help-center topics