Level 1 Registering webhooks and delivering events
Build WebhookService(transport), the core of a webhook delivery system. There is no real network and no real clock. transport(url, event_id, payload) is a function you're given; it makes one HTTP POST and returns the status code. Every time-dependent method takes now, an integer number of seconds, and now never decreases between calls.
register(user_id, event_id, url) -> str: a user has at most one webhook per event type. The first registration creates webhook"wh1", then"wh2", and so on. Registering the same(user_id, event_id)again replaces the URL, makes the webhook active again, and returns the existing id.deactivate(webhook_id) -> bool:Falseif the id is unknown or the webhook is already inactive.get_webhook(webhook_id) -> dict | None:{"user_id", "event_id", "url", "active"}.publish(now, event_id, payload) -> list[str]: create one delivery for each active webhook registered forevent_id, in order of webhook creation. Delivery ids are"d1","d2", … and are shared across all publishes. Each new delivery is"pending"and due atnow. Nothing is sent yet: sending is asynchronous.process(now) -> int: work through every delivery due at or beforenow, earliest due time first and then oldest delivery id first. For each one, calltransportwith the webhook's current URL. A status in200..299sets it to"success". In this level the transport only ever returns 2xx or plain 4xx codes (never 408 or 429), and a 4xx sets it to"failed". Level 2 adds retries for the other codes. Return how many timestransportwas called. If the delivery's webhook is inactive when its turn comes, don't send it: its status becomes"cancelled", and it isn't counted.get_delivery(delivery_id) -> dict | None:{"webhook_id", "status", "attempts", "last_code"}.attemptscountstransportcalls, andlast_codeisNonebefore the first one.
calls = []
def fake(url, event_id, payload):
calls.append(url); return 200
svc = WebhookService(fake)
svc.register("u1", "user.created", "https://a") # "wh1"
svc.register("u2", "user.created", "https://b") # "wh2"
svc.publish(10, "user.created", {"id": 7}) # ["d1", "d2"]
svc.process(10) # 2 ; calls == ["https://a", "https://b"]
svc.get_delivery("d1") # {"webhook_id": "wh1", "status": "success", "attempts": 1, "last_code": 200}