feat(api): add webhook subscription lifecycle management (closes #531)

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

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),
+    }
   }
 
   /**

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
+}

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
+}

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")
+    }
+  }
 }

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
+  }
 }