ApiCore/ApiCore-Backend/test/webhooks.e2e-spec.ts

import { INestApplication } from '@nestjs/common';
import * as http from 'node:http';
import * as crypto from 'node:crypto';
import * as bcrypt from 'bcryptjs';
import request from 'supertest';
import { App } from 'supertest/types';
import { PrismaService } from '../src/prisma/prisma.service';
import { postWebhook, WebhookResponse } from '../src/webhooks/webhook-delivery';
import { createTestApp } from './utils/create-test-app';
import { login } from './utils/api';
import { drainDomainEvents } from './utils/domain-events';
import { deliverWebhooks } from './utils/webhooks';

// The real postWebhook by default; tests that need the loopback receiver
// below swap in postToLoopback (the real one refuses loopback by design).
jest.mock('../src/webhooks/webhook-delivery', () => {
  const actual = jest.requireActual<
    typeof import('../src/webhooks/webhook-delivery')
  >('../src/webhooks/webhook-delivery');
  return { ...actual, postWebhook: jest.fn(actual.postWebhook) };
});

const postWebhookMock = postWebhook as jest.MockedFunction<typeof postWebhook>;
const realPostWebhook = jest.requireActual<
  typeof import('../src/webhooks/webhook-delivery')
>('../src/webhooks/webhook-delivery').postWebhook;

interface WebhookSubscriptionBody {
  id: string;
  url: string;
  events: string[];
  secret: string;
  active: boolean;
}

interface WebhookDeliveryBody {
  id: string;
  event: string;
  status: string;
  attempts: number;
  lastStatusCode: number | null;
  lastError: string | null;
  createdAt: string;
  deliveredAt: string | null;
}

interface ReceivedRequest {
  body: string;
  headers: http.IncomingHttpHeaders;
}

/**
 * postWebhook without the connect-time address guard, for the loopback
 * receiver of this file: same method, headers and body, real HTTP.
 */
function postToLoopback(
  rawUrl: string,
  headers: Record<string, string>,
  body: string,
): Promise<WebhookResponse> {
  return new Promise((resolve, reject) => {
    const outgoing = http.request(
      rawUrl,
      {
        method: 'POST',
        headers: {
          ...headers,
          'Content-Length': Buffer.byteLength(body).toString(),
        },
      },
      (response) => {
        response.resume();
        resolve({ status: response.statusCode ?? 0 });
      },
    );
    outgoing.on('error', reject);
    outgoing.end(body);
  });
}

