From a4edcdeb2c1b48e58c5d4828ec1e7d931b7c57d2 Mon Sep 17 00:00:00 2001
From: gbengaeben <gbengaeben6@gmail.com>
Date: Mon, 24 Aug 2026 10:09:10 +0000
Subject: [PATCH] feat(api): add webhook subscription lifecycle management
 (closes #531)

Expose list, update, delete, and manual redelivery for webhook
subscriptions, reusing the existing active flag and retry budget
instead of a delete-plus-recreate workflow. The signing secret stays
creation-time-only. Extend the shared contracts and SDK with the new
surface so provider/consumer suites pin the wire shape, and align
GET /streams list serialization with the string-id contract.
---
 api/src/contract-provider.spec.ts             | 114 ++++++-
 api/src/streams/streams.controller.ts         |  14 +-
 .../webhooks/dto/list-webhooks.query.dto.ts   |  23 ++
 api/src/webhooks/dto/update-webhook.dto.ts    |  52 +++
 .../webhook-deliveries-db.repository.ts       |  24 ++
 .../webhook-deliveries.repository.ts          |  16 +
 .../webhook-subscriptions-db.repository.ts    |  99 ++++++
 .../webhook-subscriptions.repository.ts       |  48 +++
 api/src/webhooks/webhooks.controller.spec.ts  | 229 ++++++++++---
 api/src/webhooks/webhooks.controller.ts       | 177 +++++++++-
 api/src/webhooks/webhooks.module.ts           |   5 +-
 api/src/webhooks/webhooks.service.spec.ts     | 302 +++++++++++++++++-
 api/src/webhooks/webhooks.service.ts          | 106 +++++-
 tests/contracts/src/contract.ts               |   3 +
 tests/contracts/src/index.ts                  |  11 +-
 tests/contracts/src/schemas.ts                |  47 +++
 tests/contracts/src/webhooks.contract.ts      | 127 ++++++++
 xstreamroll-sdk/README.md                     |  56 ++++
 .../__tests__/contract.consumer.test.ts       | 126 ++++++++
 xstreamroll-sdk/__tests__/http.test.ts        |  36 +++
 xstreamroll-sdk/__tests__/webhooks.test.ts    |   1 +
 xstreamroll-sdk/src/client.ts                 |  93 +++++-
 xstreamroll-sdk/src/http.ts                   |  32 +-
 xstreamroll-sdk/src/index.ts                  |   2 +
 xstreamroll-sdk/src/types.ts                  |  27 ++
 25 files changed, 1677 insertions(+), 93 deletions(-)
 create mode 100644 api/src/webhooks/dto/list-webhooks.query.dto.ts
 create mode 100644 api/src/webhooks/dto/update-webhook.dto.ts
 create mode 100644 tests/contracts/src/webhooks.contract.ts

diff --git a/api/src/contract-provider.spec.ts b/api/src/contract-provider.spec.ts
index 580f8fb..40f467e 100644
--- a/api/src/contract-provider.spec.ts
+++ b/api/src/contract-provider.spec.ts
@@ -22,6 +22,7 @@ import {
   registerBody,
   resolvePath,
   streamsContracts,
+  webhooksContracts,
   type Contract,
 } from "@xstreamroll/contract-tests"
 import request from "supertest"
@@ -44,6 +45,9 @@ import { StreamsService } from "./streams/streams.service"
 import { TagsRepository } from "./tags/repository/tags.repository"
 import { StreamTagsController } from "./tags/tags.controller"
 import { TagsService } from "./tags/tags.service"
+import { WebhookDeliveriesRepository } from "./webhooks/repository/webhook-deliveries.repository"
+import { WebhookSubscriptionsRepository } from "./webhooks/repository/webhook-subscriptions.repository"
+import { WebhooksController } from "./webhooks/webhooks.controller"
 import { WebhooksService } from "./webhooks/webhooks.service"
 
 process.env.JWT_SECRET ??= "test-secret"
@@ -88,12 +92,18 @@ describe("Contract provider verification (api)", () => {
   let app: INestApplication
   let jwtService: JwtService
   let streamsRepository: StreamsRepository
+  let subscriptionsRepository: WebhookSubscriptionsRepository
+  let deliveriesRepository: WebhookDeliveriesRepository
   let accessToken: string
   let userId: number
   let existingStreamId: string
+  let existingWebhookId: string
+  let existingDeliveryId: string
 
   beforeAll(async () => {
     streamsRepository = new StreamsRepository()
+    subscriptionsRepository = new WebhookSubscriptionsRepository()
+    deliveriesRepository = new WebhookDeliveriesRepository()
 
     /** Checks ownership against the same in-memory repository StreamsService uses. */
     const streamOwnershipService = {
@@ -113,15 +123,25 @@ describe("Contract provider verification (api)", () => {
         JwtModule.registerAsync({ useFactory: () => createJwtConfig() }),
         CacheModule.register(),
       ],
-      controllers: [StreamsController, StreamTagsController, AuthController],
+      controllers: [
+        StreamsController,
+        StreamTagsController,
+        AuthController,
+        WebhooksController,
+      ],
       providers: [
         StreamsService,
         TagsService,
         TagsRepository,
         { provide: StreamsRepository, useValue: streamsRepository },
+        WebhooksService,
+        {
+          provide: WebhookSubscriptionsRepository,
+          useValue: subscriptionsRepository,
+        },
         {
-          provide: WebhooksService,
-          useValue: { dispatchStreamEvent: async () => undefined },
+          provide: WebhookDeliveriesRepository,
+          useValue: deliveriesRepository,
         },
         AuthGuard,
         JwtExtractorService,
@@ -152,8 +172,16 @@ describe("Contract provider verification (api)", () => {
     }).compile()
 
     app = moduleFixture.createNestApplication()
+    // Mirror the real bootstrap (api/src/main.ts) so DTO coercion behaves
+    // exactly as it does in production — e.g. `streamId: "1"` in a JSON
+    // body must satisfy `@IsInt()` after implicit conversion.
     app.useGlobalPipes(
-      new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true }),
+      new ValidationPipe({
+        whitelist: true,
+        forbidNonWhitelisted: true,
+        transform: true,
+        transformOptions: { enableImplicitConversion: true },
+      }),
     )
     await app.init()
 
@@ -191,6 +219,30 @@ describe("Contract provider verification (api)", () => {
       "live-streaming",
     )
     await tagsRepository.attachToStream(stream.id, seededTag.id)
+
+    // Seed one webhook subscription + a terminally failed delivery so the
+    // update/delete/retry contracts have real ids. The subscription's
+    // events deliberately exclude `stream:started` so the `update-stream`
+    // contract (which dispatches that event) never fans out to it.
+    const webhook = await subscriptionsRepository.create({
+      userId,
+      streamId: stream.id,
+      url: "https://example.com/seed-hook",
+      events: ["stream:stopped"],
+      secret: "seed-secret",
+    })
+    existingWebhookId = String(webhook.id)
+
+    const delivery = await deliveriesRepository.create(
+      webhook.id,
+      "stream:stopped",
+      { streamId: stream.id },
+    )
+    delivery.status = "failed"
+    delivery.attemptCount = 6
+    delivery.nextAttemptAt = null
+    delivery.lastError = "connection refused"
+    existingDeliveryId = String(delivery.id)
   })
 
   afterAll(async () => {
@@ -203,10 +255,36 @@ describe("Contract provider verification (api)", () => {
       if (value === PLACEHOLDER.EXISTING_STREAM_ID)
         pathParams[key] = existingStreamId
       if (value === PLACEHOLDER.MISSING_STREAM_ID) pathParams[key] = "999999"
+      if (value === PLACEHOLDER.EXISTING_WEBHOOK_ID)
+        pathParams[key] = existingWebhookId
+      if (value === PLACEHOLDER.MISSING_WEBHOOK_ID) pathParams[key] = "999999"
+      if (value === PLACEHOLDER.EXISTING_DELIVERY_ID)
+        pathParams[key] = existingDeliveryId
     }
     return resolvePath({ ...contract.request, pathParams })
   }
 
+  /** Recursively substitutes placeholders in body values (e.g. streamId). */
+  function resolveBodyPlaceholders(value: unknown): unknown {
+    if (typeof value === "string") {
+      if (value === PLACEHOLDER.EXISTING_STREAM_ID) return existingStreamId
+      if (value === PLACEHOLDER.MISSING_STREAM_ID) return "999999"
+      if (value === PLACEHOLDER.EXISTING_WEBHOOK_ID) return existingWebhookId
+      if (value === PLACEHOLDER.MISSING_WEBHOOK_ID) return "999999"
+      if (value === PLACEHOLDER.EXISTING_DELIVERY_ID) return existingDeliveryId
+      return value
+    }
+    if (Array.isArray(value)) return value.map(resolveBodyPlaceholders)
+    if (value !== null && typeof value === "object") {
+      return Object.fromEntries(
+        Object.entries(value as Record<string, unknown>).map(
+          ([k, v]) => [k, resolveBodyPlaceholders(v)],
+        ),
+      )
+    }
+    return value
+  }
+
   async function execute(contract: Contract) {
     const path = resolveContractPath(contract)
     let req = request(app.getHttpServer())[
@@ -219,17 +297,7 @@ describe("Contract provider verification (api)", () => {
       req = req.set("X-Stream-Api-Key", process.env.STREAM_API_KEY ?? "")
     }
     if (contract.request.body !== undefined) {
-      // Substitute stream-id placeholders inside the request body (e.g. the
-      // `streamId` field of the ingest contract) the same way path params are.
-      const body = JSON.parse(JSON.stringify(contract.request.body)) as Record<
-        string,
-        unknown
-      >
-      for (const [k, v] of Object.entries(body)) {
-        if (v === PLACEHOLDER.EXISTING_STREAM_ID) body[k] = existingStreamId
-        if (v === PLACEHOLDER.MISSING_STREAM_ID) body[k] = "999999"
-      }
-      req = req.send(body)
+      req = req.send(resolveBodyPlaceholders(contract.request.body) as object)
     }
     return req
   }
@@ -268,6 +336,22 @@ describe("Contract provider verification (api)", () => {
     })
   })
 
+  describe.each(webhooksContracts)("$name", (contract) => {
+    it(contract.description, async () => {
+      const res = await execute(contract)
+
+      expect(res.status).toBe(contract.response.status)
+      const result = contract.response.schema.safeParse(res.body)
+      if (!result.success) {
+        throw new Error(
+          `${contract.name}: response did not satisfy the contract schema\n` +
+            `${JSON.stringify(result.error.format(), null, 2)}\n` +
+            `body: ${JSON.stringify(res.body, null, 2)}`,
+        )
+      }
+    })
+  })
+
   it("login contract uses credentials the register contract actually created", () => {
     // Sanity check that the two contract fixtures stay in sync with each
     // other — if this ever fails, `auth.contract.ts` was edited so the
diff --git a/api/src/streams/streams.controller.ts b/api/src/streams/streams.controller.ts
index 858e139..f87e006 100644
--- a/api/src/streams/streams.controller.ts
+++ b/api/src/streams/streams.controller.ts
@@ -43,7 +43,7 @@ import { StreamsService } from "./streams.service"
 import { AuthGuard } from "../common/guards/auth.guard"
 import { StreamOwnershipGuard } from "../common/guards/stream-ownership.guard"
 
-import type { PaginatedResponse, Stream } from "@xstreamroll/types"
+import type { Stream } from "@xstreamroll/types"
 import type { Request } from "express"
 
 const STREAM_ANALYTICS_CACHE_TTL_MS = 60_000
@@ -185,17 +185,25 @@ export class StreamsController {
   })
   @ApiOkResponse({ description: "Paginated list of streams." })
   @ApiUnauthorizedResponse({ description: "Authentication required." })
-  list(
+  async list(
     @Query() query: ListStreamsQueryDto,
     @Req() req: Request & { auth?: { userId: number } },
   ) {
     const page = query.page ?? 1
     const limit = query.limit ?? 20
-    return this.streamsService.list(page, limit, req.auth!.userId, {
+    const result = await this.streamsService.list(page, limit, req.auth!.userId, {
       status: query.status,
       visibility: query.visibility,
       ownerOnly: query.ownerOnly,
     })
+    // Single-stream endpoints serialize ids to strings via
+    // `toStreamResponse`; the list endpoint must do the same so the
+    // wire shape is consistent across the whole API (the shared
+    // `@xstreamroll/types#Stream` contract declares string ids).
+    return {
+      ...result,
+      data: result.data.map(toStreamResponse),
+    }
   }
 
   /**
diff --git a/api/src/webhooks/dto/list-webhooks.query.dto.ts b/api/src/webhooks/dto/list-webhooks.query.dto.ts
new file mode 100644
index 0000000..07a6298
--- /dev/null
+++ b/api/src/webhooks/dto/list-webhooks.query.dto.ts
@@ -0,0 +1,23 @@
+import { ApiPropertyOptional } from "@nestjs/swagger"
+import { Type } from "class-transformer"
+import { IsInt, IsOptional, Min } from "class-validator"
+
+import { PaginationQueryDto } from "../../common/dto/pagination.dto"
+
+/**
+ * Query parameters for `GET /webhooks`. Paging behaviour matches the
+ * rest of the API (1-indexed page, limit capped at 100). `streamId`
+ * narrows the result to the caller's subscriptions on one stream.
+ */
+export class ListWebhooksQueryDto extends PaginationQueryDto {
+  @ApiPropertyOptional({
+    description:
+      "Only return the caller's subscriptions on this stream.",
+    example: 1,
+  })
+  @IsOptional()
+  @Type(() => Number)
+  @IsInt({ message: "streamId must be an integer" })
+  @Min(1, { message: "streamId must be >= 1" })
+  streamId?: number
+}
diff --git a/api/src/webhooks/dto/update-webhook.dto.ts b/api/src/webhooks/dto/update-webhook.dto.ts
new file mode 100644
index 0000000..80495fe
--- /dev/null
+++ b/api/src/webhooks/dto/update-webhook.dto.ts
@@ -0,0 +1,52 @@
+import { ApiPropertyOptional } from "@nestjs/swagger"
+import { IsBoolean, IsIn, IsOptional, IsString, Matches } from "class-validator"
+
+import { STREAM_EVENTS } from "../../gateways/stream-events"
+
+const ALLOWED_EVENTS = Object.values(STREAM_EVENTS)
+
+/**
+ * Payload accepted by `PATCH /webhooks/:id`.
+ *
+ * Every field is optional — a PATCH updates only the fields present.
+ * There is deliberately **no `secret` field**: the signing secret is
+ * creation-time-only and can never be changed through the API (see the
+ * controller JSDoc). Event and URL validation mirrors
+ * {@link CreateWebhookDto} so the update path accepts exactly the same
+ * values the create path does.
+ */
+export class UpdateWebhookDto {
+  @ApiPropertyOptional({
+    description: "New URL that receives the signed POST on matching events.",
+    example: "https://example.com/webhooks/xstreamroll",
+  })
+  @IsOptional()
+  @Matches(/^https?:\/\/.+/, {
+    message: "url must be a valid absolute URL",
+  })
+  url?: string
+
+  @ApiPropertyOptional({
+    description: "Stream lifecycle events this webhook should fire on.",
+    example: ["stream:started", "stream:stopped"],
+    enum: ALLOWED_EVENTS,
+    isArray: true,
+  })
+  @IsOptional()
+  @IsString({ each: true, message: "events must be an array of strings" })
+  @IsIn(ALLOWED_EVENTS, {
+    each: true,
+    message: `each event must be one of: ${ALLOWED_EVENTS.join(", ")}`,
+  })
+  events?: string[]
+
+  @ApiPropertyOptional({
+    description:
+      "Deactivate (false) or reactivate (true) the subscription. Deactivation " +
+      "stops new fan-out and the retry sweep immediately; reactivation resumes both.",
+    example: false,
+  })
+  @IsOptional()
+  @IsBoolean({ message: "active must be a boolean" })
+  active?: boolean
+}
diff --git a/api/src/webhooks/repository/webhook-deliveries-db.repository.ts b/api/src/webhooks/repository/webhook-deliveries-db.repository.ts
index d050f7f..7fda4ba 100644
--- a/api/src/webhooks/repository/webhook-deliveries-db.repository.ts
+++ b/api/src/webhooks/repository/webhook-deliveries-db.repository.ts
@@ -5,6 +5,7 @@ import {
   ServiceUnavailableException,
 } from "@nestjs/common"
 import { Pool } from "pg"
+
 import { PG_POOL } from "../../database/database.module"
 import { WebhookDelivery } from "../webhook-delivery.entity"
 import { RecordAttemptInput } from "./webhook-deliveries.repository"
@@ -165,4 +166,27 @@ export class WebhookDeliveriesDbRepository {
       this.handleDbError(err, "recordAttempt")
     }
   }
+
+  /**
+   * Re-queues a delivery for a manual retry: marks it `pending` with
+   * `nextAttemptAt` set to now so the retry sweep picks it up
+   * immediately. `attemptCount` is deliberately kept — a manual retry
+   * must not silently grant a fresh retry budget beyond `MAX_RETRIES`
+   * (the next `nextAttemptAfter` computation still applies the cap).
+   */
+  async requeue(id: number): Promise<WebhookDelivery | undefined> {
+    try {
+      const { rows } = await this.pool.query<Record<string, unknown>>(
+        `UPDATE webhook_deliveries
+         SET status = 'pending',
+             next_attempt_at = CURRENT_TIMESTAMP
+         WHERE id = $1
+         RETURNING ${WebhookDeliveriesDbRepository.SELECT_COLUMNS}`,
+        [id],
+      )
+      return rows[0] ? this.rowToDelivery(rows[0]) : undefined
+    } catch (err) {
+      this.handleDbError(err, "requeue")
+    }
+  }
 }
diff --git a/api/src/webhooks/repository/webhook-deliveries.repository.ts b/api/src/webhooks/repository/webhook-deliveries.repository.ts
index 0be6613..c29a34f 100644
--- a/api/src/webhooks/repository/webhook-deliveries.repository.ts
+++ b/api/src/webhooks/repository/webhook-deliveries.repository.ts
@@ -1,4 +1,5 @@
 import { Injectable } from "@nestjs/common"
+
 import { WebhookDelivery } from "../webhook-delivery.entity"
 
 export interface RecordAttemptInput {
@@ -100,4 +101,19 @@ export class WebhookDeliveriesRepository {
     }
     return delivery
   }
+
+  /**
+   * Re-queues a delivery for a manual retry: marks it `pending` with
+   * `nextAttemptAt` set to now so the retry sweep picks it up
+   * immediately. `attemptCount` is deliberately kept — a manual retry
+   * must not silently grant a fresh retry budget beyond `MAX_RETRIES`.
+   */
+  async requeue(id: number): Promise<WebhookDelivery | undefined> {
+    const delivery = this.byId.get(id)
+    if (!delivery) return undefined
+
+    delivery.status = "pending"
+    delivery.nextAttemptAt = new Date()
+    return delivery
+  }
 }
diff --git a/api/src/webhooks/repository/webhook-subscriptions-db.repository.ts b/api/src/webhooks/repository/webhook-subscriptions-db.repository.ts
index 302d4de..bbd5092 100644
--- a/api/src/webhooks/repository/webhook-subscriptions-db.repository.ts
+++ b/api/src/webhooks/repository/webhook-subscriptions-db.repository.ts
@@ -5,6 +5,7 @@ import {
   ServiceUnavailableException,
 } from "@nestjs/common"
 import { Pool } from "pg"
+
 import { PG_POOL } from "../../database/database.module"
 import { WebhookSubscription } from "../webhook-subscription.entity"
 
@@ -92,4 +93,102 @@ export class WebhookSubscriptionsDbRepository {
       this.handleDbError(err, "findActiveByStreamAndEvent")
     }
   }
+
+  /**
+   * Paginated list of a user's subscriptions, newest first. Pass
+   * `streamId` to narrow to a single stream. The caller's ownership is
+   * enforced by the `user_id` filter itself — a user can only ever
+   * list their own subscriptions.
+   */
+  async listByUser(
+    userId: number,
+    page: number,
+    limit: number,
+    streamId?: number,
+  ): Promise<{ items: WebhookSubscription[]; total: number }> {
+    const offset = (page - 1) * limit
+    const where = [`user_id = $1`]
+    const params: Array<string | number> = [userId]
+    if (streamId !== undefined) {
+      params.push(streamId)
+      where.push(`stream_id = $${params.length}`)
+    }
+
+    try {
+      const { rows: countRows } = await this.pool.query<{ count: string }>(
+        `SELECT COUNT(*)::int AS count FROM webhook_subscriptions
+         WHERE ${where.join(" AND ")}`,
+        params,
+      )
+      const total = Number(countRows[0]?.count ?? 0)
+
+      const { rows } = await this.pool.query<Record<string, unknown>>(
+        `SELECT id, user_id, stream_id, url, events, secret, active, created_at
+         FROM webhook_subscriptions
+         WHERE ${where.join(" AND ")}
+         ORDER BY created_at DESC
+         LIMIT $${params.length + 1} OFFSET $${params.length + 2}`,
+        [...params, limit, offset],
+      )
+
+      return { items: rows.map((r) => this.rowToSubscription(r)), total }
+    } catch (err) {
+      this.handleDbError(err, "listByUser")
+    }
+  }
+
+  /**
+   * Applies a partial update. Only the fields present in `changes` are
+   * touched — notably there is no `secret` key, so a caller can never
+   * rotate the signing secret through this path. All values are bound
+   * as parameters, never interpolated.
+   */
+  async update(
+    id: number,
+    changes: { url?: string; events?: string[]; active?: boolean },
+  ): Promise<WebhookSubscription | undefined> {
+    const assignments: string[] = []
+    const params: Array<string | number | boolean | string[]> = [id]
+    if (changes.url !== undefined) {
+      params.push(changes.url)
+      assignments.push(`url = $${params.length}`)
+    }
+    if (changes.events !== undefined) {
+      params.push(changes.events)
+      assignments.push(`events = $${params.length}`)
+    }
+    if (changes.active !== undefined) {
+      params.push(changes.active)
+      assignments.push(`active = $${params.length}`)
+    }
+    if (assignments.length === 0) return undefined
+
+    try {
+      const { rows } = await this.pool.query<Record<string, unknown>>(
+        `UPDATE webhook_subscriptions
+         SET ${assignments.join(", ")}
+         WHERE id = $1
+         RETURNING id, user_id, stream_id, url, events, secret, active, created_at`,
+        params,
+      )
+      return rows[0] ? this.rowToSubscription(rows[0]) : undefined
+    } catch (err) {
+      this.handleDbError(err, "update")
+    }
+  }
+
+  /** Returns true when a subscription was deleted, false when it didn't exist. */
+  async delete(id: number): Promise<boolean> {
+    try {
+      // webhook_deliveries rows cascade on subscription deletion (see
+      // database/schema.sql), so one DELETE removes the whole record.
+      const { rowCount } = await this.pool.query(
+        `DELETE FROM webhook_subscriptions WHERE id = $1`,
+        [id],
+      )
+      return (rowCount ?? 0) > 0
+    } catch (err) {
+      this.handleDbError(err, "delete")
+    }
+  }
 }
diff --git a/api/src/webhooks/repository/webhook-subscriptions.repository.ts b/api/src/webhooks/repository/webhook-subscriptions.repository.ts
index 47383e8..9208382 100644
--- a/api/src/webhooks/repository/webhook-subscriptions.repository.ts
+++ b/api/src/webhooks/repository/webhook-subscriptions.repository.ts
@@ -1,4 +1,5 @@
 import { Injectable } from "@nestjs/common"
+
 import { WebhookSubscription } from "../webhook-subscription.entity"
 
 /**
@@ -51,4 +52,51 @@ export class WebhookSubscriptionsRepository {
       (s) => s.streamId === streamId && s.active && s.events.includes(event),
     )
   }
+
+  /**
+   * Paginated list of a user's subscriptions, newest first. Pass
+   * `streamId` to narrow to a single stream. The caller's ownership is
+   * enforced by the `user_id` filter itself — a user can only ever
+   * list their own subscriptions.
+   */
+  async listByUser(
+    userId: number,
+    page: number,
+    limit: number,
+    streamId?: number,
+  ): Promise<{ items: WebhookSubscription[]; total: number }> {
+    const matching = Array.from(this.byId.values())
+      .filter((s) => s.userId === userId)
+      .filter((s) => streamId === undefined || s.streamId === streamId)
+      .sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())
+
+    const offset = (page - 1) * limit
+    return {
+      items: matching.slice(offset, offset + limit),
+      total: matching.length,
+    }
+  }
+
+  /**
+   * Applies a partial update. Only the fields present in `changes` are
+   * touched — notably there is no `secret` key, so a caller can never
+   * rotate the signing secret through this path.
+   */
+  async update(
+    id: number,
+    changes: { url?: string; events?: string[]; active?: boolean },
+  ): Promise<WebhookSubscription | undefined> {
+    const subscription = this.byId.get(id)
+    if (!subscription) return undefined
+
+    if (changes.url !== undefined) subscription.url = changes.url
+    if (changes.events !== undefined) subscription.events = changes.events
+    if (changes.active !== undefined) subscription.active = changes.active
+    return subscription
+  }
+
+  /** Returns true when a subscription was deleted, false when it didn't exist. */
+  async delete(id: number): Promise<boolean> {
+    return this.byId.delete(id)
+  }
 }
diff --git a/api/src/webhooks/webhooks.controller.spec.ts b/api/src/webhooks/webhooks.controller.spec.ts
index fdcf642..8da5a63 100644
--- a/api/src/webhooks/webhooks.controller.spec.ts
+++ b/api/src/webhooks/webhooks.controller.spec.ts
@@ -17,18 +17,41 @@ jest.mock("../common/guards/stream-ownership.service", () => ({
   },
 }))
 
-import { ForbiddenException, NotFoundException } from "@nestjs/common"
-import type { Request } from "express"
+import { BadRequestException, ConflictException, ForbiddenException, NotFoundException } from "@nestjs/common"
+
+import { WebhookSubscription } from "./webhook-subscription.entity"
 import { WebhooksController } from "./webhooks.controller"
 import { WebhooksService } from "./webhooks.service"
 import { StreamOwnershipService } from "../common/guards/stream-ownership.service"
 
+import type { Request } from "express"
+
 const makeReq = (userId: number): Request & { auth: { userId: number } } =>
   ({ auth: { userId } }) as Request & { auth: { userId: number } }
 
+/** Fully-typed subscription fixture, including the secret. */
+function subscriptionFixture(overrides: Partial<WebhookSubscription> = {}): WebhookSubscription {
+  return {
+    id: 1,
+    userId: 7,
+    streamId: 5,
+    url: "https://example.com/hook",
+    events: ["stream:started"],
+    secret: "top-secret",
+    active: true,
+    createdAt: new Date("2026-01-01T00:00:00.000Z"),
+    ...overrides,
+  }
+}
+
 describe("WebhooksController", () => {
   let controller: WebhooksController
-  let service: jest.Mocked<Pick<WebhooksService, "register" | "findById" | "listDeliveries">>
+  let service: jest.Mocked<
+    Pick<
+      WebhooksService,
+      "register" | "findById" | "listDeliveries" | "listByUser" | "update" | "delete" | "retryDelivery"
+    >
+  >
   let ownership: jest.Mocked<Pick<StreamOwnershipService, "ownsStream">>
 
   beforeEach(() => {
@@ -36,6 +59,10 @@ describe("WebhooksController", () => {
       register: jest.fn(),
       findById: jest.fn(),
       listDeliveries: jest.fn(),
+      listByUser: jest.fn(),
+      update: jest.fn(),
+      delete: jest.fn(),
+      retryDelivery: jest.fn(),
     }
     ownership = {
       ownsStream: jest.fn(),
@@ -55,16 +82,7 @@ describe("WebhooksController", () => {
 
     it("registers a webhook when the caller owns the stream", async () => {
       ownership.ownsStream.mockResolvedValue(true)
-      service.register.mockResolvedValue({
-        id: 1,
-        userId: 7,
-        streamId: 5,
-        url: dto.url,
-        events: dto.events,
-        secret: "abc",
-        active: true,
-        createdAt: new Date(),
-      })
+      service.register.mockResolvedValue(subscriptionFixture())
 
       await controller.create(dto, makeReq(7))
 
@@ -87,18 +105,105 @@ describe("WebhooksController", () => {
     })
   })
 
+  describe("list", () => {
+    it("returns the caller's subscriptions with the secret stripped", async () => {
+      service.listByUser.mockResolvedValue({
+        data: [subscriptionFixture(), subscriptionFixture({ id: 2 })],
+        page: 1,
+        limit: 20,
+        total: 2,
+      })
+
+      const result = await controller.list({}, makeReq(7))
+
+      expect(service.listByUser).toHaveBeenCalledWith(7, 1, 20, undefined)
+      expect(result.total).toBe(2)
+      for (const item of result.data) {
+        expect(item).not.toHaveProperty("secret")
+      }
+    })
+
+    it("forwards page, limit, and the streamId filter", async () => {
+      service.listByUser.mockResolvedValue({ data: [], page: 2, limit: 5, total: 0 })
+
+      await controller.list({ page: 2, limit: 5, streamId: 9 }, makeReq(7))
+
+      expect(service.listByUser).toHaveBeenCalledWith(7, 2, 5, 9)
+    })
+  })
+
+  describe("update", () => {
+    it("updates url/events/active and returns the subscription without a secret", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture())
+      service.update.mockResolvedValue(
+        subscriptionFixture({ url: "https://example.com/new", active: false }),
+      )
+
+      const result = await controller.update(
+        1,
+        { url: "https://example.com/new", active: false },
+        makeReq(7),
+      )
+
+      expect(service.update).toHaveBeenCalledWith(1, {
+        url: "https://example.com/new",
+        events: undefined,
+        active: false,
+      })
+      expect(result).not.toHaveProperty("secret")
+      expect(result.url).toBe("https://example.com/new")
+    })
+
+    it("rejects an empty body with BadRequestException", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture())
+
+      await expect(controller.update(1, {}, makeReq(7))).rejects.toThrow(
+        BadRequestException,
+      )
+      expect(service.update).not.toHaveBeenCalled()
+    })
+
+    it("rejects when the caller does not own the webhook", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture({ userId: 42 }))
+
+      await expect(
+        controller.update(1, { active: false }, makeReq(7)),
+      ).rejects.toThrow(ForbiddenException)
+      expect(service.update).not.toHaveBeenCalled()
+    })
+  })
+
+  describe("delete", () => {
+    it("deletes the subscription when the caller owns it", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture())
+      service.delete.mockResolvedValue(undefined)
+
+      await controller.delete(1, makeReq(7))
+
+      expect(service.delete).toHaveBeenCalledWith(1)
+    })
+
+    it("rejects when the caller does not own the webhook", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture({ userId: 42 }))
+
+      await expect(controller.delete(1, makeReq(7))).rejects.toThrow(
+        ForbiddenException,
+      )
+      expect(service.delete).not.toHaveBeenCalled()
+    })
+
+    it("propagates NotFoundException for an unknown webhook", async () => {
+      service.findById.mockRejectedValue(new NotFoundException("webhook 999 not found"))
+
+      await expect(controller.delete(999, makeReq(7))).rejects.toThrow(
+        NotFoundException,
+      )
+    })
+  })
+
   describe("listDeliveries", () => {
     it("returns the delivery log when the caller owns the webhook", async () => {
-      service.findById.mockResolvedValue({
-        id: 1,
-        userId: 7,
-        streamId: 5,
-        url: "https://example.com/hook",
-        events: ["stream:started"],
-        secret: "abc",
-        active: true,
-        createdAt: new Date(),
-      })
+      service.findById.mockResolvedValue(subscriptionFixture())
       service.listDeliveries.mockResolvedValue({
         data: [],
         page: 1,
@@ -112,16 +217,7 @@ describe("WebhooksController", () => {
     })
 
     it("forwards explicit page and limit", async () => {
-      service.findById.mockResolvedValue({
-        id: 1,
-        userId: 7,
-        streamId: 5,
-        url: "https://example.com/hook",
-        events: ["stream:started"],
-        secret: "abc",
-        active: true,
-        createdAt: new Date(),
-      })
+      service.findById.mockResolvedValue(subscriptionFixture())
       service.listDeliveries.mockResolvedValue({
         data: [],
         page: 2,
@@ -135,16 +231,7 @@ describe("WebhooksController", () => {
     })
 
     it("rejects when the caller does not own the webhook", async () => {
-      service.findById.mockResolvedValue({
-        id: 1,
-        userId: 42,
-        streamId: 5,
-        url: "https://example.com/hook",
-        events: ["stream:started"],
-        secret: "abc",
-        active: true,
-        createdAt: new Date(),
-      })
+      service.findById.mockResolvedValue(subscriptionFixture({ userId: 42 }))
 
       await expect(
         controller.listDeliveries(1, {}, makeReq(7)),
@@ -160,4 +247,60 @@ describe("WebhooksController", () => {
       ).rejects.toThrow(NotFoundException)
     })
   })
+
+  describe("retryDelivery", () => {
+    it("re-queues a failed delivery when the caller owns the webhook", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture())
+      service.retryDelivery.mockResolvedValue({
+        id: 10,
+        webhookSubscriptionId: 1,
+        event: "stream:started",
+        payload: { streamId: 5 },
+        status: "pending",
+        attemptCount: 6,
+        lastStatusCode: null,
+        lastResponseBody: null,
+        lastError: "connection refused",
+        nextAttemptAt: new Date(),
+        deliveredAt: null,
+        createdAt: new Date(),
+      })
+
+      const result = await controller.retryDelivery(1, 10, makeReq(7))
+
+      expect(service.retryDelivery).toHaveBeenCalledWith(1, 10)
+      expect(result.status).toBe("pending")
+    })
+
+    it("rejects when the caller does not own the webhook", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture({ userId: 42 }))
+
+      await expect(controller.retryDelivery(1, 10, makeReq(7))).rejects.toThrow(
+        ForbiddenException,
+      )
+      expect(service.retryDelivery).not.toHaveBeenCalled()
+    })
+
+    it("propagates NotFoundException for an unknown delivery", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture())
+      service.retryDelivery.mockRejectedValue(
+        new NotFoundException("delivery 10 not found for webhook 1"),
+      )
+
+      await expect(controller.retryDelivery(1, 10, makeReq(7))).rejects.toThrow(
+        NotFoundException,
+      )
+    })
+
+    it("propagates ConflictException for an already-delivered delivery", async () => {
+      service.findById.mockResolvedValue(subscriptionFixture())
+      service.retryDelivery.mockRejectedValue(
+        new ConflictException("delivery 10 was already delivered"),
+      )
+
+      await expect(controller.retryDelivery(1, 10, makeReq(7))).rejects.toThrow(
+        ConflictException,
+      )
+    })
+  })
 })
diff --git a/api/src/webhooks/webhooks.controller.ts b/api/src/webhooks/webhooks.controller.ts
index 09fd919..578c690 100644
--- a/api/src/webhooks/webhooks.controller.ts
+++ b/api/src/webhooks/webhooks.controller.ts
@@ -1,12 +1,15 @@
 import {
+  BadRequestException,
   Body,
   Controller,
+  Delete,
   ForbiddenException,
   Get,
   HttpCode,
   HttpStatus,
   Param,
   ParseIntPipe,
+  Patch,
   Post,
   Query,
   Req,
@@ -14,28 +17,61 @@ import {
 } from "@nestjs/common"
 import {
   ApiBearerAuth,
+  ApiConflictResponse,
   ApiCreatedResponse,
   ApiForbiddenResponse,
+  ApiNoContentResponse,
   ApiNotFoundResponse,
   ApiOkResponse,
   ApiOperation,
   ApiTags,
   ApiUnauthorizedResponse,
 } from "@nestjs/swagger"
-import type { Request } from "express"
-import { AuthGuard } from "../common/guards/auth.guard"
-import { StreamOwnershipService } from "../common/guards/stream-ownership.service"
+
 import { CreateWebhookDto } from "./dto/create-webhook.dto"
 import { ListDeliveriesQueryDto } from "./dto/list-deliveries.query.dto"
+import { ListWebhooksQueryDto } from "./dto/list-webhooks.query.dto"
+import { UpdateWebhookDto } from "./dto/update-webhook.dto"
+import { WebhookSubscription } from "./webhook-subscription.entity"
 import { WebhooksService } from "./webhooks.service"
+import { AuthGuard } from "../common/guards/auth.guard"
+import { StreamOwnershipService } from "../common/guards/stream-ownership.service"
+
+import type { Request } from "express"
 
 type AuthedRequest = Request & { auth?: { userId: number } }
 
+/** Wire shape of a subscription on every endpoint except creation. */
+type WebhookSubscriptionResponse = Omit<WebhookSubscription, "secret">
+
 /**
- * Webhook subscription registration and delivery log.
+ * Webhook subscription registration, lifecycle management, and delivery
+ * log.
  *
- *   POST /webhooks                   Register a webhook (auth required, must own the stream)
- *   GET  /webhooks/:id/deliveries    Delivery log for a webhook (auth required, must own the webhook)
+ *   POST   /webhooks                                Register a webhook (auth required, must own the stream)
+ *   GET    /webhooks                                List the caller's subscriptions (auth required, paginated)
+ *   PATCH  /webhooks/:id                            Update URL, events, or active flag (auth required, must own the webhook)
+ *   DELETE /webhooks/:id                            Delete a subscription and its delivery history (auth required, must own the webhook)
+ *   GET    /webhooks/:id/deliveries                 Delivery log for a webhook (auth required, must own the webhook)
+ *   POST   /webhooks/:id/deliveries/:deliveryId/retry   Manually re-queue a failed/pending delivery (auth required, must own the webhook)
+ *
+ * ## `active` flag semantics
+ *
+ * Deactivating a subscription (`PATCH /webhooks/:id` with `active:
+ * false`) stops new fan-out immediately — `dispatchStreamEvent` only
+ * fans out to subscriptions matching `active = true` — and the retry
+ * sweep skips its pending deliveries while it stays inactive. Those
+ * pending deliveries are **left pending**, not cancelled: reactivating
+ * the subscription resumes their retry schedule unchanged. Deleting the
+ * subscription removes its deliveries entirely (ON DELETE CASCADE).
+ *
+ * ## Secret handling
+ *
+ * The signing secret is returned exactly once, in the `POST /webhooks`
+ * creation response. Every other endpoint (`GET /webhooks`,
+ * `PATCH /webhooks/:id`) omits it, and there is no endpoint that can
+ * change it — correcting a leaked secret means deleting the
+ * subscription and registering a new one.
  */
 @ApiTags("webhooks")
 @Controller("webhooks")
@@ -78,6 +114,87 @@ export class WebhooksController {
     })
   }
 
+  @Get()
+  @ApiOperation({
+    summary: "List webhook subscriptions",
+    description:
+      "Returns a paginated list of the caller's webhook subscriptions, " +
+      "newest first. Optionally narrows to a single stream via `streamId`. " +
+      "The signing secret is not included — it is creation-time-only.",
+  })
+  @ApiOkResponse({ description: "Paginated list of subscriptions." })
+  @ApiUnauthorizedResponse({ description: "Authentication required." })
+  async list(
+    @Query() query: ListWebhooksQueryDto,
+    @Req() req: AuthedRequest,
+  ) {
+    const page = query.page ?? 1
+    const limit = query.limit ?? 20
+    const result = await this.webhooksService.listByUser(
+      req.auth!.userId,
+      page,
+      limit,
+      query.streamId,
+    )
+    return { ...result, data: result.data.map(toSubscriptionResponse) }
+  }
+
+  @Patch(":id")
+  @ApiOperation({
+    summary: "Update a webhook subscription",
+    description:
+      "Partially updates the URL, event list, and/or `active` flag of a " +
+      "subscription. The signing secret cannot be changed — it is " +
+      "creation-time-only. Requires ownership.",
+  })
+  @ApiOkResponse({ description: "Subscription updated." })
+  @ApiNotFoundResponse({ description: "Webhook not found." })
+  @ApiUnauthorizedResponse({ description: "Authentication required." })
+  @ApiForbiddenResponse({ description: "You do not own this webhook." })
+  async update(
+    @Param("id", ParseIntPipe) id: number,
+    @Body() body: UpdateWebhookDto,
+    @Req() req: AuthedRequest,
+  ) {
+    await this.assertOwnership(id, req.auth!.userId)
+    if (
+      body.url === undefined &&
+      body.events === undefined &&
+      body.active === undefined
+    ) {
+      throw new BadRequestException(
+        "at least one of url, events, or active must be provided",
+      )
+    }
+
+    const updated = await this.webhooksService.update(id, {
+      url: body.url,
+      events: body.events,
+      active: body.active,
+    })
+    return toSubscriptionResponse(updated)
+  }
+
+  @Delete(":id")
+  @HttpCode(HttpStatus.NO_CONTENT)
+  @ApiOperation({
+    summary: "Delete a webhook subscription",
+    description:
+      "Deletes a subscription and, via the schema's ON DELETE CASCADE, its " +
+      "entire delivery history. Requires ownership.",
+  })
+  @ApiNoContentResponse({ description: "Subscription deleted." })
+  @ApiNotFoundResponse({ description: "Webhook not found." })
+  @ApiUnauthorizedResponse({ description: "Authentication required." })
+  @ApiForbiddenResponse({ description: "You do not own this webhook." })
+  async delete(
+    @Param("id", ParseIntPipe) id: number,
+    @Req() req: AuthedRequest,
+  ): Promise<void> {
+    await this.assertOwnership(id, req.auth!.userId)
+    await this.webhooksService.delete(id)
+  }
+
   @Get(":id/deliveries")
   @ApiOperation({
     summary: "List webhook deliveries",
@@ -93,15 +210,51 @@ export class WebhooksController {
     @Query() query: ListDeliveriesQueryDto,
     @Req() req: AuthedRequest,
   ) {
-    const subscription = await this.webhooksService.findById(id)
-    if (subscription.userId !== req.auth!.userId) {
-      throw new ForbiddenException(
-        `user ${req.auth!.userId} does not own webhook ${id}`,
-      )
-    }
+    await this.assertOwnership(id, req.auth!.userId)
 
     const page = query.page ?? 1
     const limit = query.limit ?? 20
     return this.webhooksService.listDeliveries(id, page, limit)
   }
+
+  @Post(":id/deliveries/:deliveryId/retry")
+  @HttpCode(HttpStatus.OK)
+  @ApiOperation({
+    summary: "Manually retry a webhook delivery",
+    description:
+      "Re-queues a failed or pending delivery so the retry sweep picks it up " +
+      "immediately. The retry budget (`MAX_RETRIES`) still applies — the " +
+      "attempt count is kept, not reset. Requires ownership of the webhook.",
+  })
+  @ApiOkResponse({ description: "Delivery re-queued." })
+  @ApiNotFoundResponse({
+    description: "Webhook or delivery not found (or delivery belongs to another webhook).",
+  })
+  @ApiConflictResponse({ description: "Delivery was already delivered." })
+  @ApiUnauthorizedResponse({ description: "Authentication required." })
+  @ApiForbiddenResponse({ description: "You do not own this webhook." })
+  async retryDelivery(
+    @Param("id", ParseIntPipe) id: number,
+    @Param("deliveryId", ParseIntPipe) deliveryId: number,
+    @Req() req: AuthedRequest,
+  ) {
+    await this.assertOwnership(id, req.auth!.userId)
+    return this.webhooksService.retryDelivery(id, deliveryId)
+  }
+
+  /** Loads the subscription and enforces that the caller owns it. */
+  private async assertOwnership(id: number, userId: number): Promise<void> {
+    const subscription = await this.webhooksService.findById(id)
+    if (subscription.userId !== userId) {
+      throw new ForbiddenException(`user ${userId} does not own webhook ${id}`)
+    }
+  }
+}
+
+/** Strips the signing secret for every non-creation response. */
+function toSubscriptionResponse(
+  subscription: WebhookSubscription,
+): WebhookSubscriptionResponse {
+  const { secret: _secret, ...rest } = subscription
+  return rest
 }
diff --git a/api/src/webhooks/webhooks.module.ts b/api/src/webhooks/webhooks.module.ts
index 5c84c54..e22d033 100644
--- a/api/src/webhooks/webhooks.module.ts
+++ b/api/src/webhooks/webhooks.module.ts
@@ -1,14 +1,15 @@
 import { Module } from "@nestjs/common"
 import { ScheduleModule } from "@nestjs/schedule"
+
 import { AuthModule } from "../auth/auth.module"
-import { AuthGuard } from "../common/guards/auth.guard"
-import { StreamOwnershipService } from "../common/guards/stream-ownership.service"
 import { WebhookDeliveriesDbRepository } from "./repository/webhook-deliveries-db.repository"
 import { WebhookDeliveriesRepository } from "./repository/webhook-deliveries.repository"
 import { WebhookSubscriptionsDbRepository } from "./repository/webhook-subscriptions-db.repository"
 import { WebhookSubscriptionsRepository } from "./repository/webhook-subscriptions.repository"
 import { WebhooksController } from "./webhooks.controller"
 import { WebhooksService } from "./webhooks.service"
+import { AuthGuard } from "../common/guards/auth.guard"
+import { StreamOwnershipService } from "../common/guards/stream-ownership.service"
 
 /**
  * Injection token used to swap the webhooks repository implementations.
diff --git a/api/src/webhooks/webhooks.service.spec.ts b/api/src/webhooks/webhooks.service.spec.ts
index 454f3e1..f6e27fb 100644
--- a/api/src/webhooks/webhooks.service.spec.ts
+++ b/api/src/webhooks/webhooks.service.spec.ts
@@ -1,13 +1,15 @@
-import { NotFoundException } from "@nestjs/common"
 import * as crypto from "crypto"
+
+import { ConflictException, NotFoundException } from "@nestjs/common"
+
+import { WebhookDeliveriesRepository } from "./repository/webhook-deliveries.repository"
+import { WebhookSubscriptionsRepository } from "./repository/webhook-subscriptions.repository"
 import {
   MAX_RETRIES,
   WebhooksService,
   nextAttemptAfter,
   signPayload,
 } from "./webhooks.service"
-import { WebhookDeliveriesRepository } from "./repository/webhook-deliveries.repository"
-import { WebhookSubscriptionsRepository } from "./repository/webhook-subscriptions.repository"
 
 /** Flushes the microtask queue so fire-and-forget attemptDelivery() settles. */
 function flushPromises(): Promise<void> {
@@ -126,6 +128,44 @@ describe("WebhooksService", () => {
       expect(fetchMock).not.toHaveBeenCalled()
     })
 
+    it("does not deliver to a deactivated subscription (fan-out stops immediately)", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+      await service.update(sub.id, { active: false })
+
+      await service.dispatchStreamEvent(5, "stream:started", {})
+      await flushPromises()
+
+      expect(fetchMock).not.toHaveBeenCalled()
+      const list = await service.listDeliveries(sub.id, 1, 20)
+      expect(list.data).toHaveLength(0)
+    })
+
+    it("reactivating a deactivated subscription resumes fan-out", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+      fetchMock.mockResolvedValue({ status: 200, text: async () => "ok" })
+
+      await service.update(sub.id, { active: false })
+      await service.dispatchStreamEvent(5, "stream:started", {})
+      await flushPromises()
+      expect(fetchMock).not.toHaveBeenCalled()
+
+      await service.update(sub.id, { active: true })
+      await service.dispatchStreamEvent(5, "stream:started", {})
+      await flushPromises()
+
+      expect(fetchMock).toHaveBeenCalledTimes(1)
+    })
+
     it("records a non-2xx response as pending with a future next attempt", async () => {
       const sub = await service.register({
         userId: 1,
@@ -167,6 +207,221 @@ describe("WebhooksService", () => {
     })
   })
 
+  describe("listByUser", () => {
+    it("returns only the caller's subscriptions, newest first", async () => {
+      const own = await service.register({
+        userId: 1,
+        streamId: 2,
+        url: "https://example.com/own",
+        events: ["stream:started"],
+      })
+      await service.register({
+        userId: 99,
+        streamId: 2,
+        url: "https://example.com/other",
+        events: ["stream:started"],
+      })
+
+      const res = await service.listByUser(1, 1, 20)
+      expect(res.data).toHaveLength(1)
+      expect(res.data[0].id).toBe(own.id)
+      expect(res.total).toBe(1)
+    })
+
+    it("filters by streamId when provided", async () => {
+      await service.register({
+        userId: 1,
+        streamId: 2,
+        url: "https://example.com/a",
+        events: ["stream:started"],
+      })
+      await service.register({
+        userId: 1,
+        streamId: 3,
+        url: "https://example.com/b",
+        events: ["stream:started"],
+      })
+
+      const res = await service.listByUser(1, 1, 20, 3)
+      expect(res.data).toHaveLength(1)
+      expect(res.data[0].streamId).toBe(3)
+    })
+
+    it("paginates like the other list endpoints", async () => {
+      for (let i = 0; i < 3; i++) {
+        await service.register({
+          userId: 1,
+          streamId: 2,
+          url: `https://example.com/${i}`,
+          events: ["stream:started"],
+        })
+      }
+
+      const page1 = await service.listByUser(1, 1, 2)
+      const page2 = await service.listByUser(1, 2, 2)
+      expect(page1.data).toHaveLength(2)
+      expect(page2.data).toHaveLength(1)
+      expect(page1.total).toBe(3)
+      expect(page2.total).toBe(3)
+    })
+  })
+
+  describe("update", () => {
+    it("updates url, events, and active", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 2,
+        url: "https://example.com/old",
+        events: ["stream:started"],
+      })
+
+      const updated = await service.update(sub.id, {
+        url: "https://example.com/new",
+        events: ["stream:stopped"],
+        active: false,
+      })
+
+      expect(updated.url).toBe("https://example.com/new")
+      expect(updated.events).toEqual(["stream:stopped"])
+      expect(updated.active).toBe(false)
+      // Secret is never touched by an update.
+      expect(updated.secret).toBe(sub.secret)
+    })
+
+    it("leaves omitted fields unchanged", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 2,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+
+      const updated = await service.update(sub.id, { active: false })
+      expect(updated.active).toBe(false)
+      expect(updated.url).toBe("https://example.com/hook")
+      expect(updated.events).toEqual(["stream:started"])
+    })
+
+    it("throws NotFoundException for an unknown webhook", async () => {
+      await expect(service.update(999, { active: false })).rejects.toThrow(
+        NotFoundException,
+      )
+    })
+  })
+
+  describe("delete", () => {
+    it("deletes an existing subscription", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 2,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+
+      await expect(service.delete(sub.id)).resolves.toBeUndefined()
+      await expect(service.findById(sub.id)).rejects.toThrow(NotFoundException)
+    })
+
+    it("throws NotFoundException for an unknown webhook", async () => {
+      await expect(service.delete(999)).rejects.toThrow(NotFoundException)
+    })
+  })
+
+  describe("retryDelivery", () => {
+    it("re-queues a terminally failed delivery keeping its attempt count", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+      // Simulate an exhausted delivery: 6 attempts, terminally failed.
+      const delivery = await deliveries.create(sub.id, "stream:started", {})
+      delivery.status = "failed"
+      delivery.attemptCount = MAX_RETRIES + 1
+      delivery.nextAttemptAt = null
+
+      const before = Date.now()
+      const requeued = await service.retryDelivery(sub.id, delivery.id)
+
+      expect(requeued.status).toBe("pending")
+      expect(requeued.attemptCount).toBe(MAX_RETRIES + 1)
+      expect(requeued.nextAttemptAt).not.toBeNull()
+      expect(requeued.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual(before)
+    })
+
+    it("pulls a pending delivery forward so the sweep picks it up immediately", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+      const delivery = await deliveries.create(sub.id, "stream:started", {})
+      delivery.status = "pending"
+      delivery.nextAttemptAt = new Date(Date.now() + 60_000)
+
+      const requeued = await service.retryDelivery(sub.id, delivery.id)
+      expect(requeued.status).toBe("pending")
+      expect(requeued.nextAttemptAt!.getTime()).toBeLessThanOrEqual(Date.now())
+    })
+
+    it("throws NotFoundException for an unknown webhook", async () => {
+      await expect(service.retryDelivery(999, 1)).rejects.toThrow(
+        NotFoundException,
+      )
+    })
+
+    it("throws NotFoundException for an unknown delivery", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+      await expect(service.retryDelivery(sub.id, 999)).rejects.toThrow(
+        NotFoundException,
+      )
+    })
+
+    it("throws NotFoundException when the delivery belongs to another webhook", async () => {
+      const subA = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/a",
+        events: ["stream:started"],
+      })
+      const subB = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/b",
+        events: ["stream:started"],
+      })
+      const delivery = await deliveries.create(subA.id, "stream:started", {})
+      delivery.status = "failed"
+
+      await expect(service.retryDelivery(subB.id, delivery.id)).rejects.toThrow(
+        NotFoundException,
+      )
+    })
+
+    it("throws ConflictException for an already-delivered delivery", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+      const delivery = await deliveries.create(sub.id, "stream:started", {})
+      delivery.status = "success"
+      delivery.deliveredAt = new Date()
+
+      await expect(service.retryDelivery(sub.id, delivery.id)).rejects.toThrow(
+        ConflictException,
+      )
+    })
+  })
+
   describe("sweepRetries", () => {
     it("re-attempts due pending deliveries and marks a delivery failed once retries are exhausted", async () => {
       const sub = await service.register({
@@ -223,6 +478,40 @@ describe("WebhooksService", () => {
 
       expect(fetchMock).not.toHaveBeenCalled()
     })
+
+    it("reactivating a deactivated subscription resumes its pending delivery retries", async () => {
+      const sub = await service.register({
+        userId: 1,
+        streamId: 5,
+        url: "https://example.com/hook",
+        events: ["stream:started"],
+      })
+      fetchMock.mockResolvedValue({ status: 500, text: async () => "boom" })
+      await service.dispatchStreamEvent(5, "stream:started", {})
+      await flushPromises()
+
+      // Deactivate, make the pending delivery due, and sweep — nothing
+      // is attempted while the subscription stays inactive.
+      await service.update(sub.id, { active: false })
+      fetchMock.mockClear()
+      let list = await service.listDeliveries(sub.id, 1, 20)
+      list.data[0].nextAttemptAt = new Date(Date.now() - 1_000)
+
+      await service.sweepRetries()
+      await flushPromises()
+      expect(fetchMock).not.toHaveBeenCalled()
+
+      // Reactivate: the same delivery is still pending with its retry
+      // schedule intact, so the sweep picks it back up.
+      await service.update(sub.id, { active: true })
+      list = await service.listDeliveries(sub.id, 1, 20)
+      list.data[0].nextAttemptAt = new Date(Date.now() - 1_000)
+
+      await service.sweepRetries()
+      await flushPromises()
+
+      expect(fetchMock).toHaveBeenCalledTimes(1)
+    })
   })
 })
 
@@ -257,13 +546,16 @@ describe("signPayload", () => {
     const secret = "test-secret"
     const body = JSON.stringify({ hello: "world" })
     const expected =
-      "sha256=" + crypto.createHmac("sha256", secret).update(body, "utf8").digest("hex")
+      "sha256=" +
+      crypto.createHmac("sha256", secret).update(body, "utf8").digest("hex")
     expect(signPayload(secret, body)).toBe(expected)
   })
 
   it("produces a different signature for a different secret", () => {
     const body = JSON.stringify({ hello: "world" })
-    expect(signPayload("secret-a", body)).not.toBe(signPayload("secret-b", body))
+    expect(signPayload("secret-a", body)).not.toBe(
+      signPayload("secret-b", body),
+    )
   })
 
   it("produces a different signature if the body changes by a single byte", () => {
diff --git a/api/src/webhooks/webhooks.service.ts b/api/src/webhooks/webhooks.service.ts
index ad4c75a..ec464d4 100644
--- a/api/src/webhooks/webhooks.service.ts
+++ b/api/src/webhooks/webhooks.service.ts
@@ -1,11 +1,18 @@
-import { Injectable, Logger, NotFoundException } from "@nestjs/common"
-import { Interval } from "@nestjs/schedule"
 import * as crypto from "crypto"
-import { PaginatedResult } from "../common/dto/pagination.dto"
-import { WebhookDelivery } from "./webhook-delivery.entity"
-import { WebhookSubscription } from "./webhook-subscription.entity"
+
+import {
+  ConflictException,
+  Injectable,
+  Logger,
+  NotFoundException,
+} from "@nestjs/common"
+import { Interval } from "@nestjs/schedule"
+
 import { WebhookDeliveriesRepository } from "./repository/webhook-deliveries.repository"
 import { WebhookSubscriptionsRepository } from "./repository/webhook-subscriptions.repository"
+import { WebhookDelivery } from "./webhook-delivery.entity"
+import { WebhookSubscription } from "./webhook-subscription.entity"
+import { PaginatedResult } from "../common/dto/pagination.dto"
 
 const WEBHOOK_SECRET_BYTES = 32
 const WEBHOOK_DELIVERY_TIMEOUT_MS = 10_000
@@ -78,6 +85,95 @@ export class WebhooksService {
     return { data: items, page, limit, total }
   }
 
+  /**
+   * Paginated list of a caller's own subscriptions, newest first. The
+   * `userId` filter makes the listing inherently owner-scoped — there is
+   * no way to enumerate another user's webhooks.
+   */
+  async listByUser(
+    userId: number,
+    page: number,
+    limit: number,
+    streamId?: number,
+  ): Promise<PaginatedResult<WebhookSubscription>> {
+    const { items, total } = await this.subscriptions.listByUser(
+      userId,
+      page,
+      limit,
+      streamId,
+    )
+    return { data: items, page, limit, total }
+  }
+
+  /**
+   * Partially updates a subscription (URL, events, and/or `active`).
+   * The signing secret is deliberately not updatable — it is
+   * creation-time-only (see `WebhooksController`). Returns the updated
+   * subscription or throws 404 when the webhook does not exist.
+   */
+  async update(
+    id: number,
+    changes: { url?: string; events?: string[]; active?: boolean },
+  ): Promise<WebhookSubscription> {
+    await this.findById(id)
+    const updated = await this.subscriptions.update(id, changes)
+    if (!updated) {
+      // The findById above succeeded, so this can only be a race with a
+      // concurrent delete; surface the same 404 the caller expects.
+      throw new NotFoundException(`webhook ${id} not found`)
+    }
+    return updated
+  }
+
+  /**
+   * Deletes a subscription and (via the schema's ON DELETE CASCADE) all
+   * of its deliveries. Throws 404 when the webhook does not exist.
+   */
+  async delete(id: number): Promise<void> {
+    const deleted = await this.subscriptions.delete(id)
+    if (!deleted) {
+      throw new NotFoundException(`webhook ${id} not found`)
+    }
+  }
+
+  /**
+   * Manually re-queues a failed or pending delivery so the retry sweep
+   * picks it up on its next pass. `attemptCount` is kept, so the manual
+   * retry still operates inside the `MAX_RETRIES` budget — once the
+   * budget is exhausted the delivery goes terminally failed again after
+   * this one extra attempt.
+   *
+   * @throws NotFoundException  when the webhook or delivery does not exist,
+   *                            or the delivery belongs to another webhook.
+   * @throws ConflictException  when the delivery already succeeded.
+   */
+  async retryDelivery(
+    webhookId: number,
+    deliveryId: number,
+  ): Promise<WebhookDelivery> {
+    await this.findById(webhookId)
+
+    const delivery = await this.deliveries.findById(deliveryId)
+    if (!delivery || delivery.webhookSubscriptionId !== webhookId) {
+      throw new NotFoundException(
+        `delivery ${deliveryId} not found for webhook ${webhookId}`,
+      )
+    }
+    if (delivery.status === "success") {
+      throw new ConflictException(
+        `delivery ${deliveryId} was already delivered`,
+      )
+    }
+
+    const requeued = await this.deliveries.requeue(deliveryId)
+    if (!requeued) {
+      throw new NotFoundException(
+        `delivery ${deliveryId} not found for webhook ${webhookId}`,
+      )
+    }
+    return requeued
+  }
+
   /**
    * Entry point called by application services (e.g. `StreamsService`) when
    * a stream lifecycle event occurs. Fans out to every active subscription
diff --git a/tests/contracts/src/contract.ts b/tests/contracts/src/contract.ts
index 3afbe0e..85cda8d 100644
--- a/tests/contracts/src/contract.ts
+++ b/tests/contracts/src/contract.ts
@@ -55,6 +55,9 @@ export interface Contract {
 export const PLACEHOLDER = {
   EXISTING_STREAM_ID: "__EXISTING_STREAM_ID__",
   MISSING_STREAM_ID: "__MISSING_STREAM_ID__",
+  EXISTING_WEBHOOK_ID: "__EXISTING_WEBHOOK_ID__",
+  MISSING_WEBHOOK_ID: "__MISSING_WEBHOOK_ID__",
+  EXISTING_DELIVERY_ID: "__EXISTING_DELIVERY_ID__",
 } as const
 
 /** Substitutes `:param` placeholders and appends the query string. */
diff --git a/tests/contracts/src/index.ts b/tests/contracts/src/index.ts
index 417f35c..0f3bcc9 100644
--- a/tests/contracts/src/index.ts
+++ b/tests/contracts/src/index.ts
@@ -17,10 +17,17 @@ export * from "./contract"
 export * from "./schemas"
 export * from "./streams.contract"
 export * from "./auth.contract"
+export * from "./webhooks.contract"
 
 import { authContracts } from "./auth.contract"
-import type { Contract } from "./contract"
 import { streamsContracts } from "./streams.contract"
+import { webhooksContracts } from "./webhooks.contract"
+
+import type { Contract } from "./contract"
 
 /** Every contract in the suite, across all resources. */
-export const allContracts: Contract[] = [...streamsContracts, ...authContracts]
+export const allContracts: Contract[] = [
+  ...streamsContracts,
+  ...authContracts,
+  ...webhooksContracts,
+]
diff --git a/tests/contracts/src/schemas.ts b/tests/contracts/src/schemas.ts
index a4a2785..15ae2e9 100644
--- a/tests/contracts/src/schemas.ts
+++ b/tests/contracts/src/schemas.ts
@@ -96,6 +96,53 @@ export const pendingStreamEventSchema = z.object({
   timestamp: z.string(),
 })
 
+/**
+ * A webhook subscription as returned by `POST /webhooks` — the only
+ * response that includes the signing `secret`.
+ */
+export const webhookSubscriptionSchema = z.object({
+  id: z.union([z.string(), z.number()]),
+  userId: z.union([z.string(), z.number()]),
+  streamId: z.union([z.string(), z.number()]),
+  url: z.string(),
+  events: z.array(z.string()),
+  secret: z.string(),
+  active: z.boolean(),
+  createdAt: z.string(),
+})
+
+/**
+ * A webhook subscription on every non-creation endpoint (`GET /webhooks`,
+ * `PATCH /webhooks/:id`) — identical to `webhookSubscriptionSchema` minus
+ * the secret, which is creation-time-only.
+ */
+export const webhookSubscriptionSummarySchema = webhookSubscriptionSchema.omit({
+  secret: true,
+})
+
+export const paginatedWebhookSubscriptionsSchema = z.object({
+  data: z.array(webhookSubscriptionSummarySchema),
+  total: z.number(),
+  page: z.number(),
+  limit: z.number(),
+})
+
+/** A single webhook delivery, as returned by the deliveries endpoints. */
+export const webhookDeliverySchema = z.object({
+  id: z.union([z.string(), z.number()]),
+  webhookSubscriptionId: z.union([z.string(), z.number()]),
+  event: z.string(),
+  payload: z.record(z.string(), z.unknown()),
+  status: z.enum(["pending", "success", "failed"]),
+  attemptCount: z.number(),
+  lastStatusCode: z.number().nullable(),
+  lastResponseBody: z.string().nullable(),
+  lastError: z.string().nullable(),
+  nextAttemptAt: z.string().nullable(),
+  deliveredAt: z.string().nullable(),
+  createdAt: z.string(),
+})
+
 export const apiErrorSchema = typed<ApiErrorResponse>()(
   z.object({
     statusCode: z.number(),
diff --git a/tests/contracts/src/webhooks.contract.ts b/tests/contracts/src/webhooks.contract.ts
new file mode 100644
index 0000000..b72ddff
--- /dev/null
+++ b/tests/contracts/src/webhooks.contract.ts
@@ -0,0 +1,127 @@
+import { z } from "zod"
+
+import { PLACEHOLDER, type Contract } from "./contract"
+import {
+  apiErrorSchema,
+  paginatedWebhookSubscriptionsSchema,
+  webhookDeliverySchema,
+  webhookSubscriptionSchema,
+  webhookSubscriptionSummarySchema,
+} from "./schemas"
+
+/** Empty (204) responses have no body; supertest yields `{}`. */
+const emptyBodySchema = z.object({})
+
+const createBody = {
+  streamId: PLACEHOLDER.EXISTING_STREAM_ID,
+  url: "https://example.com/webhooks/contract",
+  events: ["stream:started"],
+}
+
+export const webhooksContracts: Contract[] = [
+  {
+    name: "create-webhook",
+    description: "POST /webhooks registers a subscription and returns the secret once",
+    consumer: "xstreamroll-sdk",
+    provider: "api",
+    request: {
+      method: "POST",
+      path: "/webhooks",
+      body: createBody,
+      authenticated: true,
+    },
+    response: {
+      status: 201,
+      schema: webhookSubscriptionSchema,
+    },
+  },
+  {
+    name: "list-webhooks",
+    description: "GET /webhooks returns the caller's subscriptions without secrets",
+    consumer: "xstreamroll-sdk",
+    provider: "api",
+    request: {
+      method: "GET",
+      path: "/webhooks",
+      query: { page: 1, limit: 20 },
+      authenticated: true,
+    },
+    response: {
+      status: 200,
+      schema: paginatedWebhookSubscriptionsSchema,
+    },
+  },
+  {
+    name: "update-webhook",
+    description: "PATCH /webhooks/:id updates url/events/active but never the secret",
+    consumer: "xstreamroll-sdk",
+    provider: "api",
+    request: {
+      method: "PATCH",
+      path: "/webhooks/:id",
+      pathParams: { id: PLACEHOLDER.EXISTING_WEBHOOK_ID },
+      body: { active: false },
+      authenticated: true,
+    },
+    response: {
+      status: 200,
+      schema: webhookSubscriptionSummarySchema,
+    },
+  },
+  {
+    name: "retry-webhook-delivery",
+    description: "POST /webhooks/:id/deliveries/:deliveryId/retry re-queues a failed delivery",
+    consumer: "xstreamroll-sdk",
+    provider: "api",
+    request: {
+      method: "POST",
+      path: "/webhooks/:id/deliveries/:deliveryId/retry",
+      pathParams: {
+        id: PLACEHOLDER.EXISTING_WEBHOOK_ID,
+        deliveryId: PLACEHOLDER.EXISTING_DELIVERY_ID,
+      },
+      authenticated: true,
+    },
+    response: {
+      status: 200,
+      schema: webhookDeliverySchema,
+    },
+  },
+  {
+    name: "retry-webhook-delivery-not-found",
+    description: "POST retry on a nonexistent webhook returns the standard API error body",
+    consumer: "xstreamroll-sdk",
+    provider: "api",
+    request: {
+      method: "POST",
+      path: "/webhooks/:id/deliveries/:deliveryId/retry",
+      pathParams: {
+        id: PLACEHOLDER.MISSING_WEBHOOK_ID,
+        deliveryId: PLACEHOLDER.EXISTING_DELIVERY_ID,
+      },
+      authenticated: true,
+    },
+    response: {
+      status: 404,
+      schema: apiErrorSchema,
+    },
+  },
+  {
+    // Runs last on purpose: it removes the seeded subscription the retry
+    // contracts above depend on. Contracts execute in array order.
+    name: "delete-webhook",
+    description: "DELETE /webhooks/:id removes the subscription and returns 204",
+    consumer: "xstreamroll-sdk",
+    provider: "api",
+    request: {
+      method: "DELETE",
+      path: "/webhooks/:id",
+      pathParams: { id: PLACEHOLDER.EXISTING_WEBHOOK_ID },
+      authenticated: true,
+    },
+    response: {
+      status: 204,
+      schema: emptyBodySchema,
+    },
+  },
+]
diff --git a/xstreamroll-sdk/README.md b/xstreamroll-sdk/README.md
index 50901c0..e002d4a 100644
--- a/xstreamroll-sdk/README.md
+++ b/xstreamroll-sdk/README.md
@@ -257,6 +257,59 @@ await client.publishEvent({
 
 ---
 
+## Webhooks
+
+Webhooks notify an external URL whenever a stream lifecycle event
+happens. The SDK covers the full subscription lifecycle: create, list,
+deactivate/reactivate, update, delete, and manual redelivery.
+
+```ts
+// 1. Subscribe — the only response that ever contains the signing secret.
+const created = await client.subscribeWebhook({
+  streamId: "stream_abc",
+  url: "https://example.com/webhooks/xstreamroll",
+  events: ["stream:started", "stream:stopped"],
+})
+// Store `created.secret` immediately — it is never returned again.
+
+// 2. Manage subscriptions (the secret is never included in these responses).
+const all = await client.listWebhooks()
+const onStream = await client.listWebhooks({ streamId: "stream_abc" }) // optional filter
+
+// 3. Deactivate (stops new deliveries and retries immediately)…
+const paused = await client.updateWebhook(created.id, { active: false })
+// …and reactivate later; the URL and event list can be changed too.
+const resumed = await client.updateWebhook(created.id, {
+  url: "https://example.com/webhooks/xstreamroll-v2",
+  events: ["stream:started"],
+  active: true,
+})
+
+// 4. Inspect deliveries and manually re-queue a failed one. The retry
+//    budget still applies — the attempt count is kept, not reset.
+const { data: deliveries } = await client.paginateAll<WebhookDelivery>(
+  "/webhooks/1/deliveries",
+)
+const failed = deliveries.find((d) => d.status === "failed")
+if (failed) {
+  await client.retryWebhookDelivery(created.id, failed.id)
+}
+
+// 5. Remove a subscription and its delivery history.
+await client.deleteWebhook(created.id)
+```
+
+Verify inbound deliveries with `verifyWebhookSignature(secret, rawBody,
+signature)` — the `X-Webhook-Signature` header must be checked against
+the **exact raw request body bytes**, not a re-serialized JSON object.
+
+> **Secret handling:** the signing secret is returned exactly once, by
+> `subscribeWebhook()`. Every other endpoint omits it, and there is no
+> way to change it — correcting a leaked secret means deleting the
+> subscription and registering a new one.
+
+---
+
 ## HTTP transport
 
 `HttpClient` is a small, `fetch`-based wrapper that:
@@ -400,6 +453,9 @@ The SDK ships full type definitions. The most useful are:
   — stream CRUD shapes.
 * `StreamEvent`, `StreamEventRecord`, `StreamEventType` — event
   shapes.
+* `WebhookSubscription`, `WebhookSubscriptionSummary`,
+  `UpdateWebhookDto`, `WebhookDelivery` — webhook shapes (the
+  subscription summary omits the creation-time-only `secret`).
 * `AuthTokens`, `User`, `CreateUserDto`, `UpdateUserDto` — auth
   shapes.
 * `PaginatedResponse<T>`, `PaginationParams` — list helpers.
diff --git a/xstreamroll-sdk/__tests__/contract.consumer.test.ts b/xstreamroll-sdk/__tests__/contract.consumer.test.ts
index 0f06a49..eaf18dc 100644
--- a/xstreamroll-sdk/__tests__/contract.consumer.test.ts
+++ b/xstreamroll-sdk/__tests__/contract.consumer.test.ts
@@ -20,8 +20,12 @@
 import {
   allContracts,
   authResponseSchema,
+  paginatedWebhookSubscriptionsSchema,
   pendingStreamEventSchema,
   streamSchema,
+  webhookDeliverySchema,
+  webhookSubscriptionSchema,
+  webhookSubscriptionSummarySchema,
   type Contract,
 } from "@xstreamroll/contract-tests"
 import nock from "nock"
@@ -57,6 +61,7 @@ describe("Consumer contract verification (xstreamroll-sdk)", () => {
       name: "Contract stream",
       description: null,
       status: "active",
+      visibility: "private",
       createdAt: "2026-01-01T00:00:00.000Z",
       updatedAt: "2026-01-01T00:00:00.000Z",
     }
@@ -156,4 +161,125 @@ describe("Consumer contract verification (xstreamroll-sdk)", () => {
     expect(result.accessToken).toBe(example.accessToken)
     expect(result.refreshToken).toBe(example.refreshToken)
   })
+
+  // ── Webhooks (issue #531) ────────────────────────────────────────────────
+
+  it("subscribeWebhook() sends the body the create-webhook contract expects", async () => {
+    const c = contract("create-webhook")
+    const example = {
+      id: "1",
+      userId: "7",
+      streamId: "42",
+      url: "https://example.com/webhooks/contract",
+      events: ["stream:started"],
+      secret: "a1b2c3",
+      active: true,
+      createdAt: "2026-01-01T00:00:00.000Z",
+    }
+    expect(() => webhookSubscriptionSchema.parse(example)).not.toThrow()
+
+    const scope = nock(BASE_URL)
+      .post(c.request.path, c.request.body as nock.DataMatcherMap)
+      .reply(c.response.status, example)
+
+    const result = await client.subscribeWebhook(
+      c.request.body as Parameters<typeof client.subscribeWebhook>[0],
+    )
+
+    expect(scope.isDone()).toBe(true)
+    expect(result).toEqual(example)
+  })
+
+  it("listWebhooks() requests the list-webhooks path and returns a contract-valid page", async () => {
+    const c = contract("list-webhooks")
+    const example = {
+      data: [
+        {
+          id: "1",
+          userId: "7",
+          streamId: "42",
+          url: "https://example.com/hook",
+          events: ["stream:started"],
+          active: true,
+          createdAt: "2026-01-01T00:00:00.000Z",
+        },
+      ],
+      total: 1,
+      page: 1,
+      limit: 20,
+    }
+    expect(() => paginatedWebhookSubscriptionsSchema.parse(example)).not.toThrow()
+
+    const scope = nock(BASE_URL)
+      .get("/webhooks?page=1&limit=20")
+      .reply(c.response.status, example)
+
+    const result = await client.listWebhooks({ page: 1, limit: 20 })
+
+    expect(scope.isDone()).toBe(true)
+    expect(result.data).toHaveLength(1)
+    expect(result.data[0]).not.toHaveProperty("secret")
+  })
+
+  it("updateWebhook() PATCHes the update-webhook path and returns a summary", async () => {
+    const c = contract("update-webhook")
+    const example = {
+      id: "1",
+      userId: "7",
+      streamId: "42",
+      url: "https://example.com/hook",
+      events: ["stream:started"],
+      active: false,
+      createdAt: "2026-01-01T00:00:00.000Z",
+    }
+    expect(() => webhookSubscriptionSummarySchema.parse(example)).not.toThrow()
+
+    const scope = nock(BASE_URL)
+      .patch("/webhooks/1", { active: false } as nock.DataMatcherMap)
+      .reply(c.response.status, example)
+
+    const result = await client.updateWebhook("1", { active: false })
+
+    expect(scope.isDone()).toBe(true)
+    expect(result.active).toBe(false)
+    expect(result).not.toHaveProperty("secret")
+  })
+
+  it("deleteWebhook() DELETEs the delete-webhook path and resolves on 204", async () => {
+    const c = contract("delete-webhook")
+    const scope = nock(BASE_URL).delete("/webhooks/1").reply(204)
+
+    await expect(client.deleteWebhook("1")).resolves.toBeUndefined()
+
+    expect(scope.isDone()).toBe(true)
+    expect(c.response.status).toBe(204)
+  })
+
+  it("retryWebhookDelivery() POSTs the retry path and returns a contract-valid delivery", async () => {
+    const c = contract("retry-webhook-delivery")
+    const example = {
+      id: "10",
+      webhookSubscriptionId: "1",
+      event: "stream:started",
+      payload: { streamId: 42 },
+      status: "pending",
+      attemptCount: 6,
+      lastStatusCode: null,
+      lastResponseBody: null,
+      lastError: "connection refused",
+      nextAttemptAt: "2026-01-01T00:05:00.000Z",
+      deliveredAt: null,
+      createdAt: "2026-01-01T00:00:00.000Z",
+    }
+    expect(() => webhookDeliverySchema.parse(example)).not.toThrow()
+
+    const scope = nock(BASE_URL)
+      .post("/webhooks/1/deliveries/10/retry")
+      .reply(c.response.status, example)
+
+    const result = await client.retryWebhookDelivery("1", "10")
+
+    expect(scope.isDone()).toBe(true)
+    expect(result.status).toBe("pending")
+  })
 })
diff --git a/xstreamroll-sdk/__tests__/http.test.ts b/xstreamroll-sdk/__tests__/http.test.ts
index f061b7c..a2b9b45 100644
--- a/xstreamroll-sdk/__tests__/http.test.ts
+++ b/xstreamroll-sdk/__tests__/http.test.ts
@@ -138,4 +138,40 @@ describe("HttpClient interceptors", () => {
       (init.headers as Record<string, string>)["Content-Type"],
     ).toBeUndefined()
   })
+
+  it("patch() issues a PATCH with a JSON-serialised body", async () => {
+    const mockFetch = makeFetchMock()
+    global.fetch = mockFetch
+    const http = new HttpClient("http://localhost:3001")
+
+    await http.patch("/webhooks/1", { active: false })
+    const init = mockFetch.mock.calls[0][1] as RequestInit
+    expect(init.method).toBe("PATCH")
+    expect(init.body).toBe(JSON.stringify({ active: false }))
+    expect((init.headers as Record<string, string>)["Content-Type"]).toBe(
+      "application/json",
+    )
+  })
+
+  it("delete() issues a DELETE without a body by default", async () => {
+    const mockFetch = makeFetchMock()
+    global.fetch = mockFetch
+    const http = new HttpClient("http://localhost:3001")
+
+    await http.delete("/webhooks/1")
+    const init = mockFetch.mock.calls[0][1] as RequestInit
+    expect(init.method).toBe("DELETE")
+    expect(init.body).toBeUndefined()
+  })
+
+  it("delete() with a body JSON-serialises it", async () => {
+    const mockFetch = makeFetchMock()
+    global.fetch = mockFetch
+    const http = new HttpClient("http://localhost:3001")
+
+    await http.delete("/webhooks/1", { reason: "cleanup" })
+    const init = mockFetch.mock.calls[0][1] as RequestInit
+    expect(init.method).toBe("DELETE")
+    expect(init.body).toBe(JSON.stringify({ reason: "cleanup" }))
+  })
 })
diff --git a/xstreamroll-sdk/__tests__/webhooks.test.ts b/xstreamroll-sdk/__tests__/webhooks.test.ts
index 2fbafb9..35953de 100644
--- a/xstreamroll-sdk/__tests__/webhooks.test.ts
+++ b/xstreamroll-sdk/__tests__/webhooks.test.ts
@@ -1,4 +1,5 @@
 import * as crypto from "crypto"
+
 import { computeWebhookSignature, verifyWebhookSignature } from "../src/webhooks"
 
 /** Reference implementation using Node's crypto module, for cross-checking. */
diff --git a/xstreamroll-sdk/src/client.ts b/xstreamroll-sdk/src/client.ts
index 4a838bf..3bcb885 100644
--- a/xstreamroll-sdk/src/client.ts
+++ b/xstreamroll-sdk/src/client.ts
@@ -7,11 +7,14 @@ import {
   type CreateUserDto,
   type CreateWebhookDto,
   type PagedTags,
+  type PaginatedResponse,
   type Stream,
   type StreamConfig,
   type StreamEvent,
-  type PaginatedResponse,
+  type UpdateWebhookDto,
+  type WebhookDelivery,
   type WebhookSubscription,
+  type WebhookSubscriptionSummary,
 } from "./types"
 
 /** Named environment presets for base URL resolution. */
@@ -159,6 +162,65 @@ export class StreamingClient {
     })
   }
 
+  /**
+   * Lists the caller's webhook subscriptions, newest first. Optionally
+   * narrows to a single stream via `streamId`. The signing `secret` is
+   * not included in list responses — it is creation-time-only.
+   */
+  async listWebhooks(params: {
+    streamId?: string | number
+    page?: number
+    limit?: number
+  } = {}): Promise<PaginatedResponse<WebhookSubscriptionSummary>> {
+    const qs = new URLSearchParams()
+    if (params.streamId !== undefined) qs.set("streamId", String(params.streamId))
+    if (params.page !== undefined) qs.set("page", String(params.page))
+    if (params.limit !== undefined) qs.set("limit", String(params.limit))
+    const query = qs.toString()
+    return this.requestJson<PaginatedResponse<WebhookSubscriptionSummary>>(
+      `/webhooks${query ? `?${query}` : ""}`,
+      { method: "GET" },
+    )
+  }
+
+  /**
+   * Partially updates a webhook subscription: URL, event list, and/or
+   * `active`. Deactivating (`active: false`) stops new deliveries and
+   * retries immediately; reactivating resumes them. The signing secret
+   * cannot be changed — it is creation-time-only.
+   */
+  async updateWebhook(
+    id: string | number,
+    changes: UpdateWebhookDto,
+  ): Promise<WebhookSubscriptionSummary> {
+    return this.requestJson<WebhookSubscriptionSummary>(`/webhooks/${id}`, {
+      method: "PATCH",
+      body: changes,
+    })
+  }
+
+  /**
+   * Deletes a webhook subscription and its entire delivery history.
+   */
+  async deleteWebhook(id: string | number): Promise<void> {
+    await this.requestJson<void>(`/webhooks/${id}`, { method: "DELETE" })
+  }
+
+  /**
+   * Manually re-queues a failed or pending delivery so the retry sweep
+   * picks it up immediately. The retry budget still applies — the
+   * attempt count is kept, not reset.
+   */
+  async retryWebhookDelivery(
+    webhookId: string | number,
+    deliveryId: string | number,
+  ): Promise<WebhookDelivery> {
+    return this.requestJson<WebhookDelivery>(
+      `/webhooks/${webhookId}/deliveries/${deliveryId}/retry`,
+      { method: "POST" },
+    )
+  }
+
   // ── Pagination (#390) ─────────────────────────────────────────────────────
 
   /**
@@ -207,16 +269,37 @@ export class StreamingClient {
    * Maps non-2xx responses (and exhausted HttpClient retries) to ApiError,
    * and optionally retries once after a token refresh on 401.
    */
+  /** Routes a request to the matching HttpClient convenience method. */
+  private async dispatch(
+    path: string,
+    method: string,
+    body: unknown,
+    headers?: Record<string, string>,
+  ): Promise<Response> {
+    switch (method) {
+      case "POST":
+        return this.http.post(path, body, { headers })
+      case "PATCH":
+        return this.http.patch(path, body, { headers })
+      case "DELETE":
+        return this.http.delete(path, body, { headers })
+      default:
+        return this.http.get(path, { headers })
+    }
+  }
+
   private async requestJson<T>(
     path: string,
     init: { method?: string; body?: unknown; headers?: Record<string, string> } = {},
     options: { skipAuthRefresh?: boolean; retried?: boolean } = {},
   ): Promise<T> {
     try {
-      const response =
-        init.method === "POST" || init.body !== undefined
-          ? await this.http.post(path, init.body, { headers: init.headers })
-          : await this.http.get(path, { headers: init.headers })
+      const response = await this.dispatch(
+        path,
+        init.method ?? (init.body !== undefined ? "POST" : "GET"),
+        init.body,
+        init.headers,
+      )
 
       if (
         response.status === 401 &&
diff --git a/xstreamroll-sdk/src/http.ts b/xstreamroll-sdk/src/http.ts
index 7c2acbd..80b9cca 100644
--- a/xstreamroll-sdk/src/http.ts
+++ b/xstreamroll-sdk/src/http.ts
@@ -145,6 +145,36 @@ export class HttpClient {
     path: string,
     body?: unknown,
     init: RequestInit = {},
+  ): Promise<Response> {
+    return this.sendWithBody("POST", path, body, init)
+  }
+
+  /**
+   * PATCH convenience wrapper with the same JSON-serialisation
+   * behaviour as {@link post}.
+   */
+  patch(
+    path: string,
+    body?: unknown,
+    init: RequestInit = {},
+  ): Promise<Response> {
+    return this.sendWithBody("PATCH", path, body, init)
+  }
+
+  /**
+   * DELETE convenience wrapper. Accepts an optional body for the rare
+   * DELETE-with-payload endpoints; usually called without one.
+   */
+  delete(path: string, body?: unknown, init: RequestInit = {}): Promise<Response> {
+    return this.sendWithBody("DELETE", path, body, init)
+  }
+
+  /** Shared body-serialising implementation for POST/PATCH/DELETE. */
+  private sendWithBody(
+    method: "POST" | "PATCH" | "DELETE",
+    path: string,
+    body: unknown,
+    init: RequestInit,
   ): Promise<Response> {
     const headers: Record<string, string> = {
       ...(init.headers as Record<string, string> | undefined),
@@ -156,7 +186,7 @@ export class HttpClient {
     }
     return this.request(path, {
       ...init,
-      method: "POST",
+      method,
       headers,
       body: requestBody,
     })
diff --git a/xstreamroll-sdk/src/index.ts b/xstreamroll-sdk/src/index.ts
index 0d48726..c6013b3 100644
--- a/xstreamroll-sdk/src/index.ts
+++ b/xstreamroll-sdk/src/index.ts
@@ -27,7 +27,9 @@ export type {
   StreamEventRecord,
   // Webhooks
   CreateWebhookDto,
+  UpdateWebhookDto,
   WebhookSubscription,
+  WebhookSubscriptionSummary,
   WebhookDelivery,
   // Pagination
   PaginatedResponse,
diff --git a/xstreamroll-sdk/src/types.ts b/xstreamroll-sdk/src/types.ts
index ae37fd2..dc75692 100644
--- a/xstreamroll-sdk/src/types.ts
+++ b/xstreamroll-sdk/src/types.ts
@@ -109,6 +109,33 @@ export interface WebhookSubscription {
   createdAt: string
 }
 
+/**
+ * A webhook subscription as returned by every endpoint except creation
+ * (`GET /webhooks`, `PATCH /webhooks/:id`). Identical to
+ * {@link WebhookSubscription} minus the `secret`, which the API only
+ * ever returns once, from `subscribeWebhook()`.
+ */
+export interface WebhookSubscriptionSummary {
+  id: string | number
+  userId: string | number
+  streamId: string | number
+  url: string
+  events: StreamEventType[]
+  active: boolean
+  createdAt: string
+}
+
+/**
+ * Partial payload accepted by `updateWebhook()` / `PATCH /webhooks/:id`.
+ * The signing secret cannot be changed through this endpoint — it is
+ * creation-time-only.
+ */
+export interface UpdateWebhookDto {
+  url?: string
+  events?: StreamEventType[]
+  active?: boolean
+}
+
 /** A single delivery attempt, as returned by `GET /webhooks/:id/deliveries`. */
 export interface WebhookDelivery {
   id: string | number