"""Parcel Post — a delivery log for the pack bench. What this application is: it listens for the fulfillment and order events and keeps an append-only record of what happened to each parcel, so the bench has a list that is current the moment something changes rather than on the next sweep. `services/webhooks.py` names exactly this case as the reason webhooks exist — a 3PL polling `GET /v1/fulfillments` learns about a parcel on its next poll, not when it appears. Everything about VERIFYING a delivery lives in `sipspos_webhook.py`, which is the file to copy. This file is what one particular application does with an event once it is known to be genuine. export SIPSPOS_WEBHOOK_SECRET=whsec_… python parcels.py # listens on :8400 TWO PROPERTIES THAT ARE NOT ABOUT CRYPTOGRAPHY, and are the ones that bite in production: DELIVERY IS AT-LEAST-ONCE. `WebhookDelivery` is one row per (event, endpoint) with an attempt counter, and anything that is not a 2xx — including a response that was merely slow — is rescheduled. So you WILL see the same delivery twice. `X-SipsPOS-Delivery` is the idempotency key, and here it is the PRIMARY KEY of the log: a repeat cannot be double-recorded because the database will not have it, rather than because a branch remembered to check. A receiver that treats delivery as exactly-once ships the parcel twice. ANSWER FIRST, WORK AFTER. The sender applies a timeout and treats a non-2xx as a failure to retry, escalating to disabling the endpoint after 1m/5m/30m/2h/12h of them. A handler that does its work inline is one slow query away from being retried, and a few of those from being switched off. So this one records the fact and answers; the work happens off the response path. """ import json import os import sqlite3 import threading from http.server import BaseHTTPRequestHandler, HTTPServer from sipspos_webhook import Rejected, verify #: The events this receiver acts on. Every one of these is in #: `services.webhooks.EVENT_TYPES` — `scripts/parcel_post_test.py` asserts it, #: because a receiver subscribed to an event the platform does not send waits #: forever and looks, from the inside, exactly like one that works. HANDLED = { "fulfillment.created": "waiting to pack", "fulfillment.shipped": "shipped", "order.paid": "paid", "order.fulfilled": "fulfilled", # The one worth having a bench see. A refund AFTER the parcel left is the # case where somebody has to go and stop it, and it is invisible on a # pick list that only ever grows. "order.refunded": "REFUNDED — check before it leaves", } DEFAULT_PORT = 8400 class Store: """The log. SQLite because it is in the standard library and because the dedupe wants a real UNIQUE constraint, not a dictionary that dies with the process.""" def __init__(self, path=":memory:"): # `check_same_thread=False` because the work happens on a worker # thread (see `handle`); every write below is a single statement in # its own transaction, which SQLite serialises for us. self.db = sqlite3.connect(path, check_same_thread=False) self.db.execute(""" CREATE TABLE IF NOT EXISTS parcel_log ( delivery_id TEXT PRIMARY KEY, event TEXT NOT NULL, status TEXT NOT NULL, subject TEXT, received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP )""") self.db.commit() def record(self, delivery_id, event, subject): """True if this is new, False if it is a redelivery. The INSERT is what decides, not a preceding SELECT: two copies of the same retry arriving together would both pass a check-then-insert, and this is a receiver whose entire job is to be safe under retry. """ try: self.db.execute( "INSERT INTO parcel_log (delivery_id, event, status, subject)" " VALUES (?, ?, ?, ?)", (delivery_id, event, HANDLED.get(event, event), subject)) self.db.commit() return True except sqlite3.IntegrityError: return False def queue(self): """Everything logged, newest first — what the bench reads.""" return self.db.execute( "SELECT delivery_id, event, status, subject, received_at" " FROM parcel_log ORDER BY received_at DESC, rowid DESC" ).fetchall() def count(self): return self.db.execute("SELECT COUNT(*) FROM parcel_log").fetchone()[0] def subject_of(payload): """A human label for the thing the event is about. Deliberately forgiving: an event's `data` is whatever the serializer sends, and a receiver that raises on an unexpected shape turns a new field into an outage. Unknown shapes get a blank label and are still logged — the delivery id is what matters. """ data = (payload or {}).get("data") or {} for key in ("public_id", "id", "order_number", "tracking_number"): if data.get(key): return str(data[key]) return "" def handle(store, event, delivery_id, payload): """The application logic, with no HTTP in it. `"recorded"` or `"duplicate"`.""" if event not in HANDLED: # Not an error: an endpoint may be subscribed to more than this # receiver cares about, and refusing would earn a retry forever. return "ignored" return "recorded" if store.record(delivery_id, event, subject_of(payload)) \ else "duplicate" def make_handler(secret, store, on_work=None): """A `BaseHTTPRequestHandler` for `POST /webhooks`. `on_work` is the seam the test uses to run the work synchronously; in production it is a thread (see below). """ class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt, *args): # quiet by default pass def _reply(self, code, payload): body = json.dumps(payload).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_POST(self): # noqa: N802 (stdlib name) if self.path.rstrip("/") != "/webhooks": return self._reply(404, {"error": "not found"}) length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length) # RAW BYTES, kept as bytes # VERIFY BEFORE PARSING. Not a style preference: until this # returns, the body is a string a stranger sent us, and # `json.loads` on it is the first place we would have trusted it. try: verify(secret, self.headers.get("X-SipsPOS-Signature"), body) except Rejected as e: code = 400 if e.reason == "malformed" else 401 return self._reply(code, {"error": e.reason}) try: payload = json.loads(body) except ValueError: # Signed by us and still not JSON: our bug, not theirs. 400 # rather than 500 so it is not retried into a loop. return self._reply(400, {"error": "not json"}) event = self.headers.get("X-SipsPOS-Event") or payload.get("event") delivery_id = self.headers.get("X-SipsPOS-Delivery") if not delivery_id: return self._reply(400, {"error": "no delivery id"}) # ANSWER FIRST. The record itself is one INSERT and is done here so # the reply can honestly say what happened; anything SLOW belongs # in `on_work`. A real deployment puts that on a queue — the # thread below is the smallest thing that shows the shape without # dragging a broker into an example. result = handle(store, event, delivery_id, payload) self._reply(200, {"ok": True, "result": result}) if result == "recorded" and on_work is not None: on_work(event, delivery_id, payload) return Handler def serve(secret, store, port=DEFAULT_PORT): def work(event, delivery_id, payload): threading.Thread(target=lambda: None, daemon=True).start() server = HTTPServer(("127.0.0.1", port), make_handler(secret, store, work)) print(f"parcel-post: listening on http://127.0.0.1:{port}/webhooks") print(f"parcel-post: logging {len(HANDLED)} event types") server.serve_forever() def main(argv=None): secret = os.environ.get("SIPSPOS_WEBHOOK_SECRET") if not secret: print("parcel-post: SIPSPOS_WEBHOOK_SECRET is not set. The winery's " "admin shows it beside the endpoint in Admin -> Setup -> " "Webhooks.") return 2 serve(secret, Store(os.environ.get("PARCEL_POST_DB", "parcel-post.db"))) return 0 if __name__ == "__main__": raise SystemExit(main())