diff --git a/README.md b/README.md index f1ab0cd..82b260c 100644 --- a/README.md +++ b/README.md @@ -38,12 +38,16 @@ open-source. **Early development.** This build registers endpoints, stores the submissions sent to them together with the durable obligation to deliver them, and runs a -separate worker that performs the signed webhook delivery and retries it. +separate worker that performs the signed webhook delivery and retries it. An +operator can now list and reconfigure endpoints, read the delivery queue and its +attempt history, and put a failed delivery back in the queue, all through the +same authenticated management API. -Endpoint configuration now requires a management API key. Form ingestion stays -public, because an ingestion URL is meant to sit in the `action` attribute of -somebody's HTML form. There is still no rate limiting and no spam protection, -so a public deployment is exposed to whatever volume the internet sends it. +Everything that administers the service requires a management API key. Form +ingestion stays public, because an ingestion URL is meant to sit in the `action` +attribute of somebody's HTML form. There is still no rate limiting and no spam +protection, so a public deployment is exposed to whatever volume the internet +sends it. | Capability | Status | | ----------------------------- | ------------------------- | @@ -58,7 +62,11 @@ so a public deployment is exposed to whatever volume the internet sends it. | Retries with backoff | Implemented | | Schema migrations | Implemented | | API keys / authentication | Implemented | -| Manual delivery replay | **Not implemented** | +| Endpoint management | Implemented | +| Delivery inspection | Implemented | +| Manual delivery replay | Implemented | +| Endpoint deletion | **Not implemented** | +| Submission retrieval | **Not implemented** | | Rate limiting, spam handling | **Not implemented** | | Export, retention, dashboards | **Not implemented** | @@ -126,7 +134,7 @@ database is reachable and at the migration revision the build was written against, and refuse to start otherwise: ``` -the database is at migration '0001' but this build expects '0002'. +the database is at migration '0002' but this build expects '0003'. Run 'alembic upgrade head' before starting. ``` @@ -210,11 +218,17 @@ Interactive API documentation is served at `http://127.0.0.1:8000/docs`. Administering the service needs a credential. Submitting a form does not. -| Route | Credential | -| ---------------------- | -------------------------------- | -| `POST /endpoints` | Management API key required | -| `POST /f/{endpoint_id}`| **Public**, no credential | -| `GET /health` | **Public**, no credential | +| Route | Credential | +| --------------------------------------- | --------------------------- | +| `POST /endpoints` | Management API key required | +| `GET /endpoints` | Management API key required | +| `GET /endpoints/{endpoint_id}` | Management API key required | +| `PATCH /endpoints/{endpoint_id}` | Management API key required | +| `GET /deliveries` | Management API key required | +| `GET /deliveries/{delivery_id}` | Management API key required | +| `POST /deliveries/{delivery_id}/replay` | Management API key required | +| `POST /f/{endpoint_id}` | **Public**, no credential | +| `GET /health` | **Public**, no credential | A management API key is an opaque bearer token that administers the whole service. It is not a user account, not a login, and not scoped to a tenant: @@ -382,14 +396,141 @@ Returns `201 Created`: > this response, and there is no route that reads it back. Losing it means > creating a new endpoint. -Reusing an ID returns `409 endpoint_already_exists`. There is no route to list, -update or delete endpoints yet, so a webhook can only be configured at creation -time. +Reusing an ID returns `409 endpoint_already_exists`. To change an endpoint +afterwards, see [Managing endpoints](#managing-endpoints). The key that created an endpoint is not recorded on it. A management key administers the service rather than owning a slice of it, and there is no tenancy model for an owner column to belong to. +### Managing endpoints + +Every route below takes the same credential as `POST /endpoints`. The examples +use an obvious placeholder for it: + +```bash +export HYMICAL_KEY=hym_live_REPLACE_WITH_YOUR_KEY +``` + +#### `GET /endpoints` + +```bash +curl "http://127.0.0.1:8000/endpoints?limit=2" \ + -H "Authorization: Bearer $HYMICAL_KEY" +``` + +```json +{ + "items": [ + { + "id": "contact-form", + "name": "Contact form", + "is_active": true, + "created_at": "2026-08-24T14:34:27.432598Z", + "webhook_url": "https://example.com/hooks/forms" + } + ], + "next_cursor": null +} +``` + +`webhook_url` is returned because it is configuration rather than a credential: +knowing where an endpoint delivers proves nothing and forges nothing. The +signing secret is not returned, here or anywhere else. See +[Pagination](#pagination) for what `next_cursor` does. + +#### `GET /endpoints/{endpoint_id}` + +```bash +curl http://127.0.0.1:8000/endpoints/contact-form \ + -H "Authorization: Bearer $HYMICAL_KEY" +``` + +The same representation as one item of the listing, for one endpoint, so a +caller that already knows an ID does not have to page to find it. An unknown ID +returns `404 endpoint_not_found`. + +#### `PATCH /endpoints/{endpoint_id}` + +`PATCH` rather than `PUT`, because only a few fields are yours to change. Send +only what should change; anything omitted is left exactly as it was. + +```bash +curl -X PATCH http://127.0.0.1:8000/endpoints/contact-form \ + -H "Authorization: Bearer $HYMICAL_KEY" \ + -H 'Content-Type: application/json' \ + -d '{"name": "Support form", "is_active": false}' +``` + +| Field | Meaning | +| ------------- | ---------------------------------------------------------------- | +| `name` | New label, 1 to 200 characters | +| `is_active` | Whether the endpoint accepts submissions | +| `webhook_url` | New destination, or `null` to remove the webhook entirely | + +**The endpoint ID is not changeable.** It is the primary key, and it appears in +the `action` URL of every HTML form pointing at the endpoint, so changing it +would break deployed forms. An `id` in the body is ignored. + +**Deleting an endpoint is not implemented.** Disabling one is enough for now: +deletion immediately raises what happens to its stored submissions, its delivery +history and the foreign keys between them, and none of that is decided yet. + +#### Disabling and re-enabling + +```bash +curl -X PATCH http://127.0.0.1:8000/endpoints/contact-form \ + -H "Authorization: Bearer $HYMICAL_KEY" \ + -H 'Content-Type: application/json' \ + -d '{"is_active": false}' +``` + +Disabling takes effect on the very next submission, which is refused with +`409 endpoint_inactive` and stores nothing. Nothing caches the endpoint, so +there is no window in which a disabled endpoint still accepts a form. Sending +`{"is_active": true}` restores acceptance just as immediately. + +Deliveries that were already queued are **not** affected. They are work this +service already promised to do, and disabling an endpoint stops it taking on +more rather than abandoning what it owes. + +#### Changing the webhook + +The destination and its signing secret change together, because a secret belongs +to a receiver rather than to an endpoint: + +| Change | What happens to the secret | +| ----------------------------------- | ------------------------------------- | +| `webhook_url` omitted | Unchanged | +| `webhook_url` set to the same URL | Unchanged | +| `webhook_url` set to a different URL | A new one is generated and returned | +| `webhook_url` set on an endpoint with no webhook | A new one is generated and returned | +| `webhook_url` set to `null` | Removed along with the destination | + +```json +{ + "id": "contact-form", + "name": "Contact form", + "is_active": true, + "created_at": "2026-08-24T14:34:27.432598Z", + "webhook_url": "https://example.com/hooks/forms-v2", + "webhook_secret": "whsec_6f1c... (64 hex characters)" +} +``` + +> **Save `webhook_secret` now.** As at creation, it is returned only in the +> response of the request that generated it. `webhook_secret` is `null` when the +> request did not generate one, and the field does not exist at all on a read. + +Carrying the old secret over to a new destination would hand a receiver that +never had it the ability to verify signatures, and leave the previous receiver +holding a live secret. So the two always move together. + +**Deliveries already queued keep the destination and secret they snapshotted** +when their submission was accepted. Changing configuration here never redirects +work that is already owed, and never leaves a queued payload signed with a +secret its receiver never had. + ### `POST /f/{endpoint_id}` Accepts a form submission for a registered endpoint and stores it. @@ -600,8 +741,12 @@ Delivery is attempted immediately, then backs off by doubling, capped: | 5 | 80s | After `FORMS_WEBHOOK_MAX_ATTEMPTS` (default 5) the delivery becomes `failed` and -is never retried. There is no jitter: the schedule is deliberately exactly -predictable. Every attempt, including the last, stays in `delivery_attempts`. +is never retried automatically. There is no jitter: the schedule is deliberately +exactly predictable. Every attempt, including the last, stays in +`delivery_attempts`. + +The allowance is per retry cycle, not per lifetime, so a +[manual replay](#manual-delivery-replay) starts the schedule again from the top. #### At-least-once, not exactly-once @@ -624,8 +769,8 @@ attempts it has had, when it is next due, and when it finished. `delivery_attempts` holds one row per request that actually went out, numbered, with the outcome, the HTTP status when there was one, and a bounded failure message. Response bodies are **not** stored, and neither is the signing secret. -A job that is inspected and found not due records nothing. There is no API to -read either table yet; query them directly. +A job that is inspected and found not due records nothing. Both are readable +through [Inspecting deliveries](#inspecting-deliveries). #### Destinations that are refused @@ -633,7 +778,8 @@ A webhook URL must use `http` or `https` and must not name a loopback, private, link-local, multicast, reserved or unspecified address. That covers `localhost`, `127.0.0.1`, `[::1]`, `[::ffff:127.0.0.1]`, `10.0.0.0/8`, `192.168.0.0/16`, and the `169.254.169.254` cloud metadata endpoint. Rejections return -`422 invalid_webhook_url`. +`422 invalid_webhook_url`. The same check runs on a destination supplied through +`PATCH /endpoints/{endpoint_id}`. This is **not** complete SSRF protection; see [Limitations](#limitations). @@ -641,6 +787,190 @@ For local development, `FORMS_ALLOW_PRIVATE_WEBHOOK_TARGETS=true` lifts the address restriction so you can point a webhook at a server on your own machine. Do not enable it in production. +### Inspecting deliveries + +The delivery queue is operational data, so reading it needs a management API key. +The examples below use the same placeholder as the endpoint routes: + +```bash +export HYMICAL_KEY=hym_live_REPLACE_WITH_YOUR_KEY +``` + +#### `GET /deliveries` + +```bash +curl "http://127.0.0.1:8000/deliveries?state=failed&endpoint_id=contact-form" \ + -H "Authorization: Bearer $HYMICAL_KEY" +``` + +| Filter | Meaning | +| ------------- | ---------------------------------------------------------------- | +| `endpoint_id` | Only deliveries for one endpoint | +| `state` | One of `pending`, `processing`, `delivered`, `failed` | +| `limit` | Page size, 1 to 100, default 50 | +| `cursor` | The previous page's `next_cursor` | + +```json +{ + "items": [ + { + "id": "whd_9a3f...", + "submission_id": "sub_48984534f33749c49a88de2d59400dce", + "endpoint_id": "contact-form", + "state": "failed", + "destination_url": "https://example.com/hooks/forms", + "attempt_count": 5, + "cycle_attempt_count": 5, + "next_attempt_at": "2026-08-24T15:12:07.117440Z", + "created_at": "2026-08-24T14:34:27.651841Z", + "completed_at": "2026-08-24T15:12:07.204118Z" + } + ], + "next_cursor": null +} +``` + +An unknown `state` is refused with `422 invalid_request`. An `endpoint_id` that +matches nothing is not an error: it is a filter that selected no rows, and the +answer is an empty page. + +**Submitted field values are not returned**, in the listing or the detail. There +is no route that reads a submission back yet, and a delivery view is not a way +around that. + +#### `GET /deliveries/{delivery_id}` + +The same fields as one listed item, plus the ordered attempt history: + +```json +{ + "id": "whd_9a3f...", + "state": "failed", + "attempt_count": 5, + "cycle_attempt_count": 5, + "attempts": [ + { + "attempt_number": 1, + "attempted_at": "2026-08-24T14:34:27.884210Z", + "outcome": "http_error", + "response_status": 503, + "error": "destination responded with HTTP 503" + } + ] +} +``` + +Attempts are ordered by `attempt_number`, ascending. `outcome` is one of +`succeeded`, `http_error`, `timeout` or `network_error`. `response_status` is +null when the destination never answered at all. The snapshotted signing secret, +the request headers and the response body are not there: the first two are never +returned by any route, and the third is never stored. + +An unknown ID returns `404 delivery_not_found`. + +### Manual delivery replay + +#### `POST /deliveries/{delivery_id}/replay` + +```bash +curl -X POST http://127.0.0.1:8000/deliveries/whd_9a3f.../replay \ + -H "Authorization: Bearer $HYMICAL_KEY" +``` + +Returns `200 OK` with the delivery's new state: + +```json +{ + "id": "whd_9a3f...", + "state": "pending", + "attempt_count": 5, + "cycle_attempt_count": 0, + "next_attempt_at": "2026-08-24T16:02:11.006318Z", + "completed_at": null +} +``` + +**Only a delivery in terminal `failed` can be replayed.** A `pending` one is +already queued, a `processing` one is already being sent, and a `delivered` one +already reached its receiver. All three are refused with +`409 delivery_not_replayable`, and the error names the state that made it +ineligible. + +**`200`, not `202`.** A `202` would say this request will be carried out later, +which invites reading the replay as the delivery. It is not. This request +finishes here, and what it leaves behind is a queued row. + +#### What replay actually does + +Replay is a state change. **The API process sends nothing**: it has no outbound +HTTP client at all, which is what makes that structural rather than a promise. +The delivery becomes due, the ordinary worker claims it on its next poll through +its ordinary claiming path, and the ordinary retry rules apply from there. + +``` +failed -> replay -> pending and due -> worker claims -> delivered, or retries +``` + +Everything that identifies the delivery is preserved: + +- the **same logical delivery**, not a second one; +- the **same submission**, unmodified and not duplicated; +- the **snapshotted destination and signing secret**, so the receiver verifies + the replayed payload exactly as it would have verified the original; +- **every historical attempt row**, untouched. + +#### Attempt numbering and the retry budget + +A delivery carries two counters, because they answer two different questions: + +| Field | Meaning | +| --------------------- | ------------------------------------------------------ | +| `attempt_count` | Every request ever made for this delivery | +| `cycle_attempt_count` | Requests made since it last entered the queue | + +`attempt_count` only ever goes up, and it is what numbers the attempt history, +so **an attempt number is never reused** however often a delivery is replayed. +`cycle_attempt_count` is what the retry allowance is measured against, and a +replay resets it to zero. + +That is the whole model: a replay starts a fresh retry cycle while preserving +the history. A delivery that exhausted five automatic attempts and is then +replayed gets five more, numbered 6 through 10, rather than failing immediately +because five have already been spent. Until something is replayed the two +counters are equal. + +#### Two operators replaying at once + +The transition is a single conditional `UPDATE` on the delivery's state, so the +database decides who wins, not a check in application code. Exactly one request +requeues the delivery. The other reads back the state that was actually settled +on and is refused with `409 delivery_not_replayable`, the same answer it would +get for a delivery that was never failed. No duplicate work is created either +way. This is tested against real PostgreSQL with independent connections. + +### Pagination + +Both list routes page the same way. Items come back newest first, ordered by +creation time with the identifier as a tie-break, so a page is exactly +reproducible. `limit` is 1 to 100 and defaults to 50; anything outside that is +refused with `422 invalid_request` rather than quietly clamped. + +`next_cursor` is the identifier of the last item on the page. Pass it as +`cursor` to continue: + +```bash +curl "http://127.0.0.1:8000/deliveries?limit=50&cursor=whd_9a3f..." \ + -H "Authorization: Bearer $HYMICAL_KEY" +``` + +A full page always carries a cursor, even when it happens to be the last one, +because knowing otherwise would cost an extra read on every page. A walk +therefore ends on one empty page rather than on a null cursor. A cursor that +names no row is refused with `422 invalid_cursor`. + +There is deliberately **no total count**. Counting an operational table on every +page is not free, and nothing here needs the number. + ### Try it ```bash @@ -694,16 +1024,19 @@ add. | 401 | `invalid_api_key` | The management API key is malformed, unknown or revoked | | 404 | `invalid_endpoint_id` | Submission path is not a well-formed endpoint ID | | 404 | `endpoint_not_found` | Endpoint ID is well formed but no such endpoint exists | +| 404 | `delivery_not_found` | No delivery with that ID exists | | 404 | `not_found` | Unknown path | | 405 | `method_not_allowed` | Wrong method for a known path | | 409 | `endpoint_inactive` | Endpoint exists but is not accepting submissions | | 409 | `endpoint_already_exists` | Endpoint ID is already taken | | 409 | `idempotency_conflict` | Idempotency key already used for different content | +| 409 | `delivery_not_replayable` | Delivery has not terminally failed | | 413 | `request_body_too_large` | Body exceeded `FORMS_MAX_BODY_BYTES` | | 415 | `unsupported_media_type` | Content type is not a supported form encoding | | 422 | `empty_submission` | No fields were submitted | +| 422 | `invalid_cursor` | Pagination cursor does not continue from a known row | | 422 | `invalid_endpoint_id` | Endpoint ID in a request body breaks the ID rules | -| 422 | `invalid_request` | Request body failed schema validation | +| 422 | `invalid_request` | Request body or query parameters failed schema validation | | 422 | `invalid_webhook_url` | Webhook destination is malformed or not permitted | | 422 | `file_upload_not_supported`| A multipart part carried a file | | 422 | ingestion rule codes | See below | @@ -782,7 +1115,8 @@ schema, which is what keeps the fast suite's shortcut honest. Others upgrade a PostgreSQL database that already holds endpoints, submissions, deliveries and attempts, and check that the data is exactly what it was afterwards, that the downgrade removes only what the newer revision added, and that the data survives -that too. +that too. Another settles the manual replay race: several real connections +replay one failed delivery at the same instant, and exactly one of them wins. CI runs the lint, format and type checks once, the fast suite across Python 3.11 to 3.13, and the PostgreSQL suite once against a PostgreSQL 17 service. @@ -807,7 +1141,11 @@ src/hymical_forms/ schema.py the boundary between the application and Alembic main.py ASGI entrypoint api/ HTTP routes and response models + endpoints.py creating, listing, inspecting and changing endpoints + deliveries.py reading the delivery queue and replaying a failed delivery + submissions.py public form ingestion security.py the management authentication dependency + pagination.py the one cursor design both list routes share migrations/ Alembic environment and revisions ``` @@ -861,6 +1199,21 @@ This is covered by real integration tests: concurrent PostgreSQL sessions claim disjoint work, a row another worker holds is skipped rather than waited on, and an expired lease becomes reclaimable by exactly one worker. +Manual replay is settled the same way, by the database rather than by the +application: one conditional `UPDATE` moves a delivery out of `failed`, and a +second simultaneous request matches no row and is refused. Reading the state, +judging it in Python and then writing it would let both requests pass the check +and both reset the retry cycle, which is the duplicated work this exists to +prevent. + +The two attempt counters on a delivery are what let a replay be both honest and +useful. `attempts` is the lifetime total and only ever rises, so it can number +the audit trail without a number ever being reused. `cycle_attempts` is the +count since the delivery last entered the queue, so the retry policy has +something to measure that a replay is allowed to reset. Before anything is +replayed the two are the same number, which is why upgrading an existing +database sets one from the other. + ## Limitations - **Delivery is at-least-once, never exactly-once.** A worker that delivers @@ -872,8 +1225,13 @@ an expired lease becomes reclaimable by exactly one worker. and every valid key can do everything a management key can do. Separate keys are useful for revoking one caller without disturbing another, and for nothing else yet. -- **A failed delivery is final and cannot be replayed.** Once a delivery reaches - `failed`, nothing retries it and there is no manual replay route. +- **A failed delivery is never retried on its own.** It stays `failed` until an + operator replays it, and nothing notices that it failed for you: there is no + alerting, no dead-letter notification and no automatic sweep. +- **A concurrent `PATCH` on one endpoint is last-write-wins.** The row is locked + for the transaction on PostgreSQL, so two operators rotating a signing secret + at the same moment are serialised and both get an answer that matches what was + stored. SQLite has no such locking and is not a production target. - **The lease must outlast a delivery attempt.** A batch is delivered concurrently, so it takes about as long as its slowest single delivery rather than the sum, but if `FORMS_WORKER_LEASE_SECONDS` were set below the connect @@ -899,18 +1257,20 @@ an expired lease becomes reclaimable by exactly one worker. to absorb and the ingestion path never pays because it is not authenticated. A failure to write it is logged and ignored rather than allowed to turn a valid credential into a `401`. -- **A webhook can only be set when the endpoint is created.** There is no route - to change a destination or rotate a signing secret. -- **No API for delivery attempts.** They are recorded, but reading them means - querying the database directly. +- **A signing secret cannot be rotated in place.** Rotation happens only as a + side effect of changing the destination, so re-keying a receiver that stays at + the same URL means pointing the endpoint elsewhere and back, or standing up a + second URL. A dedicated rotate action is not implemented. - **Migrations are applied by hand, one command at a time.** There is no zero-downtime story and none is claimed: a migration that rewrites a table will lock it, and a build whose expected revision does not match the database refuses to start rather than serving against a schema it does not understand. Plan a deploy as migrate-then-restart. -- **No way to read submissions back over the API.** They are stored, but - retrieval, export and retention are not implemented. -- **No route to list, update or delete endpoints.** +- **No way to read submissions back over the API.** They are stored, and a + delivery can be inspected, but the submitted values themselves are deliberately + not exposed by any route. Retrieval, export and retention are not implemented. +- **Endpoints cannot be deleted, and an endpoint ID cannot be changed.** + Disabling an endpoint is the way to stop it accepting submissions. - **No file uploads.** Multipart text fields are accepted; file parts are rejected. - **`multipart/form-data` bodies are buffered in memory,** bounded by diff --git a/src/hymical_forms/api/deliveries.py b/src/hymical_forms/api/deliveries.py new file mode 100644 index 0000000..7304de9 --- /dev/null +++ b/src/hymical_forms/api/deliveries.py @@ -0,0 +1,352 @@ +""" +delivery inspection and manual replay: the operational view of the outbox + +These routes read the durable delivery queue and can put one terminally failed +delivery back into it. They never send anything: the API process holds no +outbound HTTP client, and replay is a state change that makes existing work +claimable by the existing worker again. + +Nothing here exposes the signing secret a delivery snapshotted, and nothing here +exposes the submitted form contents the delivery carries. +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from http import HTTPStatus +from typing import Annotated + +from fastapi import APIRouter, Query +from pydantic import BaseModel, Field + +from hymical_forms import models, storage +from hymical_forms.api.pagination import ( + DEFAULT_PAGE_SIZE, + CursorQuery, + InvalidCursor, + LimitQuery, + next_cursor, +) +from hymical_forms.api.security import ManagementKeyDep +from hymical_forms.db import SessionDep +from hymical_forms.errors import ApiError, ErrorResponse +from hymical_forms.ingestion import ENDPOINT_ID_MAX_LENGTH +from hymical_forms.models import utcnow +from hymical_forms.webhooks import DeliveryOutcome, DeliveryState + +logger = logging.getLogger(__name__) + +router = APIRouter(tags=["deliveries"]) + +UNAUTHENTICATED = { + "model": ErrorResponse, + "description": "Missing or invalid management API key", +} + +EndpointFilter = Annotated[ + str | None, + Query( + max_length=ENDPOINT_ID_MAX_LENGTH, + description="Only deliveries for this endpoint. Omit for every endpoint.", + ), +] + +StateFilter = Annotated[ + DeliveryState | None, + Query(description="Only deliveries in this state. Omit for every state."), +] + + +class DeliveryNotFound(ApiError): + """ + raised when a management route addresses a delivery that does not exist + """ + + status_code = HTTPStatus.NOT_FOUND + code = "delivery_not_found" + + def __init__(self, delivery_id: str) -> None: + """ + name the delivery identifier that could not be resolved + :param delivery_id: the identifier taken from the request path + """ + super().__init__( + f"No webhook delivery with the ID {delivery_id!r} exists.", + details={"delivery_id": delivery_id}, + ) + + +class DeliveryNotReplayable(ApiError): + """ + raised when a replay was asked for a delivery that has not terminally failed + """ + + # A 409 rather than a 422: the body was fine and the request is refused + # because of the state the resource is in, which is what 409 describes. It is + # deliberately not a 500, because this is an ordinary answer rather than a + # fault. + status_code = HTTPStatus.CONFLICT + code = "delivery_not_replayable" + + def __init__(self, delivery_id: str, state: str) -> None: + """ + report the state that makes the delivery ineligible for replay + :param delivery_id: the delivery the caller asked to replay + :param state: the state the delivery is actually in + """ + # The state is named because it is what tells an operator whether to wait, + # to stop, or that somebody else replayed this a moment ago. + super().__init__( + f"Only a delivery that has terminally failed can be replayed. Delivery " + f"{delivery_id!r} is {state!r}.", + details={"delivery_id": delivery_id, "state": state}, + ) + + +class DeliveryAttemptView(BaseModel): + """ + one recorded outbound request, as returned by the API + """ + + # Response bodies are not stored by this service and so cannot be reported + # here. Neither the signing secret nor any request header is stored either. + attempt_number: int = Field( + description=( + "Position of this attempt in the delivery's lifetime history, starting at 1. " + "Numbers are never reused, including across a manual replay." + ) + ) + attempted_at: datetime = Field(description="UTC timestamp of when the request was made.") + outcome: DeliveryOutcome = Field(description="What the attempt produced.") + response_status: int | None = Field( + description="HTTP status the destination answered with, or null if it never answered." + ) + error: str | None = Field( + description="Bounded failure message, or null when the attempt succeeded." + ) + + +class DeliveryView(BaseModel): + """ + one logical delivery, as returned by the API + """ + + id: str = Field(description="Opaque identifier for this logical delivery.") + submission_id: str = Field(description="The submission this delivery carries.") + endpoint_id: str = Field(description="The endpoint that submission was addressed to.") + state: DeliveryState = Field(description="Where this delivery has got to.") + destination_url: str = Field( + description=( + "Where this delivery is sent, as snapshotted when the submission was " + "accepted. Changing the endpoint's webhook does not change it." + ) + ) + attempt_count: int = Field( + description="How many requests have ever been made for this delivery." + ) + cycle_attempt_count: int = Field( + description=( + "How many requests have been made since the delivery last entered the queue. " + "The retry allowance is measured against this, and a manual replay resets it." + ) + ) + next_attempt_at: datetime = Field( + description="UTC timestamp of when this delivery next becomes due." + ) + created_at: datetime = Field(description="UTC timestamp of when the delivery was queued.") + completed_at: datetime | None = Field( + description="UTC timestamp of when the delivery finished, or null if it has not." + ) + + +class DeliveryDetail(DeliveryView): + """ + one logical delivery together with its ordered attempt history + """ + + attempts: list[DeliveryAttemptView] = Field( + description="Every request made for this delivery, lowest attempt number first." + ) + + +class DeliveryPage(BaseModel): + """ + one page of deliveries + """ + + items: list[DeliveryView] = Field(description="The deliveries on this page, newest first.") + next_cursor: str | None = Field( + description=( + "Pass as `cursor` to read the next page, or null when this is certainly " + "the last one. A full page always carries a cursor, so the final request " + "of a walk returns an empty page." + ) + ) + + +@router.get( + "/deliveries", + summary="List webhook deliveries", + responses={ + 401: UNAUTHENTICATED, + 422: {"model": ErrorResponse, "description": "Invalid filter, page size or cursor"}, + 503: {"model": ErrorResponse, "description": "Database unavailable"}, + }, +) +def list_deliveries( + session: SessionDep, + principal: ManagementKeyDep, + endpoint_id: EndpointFilter = None, + state: StateFilter = None, + limit: LimitQuery = DEFAULT_PAGE_SIZE, + cursor: CursorQuery = None, +) -> DeliveryPage: + """ + read a page of the delivery queue + :param session: the session this request does its database work through + :param principal: the management key this request authenticated as + :param endpoint_id: only deliveries for this endpoint, or None for every endpoint + :param state: only deliveries in this state, or None for every state + :param limit: the most deliveries to return + :param cursor: the previous page's cursor, or None to read the first page + :returns: one page of deliveries, newest first + :raises InvalidCursor: if the cursor does not continue from a known delivery + """ + # The submitted fields are not summarised, counted or echoed here. Reading a + # submission back is not something this API does yet, and a delivery listing + # is not the place to start. + try: + page = storage.list_deliveries( + session, + limit=limit, + after=cursor, + endpoint_id=endpoint_id, + state=state.value if state is not None else None, + ) + except storage.UnknownCursor as exc: + raise InvalidCursor() from exc + + return DeliveryPage( + items=[_view(record) for record in page], + next_cursor=next_cursor([record.delivery.id for record in page], limit=limit), + ) + + +@router.get( + "/deliveries/{delivery_id}", + summary="Inspect one webhook delivery", + responses={ + 401: UNAUTHENTICATED, + 404: {"model": ErrorResponse, "description": "No such delivery"}, + 503: {"model": ErrorResponse, "description": "Database unavailable"}, + }, +) +def get_delivery( + delivery_id: str, + session: SessionDep, + principal: ManagementKeyDep, +) -> DeliveryDetail: + """ + read one delivery and the ordered history of the requests made for it + :param delivery_id: delivery identifier taken from the request path + :param session: the session this request does its database work through + :param principal: the management key this request authenticated as + :returns: the delivery and its attempts, carrying no signing secret + :raises DeliveryNotFound: if no delivery holds that identifier + """ + record = storage.get_delivery(session, delivery_id) + if record is None: + raise DeliveryNotFound(delivery_id) + + attempts = storage.list_delivery_attempts(session, delivery_id) + return DeliveryDetail( + **_view(record).model_dump(), + attempts=[_attempt(attempt) for attempt in attempts], + ) + + +@router.post( + "/deliveries/{delivery_id}/replay", + summary="Requeue a failed webhook delivery", + responses={ + 401: UNAUTHENTICATED, + 404: {"model": ErrorResponse, "description": "No such delivery"}, + 409: {"model": ErrorResponse, "description": "Delivery has not terminally failed"}, + 503: {"model": ErrorResponse, "description": "Database unavailable"}, + }, +) +def replay_delivery( + delivery_id: str, + session: SessionDep, + principal: ManagementKeyDep, +) -> DeliveryView: + """ + put a terminally failed delivery back into the queue for the worker + :param delivery_id: delivery identifier taken from the request path + :param session: the session this request does its database work through + :param principal: the management key this request authenticated as + :returns: the delivery as it now stands, due and waiting for a worker + :raises DeliveryNotFound: if no delivery holds that identifier + :raises DeliveryNotReplayable: if the delivery has not terminally failed + """ + # 200 rather than 202. A 202 would say "your request has been accepted and + # will be carried out later", which invites reading the replay itself as the + # delivery. It is not: this request completes here, and what it leaves behind + # is a row the ordinary worker will claim on its next poll. The body is the + # delivery's new state, which is the whole of what this request did. + # + # No outbound request is made. This process has no HTTP client to make one + # with, which is what keeps that guarantee structural rather than a promise. + outcome = storage.requeue_failed_delivery(session, delivery_id, now=utcnow()) + if outcome.record is None: + raise DeliveryNotFound(delivery_id) + if not outcome.requeued: + # Either the delivery was never failed, or two operators replayed it at + # once and this is the loser. Both are answered from the state the + # database settled on, so the answer is the same however the race went. + raise DeliveryNotReplayable(delivery_id, outcome.record.delivery.state) + + logger.info( + "delivery %s requeued by management key %s (%s)", + delivery_id, + principal.key_id, + principal.name, + ) + return _view(outcome.record) + + +def _view(record: storage.DeliveryRecord) -> DeliveryView: + """ + render a delivery for a management read + :param record: the delivery and the endpoint it belongs to + :returns: the delivery's operational state, carrying no signing secret + """ + delivery = record.delivery + return DeliveryView( + id=delivery.id, + submission_id=delivery.submission_id, + endpoint_id=record.endpoint_id, + state=DeliveryState(delivery.state), + destination_url=delivery.destination_url, + attempt_count=delivery.attempts, + cycle_attempt_count=delivery.cycle_attempts, + next_attempt_at=delivery.next_attempt_at, + created_at=delivery.created_at, + completed_at=delivery.completed_at, + ) + + +def _attempt(attempt: models.DeliveryAttempt) -> DeliveryAttemptView: + """ + render one recorded attempt for a management read + :param attempt: the persisted attempt + :returns: what the attempt produced, with no credential material + """ + return DeliveryAttemptView( + attempt_number=attempt.attempt_number, + attempted_at=attempt.attempted_at, + outcome=DeliveryOutcome(attempt.outcome), + response_status=attempt.response_status, + error=attempt.error, + ) diff --git a/src/hymical_forms/api/endpoints.py b/src/hymical_forms/api/endpoints.py index 0b68798..f3a5c57 100644 --- a/src/hymical_forms/api/endpoints.py +++ b/src/hymical_forms/api/endpoints.py @@ -1,5 +1,11 @@ """ -endpoint management: ``POST /endpoints`` +endpoint management: creating, listing, inspecting and changing form endpoints + +Every route here is behind the management authentication boundary, declared the +same way: by asking for :data:`~hymical_forms.api.security.ManagementKeyDep`. +None of them reads a submitted form's contents, and none of them reads a webhook +signing secret back out. A secret leaves this service only in the response of the +mutation that generated it. """ from __future__ import annotations @@ -12,18 +18,32 @@ from pydantic import BaseModel, Field from hymical_forms import storage, webhooks +from hymical_forms.api.pagination import ( + DEFAULT_PAGE_SIZE, + CursorQuery, + InvalidCursor, + LimitQuery, + next_cursor, +) from hymical_forms.api.security import ManagementKeyDep from hymical_forms.config import Settings from hymical_forms.db import SessionDep from hymical_forms.errors import ApiError, ErrorResponse from hymical_forms.ingestion import ENDPOINT_ID_RULE, is_valid_endpoint_id -from hymical_forms.models import ENDPOINT_NAME_MAX_LENGTH +from hymical_forms.models import ENDPOINT_NAME_MAX_LENGTH, Endpoint from hymical_forms.webhooks import WEBHOOK_URL_MAX_LENGTH logger = logging.getLogger(__name__) router = APIRouter(tags=["endpoints"]) +# Repeated on every management route in this module, so that an integrator meets +# one description of the authentication boundary rather than several. +UNAUTHENTICATED = { + "model": ErrorResponse, + "description": "Missing or invalid management API key", +} + class InvalidEndpointId(ApiError): """ @@ -62,6 +82,27 @@ def __init__(self, endpoint_id: str) -> None: ) +class EndpointNotFound(ApiError): + """ + raised when a management route addresses an endpoint that does not exist + """ + + # The same code the public ingestion route answers with for the same + # condition, because it is the same condition: no endpoint holds that ID. + status_code = HTTPStatus.NOT_FOUND + code = "endpoint_not_found" + + def __init__(self, endpoint_id: str) -> None: + """ + name the endpoint identifier that could not be resolved + :param endpoint_id: the identifier taken from the request path + """ + super().__init__( + f"No form endpoint with the ID {endpoint_id!r} exists.", + details={"endpoint_id": endpoint_id}, + ) + + class InvalidWebhookUrl(ApiError): """ raised when a webhook destination is malformed or not permitted @@ -103,22 +144,87 @@ class CreateEndpointRequest(BaseModel): ) -class EndpointResponse(BaseModel): +class UpdateEndpointRequest(BaseModel): + """ + the body accepted when changing an endpoint + + Every field is optional and an omitted one is left alone. ``webhook_url`` is + the only field where an explicit ``null`` means something of its own: it + removes the webhook. Sending ``null`` for the other two is the same as + omitting them, because neither has a meaningful empty value. + """ + + # The endpoint's ID is deliberately not here. It is the primary key and it + # appears in the action URL of every HTML form pointing at the endpoint, so + # changing it would break deployed forms and orphan stored submissions. + name: str | None = Field( + default=None, + min_length=1, + max_length=ENDPOINT_NAME_MAX_LENGTH, + description="New human-readable label. Omit to leave the label unchanged.", + ) + is_active: bool | None = Field( + default=None, + description=( + "Whether the endpoint accepts submissions. Setting it to false takes effect " + "on the very next submission; queued deliveries are unaffected." + ), + ) + webhook_url: str | None = Field( + default=None, + max_length=WEBHOOK_URL_MAX_LENGTH, + description=( + "New http or https destination. Omit to leave the destination and its " + "signing secret alone, or send null to remove the webhook entirely. A new " + "destination is given a new signing secret, returned once in the response." + ), + ) + + +class EndpointView(BaseModel): """ - an endpoint as returned by the API + an endpoint as returned by a management read """ + # No webhook_secret field exists here at all, rather than one that is always + # null. A read route that cannot name the secret cannot leak it. id: str = Field(description="The public identifier the endpoint answers on.") name: str = Field(description="Human-readable label for the endpoint.") is_active: bool = Field(description="Whether the endpoint currently accepts submissions.") created_at: datetime = Field(description="UTC timestamp of when the endpoint was created.") webhook_url: str | None = Field( - description="Where accepted submissions are delivered, or null if none is configured." + description=( + "Where accepted submissions are delivered, or null if none is configured. " + "This is configuration rather than a credential, so it is readable." + ) + ) + + +class EndpointPage(BaseModel): + """ + one page of endpoints + """ + + items: list[EndpointView] = Field(description="The endpoints on this page, newest first.") + next_cursor: str | None = Field( + description=( + "Pass as `cursor` to read the next page, or null when this is certainly " + "the last one. A full page always carries a cursor, so the final request " + "of a walk returns an empty page." + ) ) + + +class EndpointResponse(EndpointView): + """ + an endpoint as returned by a mutation that may have generated a secret + """ + webhook_secret: str | None = Field( description=( - "The signing secret for this endpoint's webhook. Returned only here, at " - "creation, and never retrievable again. Null if no webhook is configured." + "The signing secret for this endpoint's webhook, returned only in the " + "response of the request that generated it and never retrievable again. " + "Null when this request did not generate one." ) ) @@ -128,7 +234,7 @@ class EndpointResponse(BaseModel): status_code=HTTPStatus.CREATED, summary="Create a form endpoint", responses={ - 401: {"model": ErrorResponse, "description": "Missing or invalid management API key"}, + 401: UNAUTHENTICATED, 409: {"model": ErrorResponse, "description": "Endpoint ID already taken"}, 422: { "model": ErrorResponse, @@ -203,6 +309,186 @@ def create_endpoint( # The secret leaves the service exactly once, in this response. There is no # route that reads it back, so a caller that loses it has to make a new # endpoint rather than being handed the old secret again. + return _mutated(endpoint, secret) + + +@router.get( + "/endpoints", + summary="List form endpoints", + responses={ + 401: UNAUTHENTICATED, + 422: {"model": ErrorResponse, "description": "Invalid page size or cursor"}, + 503: {"model": ErrorResponse, "description": "Database unavailable"}, + }, +) +def list_endpoints( + session: SessionDep, + principal: ManagementKeyDep, + limit: LimitQuery = DEFAULT_PAGE_SIZE, + cursor: CursorQuery = None, +) -> EndpointPage: + """ + read a page of the endpoints this service holds + :param session: the session this request does its database work through + :param principal: the management key this request authenticated as + :param limit: the most endpoints to return + :param cursor: the previous page's cursor, or None to read the first page + :returns: one page of endpoints, newest first + :raises InvalidCursor: if the cursor does not continue from a known endpoint + """ + try: + page = storage.list_endpoints(session, limit=limit, after=cursor) + except storage.UnknownCursor as exc: + raise InvalidCursor() from exc + + return EndpointPage( + items=[_view(endpoint) for endpoint in page], + next_cursor=next_cursor([endpoint.id for endpoint in page], limit=limit), + ) + + +@router.get( + "/endpoints/{endpoint_id}", + summary="Inspect one form endpoint", + responses={ + 401: UNAUTHENTICATED, + 404: {"model": ErrorResponse, "description": "No such endpoint"}, + 503: {"model": ErrorResponse, "description": "Database unavailable"}, + }, +) +def get_endpoint( + endpoint_id: str, + session: SessionDep, + principal: ManagementKeyDep, +) -> EndpointView: + """ + read one endpoint's configuration + :param endpoint_id: endpoint identifier taken from the request path + :param session: the session this request does its database work through + :param principal: the management key this request authenticated as + :returns: the endpoint, carrying no signing secret + :raises EndpointNotFound: if no endpoint holds that identifier + """ + endpoint = storage.get_endpoint(session, endpoint_id) + if endpoint is None: + raise EndpointNotFound(endpoint_id) + return _view(endpoint) + + +@router.patch( + "/endpoints/{endpoint_id}", + summary="Change a form endpoint", + responses={ + 401: UNAUTHENTICATED, + 404: {"model": ErrorResponse, "description": "No such endpoint"}, + 422: {"model": ErrorResponse, "description": "Invalid name or webhook URL"}, + 503: {"model": ErrorResponse, "description": "Database unavailable"}, + }, +) +def update_endpoint( + endpoint_id: str, + payload: UpdateEndpointRequest, + request: Request, + session: SessionDep, + principal: ManagementKeyDep, +) -> EndpointResponse: + """ + change the parts of an endpoint's configuration that are safe to change + :param endpoint_id: endpoint identifier taken from the request path + :param payload: the fields to change, all of them optional + :param request: the incoming request, read for the active configuration + :param session: the session this request does its database work through + :param principal: the management key this request authenticated as + :returns: the endpoint as persisted, carrying a signing secret only if one was made + :raises EndpointNotFound: if no endpoint holds that identifier + """ + # PATCH rather than PUT, because only a few fields are mutable and a PUT would + # promise that the body describes the whole resource. It does not: the ID and + # the existing signing secret are not the caller's to send. + endpoint = storage.get_endpoint_for_update(session, endpoint_id) + if endpoint is None: + raise EndpointNotFound(endpoint_id) + + settings: Settings = request.app.state.settings + webhook_url = endpoint.webhook_url + webhook_secret = endpoint.webhook_secret + generated: str | None = None + + # ``model_fields_set`` is what separates "the caller sent null" from "the + # caller said nothing", which for the destination are two different requests. + if "webhook_url" in payload.model_fields_set and payload.webhook_url != webhook_url: + webhook_url = payload.webhook_url + if webhook_url is None: + # Removing the webhook removes the secret with it, which the paired + # check constraint requires and which is right anyway: a secret with + # no destination is a credential nobody is using. + webhook_secret = None + else: + try: + webhooks.validate_webhook_url( + webhook_url, + allow_private_targets=settings.allow_private_webhook_targets, + ) + except webhooks.WebhookUrlRejected as exc: + raise InvalidWebhookUrl(exc.reason) from exc + # A new destination gets a new secret, always. Carrying the old one + # over would hand a receiver that never had it the ability to verify + # signatures, and would leave the previous receiver holding a live + # secret for payloads it no longer gets. An unchanged destination is + # caught by the comparison above and keeps its secret untouched. + webhook_secret = webhooks.new_signing_secret() + generated = webhook_secret + + updated = storage.update_endpoint( + session, + endpoint, + name=payload.name if payload.name is not None else endpoint.name, + is_active=payload.is_active if payload.is_active is not None else endpoint.is_active, + webhook_url=webhook_url, + webhook_secret=webhook_secret, + ) + + logger.info( + "endpoint %s updated by management key %s (%s)", + updated.id, + principal.key_id, + principal.name, + ) + if generated is not None: + # Says that a secret was made, never what it is. + logger.info( + "endpoint %s has a new webhook destination and a new signing secret", updated.id + ) + + # Deliveries already queued keep the destination and secret they snapshotted + # when their submission was accepted. Changing configuration here does not + # redirect work that is already owed, and does not leave a queued payload + # signed with a secret its receiver never had. + return _mutated(updated, generated) + + +def _view(endpoint: Endpoint) -> EndpointView: + """ + render an endpoint for a management read + :param endpoint: the persisted endpoint + :returns: the endpoint's safe configuration + """ + return EndpointView( + id=endpoint.id, + name=endpoint.name, + is_active=endpoint.is_active, + created_at=endpoint.created_at, + webhook_url=endpoint.webhook_url, + ) + + +def _mutated(endpoint: Endpoint, secret: str | None) -> EndpointResponse: + """ + render an endpoint for the response of a request that changed it + :param endpoint: the persisted endpoint + :param secret: a signing secret this request generated, or None if it made none + :returns: the endpoint, carrying the new secret exactly once + """ return EndpointResponse( id=endpoint.id, name=endpoint.name, diff --git a/src/hymical_forms/api/pagination.py b/src/hymical_forms/api/pagination.py new file mode 100644 index 0000000..f68b674 --- /dev/null +++ b/src/hymical_forms/api/pagination.py @@ -0,0 +1,85 @@ +""" +the one pagination design the management list routes share + +A page is a bounded number of items in a fixed order, newest first, and the +cursor is the opaque identifier of the last item on the page. There is no total +count: counting an operational table on every page is not free, and nothing here +needs the number. + +This is deliberately not a framework. It is the two query parameters, the one +error, and the single rule for deciding whether to hand back a cursor, kept in +one place so that both list routes cannot drift apart. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from http import HTTPStatus +from typing import Annotated + +from fastapi import Query + +from hymical_forms.errors import ApiError + +DEFAULT_PAGE_SIZE = 50 + +# The ceiling exists so that no request can ask for an unbounded read, whatever +# the caller passes. A limit above it is refused rather than quietly clamped: a +# caller that asked for a thousand rows and silently received a hundred would +# page wrongly and never find out. +MAX_PAGE_SIZE = 100 + +# Cursors are identifiers this service generated, so this only has to be wide +# enough for the longest of them. +CURSOR_MAX_LENGTH = 64 + + +class InvalidCursor(ApiError): + """ + raised when a cursor does not name a row that can be continued from + """ + + status_code = HTTPStatus.UNPROCESSABLE_ENTITY + code = "invalid_cursor" + + def __init__(self) -> None: + """ + say that the cursor is unusable, without describing what it missed + """ + super().__init__( + "The cursor does not continue from a known row. Request the first page " + "without one to start again.", + details={"field": "cursor"}, + ) + + +LimitQuery = Annotated[ + int, + Query( + ge=1, + le=MAX_PAGE_SIZE, + description=f"How many items to return, at most {MAX_PAGE_SIZE}.", + ), +] + +CursorQuery = Annotated[ + str | None, + Query( + max_length=CURSOR_MAX_LENGTH, + description="The `next_cursor` of the previous page. Omit it to read the first page.", + ), +] + + +def next_cursor(page: Sequence[str], *, limit: int) -> str | None: + """ + decide what cursor, if any, the caller should continue from + :param page: the identifiers of the items being returned, in page order + :param limit: the page size the request asked for + :returns: the cursor to continue from, or None when there is certainly no more + """ + # A full page hands back a cursor even when it happens to be the last one, + # because knowing that would cost an extra row read on every page. The caller + # therefore ends on one empty page rather than on a null cursor, which is the + # ordinary shape of cursor pagination and is documented as such. + return page[-1] if len(page) == limit else None diff --git a/src/hymical_forms/app.py b/src/hymical_forms/app.py index 8d817f6..edbad6f 100644 --- a/src/hymical_forms/app.py +++ b/src/hymical_forms/app.py @@ -10,7 +10,7 @@ from fastapi import FastAPI from hymical_forms import __version__ -from hymical_forms.api import endpoints, health, submissions +from hymical_forms.api import deliveries, endpoints, health, submissions from hymical_forms.config import Settings from hymical_forms.db import create_engine_from_url, create_session_factory from hymical_forms.errors import register_exception_handlers @@ -77,6 +77,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: app.include_router(health.router) app.include_router(endpoints.router) + app.include_router(deliveries.router) app.include_router(submissions.router) return app diff --git a/src/hymical_forms/migrations/versions/0003_20260824_delivery_retry_cycles.py b/src/hymical_forms/migrations/versions/0003_20260824_delivery_retry_cycles.py new file mode 100644 index 0000000..b412b1f --- /dev/null +++ b/src/hymical_forms/migrations/versions/0003_20260824_delivery_retry_cycles.py @@ -0,0 +1,63 @@ +""" +delivery retry cycles + +Manual replay needs a delivery that has already used up its allowance to be able +to earn a fresh one. ``attempts`` cannot do that job on its own: it numbers the +attempt history, so resetting it would make an attempt number repeat, and +leaving it alone would make a replayed delivery fail on its very next attempt. + +This adds one column, ``cycle_attempts``, holding the attempts made since the +delivery last entered the queue. The retry policy is measured against it, and a +replay resets it to zero. ``attempts`` keeps counting every request ever made. + +Existing deliveries have never been replayed, so their current cycle is the whole +of their history and the backfill sets the new column from ``attempts``. The +downgrade drops the column and nothing else: ``attempts`` still holds every +delivery's lifetime total, which is what the 0002 build reads it as. + +Timestamps are not touched here. Nothing in this revision imports application +code, for the reason given in ``0001``: a migration is a frozen record of a +change, not a view of the current models. + +revision: 0003 +revises: 0002 +created: 2026-08-24 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0003" +down_revision: str | None = "0002" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """ + add the per-cycle attempt counter a manual replay resets + """ + # Added nullable so a populated table can take the column at all, then + # backfilled, then tightened. The tightening is the only step that needs an + # ALTER, which is why it goes through batch mode: SQLite cannot change a + # column in place and has to rebuild the table instead. Batch mode is inert + # on PostgreSQL. + op.add_column("webhook_deliveries", sa.Column("cycle_attempts", sa.Integer(), nullable=True)) + op.execute("UPDATE webhook_deliveries SET cycle_attempts = attempts") + with op.batch_alter_table("webhook_deliveries") as batch: + batch.alter_column("cycle_attempts", existing_type=sa.Integer(), nullable=False) + + +def downgrade() -> None: + """ + remove the per-cycle attempt counter + """ + # No data is lost that the 0002 build can tell. A delivery replayed while + # this revision was applied comes back with its lifetime attempt total in + # ``attempts``, which the older build reads as its retry allowance, so a + # replayed delivery simply has no allowance left again. + op.drop_column("webhook_deliveries", "cycle_attempts") diff --git a/src/hymical_forms/models.py b/src/hymical_forms/models.py index 2880904..de0e4c8 100644 --- a/src/hymical_forms/models.py +++ b/src/hymical_forms/models.py @@ -280,8 +280,18 @@ class WebhookDelivery(Base): state: Mapped[str] = mapped_column( String(DELIVERY_STATE_MAX_LENGTH), default=DeliveryState.PENDING ) + + # Every request ever made for this delivery. It only ever goes up, which is + # what lets it number the attempt history without a number being reused. attempts: Mapped[int] = mapped_column(default=0) + # The requests made since this delivery last entered the queue. A manual + # replay resets this and nothing else, so a delivery that exhausted its + # allowance gets a whole fresh retry cycle rather than one last attempt, + # while the history it already has keeps its numbering. Until something is + # replayed the two counters are equal. + cycle_attempts: Mapped[int] = mapped_column(default=0) + # When this delivery next becomes due. Indexed with the state because that # pair is exactly what a worker scans for on every poll. next_attempt_at: Mapped[datetime] = mapped_column(UtcDateTime, index=True) diff --git a/src/hymical_forms/storage.py b/src/hymical_forms/storage.py index 0cc57a8..f1d2a6a 100644 --- a/src/hymical_forms/storage.py +++ b/src/hymical_forms/storage.py @@ -8,18 +8,20 @@ deliver it as one atomic unit and must roll both back together to settle an idempotency race; :func:`claim_due_deliveries`, because a claim is only worth anything once it is committed; :func:`complete_attempt`, because the audit -record and the state it justifies have to land together; and the two management -key writes, :func:`revoke_management_key` and :func:`record_management_key_use`, -because each is the whole of what its caller came to do. +record and the state it justifies have to land together; :func:`update_endpoint` +and :func:`requeue_failed_delivery`, because each is the whole of what its +management request came to do; and the two management key writes, +:func:`revoke_management_key` and :func:`record_management_key_use`, for the +same reason. """ from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta -from typing import Any, cast +from typing import Any, TypeVar, cast -from sqlalchemy import ColumnElement, and_, or_, select, update +from sqlalchemy import ColumnElement, Select, and_, or_, select, tuple_, update from sqlalchemy.engine import CursorResult from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -37,6 +39,25 @@ new_webhook_delivery_id, ) +# The two tables management routes page through. Both are keyed by an opaque +# identifier and carry a creation timestamp, which is all cursor pagination here +# asks of a table. +_Paged = TypeVar("_Paged", models.Endpoint, models.WebhookDelivery) + + +class UnknownCursor(Exception): + """ + raised when a pagination cursor does not name a row to continue from + """ + + def __init__(self, cursor: str) -> None: + """ + record the cursor that could not be resolved + :param cursor: the opaque cursor the caller asked to continue from + """ + super().__init__("the pagination cursor does not name a known row") + self.cursor = cursor + class EndpointAlreadyExists(Exception): """ @@ -100,6 +121,73 @@ def get_endpoint(session: Session, endpoint_id: str) -> models.Endpoint | None: return session.get(models.Endpoint, endpoint_id) +def list_endpoints( + session: Session, *, limit: int, after: str | None = None +) -> list[models.Endpoint]: + """ + read one page of endpoints, newest first + :param session: the session to query through + :param limit: the most endpoints to return + :param after: identifier of the last endpoint on the previous page, or None to start + :returns: the page, at most ``limit`` long + :raises UnknownCursor: if ``after`` does not name an existing endpoint + """ + endpoint = models.Endpoint + statement = select(endpoint).order_by(endpoint.created_at.desc(), endpoint.id.desc()) + if after is not None: + statement = statement.where(_page_after(session, endpoint, after)) + return list(session.scalars(statement.limit(limit))) + + +def get_endpoint_for_update(session: Session, endpoint_id: str) -> models.Endpoint | None: + """ + read an endpoint with the intention of changing it in the same transaction + :param session: the session to query through + :param endpoint_id: the public identifier to resolve + :returns: the endpoint, or None if no endpoint holds that identifier + """ + # Locked on PostgreSQL, because two operators changing one endpoint's webhook + # at the same moment would otherwise both read the old destination, both mint + # a secret, and leave one of them holding a secret this service never stored. + # SQLite has no row locking and serialises writers anyway; it is not a + # production target. + statement = select(models.Endpoint).where(models.Endpoint.id == endpoint_id) + if session.get_bind().dialect.name == "postgresql": + statement = statement.with_for_update() + return session.scalars(statement).one_or_none() + + +def update_endpoint( + session: Session, + endpoint: models.Endpoint, + *, + name: str, + is_active: bool, + webhook_url: str | None, + webhook_secret: str | None, +) -> models.Endpoint: + """ + write a resolved endpoint configuration and commit it + :param session: the session to write through + :param endpoint: the endpoint being changed, already loaded in this transaction + :param name: the label the endpoint should end up with + :param is_active: whether the endpoint should accept submissions + :param webhook_url: the destination it should end up with, or None for no webhook + :param webhook_secret: the signing secret for that destination, paired with the URL + :returns: the endpoint, changed and committed + """ + # Every value is resolved by the caller, so the decision about what "unchanged" + # means for a partial update stays in one place and this function has no + # opinion about it. Deliveries already queued are untouched on purpose: they + # snapshotted their destination and secret when the submission was accepted. + endpoint.name = name + endpoint.is_active = is_active + endpoint.webhook_url = webhook_url + endpoint.webhook_secret = webhook_secret + session.commit() + return endpoint + + class IdempotencyKeyReused(Exception): """ raised when an idempotency key was already used for different content @@ -348,7 +436,15 @@ def complete_attempt( """ # The audit row and the state it justifies are written together, so the # history can never disagree with the job about how many attempts happened. + # + # Two counters, because they answer two different questions. The attempt is + # numbered from the lifetime total, so a number is never reused however often + # the delivery has been replayed. The retry allowance is measured against the + # current cycle, so a replayed delivery gets a whole schedule again rather + # than failing immediately on a total that is already spent. Until something + # is replayed the two are the same number. attempt_number = delivery.attempts + 1 + cycle_attempt = delivery.cycle_attempts + 1 attempt = models.DeliveryAttempt( id=new_delivery_attempt_id(), delivery_id=delivery.id, @@ -363,14 +459,15 @@ def complete_attempt( session.add(attempt) delivery.attempts = attempt_number + delivery.cycle_attempts = cycle_attempt delivery.claim_expires_at = None if result.outcome is DeliveryOutcome.SUCCEEDED: delivery.state = DeliveryState.DELIVERED delivery.completed_at = now - elif is_retryable(result) and not policy.is_exhausted(attempt_number): + elif is_retryable(result) and not policy.is_exhausted(cycle_attempt): delivery.state = DeliveryState.PENDING - delivery.next_attempt_at = now + policy.delay_after(attempt_number) + delivery.next_attempt_at = now + policy.delay_after(cycle_attempt) else: # Either the receiver said something repeating will not fix, or the # allowance ran out. Either way this is the last word on the delivery. @@ -381,6 +478,150 @@ def complete_attempt( return attempt +@dataclass(frozen=True, slots=True) +class DeliveryRecord: + """ + a delivery together with the endpoint whose submission it carries + """ + + # The endpoint is not a column on the delivery: it is reached through the + # submission. Carrying it alongside means a management response can report + # which endpoint a delivery belongs to without a second query per row. + delivery: models.WebhookDelivery + endpoint_id: str + + +def list_deliveries( + session: Session, + *, + limit: int, + after: str | None = None, + endpoint_id: str | None = None, + state: str | None = None, +) -> list[DeliveryRecord]: + """ + read one page of the delivery queue, newest first + :param session: the session to query through + :param limit: the most deliveries to return + :param after: identifier of the last delivery on the previous page, or None to start + :param endpoint_id: only deliveries for this endpoint, or None for every endpoint + :param state: only deliveries in this state, or None for every state + :returns: the page, at most ``limit`` long + :raises UnknownCursor: if ``after`` does not name an existing delivery + """ + delivery = models.WebhookDelivery + statement = _delivery_query().order_by(delivery.created_at.desc(), delivery.id.desc()) + if endpoint_id is not None: + statement = statement.where(models.Submission.endpoint_id == endpoint_id) + if state is not None: + statement = statement.where(delivery.state == state) + if after is not None: + statement = statement.where(_page_after(session, delivery, after)) + + return [ + DeliveryRecord(row, endpoint) for row, endpoint in session.execute(statement.limit(limit)) + ] + + +def get_delivery(session: Session, delivery_id: str) -> DeliveryRecord | None: + """ + look one delivery up by its identifier + :param session: the session to query through + :param delivery_id: the identifier to resolve + :returns: the delivery and its endpoint, or None if no delivery holds that identifier + """ + row = session.execute( + _delivery_query().where(models.WebhookDelivery.id == delivery_id) + ).one_or_none() + return DeliveryRecord(row[0], row[1]) if row is not None else None + + +def list_delivery_attempts(session: Session, delivery_id: str) -> list[models.DeliveryAttempt]: + """ + read the ordered attempt history of one delivery + :param session: the session to query through + :param delivery_id: the delivery whose history to read + :returns: every recorded attempt, lowest attempt number first + """ + # Attempt numbers never repeat within a delivery, even across a manual replay, + # so ordering by one is already total. The identifier is a tie-break that + # cannot be reached rather than one that is expected to matter. + attempt = models.DeliveryAttempt + return list( + session.scalars( + select(attempt) + .where(attempt.delivery_id == delivery_id) + .order_by(attempt.attempt_number, attempt.id) + ) + ) + + +@dataclass(frozen=True, slots=True) +class ReplayOutcome: + """ + what a manual replay request did, and what the delivery looks like now + """ + + # ``record`` is None only when the delivery does not exist. ``requeued`` is + # False for a delivery that was not failed, which includes the loser of two + # simultaneous replays: it reads back the pending delivery the winner made. + record: DeliveryRecord | None + requeued: bool + + +def requeue_failed_delivery(session: Session, delivery_id: str, *, now: datetime) -> ReplayOutcome: + """ + put a terminally failed delivery back in the queue for the worker to claim + :param session: the session to write through + :param delivery_id: the delivery to requeue + :param now: the instant the delivery should become due again + :returns: the resulting delivery and whether this request is what requeued it + """ + # The transition is one conditional UPDATE, so the database decides who wins. + # Reading the state, judging it in Python and then writing would let two + # simultaneous replays both pass the check and both reset the retry cycle, + # which is exactly the duplicated work this is meant to prevent. + # + # Only the queueing state is touched. The snapshotted destination and signing + # secret, the submission, the lifetime attempt count and every recorded + # attempt are left exactly as they are: a replay resumes a delivery, it does + # not start a new one. Clearing ``cycle_attempts`` is what grants the fresh + # retry allowance, and clearing ``completed_at`` is required by the check + # constraint that says a delivery is finished exactly when it says it is. + delivery = models.WebhookDelivery + result = cast( + "CursorResult[Any]", + session.execute( + update(delivery) + .where(delivery.id == delivery_id) + .where(delivery.state == DeliveryState.FAILED) + .values( + state=DeliveryState.PENDING, + cycle_attempts=0, + next_attempt_at=now, + claim_expires_at=None, + completed_at=None, + ) + .execution_options(synchronize_session=False) + ), + ) + session.commit() + + # Read after the commit, so both the winner and the loser describe the state + # the database actually settled on rather than the one they hoped for. + return ReplayOutcome(get_delivery(session, delivery_id), requeued=result.rowcount == 1) + + +def _delivery_query() -> Select[tuple[models.WebhookDelivery, str]]: + """ + build the read every delivery management route starts from + :returns: a select of deliveries joined to the endpoint they belong to + """ + return select(models.WebhookDelivery, models.Submission.endpoint_id).join( + models.Submission, models.Submission.id == models.WebhookDelivery.submission_id + ) + + def create_management_key( session: Session, *, @@ -493,6 +734,29 @@ def record_management_key_use(session: Session, key_id: str, *, now: datetime) - session.commit() +def _page_after(session: Session, model: type[_Paged], cursor: str) -> ColumnElement[bool]: + """ + build the test for rows that come after a cursor, in newest-first order + :param session: the session to resolve the cursor through + :param model: the table being paged + :param cursor: the identifier of the last row on the previous page + :returns: a SQL condition matching only the rows that follow it + :raises UnknownCursor: if the cursor does not name an existing row + """ + anchor = session.get(model, cursor) + if anchor is None: + # Refused rather than treated as the first page, so a caller that pages + # past a row somebody deleted learns about it instead of silently + # starting again from the top. + raise UnknownCursor(cursor) + + # A row-value comparison, so the tie-break and the ordering are one expression + # rather than two that have to be kept in agreement. Both timestamps and both + # identifiers are compared, which is what makes a page boundary total even + # when several rows were created in the same transaction. + return tuple_(model.created_at, model.id) < (anchor.created_at, anchor.id) + + def _settle(existing: models.Submission, payload_fingerprint: str | None) -> Submission: """ decide whether an earlier submission is a replay of this one or a clash diff --git a/tests/integration/support.py b/tests/integration/support.py index 98487bc..f294b76 100644 --- a/tests/integration/support.py +++ b/tests/integration/support.py @@ -113,6 +113,67 @@ def seed_endpoint(session: Session, endpoint_id: str = "contact-form") -> models return endpoint +def seed_failed_delivery( + session: Session, + *, + now: datetime, + endpoint_id: str = "contact-form", + attempts: int = 5, +) -> str: + """ + insert a submission whose delivery has already been given up on + :param session: the session to insert through + :param now: the instant the delivery was queued and completed + :param endpoint_id: the endpoint it belongs to + :param attempts: how many requests were made before it was given up on + :returns: the delivery id + """ + submission_id = f"sub_{uuid.uuid4().hex}" + delivery_id = f"whd_{uuid.uuid4().hex}" + session.add( + models.Submission( + id=submission_id, + endpoint_id=endpoint_id, + received_at=now, + fields={"email": ["dev@example.com"]}, + ) + ) + session.add( + models.WebhookDelivery( + id=delivery_id, + submission_id=submission_id, + destination_url="https://example.invalid/hook", + signing_secret="whsec_" + "a" * 64, + state=DeliveryState.FAILED, + attempts=attempts, + cycle_attempts=attempts, + next_attempt_at=now, + created_at=now, + completed_at=now, + ) + ) + # Flushed before the attempts, because nothing in this schema declares an ORM + # relationship, so the unit of work has no reason to insert the rows the + # attempts reference first. + session.flush() + + for number in range(1, attempts + 1): + session.add( + models.DeliveryAttempt( + id=f"att_{uuid.uuid4().hex}", + delivery_id=delivery_id, + submission_id=submission_id, + attempt_number=number, + destination_url="https://example.invalid/hook", + attempted_at=now, + outcome="http_error", + response_status=503, + ) + ) + session.commit() + return delivery_id + + def seed_due_deliveries( session: Session, count: int, *, now: datetime, endpoint_id: str = "contact-form" ) -> list[str]: diff --git a/tests/integration/test_constraints_postgres.py b/tests/integration/test_constraints_postgres.py index 03e5f81..0635429 100644 --- a/tests/integration/test_constraints_postgres.py +++ b/tests/integration/test_constraints_postgres.py @@ -58,6 +58,7 @@ def a_delivery(delivery_id: str, submission_id: str) -> models.WebhookDelivery: signing_secret="whsec_" + "a" * 64, state=DeliveryState.PENDING, attempts=0, + cycle_attempts=0, next_attempt_at=NOW, created_at=NOW, ) diff --git a/tests/integration/test_migrations_postgres.py b/tests/integration/test_migrations_postgres.py index d34abb9..739f9b6 100644 --- a/tests/integration/test_migrations_postgres.py +++ b/tests/integration/test_migrations_postgres.py @@ -236,6 +236,98 @@ def test_a_populated_database_upgraded_again_still_matches_the_models(postgres_u assert difference == [], f"migrated schema differs from the models: {difference}" +# --- the retry cycle counter 0003 added -------------------------------------- +# +# 0003 is the first revision that changes an existing table rather than adding a +# new one, so an upgrade has to preserve rows it is also rewriting. + + +def test_upgrading_a_populated_0002_adds_the_cycle_counter(postgres_url: str) -> None: + """ + a delivery that has never been replayed must come out with its whole history as its cycle + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with _database_at_baseline(postgres_url) as (config, engine): + with engine.begin() as connection: + _seed_baseline_data(connection) + command.upgrade(config, "0002") + + command.upgrade(config, "0003") + + assert current_revision(engine) == "0003" + with engine.connect() as connection: + _assert_baseline_data_intact(connection) + row = connection.execute( + text("select attempts, cycle_attempts from webhook_deliveries where id = :id"), + {"id": SEEDED_DELIVERY}, + ).one() + assert row.attempts == 1 + assert row.cycle_attempts == 1 + + +def test_the_cycle_counter_is_not_nullable(postgres_url: str) -> None: + with _database_at_baseline(postgres_url) as (config, engine): + command.upgrade(config, "0003") + + with engine.connect() as connection: + nullable = connection.scalar( + text( + "select is_nullable from information_schema.columns " + "where table_name = 'webhook_deliveries' and column_name = 'cycle_attempts'" + ) + ) + + assert nullable == "NO" + + +def test_downgrading_from_0003_leaves_the_delivery_data_alone(postgres_url: str) -> None: + """ + the downgrade must remove the column 0003 added and nothing else + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with _database_at_baseline(postgres_url) as (config, engine): + with engine.begin() as connection: + _seed_baseline_data(connection) + command.upgrade(config, "0003") + + command.downgrade(config, "0002") + + assert current_revision(engine) == "0002" + columns = {column["name"] for column in inspect(engine).get_columns("webhook_deliveries")} + assert "cycle_attempts" not in columns + assert "attempts" in columns + with engine.connect() as connection: + _assert_baseline_data_intact(connection) + + +def test_a_populated_0002_survives_the_whole_round_trip(postgres_url: str) -> None: + """ + populated 0002 to 0003 and back and forward again must end with zero drift + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with _database_at_baseline(postgres_url) as (config, engine): + with engine.begin() as connection: + _seed_baseline_data(connection) + command.upgrade(config, "0002") + + command.upgrade(config, "0003") + command.downgrade(config, "0002") + command.upgrade(config, "0003") + + assert current_revision(engine) == head_revision() + with engine.connect() as connection: + _assert_baseline_data_intact(connection) + assert ( + connection.scalar( + text("select cycle_attempts from webhook_deliveries where id = :id"), + {"id": SEEDED_DELIVERY}, + ) + == 1 + ) + difference = compare_metadata(MigrationContext.configure(connection), Base.metadata) + assert difference == [], f"migrated schema differs from the models: {difference}" + + @contextmanager def _database_at_baseline(postgres_url: str) -> Iterator[tuple[Config, Engine]]: """ diff --git a/tests/integration/test_replay_postgres.py b/tests/integration/test_replay_postgres.py new file mode 100644 index 0000000..82a7ea2 --- /dev/null +++ b/tests/integration/test_replay_postgres.py @@ -0,0 +1,174 @@ +""" +manual replay against real PostgreSQL + +Replay is a conditional state transition, so what matters is what two operators +pressing the button at the same moment get. SQLite serialises writers and would +make that race look safe whether or not it is, which is why it is settled here, +with independent connections and a real row lock. +""" + +from __future__ import annotations + +import threading +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta + +from fastapi.testclient import TestClient +from sqlalchemy import func, select +from sqlalchemy.orm import Session, sessionmaker + +from hymical_forms import models, storage +from hymical_forms.webhooks import DeliveryState +from integration.support import seed_endpoint, seed_failed_delivery + +NOW = datetime(2026, 8, 24, 12, 0, tzinfo=UTC) +LATER = NOW + timedelta(hours=1) + + +def test_a_failed_delivery_is_requeued_once(sessions: sessionmaker[Session]) -> None: + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_failed_delivery(setup, now=NOW) + + with sessions() as operator: + outcome = storage.requeue_failed_delivery(operator, delivery_id, now=LATER) + + assert outcome.requeued is True + with sessions() as observer: + delivery = observer.get(models.WebhookDelivery, delivery_id) + assert delivery is not None + assert delivery.state == DeliveryState.PENDING + assert delivery.completed_at is None + assert delivery.cycle_attempts == 0 + assert delivery.attempts == 5 + assert delivery.next_attempt_at == LATER + + +def test_two_operators_replaying_at_once_do_not_both_win( + sessions: sessionmaker[Session], +) -> None: + """ + the losing request must be refused by the database, not by a check in Python + :param sessions: factory handing out independent connections + """ + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_failed_delivery(setup, now=NOW) + + barrier = threading.Barrier(2) + + def replay() -> storage.ReplayOutcome: + barrier.wait() + with sessions() as session: + return storage.requeue_failed_delivery(session, delivery_id, now=LATER) + + with ThreadPoolExecutor(max_workers=2) as pool: + first, second = (future.result() for future in [pool.submit(replay) for _ in range(2)]) + + assert [first.requeued, second.requeued].count(True) == 1, "both replays claimed to win" + # And the loser describes the state the database actually settled on, rather + # than the one it hoped for, so its answer is the same however the race went. + loser = first if not first.requeued else second + assert loser.record is not None + assert loser.record.delivery.state == DeliveryState.PENDING + + +def test_a_concurrent_replay_duplicates_no_work(sessions: sessionmaker[Session]) -> None: + """ + six simultaneous replays must leave one delivery with one fresh retry cycle + :param sessions: factory handing out independent connections + """ + operators = 6 + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_failed_delivery(setup, now=NOW) + + barrier = threading.Barrier(operators) + + def replay() -> bool: + barrier.wait() + with sessions() as session: + return storage.requeue_failed_delivery(session, delivery_id, now=LATER).requeued + + with ThreadPoolExecutor(max_workers=operators) as pool: + results = [future.result() for future in [pool.submit(replay) for _ in range(operators)]] + + assert results.count(True) == 1 + with sessions() as observer: + assert observer.scalar(select(func.count()).select_from(models.WebhookDelivery)) == 1 + assert observer.scalar(select(func.count()).select_from(models.Submission)) == 1 + # The history is untouched: a replay resumes a delivery, it does not + # rewrite what already happened to it. + assert observer.scalar(select(func.count()).select_from(models.DeliveryAttempt)) == 5 + + +def test_a_replayed_delivery_is_claimable_by_the_worker( + sessions: sessionmaker[Session], +) -> None: + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_failed_delivery(setup, now=NOW) + with sessions() as operator: + storage.requeue_failed_delivery(operator, delivery_id, now=LATER) + + with sessions() as worker: + claimed = storage.claim_due_deliveries(worker, now=LATER, lease_seconds=60, limit=10) + + assert [job.id for job in claimed] == [delivery_id] + + +def test_a_delivery_that_was_never_failed_is_refused(sessions: sessionmaker[Session]) -> None: + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_failed_delivery(setup, now=NOW) + delivery = setup.get(models.WebhookDelivery, delivery_id) + assert delivery is not None + delivery.state = DeliveryState.DELIVERED + setup.commit() + + with sessions() as operator: + outcome = storage.requeue_failed_delivery(operator, delivery_id, now=LATER) + + assert outcome.requeued is False + assert outcome.record is not None + assert outcome.record.delivery.state == DeliveryState.DELIVERED + + +def test_replay_over_http_answers_from_the_settled_state( + pg_client: TestClient, sessions: sessionmaker[Session] +) -> None: + """ + the whole authenticated path, on the database this service is meant to run on + :param pg_client: an API client backed by PostgreSQL + :param sessions: factory handing out independent connections + """ + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_failed_delivery(setup, now=NOW) + + first = pg_client.post(f"/deliveries/{delivery_id}/replay") + second = pg_client.post(f"/deliveries/{delivery_id}/replay") + + assert first.status_code == 200 + assert first.json()["state"] == "pending" + assert first.json()["cycle_attempt_count"] == 0 + assert first.json()["attempt_count"] == 5 + assert second.status_code == 409 + assert second.json()["error"]["code"] == "delivery_not_replayable" + + +def test_a_delivery_read_over_http_never_carries_the_snapshotted_secret( + pg_client: TestClient, sessions: sessionmaker[Session] +) -> None: + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_failed_delivery(setup, now=NOW) + + listing = pg_client.get("/deliveries") + detail = pg_client.get(f"/deliveries/{delivery_id}") + + assert listing.status_code == 200 + assert detail.status_code == 200 + for response in (listing, detail): + assert "whsec_" not in response.text + assert [attempt["attempt_number"] for attempt in detail.json()["attempts"]] == [1, 2, 3, 4, 5] diff --git a/tests/test_deliveries_api.py b/tests/test_deliveries_api.py new file mode 100644 index 0000000..ec32dc4 --- /dev/null +++ b/tests/test_deliveries_api.py @@ -0,0 +1,443 @@ +""" +reading the delivery queue through ``GET /deliveries`` and ``GET /deliveries/{id}`` + +Deliveries are driven into each state through the real ingestion path and the +real worker, so what is being listed is what the service would actually be +holding rather than rows a fixture invented. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from conftest import ( + ClientFactory, + bearer, + create_endpoint, + management_key, + open_session, + work_once, +) +from hymical_forms import models, storage +from hymical_forms.webhooks import DeliveryState +from webhook_server import WebhookRecorder, unused_local_url + +MANAGEMENT_ROUTES = [ + ("GET", "/deliveries"), + ("GET", "/deliveries/whd_00000000000000000000000000000000"), +] + + +def queued_client( + make_client: ClientFactory, url: str, **overrides: Any +) -> tuple[TestClient, dict[str, Any]]: + """ + build a client whose default endpoint delivers to a given destination + :param make_client: factory for clients bound to a configured app + :param url: the webhook destination to configure + :param overrides: extra setting overrides for the application + :returns: the client and the created endpoint as the API returned it + """ + overrides.setdefault("allow_private_webhook_targets", True) + client = make_client(seed_endpoint=False, **overrides) + endpoint = create_endpoint(client, webhook_url=url) + return client, endpoint + + +def submit(client: TestClient, endpoint_id: str = "contact-form", **fields: str) -> str: + """ + post a submission and read back the delivery it queued + :param client: the client to submit through + :param endpoint_id: the endpoint to address + :param fields: the form fields to send + :returns: the identifier of the delivery that was queued + """ + data = fields or {"email": "dev@example.com"} + response = client.post(f"/f/{endpoint_id}", data=data) + assert response.status_code == 202, response.text + submission_id = response.json()["submission_id"] + with open_session(client) as session: + delivery = session.scalars( + select(models.WebhookDelivery).where( + models.WebhookDelivery.submission_id == submission_id + ) + ).one() + return delivery.id + + +def states(client: TestClient) -> dict[str, str]: + """ + read every delivery's state straight from the database + :param client: the client whose application database should be inspected + :returns: the state of each delivery, keyed by delivery id + """ + with open_session(client) as session: + return {row.id: row.state for row in session.scalars(select(models.WebhookDelivery))} + + +# --- authentication ---------------------------------------------------------- + + +@pytest.mark.parametrize(("method", "path"), MANAGEMENT_ROUTES) +def test_a_delivery_read_refuses_an_unauthenticated_request( + make_client: ClientFactory, method: str, path: str +) -> None: + client = make_client(authenticate=False) + + response = client.request(method, path) + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "authentication_required" + + +@pytest.mark.parametrize(("method", "path"), MANAGEMENT_ROUTES) +def test_a_delivery_read_refuses_an_unusable_key( + make_client: ClientFactory, method: str, path: str +) -> None: + client = make_client(authenticate=False) + + response = client.request(method, path, headers=bearer("hym_live_nope")) + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "invalid_api_key" + + +# --- listing ----------------------------------------------------------------- + + +def test_listing_with_nothing_queued_is_empty(client: TestClient) -> None: + body = client.get("/deliveries").json() + + assert body == {"items": [], "next_cursor": None} + + +def test_a_listed_delivery_describes_the_work_that_is_owed( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, endpoint = queued_client(make_client, webhook.url) + delivery_id = submit(client) + + item = client.get("/deliveries").json()["items"][0] + + assert item["id"] == delivery_id + assert item["endpoint_id"] == "contact-form" + assert item["state"] == "pending" + assert item["destination_url"] == endpoint["webhook_url"] + assert item["attempt_count"] == 0 + assert item["cycle_attempt_count"] == 0 + assert item["completed_at"] is None + assert item["submission_id"].startswith("sub_") + + +def test_listing_is_ordered_newest_first( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + created = [submit(client, email=f"dev{index}@example.com") for index in range(4)] + + listed = [item["id"] for item in client.get("/deliveries").json()["items"]] + + assert listed == list(reversed(created)) + + +def test_paging_visits_every_delivery_exactly_once( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + created = [submit(client, email=f"dev{index}@example.com") for index in range(5)] + + seen: list[str] = [] + cursor: str | None = None + for _ in range(10): + params: dict[str, Any] = {"limit": 2} + if cursor is not None: + params["cursor"] = cursor + body = client.get("/deliveries", params=params).json() + seen.extend(item["id"] for item in body["items"]) + cursor = body["next_cursor"] + if cursor is None: + break + + assert cursor is None + assert seen == list(reversed(created)) + + +def test_the_same_page_request_answers_identically( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + for index in range(5): + submit(client, email=f"dev{index}@example.com") + + first = client.get("/deliveries", params={"limit": 3}).json() + again = client.get("/deliveries", params={"limit": 3}).json() + + assert first == again + + +@pytest.mark.parametrize("limit", [0, -1, 101]) +def test_an_unusable_page_size_is_refused(client: TestClient, limit: int) -> None: + response = client.get("/deliveries", params={"limit": limit}) + + assert response.status_code == 422 + assert response.json()["error"]["code"] == "invalid_request" + + +def test_an_unknown_cursor_is_refused(client: TestClient) -> None: + response = client.get("/deliveries", params={"cursor": "whd_nothing"}) + + assert response.status_code == 422 + assert response.json()["error"]["code"] == "invalid_cursor" + + +# --- filtering --------------------------------------------------------------- + + +def test_deliveries_can_be_filtered_by_endpoint( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + create_endpoint(client, "waitlist", webhook_url=webhook.url) + wanted = submit(client, "contact-form") + submit(client, "waitlist") + + body = client.get("/deliveries", params={"endpoint_id": "contact-form"}).json() + + assert [item["id"] for item in body["items"]] == [wanted] + + +def test_filtering_by_an_endpoint_with_nothing_queued_is_empty( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + submit(client) + + body = client.get("/deliveries", params={"endpoint_id": "waitlist"}).json() + + assert body == {"items": [], "next_cursor": None} + + +def test_deliveries_can_be_filtered_by_state( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + """ + every state the domain defines must be reachable through the filter + :param make_client: factory for clients bound to a configured app + :param webhook: a local server standing in for the receiver + """ + client, _ = queued_client(make_client, webhook.url, worker_batch_size=1) + delivered = submit(client, email="one@example.com") + work_once(client) + + webhook.status = 400 + failed = submit(client, email="two@example.com") + work_once(client) + + webhook.status = 200 + pending = submit(client, email="three@example.com") + + # Claimed and abandoned, exactly as a worker holding a lease would leave it. + # The two terminal deliveries are not claimable, so the only one this can + # take is the pending one. + with open_session(client) as session: + storage.claim_due_deliveries( + session, now=datetime.now(UTC) + timedelta(hours=1), lease_seconds=3600, limit=1 + ) + processing = next( + identifier + for identifier, state in states(client).items() + if state == DeliveryState.PROCESSING + ) + + assert processing == pending + for state, expected in [ + ("delivered", delivered), + ("failed", failed), + ("processing", processing), + ]: + body = client.get("/deliveries", params={"state": state}).json() + assert [item["id"] for item in body["items"]] == [expected], state + + +def test_the_two_filters_combine(make_client: ClientFactory, webhook: WebhookRecorder) -> None: + webhook.status = 400 + client, _ = queued_client(make_client, webhook.url) + create_endpoint(client, "waitlist", webhook_url=webhook.url) + wanted = submit(client, "contact-form") + submit(client, "waitlist") + work_once(client) + + body = client.get( + "/deliveries", params={"endpoint_id": "contact-form", "state": "failed"} + ).json() + + assert [item["id"] for item in body["items"]] == [wanted] + + +def test_an_unknown_state_filter_is_refused(client: TestClient) -> None: + response = client.get("/deliveries", params={"state": "exploded"}) + + assert response.status_code == 422 + body = response.json() + assert body["error"]["code"] == "invalid_request" + assert [field["field"] for field in body["error"]["details"]["fields"]] == ["state"] + + +# --- detail ------------------------------------------------------------------ + + +def test_detail_describes_one_delivery( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, endpoint = queued_client(make_client, webhook.url) + delivery_id = submit(client) + + body = client.get(f"/deliveries/{delivery_id}").json() + + assert body["id"] == delivery_id + assert body["endpoint_id"] == "contact-form" + assert body["destination_url"] == endpoint["webhook_url"] + assert body["attempts"] == [] + + +def test_an_unknown_delivery_is_a_404(client: TestClient) -> None: + response = client.get("/deliveries/whd_nothing") + + assert response.status_code == 404 + body = response.json() + assert body["error"]["code"] == "delivery_not_found" + assert body["error"]["details"]["delivery_id"] == "whd_nothing" + + +def test_the_attempt_history_is_ordered_and_complete( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + webhook.status = 503 + client, _ = queued_client( + make_client, webhook.url, webhook_max_attempts=3, webhook_retry_initial_seconds=10 + ) + delivery_id = submit(client) + + now = due_at(client, delivery_id) + for _ in range(3): + work_once(client, now=now) + now = now + timedelta(minutes=5) + + body = client.get(f"/deliveries/{delivery_id}").json() + + assert [attempt["attempt_number"] for attempt in body["attempts"]] == [1, 2, 3] + assert all(attempt["outcome"] == "http_error" for attempt in body["attempts"]) + assert all(attempt["response_status"] == 503 for attempt in body["attempts"]) + assert body["state"] == "failed" + assert body["attempt_count"] == 3 + + +def test_a_successful_attempt_reports_no_failure( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + delivery_id = submit(client) + work_once(client) + + attempt = client.get(f"/deliveries/{delivery_id}").json()["attempts"][0] + + assert attempt["outcome"] == "succeeded" + assert attempt["response_status"] == 200 + assert attempt["error"] is None + + +def test_a_network_failure_is_reported_without_a_status(make_client: ClientFactory) -> None: + client, _ = queued_client(make_client, unused_local_url()) + delivery_id = submit(client) + work_once(client) + + attempt = client.get(f"/deliveries/{delivery_id}").json()["attempts"][0] + + assert attempt["outcome"] == "network_error" + assert attempt["response_status"] is None + assert attempt["error"] is not None + assert len(attempt["error"]) <= 500 + + +def test_the_history_of_one_delivery_excludes_another( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + first = submit(client, email="one@example.com") + submit(client, email="two@example.com") + work_once(client) + + body = client.get(f"/deliveries/{first}").json() + + assert len(body["attempts"]) == 1 + assert body["attempts"][0]["attempt_number"] == 1 + + +# --- what must never be returned --------------------------------------------- + + +def test_a_delivery_read_never_carries_the_snapshotted_secret( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + """ + the secret is stored on the delivery, so its absence has to be asserted + :param make_client: factory for clients bound to a configured app + :param webhook: a local server standing in for the receiver + """ + client, endpoint = queued_client(make_client, webhook.url) + delivery_id = submit(client) + work_once(client) + secret = endpoint["webhook_secret"] + + listing = client.get("/deliveries") + detail = client.get(f"/deliveries/{delivery_id}") + + assert secret not in listing.text + assert secret not in detail.text + assert "signing_secret" not in listing.text + assert "signing_secret" not in detail.text + + +def test_a_delivery_read_never_carries_the_submitted_fields( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + delivery_id = submit(client, email="private@example.com", note="do not echo me") + + listing = client.get("/deliveries") + detail = client.get(f"/deliveries/{delivery_id}") + + for response in (listing, detail): + assert "private@example.com" not in response.text + assert "do not echo me" not in response.text + + +def test_a_delivery_read_never_carries_management_credentials( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client, _ = queued_client(make_client, webhook.url) + delivery_id = submit(client) + key = management_key(client) + + detail = client.get(f"/deliveries/{delivery_id}") + + assert key not in detail.text + assert "authorization" not in detail.text.lower() + + +def due_at(client: TestClient, delivery_id: str) -> datetime: + """ + read the instant one delivery next becomes due + :param client: the client whose application database should be inspected + :param delivery_id: the delivery to read + :returns: that delivery's next attempt time + """ + with open_session(client) as session: + delivery = session.get(models.WebhookDelivery, delivery_id) + assert delivery is not None + return delivery.next_attempt_at diff --git a/tests/test_endpoints_management.py b/tests/test_endpoints_management.py new file mode 100644 index 0000000..49b7045 --- /dev/null +++ b/tests/test_endpoints_management.py @@ -0,0 +1,510 @@ +""" +reading and changing endpoints through the management API + +``POST /endpoints`` is covered by ``test_endpoints_api.py``; this module is about +the routes interval 8 added around it. +""" + +from __future__ import annotations + +import hashlib +import hmac +import logging +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +from conftest import ( + ClientFactory, + bearer, + create_endpoint, + management_key, + open_session, + work_once, +) +from hymical_forms import models +from webhook_server import WebhookRecorder + +MANAGEMENT_ROUTES = [ + ("GET", "/endpoints", None), + ("GET", "/endpoints/contact-form", None), + ("PATCH", "/endpoints/contact-form", {"name": "Renamed"}), +] + + +def seed(client: TestClient, count: int, *, prefix: str = "form") -> list[str]: + """ + register several endpoints in a known order + :param client: the client whose application should hold them + :param count: how many to register + :param prefix: the identifier prefix each one is built from + :returns: the identifiers, oldest first + """ + return [create_endpoint(client, f"{prefix}-{index}")["id"] for index in range(count)] + + +def walk(client: TestClient, *, limit: int) -> list[str]: + """ + follow the cursor to the end and collect every endpoint identifier seen + :param client: the client to page through + :param limit: the page size to ask for on every request + :returns: the identifiers in the order the API returned them + """ + seen: list[str] = [] + cursor: str | None = None + for _ in range(100): + params: dict[str, Any] = {"limit": limit} + if cursor is not None: + params["cursor"] = cursor + body = client.get("/endpoints", params=params).json() + seen.extend(item["id"] for item in body["items"]) + cursor = body["next_cursor"] + if cursor is None: + return seen + raise AssertionError("the cursor never ran out") + + +# --- authentication ---------------------------------------------------------- + + +@pytest.mark.parametrize(("method", "path", "body"), MANAGEMENT_ROUTES) +def test_a_management_endpoint_route_refuses_an_unauthenticated_request( + make_client: ClientFactory, method: str, path: str, body: dict[str, Any] | None +) -> None: + client = make_client(authenticate=False) + + response = client.request(method, path, json=body) + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "authentication_required" + + +@pytest.mark.parametrize(("method", "path", "body"), MANAGEMENT_ROUTES) +def test_a_management_endpoint_route_refuses_an_unusable_key( + make_client: ClientFactory, method: str, path: str, body: dict[str, Any] | None +) -> None: + client = make_client(authenticate=False) + + response = client.request(method, path, json=body, headers=bearer("hym_live_nope")) + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "invalid_api_key" + + +def test_a_refused_patch_changes_nothing(make_client: ClientFactory) -> None: + client = make_client(authenticate=False) + + client.patch("/endpoints/contact-form", json={"name": "Renamed"}) + + with open_session(client) as session: + endpoint = session.get(models.Endpoint, "contact-form") + assert endpoint is not None + assert endpoint.name == "Contact form" + + +@pytest.mark.parametrize(("method", "path", "body"), MANAGEMENT_ROUTES) +def test_a_valid_key_reaches_every_management_endpoint_route( + client: TestClient, method: str, path: str, body: dict[str, Any] | None +) -> None: + response = client.request(method, path, json=body) + + assert response.status_code == 200, response.text + + +# --- listing ----------------------------------------------------------------- + + +def test_listing_returns_the_registered_endpoints(client: TestClient) -> None: + create_endpoint(client, "waitlist", name="Waitlist") + + body = client.get("/endpoints").json() + + assert {item["id"] for item in body["items"]} == {"contact-form", "waitlist"} + assert body["next_cursor"] is None + + +def test_listing_an_empty_service_returns_an_empty_page(empty_client: TestClient) -> None: + body = empty_client.get("/endpoints").json() + + assert body == {"items": [], "next_cursor": None} + + +def test_a_listed_endpoint_carries_its_configuration( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(seed_endpoint=False, allow_private_webhook_targets=True) + create_endpoint(client, "contact-form", name="Contact form", webhook_url=webhook.url) + + item = client.get("/endpoints").json()["items"][0] + + assert item["id"] == "contact-form" + assert item["name"] == "Contact form" + assert item["is_active"] is True + assert item["webhook_url"] == webhook.url + assert isinstance(item["created_at"], str) + + +def test_listing_is_ordered_newest_first(empty_client: TestClient) -> None: + seed(empty_client, 4) + + listed = [item["id"] for item in empty_client.get("/endpoints").json()["items"]] + + assert listed == ["form-3", "form-2", "form-1", "form-0"] + + +def test_paging_visits_every_endpoint_exactly_once(empty_client: TestClient) -> None: + created = seed(empty_client, 7) + + seen = walk(empty_client, limit=2) + + assert seen == list(reversed(created)) + + +def test_a_page_is_bounded_by_the_requested_limit(empty_client: TestClient) -> None: + seed(empty_client, 5) + + body = empty_client.get("/endpoints", params={"limit": 2}).json() + + assert len(body["items"]) == 2 + assert body["next_cursor"] == body["items"][-1]["id"] + + +def test_the_same_page_request_answers_identically(empty_client: TestClient) -> None: + """ + a page has to be reproducible, or a cursor walk cannot be trusted + :param empty_client: test client whose app holds no endpoints + """ + seed(empty_client, 5) + + first = empty_client.get("/endpoints", params={"limit": 3}).json() + again = empty_client.get("/endpoints", params={"limit": 3}).json() + + assert first == again + + +@pytest.mark.parametrize("limit", [0, -1, 101, 1000]) +def test_an_unusable_page_size_is_refused(empty_client: TestClient, limit: int) -> None: + response = empty_client.get("/endpoints", params={"limit": limit}) + + assert response.status_code == 422 + assert response.json()["error"]["code"] == "invalid_request" + + +def test_an_unknown_cursor_is_refused(empty_client: TestClient) -> None: + seed(empty_client, 2) + + response = empty_client.get("/endpoints", params={"cursor": "no-such-endpoint"}) + + assert response.status_code == 422 + assert response.json()["error"]["code"] == "invalid_cursor" + + +def test_listing_never_carries_a_webhook_secret( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(seed_endpoint=False, allow_private_webhook_targets=True) + created = create_endpoint(client, "contact-form", webhook_url=webhook.url) + + response = client.get("/endpoints") + + assert created["webhook_secret"] is not None + assert created["webhook_secret"] not in response.text + assert "webhook_secret" not in response.json()["items"][0] + + +# --- detail ------------------------------------------------------------------ + + +def test_detail_describes_one_endpoint(client: TestClient) -> None: + body = client.get("/endpoints/contact-form").json() + + assert body["id"] == "contact-form" + assert body["name"] == "Contact form" + assert body["is_active"] is True + assert body["webhook_url"] is None + + +def test_detail_never_carries_a_webhook_secret( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(seed_endpoint=False, allow_private_webhook_targets=True) + created = create_endpoint(client, "contact-form", webhook_url=webhook.url) + + response = client.get("/endpoints/contact-form") + + assert created["webhook_secret"] not in response.text + assert "webhook_secret" not in response.json() + + +def test_detail_of_an_unknown_endpoint_is_a_404(empty_client: TestClient) -> None: + response = empty_client.get("/endpoints/nothing-here") + + assert response.status_code == 404 + body = response.json() + assert body["error"]["code"] == "endpoint_not_found" + assert body["error"]["details"]["endpoint_id"] == "nothing-here" + + +# --- updating ---------------------------------------------------------------- + + +def test_the_name_can_be_changed(client: TestClient) -> None: + response = client.patch("/endpoints/contact-form", json={"name": "Support form"}) + + assert response.status_code == 200 + assert response.json()["name"] == "Support form" + assert client.get("/endpoints/contact-form").json()["name"] == "Support form" + + +def test_the_active_state_can_be_changed(client: TestClient) -> None: + response = client.patch("/endpoints/contact-form", json={"is_active": False}) + + assert response.json()["is_active"] is False + assert client.get("/endpoints/contact-form").json()["is_active"] is False + + +def test_an_omitted_field_is_left_alone(client: TestClient) -> None: + client.patch("/endpoints/contact-form", json={"name": "Support form"}) + + body = client.patch("/endpoints/contact-form", json={"is_active": False}).json() + + assert body["name"] == "Support form" + assert body["is_active"] is False + + +def test_an_empty_patch_changes_nothing(client: TestClient) -> None: + before = client.get("/endpoints/contact-form").json() + + response = client.patch("/endpoints/contact-form", json={}) + + assert response.status_code == 200 + assert {key: value for key, value in response.json().items() if key in before} == before + + +def test_the_endpoint_id_cannot_be_changed(client: TestClient) -> None: + """ + the ID is the primary key and sits in the action URL of every deployed form + :param client: test client whose app holds the default endpoint + """ + response = client.patch("/endpoints/contact-form", json={"id": "renamed-form"}) + + assert response.status_code == 200 + assert response.json()["id"] == "contact-form" + with open_session(client) as session: + assert session.get(models.Endpoint, "renamed-form") is None + assert session.get(models.Endpoint, "contact-form") is not None + + +def test_patching_an_unknown_endpoint_is_a_404(empty_client: TestClient) -> None: + response = empty_client.patch("/endpoints/nothing-here", json={"name": "Whatever"}) + + assert response.status_code == 404 + assert response.json()["error"]["code"] == "endpoint_not_found" + + +@pytest.mark.parametrize("payload", [{"name": ""}, {"name": "x" * 201}, {"is_active": "maybe"}]) +def test_an_unusable_patch_body_is_refused(client: TestClient, payload: dict[str, Any]) -> None: + response = client.patch("/endpoints/contact-form", json=payload) + + assert response.status_code == 422 + assert response.json()["error"]["code"] == "invalid_request" + + +# --- disabling and re-enabling ----------------------------------------------- + + +def test_disabling_stops_submissions_immediately(client: TestClient) -> None: + client.patch("/endpoints/contact-form", json={"is_active": False}) + + response = client.post("/f/contact-form", data={"email": "dev@example.com"}) + + assert response.status_code == 409 + assert response.json()["error"]["code"] == "endpoint_inactive" + + +def test_a_submission_to_a_disabled_endpoint_persists_nothing(client: TestClient) -> None: + client.patch("/endpoints/contact-form", json={"is_active": False}) + + client.post("/f/contact-form", data={"email": "dev@example.com"}) + + with open_session(client) as session: + assert session.query(models.Submission).count() == 0 + + +def test_re_enabling_restores_acceptance(client: TestClient) -> None: + client.patch("/endpoints/contact-form", json={"is_active": False}) + + client.patch("/endpoints/contact-form", json={"is_active": True}) + + response = client.post("/f/contact-form", data={"email": "dev@example.com"}) + assert response.status_code == 202 + + +# --- webhook reconfiguration -------------------------------------------------- + + +def test_adding_a_webhook_generates_a_secret( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(allow_private_webhook_targets=True) + + body = client.patch("/endpoints/contact-form", json={"webhook_url": webhook.url}).json() + + assert body["webhook_url"] == webhook.url + assert body["webhook_secret"] is not None + assert body["webhook_secret"].startswith("whsec_") + + +def test_the_generated_secret_is_the_one_deliveries_are_signed_with( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(allow_private_webhook_targets=True) + secret = client.patch("/endpoints/contact-form", json={"webhook_url": webhook.url}).json()[ + "webhook_secret" + ] + + client.post("/f/contact-form", data={"email": "dev@example.com"}) + work_once(client) + + delivered = webhook.received[0] + expected = hmac.new(secret.encode("utf-8"), delivered.body, hashlib.sha256).hexdigest() + assert delivered.headers["hymical-signature"] == f"v1={expected}" + + +def test_an_unchanged_destination_keeps_its_secret( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + """ + resending the same URL must not silently invalidate a receiver's secret + :param make_client: factory for clients bound to a configured app + :param webhook: a local server standing in for the receiver + """ + client = make_client(seed_endpoint=False, allow_private_webhook_targets=True) + created = create_endpoint(client, "contact-form", webhook_url=webhook.url) + + body = client.patch( + "/endpoints/contact-form", json={"name": "Renamed", "webhook_url": webhook.url} + ).json() + + assert body["webhook_secret"] is None + with open_session(client) as session: + endpoint = session.get(models.Endpoint, "contact-form") + assert endpoint is not None + assert endpoint.webhook_secret == created["webhook_secret"] + + +def test_changing_the_destination_generates_a_new_secret( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(seed_endpoint=False, allow_private_webhook_targets=True) + created = create_endpoint(client, "contact-form", webhook_url=webhook.url) + + body = client.patch( + "/endpoints/contact-form", json={"webhook_url": webhook.url + "/moved"} + ).json() + + assert body["webhook_secret"] is not None + assert body["webhook_secret"] != created["webhook_secret"] + with open_session(client) as session: + endpoint = session.get(models.Endpoint, "contact-form") + assert endpoint is not None + assert endpoint.webhook_secret == body["webhook_secret"] + + +def test_removing_the_webhook_removes_the_secret_too( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(seed_endpoint=False, allow_private_webhook_targets=True) + create_endpoint(client, "contact-form", webhook_url=webhook.url) + + body = client.patch("/endpoints/contact-form", json={"webhook_url": None}).json() + + assert body["webhook_url"] is None + assert body["webhook_secret"] is None + with open_session(client) as session: + endpoint = session.get(models.Endpoint, "contact-form") + assert endpoint is not None + assert endpoint.webhook_secret is None + + +def test_a_submission_after_removal_queues_nothing( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(seed_endpoint=False, allow_private_webhook_targets=True) + create_endpoint(client, "contact-form", webhook_url=webhook.url) + client.patch("/endpoints/contact-form", json={"webhook_url": None}) + + response = client.post("/f/contact-form", data={"email": "dev@example.com"}) + + assert response.json()["delivery"]["queued"] is False + assert work_once(client) == 0 + + +def test_an_unusable_destination_is_refused(client: TestClient) -> None: + response = client.patch( + "/endpoints/contact-form", json={"webhook_url": "http://127.0.0.1/hook"} + ) + + assert response.status_code == 422 + assert response.json()["error"]["code"] == "invalid_webhook_url" + + +def test_a_refused_destination_leaves_the_endpoint_alone( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = make_client(seed_endpoint=False, allow_private_webhook_targets=False) + create_endpoint(client, "contact-form", name="Contact form") + + client.patch("/endpoints/contact-form", json={"name": "Renamed", "webhook_url": "ftp://x/y"}) + + with open_session(client) as session: + endpoint = session.get(models.Endpoint, "contact-form") + assert endpoint is not None + assert endpoint.name == "Contact form" + assert endpoint.webhook_url is None + + +def test_a_queued_delivery_keeps_the_destination_it_snapshotted( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + """ + changing configuration must not redirect work that is already owed + :param make_client: factory for clients bound to a configured app + :param webhook: a local server standing in for the original receiver + """ + client = make_client(seed_endpoint=False, allow_private_webhook_targets=True) + created = create_endpoint(client, "contact-form", webhook_url=webhook.url) + client.post("/f/contact-form", data={"email": "dev@example.com"}) + + client.patch("/endpoints/contact-form", json={"webhook_url": webhook.url + "/elsewhere"}) + + with open_session(client) as session: + delivery = session.query(models.WebhookDelivery).one() + assert delivery.destination_url == webhook.url + assert delivery.signing_secret == created["webhook_secret"] + + +def test_a_patch_never_logs_the_generated_secret( + make_client: ClientFactory, webhook: WebhookRecorder, caplog: pytest.LogCaptureFixture +) -> None: + client = make_client(allow_private_webhook_targets=True) + + with caplog.at_level(logging.DEBUG): + secret = client.patch("/endpoints/contact-form", json={"webhook_url": webhook.url}).json()[ + "webhook_secret" + ] + + assert caplog.text != "" + assert secret not in caplog.text + + +def test_a_patch_does_not_log_the_management_credential( + client: TestClient, caplog: pytest.LogCaptureFixture +) -> None: + key = management_key(client) + + with caplog.at_level(logging.DEBUG): + client.patch("/endpoints/contact-form", json={"name": "Renamed"}) + + assert key not in caplog.text diff --git a/tests/test_openapi.py b/tests/test_openapi.py new file mode 100644 index 0000000..e2134d1 --- /dev/null +++ b/tests/test_openapi.py @@ -0,0 +1,137 @@ +""" +the generated OpenAPI document + +The schema is what an integrator reads first, so what it says about the +authentication boundary has to be true. A management route that forgot its +dependency would advertise itself as public here, which is precisely the mistake +worth failing a build over. +""" + +from __future__ import annotations + +from typing import Any, cast + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +# Every route that must require a management API key, and every route that must +# not. Adding a management route without adding it here would leave its +# authentication unasserted, so the two lists are also the checklist. +MANAGEMENT_OPERATIONS = [ + ("/endpoints", "post"), + ("/endpoints", "get"), + ("/endpoints/{endpoint_id}", "get"), + ("/endpoints/{endpoint_id}", "patch"), + ("/deliveries", "get"), + ("/deliveries/{delivery_id}", "get"), + ("/deliveries/{delivery_id}/replay", "post"), +] + +PUBLIC_OPERATIONS = [ + ("/health", "get"), + ("/f/{endpoint_id}", "post"), +] + + +def schema(client: TestClient) -> dict[str, Any]: + """ + read the OpenAPI document the application generates + :param client: the client whose application should be described + :returns: the generated schema + """ + return cast(FastAPI, client.app).openapi() + + +def operation(client: TestClient, path: str, method: str) -> dict[str, Any]: + """ + read one operation out of the generated document + :param client: the client whose application should be described + :param path: the templated path the operation is declared on + :param method: the HTTP method, lowercased + :returns: the operation object + """ + paths = schema(client)["paths"] + assert path in paths, f"{path} is not in the generated schema" + assert method in paths[path], f"{method.upper()} {path} is not in the generated schema" + return cast(dict[str, Any], paths[path][method]) + + +def test_every_route_is_declared(client: TestClient) -> None: + declared = { + (path, method) + for path, operations in schema(client)["paths"].items() + for method in operations + } + + assert declared == set(MANAGEMENT_OPERATIONS) | set(PUBLIC_OPERATIONS) + + +@pytest.mark.parametrize(("path", "method"), MANAGEMENT_OPERATIONS) +def test_a_management_route_advertises_bearer_authentication( + client: TestClient, path: str, method: str +) -> None: + security = operation(client, path, method).get("security") + + assert security is not None, f"{method.upper()} {path} advertises no authentication" + assert any("ManagementApiKey" in requirement for requirement in security) + + +@pytest.mark.parametrize(("path", "method"), PUBLIC_OPERATIONS) +def test_a_public_route_advertises_no_authentication( + client: TestClient, path: str, method: str +) -> None: + assert "security" not in operation(client, path, method) + + +def test_the_security_scheme_is_a_bearer_scheme(client: TestClient) -> None: + scheme = schema(client)["components"]["securitySchemes"]["ManagementApiKey"] + + assert scheme["type"] == "http" + assert scheme["scheme"] == "bearer" + + +@pytest.mark.parametrize(("path", "method"), MANAGEMENT_OPERATIONS) +def test_a_management_route_documents_its_refusal( + client: TestClient, path: str, method: str +) -> None: + responses = operation(client, path, method)["responses"] + + assert "401" in responses + + +def test_no_response_schema_mentions_a_webhook_signing_secret(client: TestClient) -> None: + """ + a read model that could name the secret is a leak waiting to be written + :param client: test client whose app holds the default endpoint + """ + components = schema(client)["components"]["schemas"] + + exposing = { + name + for name, model in components.items() + if "webhook_secret" in model.get("properties", {}) + } + # The one model that may name it is the mutation response, where a newly + # generated secret is handed over exactly once and never read back. + assert exposing == {"EndpointResponse"} + + for name, model in components.items(): + properties = set(model.get("properties", {})) + assert "signing_secret" not in properties, name + assert "key_digest" not in properties, name + assert "api_key" not in properties, name + + +def test_the_delivery_views_carry_no_submitted_fields(client: TestClient) -> None: + components = schema(client)["components"]["schemas"] + + for name in ("DeliveryView", "DeliveryDetail", "DeliveryAttemptView"): + assert "fields" not in components[name]["properties"], name + + +def test_a_page_response_has_the_shared_shape(client: TestClient) -> None: + components = schema(client)["components"]["schemas"] + + for name in ("EndpointPage", "DeliveryPage"): + assert set(components[name]["properties"]) == {"items", "next_cursor"}, name diff --git a/tests/test_replay.py b/tests/test_replay.py new file mode 100644 index 0000000..72de422 --- /dev/null +++ b/tests/test_replay.py @@ -0,0 +1,477 @@ +""" +manual replay: ``POST /deliveries/{delivery_id}/replay`` + +Replay is a state change and nothing else. Every test here proves that by +letting the ordinary worker do the sending afterwards, exactly as it would for +a delivery nobody touched. +""" + +from __future__ import annotations + +import json +import logging +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from conftest import ( + ClientFactory, + bearer, + create_endpoint, + management_key, + open_session, + work_once, +) +from hymical_forms import models, storage +from hymical_forms.webhooks import DeliveryState +from webhook_server import WebhookRecorder + +REPLAY_PATH = "/deliveries/{delivery_id}/replay" + + +def failing_client( + make_client: ClientFactory, webhook: WebhookRecorder, *, status: int = 400, **overrides: Any +) -> TestClient: + """ + build a client whose endpoint delivers to a destination that refuses everything + :param make_client: factory for clients bound to a configured app + :param webhook: the local server standing in for the receiver + :param status: the status that destination answers with, final by default + :param overrides: extra setting overrides for the application + :returns: a client with the default endpoint pointed at that destination + """ + overrides.setdefault("allow_private_webhook_targets", True) + webhook.status = status + client = make_client(seed_endpoint=False, **overrides) + create_endpoint(client, webhook_url=webhook.url) + return client + + +def delivery_of(client: TestClient) -> models.WebhookDelivery: + """ + read the single queued delivery behind a client + :param client: the client whose application database should be inspected + :returns: the one delivery row + """ + with open_session(client) as session: + return session.scalars(select(models.WebhookDelivery)).one() + + +def attempts_of(client: TestClient) -> list[models.DeliveryAttempt]: + """ + read every recorded attempt behind a client, oldest first + :param client: the client whose application database should be inspected + :returns: the attempt rows ordered by attempt number + """ + with open_session(client) as session: + return list( + session.scalars( + select(models.DeliveryAttempt).order_by(models.DeliveryAttempt.attempt_number) + ) + ) + + +def fail_once(client: TestClient, webhook: WebhookRecorder) -> str: + """ + submit a form and let the worker drive its delivery to terminal failure + :param client: the client to submit through + :param webhook: the local server refusing the delivery + :returns: the identifier of the now-failed delivery + """ + client.post("/f/contact-form", data={"email": "dev@example.com"}) + work_once(client, now=delivery_of(client).next_attempt_at) + delivery = delivery_of(client) + assert delivery.state == DeliveryState.FAILED, delivery.state + return delivery.id + + +# --- authentication ---------------------------------------------------------- + + +def test_replay_refuses_an_unauthenticated_request( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + client.headers.pop("Authorization") + + response = client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "authentication_required" + assert delivery_of(client).state == DeliveryState.FAILED + + +def test_replay_refuses_an_unusable_key( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + + response = client.post( + REPLAY_PATH.format(delivery_id=delivery_id), headers=bearer("hym_live_nope") + ) + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "invalid_api_key" + assert delivery_of(client).state == DeliveryState.FAILED + + +# --- eligibility ------------------------------------------------------------- + + +def test_an_unknown_delivery_is_a_404(client: TestClient) -> None: + response = client.post(REPLAY_PATH.format(delivery_id="whd_nothing")) + + assert response.status_code == 404 + body = response.json() + assert body["error"]["code"] == "delivery_not_found" + assert body["error"]["details"]["delivery_id"] == "whd_nothing" + + +def test_a_pending_delivery_cannot_be_replayed( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + client.post("/f/contact-form", data={"email": "dev@example.com"}) + delivery_id = delivery_of(client).id + + response = client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + assert response.status_code == 409 + body = response.json() + assert body["error"]["code"] == "delivery_not_replayable" + assert body["error"]["details"]["state"] == "pending" + + +def test_a_processing_delivery_cannot_be_replayed( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + client.post("/f/contact-form", data={"email": "dev@example.com"}) + delivery_id = delivery_of(client).id + with open_session(client) as session: + storage.claim_due_deliveries( + session, now=datetime.now(UTC) + timedelta(hours=1), lease_seconds=3600, limit=1 + ) + + response = client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + assert response.status_code == 409 + assert response.json()["error"]["details"]["state"] == "processing" + + +def test_a_delivered_delivery_cannot_be_replayed( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + """ + replay is a repair for failure, not a way to send a receiver the same event twice + :param make_client: factory for clients bound to a configured app + :param webhook: the local server recording deliveries + """ + client = failing_client(make_client, webhook, status=200) + client.post("/f/contact-form", data={"email": "dev@example.com"}) + work_once(client) + delivery_id = delivery_of(client).id + + response = client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + assert response.status_code == 409 + assert response.json()["error"]["details"]["state"] == "delivered" + assert len(webhook.received) == 1 + + +def test_a_failed_delivery_can_be_replayed( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + + response = client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + assert response.status_code == 200 + body = response.json() + assert body["id"] == delivery_id + assert body["state"] == "pending" + assert body["completed_at"] is None + assert body["cycle_attempt_count"] == 0 + assert body["attempt_count"] == 1 + + +def test_replaying_twice_is_refused_the_second_time( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + assert client.post(REPLAY_PATH.format(delivery_id=delivery_id)).status_code == 200 + + response = client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + assert response.status_code == 409 + assert response.json()["error"]["details"]["state"] == "pending" + + +# --- what replay must not do ------------------------------------------------- + + +def test_replay_creates_no_new_submission( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + with open_session(client) as session: + before = session.scalars(select(models.Submission)).one() + + client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + with open_session(client) as session: + after = session.scalars(select(models.Submission)).one() + assert after.id == before.id + assert after.received_at == before.received_at + assert after.fields == before.fields + + +def test_replay_creates_no_second_delivery( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + + client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + with open_session(client) as session: + deliveries = list(session.scalars(select(models.WebhookDelivery))) + assert [delivery.id for delivery in deliveries] == [delivery_id] + + +def test_replay_keeps_the_historical_attempts( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client( + make_client, + webhook, + status=503, + webhook_max_attempts=2, + webhook_retry_initial_seconds=10, + ) + client.post("/f/contact-form", data={"email": "dev@example.com"}) + now = delivery_of(client).next_attempt_at + work_once(client, now=now) + work_once(client, now=delivery_of(client).next_attempt_at) + before = [(attempt.id, attempt.attempt_number) for attempt in attempts_of(client)] + assert len(before) == 2 + + client.post(REPLAY_PATH.format(delivery_id=delivery_of(client).id)) + + after = [(attempt.id, attempt.attempt_number) for attempt in attempts_of(client)] + assert after == before + + +def test_replay_preserves_the_snapshotted_destination( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + before = delivery_of(client) + destination, secret = before.destination_url, before.signing_secret + + body = client.post(REPLAY_PATH.format(delivery_id=delivery_id)).json() + + after = delivery_of(client) + assert after.destination_url == destination + assert after.signing_secret == secret + assert body["destination_url"] == destination + + +def test_replay_never_returns_or_logs_the_signing_secret( + make_client: ClientFactory, webhook: WebhookRecorder, caplog: pytest.LogCaptureFixture +) -> None: + """ + the signing identity has to survive a replay without ever being shown + :param make_client: factory for clients bound to a configured app + :param webhook: the local server refusing the delivery + :param caplog: pytest fixture capturing everything that was logged + """ + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + secret = delivery_of(client).signing_secret + key = management_key(client) + + with caplog.at_level(logging.DEBUG): + response = client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + assert secret not in response.text + assert secret not in caplog.text + assert key not in caplog.text + assert key not in response.text + + +def test_replay_sends_nothing_itself(make_client: ClientFactory, webhook: WebhookRecorder) -> None: + """ + the API process makes no outbound request, and replay must not be the exception + :param make_client: factory for clients bound to a configured app + :param webhook: the local server that would record a delivery if one were sent + """ + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + webhook.status = 200 + webhook.received.clear() + + response = client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + + assert response.status_code == 200 + assert webhook.received == [], "the API process sent the webhook itself" + assert attempts_of(client)[-1].attempt_number == 1 + + +# --- the worker picks it up -------------------------------------------------- + + +def test_the_worker_claims_a_replayed_delivery( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + webhook.status = 200 + webhook.received.clear() + + client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + handled = work_once(client) + + assert handled == 1 + assert len(webhook.received) == 1 + + +def test_a_replayed_delivery_that_succeeds_becomes_delivered( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + webhook.status = 200 + + client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + work_once(client) + + delivery = delivery_of(client) + assert delivery.state == DeliveryState.DELIVERED + assert delivery.completed_at is not None + assert delivery.attempts == 2 + assert delivery.cycle_attempts == 1 + + +def test_a_replayed_attempt_continues_the_numbering( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + """ + an attempt number must never be reused, so the history stays readable + :param make_client: factory for clients bound to a configured app + :param webhook: the local server refusing then accepting the delivery + """ + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + webhook.status = 200 + + client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + work_once(client) + + recorded = attempts_of(client) + assert [attempt.attempt_number for attempt in recorded] == [1, 2] + assert recorded[0].outcome == "http_error" + assert recorded[1].outcome == "succeeded" + assert len({attempt.id for attempt in recorded}) == 2 + + +def test_a_replay_earns_a_whole_retry_cycle( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + """ + the point of the cycle counter: a spent delivery must not fail on its first retry + :param make_client: factory for clients bound to a configured app + :param webhook: the local server answering with a retryable status throughout + """ + client = failing_client( + make_client, + webhook, + status=503, + webhook_max_attempts=3, + webhook_retry_initial_seconds=10, + ) + client.post("/f/contact-form", data={"email": "dev@example.com"}) + now = delivery_of(client).next_attempt_at + for _ in range(3): + work_once(client, now=now) + now = delivery_of(client).next_attempt_at + timedelta(seconds=1) + assert delivery_of(client).state == DeliveryState.FAILED + assert delivery_of(client).attempts == 3 + + client.post(REPLAY_PATH.format(delivery_id=delivery_of(client).id)) + replayed = delivery_of(client).next_attempt_at + for _ in range(3): + work_once(client, now=replayed) + replayed = delivery_of(client).next_attempt_at + timedelta(seconds=1) + + delivery = delivery_of(client) + assert delivery.state == DeliveryState.FAILED + assert delivery.attempts == 6, "the replayed cycle did not get a full allowance" + assert delivery.cycle_attempts == 3 + assert [attempt.attempt_number for attempt in attempts_of(client)] == [1, 2, 3, 4, 5, 6] + + +def test_the_backoff_starts_again_from_the_first_delay( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client( + make_client, + webhook, + status=503, + webhook_max_attempts=2, + webhook_retry_initial_seconds=10, + ) + client.post("/f/contact-form", data={"email": "dev@example.com"}) + now = delivery_of(client).next_attempt_at + work_once(client, now=now) + work_once(client, now=delivery_of(client).next_attempt_at) + client.post(REPLAY_PATH.format(delivery_id=delivery_of(client).id)) + + replayed_at = delivery_of(client).next_attempt_at + work_once(client, now=replayed_at) + + assert delivery_of(client).next_attempt_at == replayed_at + timedelta(seconds=10) + + +def test_the_replayed_delivery_carries_the_original_submission( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + client.post("/f/contact-form", data={"email": "dev@example.com", "note": "keep me"}) + work_once(client, now=delivery_of(client).next_attempt_at) + delivery_id = delivery_of(client).id + webhook.status = 200 + webhook.received.clear() + + client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + work_once(client) + + payload = json.loads(webhook.received[0].body)["submission"] + assert payload["id"] == delivery_of(client).submission_id + assert payload["fields"] == {"email": ["dev@example.com"], "note": ["keep me"]} + + +def test_the_detail_route_shows_the_whole_history_after_a_replay( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + client = failing_client(make_client, webhook) + delivery_id = fail_once(client, webhook) + webhook.status = 200 + client.post(REPLAY_PATH.format(delivery_id=delivery_id)) + work_once(client) + + body = client.get(f"/deliveries/{delivery_id}").json() + + assert [attempt["attempt_number"] for attempt in body["attempts"]] == [1, 2] + assert body["attempt_count"] == 2 + assert body["cycle_attempt_count"] == 1 + assert body["state"] == "delivered"