class SQLiteInbox:
"""Single-service example inbox. Use a shared durable store for multiple replicas."""
def __init__(self, path):
self.path = str(path)
with sqlite3.connect(self.path) as db:
db.execute("""CREATE TABLE IF NOT EXISTS webhook_inbox (
workspace_id TEXT NOT NULL, event_type TEXT NOT NULL,
event_id TEXT NOT NULL, fingerprint TEXT NOT NULL,
body TEXT NOT NULL, received_at REAL NOT NULL,
processed INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY(workspace_id, event_type, event_id, fingerprint)
)""")
def accept(self, event):
identifiers = [event.get(key) for key in ("workspaceId", "eventType", "eventId")]
if not all(isinstance(value, str) and value for value in identifiers):
raise VerificationError("Missing event envelope identifiers")
body = canonical_payload(event)
fingerprint = hashlib.sha256(body).hexdigest()
with sqlite3.connect(self.path, timeout=3) as db:
cursor = db.execute(
"""INSERT OR IGNORE INTO webhook_inbox
(workspace_id, event_type, event_id, fingerprint, body, received_at)
VALUES (?, ?, ?, ?, ?, ?)""",
(*identifiers, fingerprint, body.decode("utf-8"), time.time()),
)
inserted = cursor.rowcount == 1
# Context manager commits before the request can be acknowledged.
return inserted
def make_wsgi_app(inbox, webhook_api_key, client_secret):
"""Verify, commit an inbox record, then acknowledge; workers process it later."""
def app(environ, start_response):
if environ.get("REQUEST_METHOD") != "POST":
start_response("405 Method Not Allowed", [("Allow", "POST")])
return [b""]
try:
length = int(environ.get("CONTENT_LENGTH") or "0")
if length < 1:
raise VerificationError("Missing JSON body")
raw_body = environ["wsgi.input"].read(length)
headers = {
"x-bridge-api-key": environ.get("HTTP_X_BRIDGE_API_KEY", ""),
"x-bridge-signature": environ.get("HTTP_X_BRIDGE_SIGNATURE", ""),
"x-bridge-timestamp": environ.get("HTTP_X_BRIDGE_TIMESTAMP", ""),
}
event = verify_webhook(raw_body, headers, webhook_api_key, client_secret)
inbox.accept(event) # A duplicate exact event is also acknowledged.
except (VerificationError, ValueError):
status, body = "400 Bad Request", b'{"accepted":false}'
except sqlite3.Error:
status, body = "503 Service Unavailable", b'{"accepted":false}'
else:
status, body = "200 OK", b'{"accepted":true}'
start_response(status, [("Content-Type", "application/json"), ("Content-Length", str(len(body)))])
return [body]
return app