function signatureOf(secret: string, received: ReceivedRequest): string {
  const timestamp = String(received.headers['x-apicore-timestamp']);
  const deliveryId = String(received.headers['x-apicore-delivery']);
  const digest = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.${deliveryId}.${received.body}`)
    .digest('hex');
  return `sha256=${digest}`;
}

describe('Webhooks (e2e)', () => {
  let app: INestApplication<App>;
  let prisma: PrismaService;
  const suffix = `${Date.now()}-${Math.round(Math.random() * 1e6)}`;
  const adminEmail = `webhooks-e2e-admin-${suffix}@example.com`;
  const staffEmail = `webhooks-e2e-staff-${suffix}@example.com`;
  const password = 'SuperSecret123!';
  // Stored in plain like a pre-encryption row; openWebhookSecret reads it.
  const loopbackSecret = crypto.randomBytes(32).toString('hex');
  let adminId: string;
  let staffId: string;
  let adminAccessToken: string;
  let staffAccessToken: string;
  let apiaryId: string;
  let subscriptionId: string;
  let blockedSubscriptionId: string;
  let colonySubscriptionId: string;
  let taskSubscriptionId: string;
  const colonyIds: string[] = [];
  const taskIds: string[] = [];

  let receiverServer: http.Server;
  let receiverUrl: string;
  let receivedRequests: ReceivedRequest[] = [];
  /** Delivery ids the receiver answers with 500 once. */
  const failOnce = new Set<string>();

  const received = (deliveryId: string): ReceivedRequest[] =>
    receivedRequests.filter(
      (entry) => entry.headers['x-apicore-delivery'] === deliveryId,
    );

  /** Creates a colony through the API; returns its id. */
  async function createColony(): Promise<string> {
    const response = await request(app.getHttpServer())
      .post('/api/v1/colonies')
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .send({
        code: `WH-${suffix}-${colonyIds.length}`,
        apiaryId,
        colonyType: 'PRODUCTION_COLONY',
      })
      .expect(201);
    const id = (response.body as { id: string }).id;
    colonyIds.push(id);
    return id;
  }

  /** Creates a task through the API; returns its id. */
  async function createTask(title: string): Promise<string> {
    const response = await request(app.getHttpServer())
      .post('/api/v1/tasks')
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .send({ title: `${title} ${suffix}` })
      .expect(201);
    const id = (response.body as { id: string }).id;
    taskIds.push(id);
    return id;
  }

  /** The delivery of `subscription` whose body is about `recordId`. */
  function deliveryAbout(subscription: string, recordId: string) {
    return prisma.webhookDelivery.findFirstOrThrow({
      where: { subscriptionId: subscription, body: { contains: recordId } },
    });
  }

  /** All deliveries of `subscription` for `event` about `recordId`. */
  function deliveriesAbout(
    subscription: string,
    event: string,
    recordId: string,
  ) {
    return prisma.webhookDelivery.findMany({
      where: {
        subscriptionId: subscription,
        event,
        body: { contains: recordId },
      },
    });
  }

  beforeAll(async () => {
    app = await createTestApp();
    prisma = app.get(PrismaService);

    const passwordHash = await bcrypt.hash(password, 12);
    const admin = await prisma.user.create({
      data: {
        email: adminEmail,
        passwordHash,
        name: 'Webhooks E2E Admin',
        role: { connect: { key: 'ADMIN' } },
      },
    });
    adminId = admin.id;
    const staff = await prisma.user.create({
      data: {
        email: staffEmail,
        passwordHash,
        name: 'Webhooks E2E Staff',
        role: { connect: { key: 'STAFF' } },
      },
    });
    staffId = staff.id;
    adminAccessToken = (await login(app, adminEmail, password)).accessToken;
    staffAccessToken = (await login(app, staffEmail, password)).accessToken;
    apiaryId = (
      await prisma.apiary.create({ data: { name: `Webhook-Stand ${suffix}` } })
    ).id;

    receivedRequests = [];
    receiverServer = http.createServer((req, res) => {
      const chunks: Buffer[] = [];
      req.on('data', (chunk: Buffer) => chunks.push(chunk));
      req.on('end', () => {
        const deliveryId = String(req.headers['x-apicore-delivery']);
        receivedRequests.push({
          body: Buffer.concat(chunks).toString('utf-8'),
          headers: req.headers,
        });
        const fail = failOnce.delete(deliveryId);
        res.writeHead(fail ? 500 : 200);
        res.end(fail ? 'receiver error body' : 'ok');
      });
    });
    await new Promise<void>((resolve) =>
      receiverServer.listen(0, '127.0.0.1', resolve),
    );
    const address = receiverServer.address();
    if (address === null || typeof address === 'string') {
      throw new Error('Failed to determine receiver server address');
    }
    receiverUrl = `http://127.0.0.1:${address.port}/hook`;
  });

  afterEach(() => {
    postWebhookMock.mockImplementation(realPostWebhook);
  });

  afterAll(async () => {
    await new Promise<void>((resolve, reject) =>
      receiverServer.close((error) => (error ? reject(error) : resolve())),
    );
    // Deliveries go with their subscription (ON DELETE CASCADE).
    await prisma.webhookSubscription.deleteMany({
      where: {
        id: {
          in: [
            subscriptionId,
            blockedSubscriptionId,
            colonySubscriptionId,
            taskSubscriptionId,
          ].filter(Boolean),
        },
      },
    });
    // Event deliveries go with their event (ON DELETE CASCADE).
    await prisma.domainEvent.deleteMany({
      where: { aggregateType: 'Task', aggregateId: { in: taskIds } },
    });
    await prisma.colony.deleteMany({ where: { id: { in: colonyIds } } });
    await prisma.apiary.deleteMany({ where: { id: apiaryId } });
    await prisma.task.deleteMany({
      where: { createdByUserId: { in: [adminId, staffId] } },
    });
    await prisma.refreshToken.deleteMany({
      where: { userId: { in: [adminId, staffId] } },
    });
    await prisma.auditLog.deleteMany({
      where: { userId: { in: [adminId, staffId] } },
    });
    await prisma.user.deleteMany({ where: { id: { in: [adminId, staffId] } } });
    await app.close();
  });

  it('rejects non-admin roles from listing webhooks', () => {
    return request(app.getHttpServer())
      .get('/api/v1/webhooks')
      .set('Authorization', `Bearer ${staffAccessToken}`)
      .expect(403);
  });

  it('rejects a webhook URL targeting a private or loopback address (SSRF guard)', async () => {
    await request(app.getHttpServer())
      .post('/api/v1/webhooks')
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .send({ url: receiverUrl, events: ['task.created'] })
      .expect(400);
    await request(app.getHttpServer())
      .post('/api/v1/webhooks')
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .send({
        url: 'http://169.254.169.254/latest/meta-data/',
        events: ['task.created'],
      })
      .expect(400);
  });

  it('lets an admin create a webhook subscription with a generated secret', async () => {
    const response = await request(app.getHttpServer())
      .post('/api/v1/webhooks')
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .send({ url: 'https://example.com/hook', events: ['task.created'] })
      .expect(201);
    const body = response.body as WebhookSubscriptionBody;
    subscriptionId = body.id;
    expect(body.secret).toHaveLength(64);
    expect(body.events).toEqual(['task.created']);
  });

  it('accepts the new event types', async () => {
    await request(app.getHttpServer())
      .patch(`/api/v1/webhooks/${subscriptionId}`)
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .send({
        events: [
          'task.created',
          'varroa.measured',
          'treatment.created',
          'feeding.created',
          'stock.changed',
        ],
      })
      .expect(200);
  });

  it('excludes the secret from the list response', async () => {
    const response = await request(app.getHttpServer())
      .get('/api/v1/webhooks')
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .expect(200);
    const body = response.body as Partial<WebhookSubscriptionBody>[];
    const listed = body.find((entry) => entry.id === subscriptionId);
    expect(listed?.secret).toBeUndefined();
  });

  it('never delivers to a private address, even from a row that bypassed the create guard', async () => {
    // Stands in for DNS rebinding or a tampered row: the stored URL points
    // at a loopback receiver. The connection-time check must refuse it.
    const subscription = await prisma.webhookSubscription.create({
      data: {
        url: receiverUrl,
        events: ['task.created'],
        secret: loopbackSecret,
        createdByUserId: adminId,
      },
    });
    blockedSubscriptionId = subscription.id;

    const taskId = await createTask('Webhook-Test');
    // The request recorded TaskCreated; its webhook handler queues the
    // delivery. Nothing is sent yet.
    await drainDomainEvents(app, [taskId]);
    const queued = await deliveryAbout(blockedSubscriptionId, taskId);
    expect(queued).toMatchObject({ status: 'PENDING', attempts: 0 });

    await deliverWebhooks(app, [blockedSubscriptionId]);

    expect(receivedRequests).toHaveLength(0);
    const after = await deliveryAbout(blockedSubscriptionId, taskId);
    expect(after).toMatchObject({
      status: 'PENDING',
      attempts: 1,
      lastStatusCode: null,
    });
    expect(after.lastError).toContain('blocked address');
  });

  it('delivers task.created and task.completed once each, through the event and the queue', async () => {
    postWebhookMock.mockImplementation(postToLoopback);
    taskSubscriptionId = (
      await prisma.webhookSubscription.create({
        data: {
          url: receiverUrl,
          events: ['task.created', 'task.completed'],
          secret: loopbackSecret,
          createdByUserId: adminId,
        },
      })
    ).id;

    const taskId = await createTask('Webhook-Aufgabe');
    // Nothing is queued before the event is dispatched.
    await expect(
      deliveriesAbout(taskSubscriptionId, 'task.created', taskId),
    ).resolves.toHaveLength(0);
    await drainDomainEvents(app, [taskId]);
    await deliverWebhooks(app, [taskSubscriptionId]);

    const [created, ...moreCreated] = await deliveriesAbout(
      taskSubscriptionId,
      'task.created',
      taskId,
    );
    expect(moreCreated).toHaveLength(0);
    expect(created).toMatchObject({ status: 'DELIVERED', lastStatusCode: 200 });
    const [sentCreated, ...resentCreated] = received(created.id);
    expect(resentCreated).toHaveLength(0);
    expect(sentCreated.headers['x-apicore-event']).toBe('task.created');
    expect(JSON.parse(sentCreated.body)).toMatchObject({
      event: 'task.created',
      data: { id: taskId, status: 'OPEN' },
    });

    // Completed twice: the second request announces nothing.
    for (let i = 0; i < 2; i++) {
      await request(app.getHttpServer())
        .post(`/api/v1/tasks/${taskId}/complete`)
        .set('Authorization', `Bearer ${adminAccessToken}`)
        .expect(201);
    }
    await drainDomainEvents(app, [taskId]);
    await deliverWebhooks(app, [taskSubscriptionId]);

    const [completed, ...moreCompleted] = await deliveriesAbout(
      taskSubscriptionId,
      'task.completed',
      taskId,
    );
    expect(moreCompleted).toHaveLength(0);
    expect(completed).toMatchObject({
      status: 'DELIVERED',
      lastStatusCode: 200,
    });
    const [sentCompleted, ...resentCompleted] = received(completed.id);
    expect(resentCompleted).toHaveLength(0);
    expect(sentCompleted.headers['x-apicore-signature']).toBe(
      signatureOf(loopbackSecret, sentCompleted),
    );
    expect(JSON.parse(sentCompleted.body)).toMatchObject({
      event: 'task.completed',
      data: { id: taskId, status: 'DONE' },
    });
    // Still the one task.created: completing sends no second one.
    await expect(
      deliveriesAbout(taskSubscriptionId, 'task.created', taskId),
    ).resolves.toHaveLength(1);
  });

  it('queues task.completed once for a PATCH into DONE, also when it is repeated', async () => {
    const taskId = await createTask('Per Bearbeiten erledigt');

    for (let i = 0; i < 2; i++) {
      await request(app.getHttpServer())
        .patch(`/api/v1/tasks/${taskId}`)
        .set('Authorization', `Bearer ${adminAccessToken}`)
        .send({ status: 'DONE' })
        .expect(200);
    }
    await expect(
      prisma.domainEvent.count({
        where: { aggregateId: taskId, type: 'TaskCompleted' },
      }),
    ).resolves.toBe(1);
    await expect(
      prisma.domainEvent.count({
        where: { aggregateId: taskId, type: 'TaskUpdated' },
      }),
    ).resolves.toBe(2);

    await drainDomainEvents(app, [taskId]);

    const completed = await deliveriesAbout(
      taskSubscriptionId,
      'task.completed',
      taskId,
    );
    expect(completed).toHaveLength(1);
    expect(JSON.parse(completed[0].body)).toMatchObject({
      event: 'task.completed',
      data: { id: taskId, status: 'DONE' },
    });
  });

  it('delivers colony.created through the queue with a stable delivery id', async () => {
    postWebhookMock.mockImplementation(postToLoopback);
    colonySubscriptionId = (
      await prisma.webhookSubscription.create({
        data: {
          url: receiverUrl,
          events: ['colony.created'],
          secret: loopbackSecret,
          createdByUserId: adminId,
        },
      })
    ).id;

    const colonyId = await createColony();
    await deliverWebhooks(app, [colonySubscriptionId]);

    const delivery = await deliveryAbout(colonySubscriptionId, colonyId);
    const [sent, ...more] = received(delivery.id);
    expect(more).toHaveLength(0);
    expect(sent.headers['x-apicore-delivery']).toBe(delivery.id);
    expect(sent.headers['x-apicore-event']).toBe('colony.created');
    expect(sent.headers['x-apicore-signature']).toBe(
      signatureOf(loopbackSecret, sent),
    );
    // The body stored at enqueue time, byte for byte.
    expect(sent.body).toBe(delivery.body);
    expect(JSON.parse(sent.body)).toMatchObject({
      event: 'colony.created',
      data: { id: colonyId, apiaryId },
    });
    expect(delivery).toMatchObject({
      status: 'DELIVERED',
      attempts: 0,
      lastStatusCode: 200,
      claimedAt: null,
    });
    expect(delivery.deliveredAt).toBeInstanceOf(Date);
  });

  it('retries a 500 with the same delivery id and then marks it DELIVERED', async () => {
    postWebhookMock.mockImplementation(postToLoopback);
    const colonyId = await createColony();
    const queued = await deliveryAbout(colonySubscriptionId, colonyId);
    failOnce.add(queued.id);

    await deliverWebhooks(app, [colonySubscriptionId]);

    const failed = await deliveryAbout(colonySubscriptionId, colonyId);
    expect(failed).toMatchObject({
      status: 'PENDING',
      attempts: 1,
      lastStatusCode: 500,
      lastError: 'Receiver answered HTTP 500',
    });
    // Backoff by the database clock: the first retry waits 30 s.
    const [{ waitMs }] = await prisma.$queryRaw<{ waitMs: number }[]>`
      SELECT (EXTRACT(EPOCH FROM ("availableAt" - now())) * 1000)::float8 AS "waitMs"
      FROM "WebhookDelivery" WHERE "id" = ${queued.id}`;
    expect(waitMs).toBeGreaterThan(25_000);
    expect(waitMs).toBeLessThanOrEqual(30_000);
    // Not due yet: another run leaves it alone.
    await deliverWebhooks(app, [colonySubscriptionId]);
    expect(received(queued.id)).toHaveLength(1);

    await prisma.$executeRaw`UPDATE "WebhookDelivery" SET "availableAt" = now() WHERE "id" = ${queued.id}`;
    await deliverWebhooks(app, [colonySubscriptionId]);

    const attempts = received(queued.id);
    expect(attempts).toHaveLength(2);
    expect(attempts[1].headers['x-apicore-delivery']).toBe(queued.id);
    expect(attempts[1].body).toBe(attempts[0].body);
    attempts.forEach((attempt) =>
      expect(attempt.headers['x-apicore-signature']).toBe(
        signatureOf(loopbackSecret, attempt),
      ),
    );
    const delivered = await deliveryAbout(colonySubscriptionId, colonyId);
    expect(delivered).toMatchObject({
      status: 'DELIVERED',
      attempts: 1,
      lastStatusCode: 200,
    });
  });

  it('lists deliveries for admins and refuses staff', async () => {
    // Every e2e file creates colonies: while the subscription is active,
    // their colony.created deliveries pile up on it and could push this
    // file's rows off the page. Deactivated, nothing new is queued.
    await prisma.webhookSubscription.update({
      where: { id: colonySubscriptionId },
      data: { active: false },
    });
    const response = await request(app.getHttpServer())
      .get(`/api/v1/webhooks/${colonySubscriptionId}/deliveries?limit=100`)
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .expect(200);
    const deliveries = response.body as WebhookDeliveryBody[];
    const ours = await prisma.webhookDelivery.findMany({
      where: {
        subscriptionId: colonySubscriptionId,
        OR: colonyIds.map((id) => ({ body: { contains: id } })),
      },
    });
    expect(ours.length).toBeGreaterThanOrEqual(2);
    for (const delivery of ours) {
      expect(deliveries.map((entry) => entry.id)).toContain(delivery.id);
    }
    for (const entry of deliveries) {
      expect(Object.keys(entry).sort()).toEqual(
        [
          'attempts',
          'createdAt',
          'deliveredAt',
          'event',
          'id',
          'lastError',
          'lastStatusCode',
          'status',
        ].sort(),
      );
    }
    // Newest first.
    const createdAt = deliveries.map((entry) => Date.parse(entry.createdAt));
    expect(createdAt).toEqual([...createdAt].sort((a, b) => b - a));

    await request(app.getHttpServer())
      .get(`/api/v1/webhooks/${colonySubscriptionId}/deliveries`)
      .set('Authorization', `Bearer ${staffAccessToken}`)
      .expect(403);
  });

  it.each(['0', '101', '1.5', 'abc'])(
    'rejects limit=%s for the delivery list',
    (limit) =>
      request(app.getHttpServer())
        .get(`/api/v1/webhooks/${colonySubscriptionId}/deliveries`)
        .query({ limit })
        .set('Authorization', `Bearer ${adminAccessToken}`)
        .expect(400),
  );

  it('answers 404 for the deliveries of an unknown subscription', () =>
    request(app.getHttpServer())
      .get('/api/v1/webhooks/does-not-exist/deliveries')
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .expect(404));

  it.each(['url', 'events', 'active'])(
    'rejects null for the required %s',
    (field) =>
      request(app.getHttpServer())
        .patch(`/api/v1/webhooks/${subscriptionId}`)
        .set('Authorization', `Bearer ${adminAccessToken}`)
        .send({ [field]: null })
        .expect(400),
  );

  it('lets an admin delete a webhook subscription', () => {
    return request(app.getHttpServer())
      .delete(`/api/v1/webhooks/${subscriptionId}`)
      .set('Authorization', `Bearer ${adminAccessToken}`)
      .expect(204);
  });
});