ApiCore/ApiCore-Backend/test/calendar-sync.e2e-spec.ts

import { INestApplication, Logger } from '@nestjs/common';
import * as bcrypt from 'bcryptjs';
import request from 'supertest';
import { App } from 'supertest/types';
import { PrismaService } from '../src/prisma/prisma.service';
import { CalendarSyncService } from '../src/calendar-sync/calendar-sync.service';
import { CalendarSyncSettingsService } from '../src/calendar-sync/calendar-sync-settings.service';
import { GraphClientService } from '../src/calendar-sync/graph-client.service';
import { withTimeout } from '../src/common/with-timeout';
import { WAKE_CHANNELS } from '../src/common/worker-wakeup';
import { PgListener } from '../src/worker/pg-listener';
import { createTestApp } from './utils/create-test-app';
import { login } from './utils/api';
import { drainDomainEvents } from './utils/domain-events';

/**
 * The M365 outbox against the real database, with Microsoft Graph replaced
 * by a recorder: several workers (worker pods, each ticking and woken by
 * "Jetzt synchronisieren") must never write the same entry twice, and an
 * edit that arrives while an entry is being synced must not get lost. The
 * API itself only queues a pass; it has no worker loop.
 */
describe('Calendar sync outbox (e2e)', () => {
  let app: INestApplication<App>;
  let prisma: PrismaService;
  let calendarSync: CalendarSyncService;
  let graph: GraphClientService;
  const suffix = `${Date.now()}-${Math.round(Math.random() * 1e6)}`;
  const taskIds: string[] = [];
  let createdEvents: string[];
  let updateEvent: jest.SpyInstance;
  let eventCounter = 0;
  const adminEmail = `calendar-sync-e2e-${suffix}@example.com`;
  const password = 'SuperSecret123!';
  let token: string;

  const settings = {
    id: 'singleton',
    enabled: true,
    groupId: 'group-e2e',
    syncTask: true,
    syncInspection: true,
    syncTreatment: true,
    syncFeeding: true,
    syncHarvest: true,
    syncQueenAssignment: true,
    updatedAt: new Date(),
  };

  const wait = (ms: number) => new Promise((done) => setTimeout(done, ms));

  /** Polls until the condition holds - for work that runs in the background. */
  async function eventually(condition: () => Promise<boolean> | boolean) {
    for (let i = 0; i < 100; i++) {
      if (await condition()) return;
      await wait(50);
    }
    throw new Error('condition not met in time');
  }

  async function queueTask(title: string) {
    const task = await prisma.task.create({
      data: { title: `${title} ${suffix}`, dueDate: new Date('2027-05-01') },
    });
    taskIds.push(task.id);
    await prisma.calendarSyncOutboxEntry.create({
      data: { sourceType: 'TASK', sourceId: task.id },
    });
    return task;
  }

  const entryFor = (taskId: string) =>
    prisma.calendarSyncOutboxEntry.findUniqueOrThrow({
      where: { sourceType_sourceId: { sourceType: 'TASK', sourceId: taskId } },
    });

  /** Graph calls made for the tasks of this spec (not other leftovers). */
  const createsFor = (title: string) =>
    createdEvents.filter((created) => created === `${title} ${suffix}`);

  beforeAll(async () => {
    app = await createTestApp();
    prisma = app.get(PrismaService);
    calendarSync = app.get(CalendarSyncService);
    graph = app.get(GraphClientService);
    await prisma.user.create({
      data: {
        email: adminEmail,
        passwordHash: await bcrypt.hash(password, 12),
        name: 'Calendar Sync Admin',
        role: { connect: { key: 'ADMIN' } },
      },
    });
    token = (await login(app, adminEmail, password)).accessToken;
  });

  beforeEach(() => {
    createdEvents = [];
    jest
      .spyOn(app.get(CalendarSyncSettingsService), 'get')
      .mockResolvedValue(settings);
    jest.spyOn(graph, 'isConfigured').mockReturnValue(true);
    // A real Graph round trip takes a moment - long enough for a second
    // worker to read the same row in the meantime.
    jest.spyOn(graph, 'createEvent').mockImplementation(async (_g, input) => {
      createdEvents.push(input.title);
      await wait(50);
      eventCounter += 1;
      return `event-${suffix}-${eventCounter}`;
    });
    updateEvent = jest.spyOn(graph, 'updateEvent').mockResolvedValue(undefined);
    jest.spyOn(graph, 'deleteEvent').mockResolvedValue(undefined);
  });

  afterEach(() => {
    jest.restoreAllMocks();
  });

  afterAll(async () => {
    await prisma.calendarSyncOutboxEntry.deleteMany({
      where: { sourceType: 'TASK', sourceId: { in: taskIds } },
    });
    await prisma.domainEvent.deleteMany({
      where: { aggregateType: 'Task', aggregateId: { in: taskIds } },
    });
    await prisma.task.deleteMany({ where: { id: { in: taskIds } } });
    const admin = await prisma.user.findUnique({
      where: { email: adminEmail },
    });
    if (admin) {
      await prisma.refreshToken.deleteMany({ where: { userId: admin.id } });
      await prisma.auditLog.deleteMany({ where: { userId: admin.id } });
      await prisma.user.delete({ where: { id: admin.id } });
    }
    await app.close();
  });

  it('creates each event once when several workers run at the same time', async () => {
    const titles = ['Parallel A', 'Parallel B', 'Parallel C'];
    const tasks: { id: string }[] = [];
    for (const title of titles) {
      tasks.push(await queueTask(title));
    }

    // Three worker pods, ticking or woken by "Jetzt synchronisieren".
    await Promise.all([
      calendarSync.processOutbox(),
      calendarSync.processOutbox(),
      calendarSync.processOutbox(),
    ]);

    for (const title of titles) {
      expect(createsFor(title)).toHaveLength(1);
    }
    for (const task of tasks) {
      const entry = await entryFor(task.id);
      expect(entry.status).toBe('SYNCED');
      expect(entry.graphEventId).toMatch(/^event-/);
      expect(entry.claimedAt).toBeNull();
    }
  });

  it('syncs an edit that arrives while the entry is being written to M365', async () => {
    const task = await queueTask('Edited during sync');
    jest
      .spyOn(graph, 'createEvent')
      .mockImplementationOnce(async (_g, input) => {
        createdEvents.push(input.title);
        // The task's due date is changed right now, on another request.
        await calendarSync.enqueue('TASK', task.id);
        return `event-${suffix}-edited`;
      });

    await calendarSync.processOutbox();

    const afterFirstPass = await entryFor(task.id);
    // Not SYNCED: the newer state still has to go out - onto the same event.
    expect(afterFirstPass.status).toBe('PENDING');
    expect(afterFirstPass.graphEventId).toBe(`event-${suffix}-edited`);
    expect(afterFirstPass.claimedAt).toBeNull();

    await calendarSync.processOutbox();

    expect(createsFor('Edited during sync')).toHaveLength(1);
    expect(updateEvent).toHaveBeenCalledWith(
      'group-e2e',
      `event-${suffix}-edited`,
      expect.objectContaining({ title: `Edited during sync ${suffix}` }),
    );
    expect((await entryFor(task.id)).status).toBe('SYNCED');
  });

  it('takes over an entry whose worker died, but not one still being worked on', async () => {
    const abandoned = await queueTask('Abandoned claim');
    const busy = await queueTask('Busy claim');
    await prisma.calendarSyncOutboxEntry.update({
      where: {
        sourceType_sourceId: { sourceType: 'TASK', sourceId: abandoned.id },
      },
      data: { claimedAt: new Date(Date.now() - 60 * 60 * 1000) },
    });
    await prisma.calendarSyncOutboxEntry.update({
      where: { sourceType_sourceId: { sourceType: 'TASK', sourceId: busy.id } },
      data: { claimedAt: new Date() },
    });

    await calendarSync.processOutbox();

    expect(createsFor('Abandoned claim')).toHaveLength(1);
    expect((await entryFor(abandoned.id)).status).toBe('SYNCED');
    expect(createsFor('Busy claim')).toHaveLength(0);
    expect((await entryFor(busy.id)).status).toBe('PENDING');

    // The busy worker finishes - its live claim would otherwise keep
    // getStatus().running true for the other tests.
    await prisma.calendarSyncOutboxEntry.update({
      where: { sourceType_sourceId: { sourceType: 'TASK', sourceId: busy.id } },
      data: { claimedAt: null },
    });
  });

  it('retries a FAILED entry after "Jetzt synchronisieren", not on a regular tick', async () => {
    const task = await queueTask('Failed before');
    await prisma.calendarSyncOutboxEntry.update({
      where: { sourceType_sourceId: { sourceType: 'TASK', sourceId: task.id } },
      data: { status: 'FAILED', attempts: 5, lastError: 'Graph down' },
    });

    await calendarSync.processOutbox();
    expect(createsFor('Failed before')).toHaveLength(0);
    expect((await entryFor(task.id)).status).toBe('FAILED');

    // The click, then the tick of the worker it woke.
    await calendarSync.requestRun();
    await calendarSync.processOutbox();

    expect(createsFor('Failed before')).toHaveLength(1);
    const entry = await entryFor(task.id);
    expect(entry.status).toBe('SYNCED');
    expect(entry.attempts).toBe(0);
    expect(entry.lastError).toBeNull();
  });

  it('removes the Graph event of a task that was completed, and settles the entry SKIPPED', async () => {
    const task = await queueTask('Completed with event');
    await calendarSync.processOutbox();
    const synced = await entryFor(task.id);
    expect(synced.status).toBe('SYNCED');
    const graphEventId = synced.graphEventId;
    expect(graphEventId).toMatch(/^event-/);
    const deleteEvent = jest.spyOn(graph, 'deleteEvent');

    // The real path: the request records TaskCompleted, the dispatcher runs
    // the calendar-sync.task handler, which queues the entry again.
    await request(app.getHttpServer())
      .post(`/api/v1/tasks/${task.id}/complete`)
      .set('Authorization', `Bearer ${token}`)
      .expect(201);
    expect((await entryFor(task.id)).status).toBe('SYNCED');
    await drainDomainEvents(app, [task.id]);
    expect((await entryFor(task.id)).status).toBe('PENDING');

    await calendarSync.processOutbox();

    expect(deleteEvent).toHaveBeenCalledWith('group-e2e', graphEventId);
    expect(await entryFor(task.id)).toMatchObject({
      status: 'SKIPPED',
      graphEventId: null,
      claimedAt: null,
    });
  });

  describe('API', () => {
    const auth = () => ({ Authorization: `Bearer ${token}` });

    it('answers "Jetzt synchronisieren" with 202, resets FAILED and wakes the worker - no Graph call from the API', async () => {
      const task = await queueTask('Requested via API');
      await prisma.calendarSyncOutboxEntry.update({
        where: {
          sourceType_sourceId: { sourceType: 'TASK', sourceId: task.id },
        },
        data: { status: 'FAILED', attempts: 5, lastError: 'Graph down' },
      });
      let wake: () => void = () => undefined;
      const woken = new Promise<void>((resolve) => {
        wake = resolve;
      });
      const worker = new PgListener({
        connectionString: process.env.DATABASE_URL ?? '',
        channels: [WAKE_CHANNELS.calendarSync],
        onNotify: () => wake(),
        logger: new Logger('CalendarSyncE2E'),
      });
      await worker.start();

      try {
        const response = await request(app.getHttpServer())
          .post('/api/v1/calendar/sync/run-now')
          .set(auth())
          .expect(202);

        // A pass is due: the client follows it through GET status.
        expect((response.body as { running: boolean }).running).toBe(true);
        await withTimeout(woken, 5_000, 'no calendar_sync wake-up');
      } finally {
        await worker.stop();
      }
      expect(await entryFor(task.id)).toMatchObject({
        status: 'PENDING',
        attempts: 0,
        lastError: null,
        claimedAt: null,
      });
      expect(createdEvents).toEqual([]);
      expect(updateEvent).not.toHaveBeenCalled();
    }, 15_000);

    it('reports running in GET status while a worker holds an entry', async () => {
      const task = await queueTask('Held by a worker');
      const title = `Held by a worker ${suffix}`;
      let releaseCreate: () => void = () => undefined;
      jest.spyOn(graph, 'createEvent').mockImplementation(async (_g, input) => {
        createdEvents.push(input.title);
        if (input.title === title) {
          await new Promise<void>((resolve) => {
            releaseCreate = resolve;
          });
        }
        eventCounter += 1;
        return `event-${suffix}-${eventCounter}`;
      });
      const status = async () =>
        (
          await request(app.getHttpServer())
            .get('/api/v1/calendar/sync/status')
            .set(auth())
            .expect(200)
        ).body as { running: boolean };

      // A worker's tick, held inside the Graph call for this entry.
      const pass = calendarSync.processOutbox();
      await eventually(() => createsFor('Held by a worker').length === 1);
      expect((await status()).running).toBe(true);

      releaseCreate();
      await pass;

      expect((await status()).running).toBe(false);
      expect((await entryFor(task.id)).status).toBe('SYNCED');
    }, 15_000);

    it('rejects a group id that is not an Entra object id with 400', async () => {
      const response = await request(app.getHttpServer())
        .patch('/api/v1/calendar/sync/settings')
        .set(auth())
        .send({ groupId: '../users/[email protected]' })
        .expect(400);

      expect(JSON.stringify(response.body)).toContain('Gruppen-ID');
    });

    // PATCH null rule: the switches cannot hold null - 400, not a 500.
    it('rejects null on a switch with 400', async () => {
      await request(app.getHttpServer())
        .patch('/api/v1/calendar/sync/settings')
        .set(auth())
        .send({ enabled: null })
        .expect(400);
    });
  });
});