feat: T4 wave 3 - notifications per user, suggestions API, varroa bands, split/queen events

- Schema III: per-user notifications (broadcast fan-out migration, category, link, requiredPermissions filter), notification preferences and settings endpoints, calendar withdrawal switch, occupiedFrames - Suggestions API: list/summary with permission visibility and paging, accept/dismiss/reopen incl. bulk, SuggestionView on all responses - Varroa traffic light by season and method, young colonies, treatment effect window, sample-based percentage; shared withdrawal-window helper on Berlin calendar days; GET /varroa-measurements/thresholds - ColonySplit and QueenAssigned domain events with calendar and webhook handlers - Outbox claims compare against now() rounded to timestamp(3); fixes the flaky e2e claims of rows written with SQL now() Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
2026-10-14 14:25:24 +02:00
1 parent d99a697739
commit f468dc5f75
@@ -7,12 +7,17 @@ import { QueensService } from './queens.service';
 import { PrismaService } from '../prisma/prisma.service';
 import { AuditLogService } from '../audit/audit-log.service';
 import { WebhooksService } from '../webhooks/webhooks.service';
-import { CalendarSyncService } from '../calendar-sync/calendar-sync.service';
+import { DomainEventsService } from '../domain-events/domain-events.service';

 describe('QueensService', () => {
  let service: QueensService;
  let tx: {
-    queen: { update: jest.Mock };
+    $queryRaw: jest.Mock;
+    queen: {
+      update: jest.Mock;
+      findUnique: jest.Mock;
+      findUniqueOrThrow: jest.Mock;
+    };
    queenColonyHistory: { updateMany: jest.Mock; create: jest.Mock };
    colony: { findUnique: jest.Mock; update: jest.Mock };
  };
@@ -29,13 +34,20 @@ describe('QueensService', () => {
    $transaction: jest.Mock;
  };
  let auditLog: { record: jest.Mock };
+  let webhooks: { dispatch: jest.Mock };
+  let events: { record: jest.Mock };

  const queen = { id: 'queen-1', code: 'Q-001', status: 'ACTIVE' };
  const colony = { id: 'colony-1', currentQueenId: null };

  beforeEach(() => {
    tx = {
-      queen: { update: jest.fn().mockResolvedValue(queen) },
+      $queryRaw: jest.fn().mockResolvedValue([]),
+      queen: {
+        update: jest.fn().mockResolvedValue(queen),
+        findUnique: jest.fn(),
+        findUniqueOrThrow: jest.fn().mockResolvedValue(queen),
+      },
      queenColonyHistory: { updateMany: jest.fn(), create: jest.fn() },
      colony: {
        findUnique: jest.fn().mockResolvedValue(null),
@@ -57,13 +69,17 @@ describe('QueensService', () => {
      ),
    };
    auditLog = { record: jest.fn().mockResolvedValue(undefined) };
-    const webhooks = { dispatch: jest.fn().mockResolvedValue(undefined) };
-    const calendarSync = { enqueue: jest.fn(), cancel: jest.fn() };
+    webhooks = { dispatch: jest.fn().mockResolvedValue(undefined) };
+    events = { record: jest.fn().mockResolvedValue(undefined) };
+    // The locked re-read sees what the pre-read saw unless a test says otherwise.
+    tx.queen.findUnique.mockImplementation(
+      (args: unknown) => prisma.queen.findUnique(args) as Promise<unknown>,
+    );
    service = new QueensService(
      prisma as unknown as PrismaService,
      auditLog as unknown as AuditLogService,
      webhooks as unknown as WebhooksService,
-      calendarSync as unknown as CalendarSyncService,
+      events as unknown as DomainEventsService,
    );
  });

@@ -74,9 +90,18 @@ describe('QueensService', () => {
    );
  });

+  // Inside the transaction: lookup by id is the target colony, by
+  // currentQueenId the colony the queen leaves.
+  function mockColonies(target: unknown, previous: unknown = null): void {
+    const lookup = (args: { where: { id?: string } }) =>
+      Promise.resolve(args.where.id === undefined ? previous : target);
+    prisma.colony.findUnique.mockImplementation(lookup);
+    tx.colony.findUnique.mockImplementation(lookup);
+  }
+
  it('assignColony throws NotFoundException when the colony is missing', async () => {
    prisma.queen.findUnique.mockResolvedValue(queen);
-    prisma.colony.findUnique.mockResolvedValue(null);
+    mockColonies(null);
    await expect(
      service.assignColony('queen-1', { colonyId: 'missing' }, 'user-1'),
    ).rejects.toBeInstanceOf(NotFoundException);
@@ -90,15 +115,13 @@ describe('QueensService', () => {
    await expect(
      service.assignColony('queen-1', { colonyId: 'colony-1' }, 'user-1'),
    ).rejects.toBeInstanceOf(ConflictException);
-    expect(prisma.colony.findUnique).not.toHaveBeenCalled();
+    expect(tx.colony.update).not.toHaveBeenCalled();
+    expect(events.record).not.toHaveBeenCalled();
  });

  it('assignColony closes the previous queen history, sets currentQueenId, and opens a new history row', async () => {
    prisma.queen.findUnique.mockResolvedValue(queen);
-    prisma.colony.findUnique.mockResolvedValue({
-      ...colony,
-      currentQueenId: 'queen-old',
-    });
+    mockColonies({ ...colony, status: 'ACTIVE', currentQueenId: 'queen-old' });
    tx.colony.update.mockResolvedValue({});
    tx.queenColonyHistory.create.mockResolvedValue({ id: 'history-1' });

@@ -122,6 +145,57 @@ describe('QueensService', () => {
    expect(historyArg.data.colonyId).toBe('colony-1');
  });

+  it('assignColony records QueenAssigned with previousColonyId and calls neither webhooks.dispatch nor calendarSync', async () => {
+    prisma.queen.findUnique.mockResolvedValue(queen);
+    mockColonies(
+      { ...colony, status: 'ACTIVE' },
+      { id: 'colony-prev', status: 'ACTIVE' },
+    );
+    tx.colony.update.mockResolvedValue({});
+    tx.queenColonyHistory.create.mockResolvedValue({ id: 'history-1' });
+
+    await service.assignColony('queen-1', { colonyId: 'colony-1' }, 'user-1');
+
+    // Target and previous colony in one statement, then the queen.
+    expect(tx.$queryRaw).toHaveBeenCalledTimes(2);
+    expect(events.record).toHaveBeenCalledTimes(1);
+    expect(events.record).toHaveBeenCalledWith(
+      tx,
+      expect.objectContaining({
+        type: 'QueenAssigned',
+        aggregateType: 'Queen',
+        aggregateId: 'queen-1',
+        actorUserId: 'user-1',
+        payload: {
+          colonyId: 'colony-1',
+          historyId: 'history-1',
+          assignedOn: expect.stringMatching(/^\d{4}-\d{2}-\d{2}$/) as string,
+          previousColonyId: 'colony-prev',
+        },
+      }),
+    );
+    expect(auditLog.record).toHaveBeenCalledWith(expect.anything(), tx);
+    expect(webhooks.dispatch).not.toHaveBeenCalled();
+  });
+
+  it('assignColony answers 409 when the queen was moved by someone else after the pre-read', async () => {
+    prisma.queen.findUnique.mockResolvedValue(queen);
+    prisma.colony.findUnique.mockResolvedValue(null);
+    tx.colony.findUnique.mockImplementation(
+      (args: { where: { id?: string } }) =>
+        Promise.resolve(
+          args.where.id === undefined
+            ? { id: 'colony-other', status: 'ACTIVE' }
+            : { ...colony, status: 'ACTIVE' },
+        ),
+    );
+
+    await expect(
+      service.assignColony('queen-1', { colonyId: 'colony-1' }, 'user-1'),
+    ).rejects.toBeInstanceOf(ConflictException);
+    expect(events.record).not.toHaveBeenCalled();
+  });
+
  it('addEvaluation creates a queen evaluation and records an audit log entry', async () => {
    prisma.queen.findUnique.mockResolvedValue(queen);
    const created = { id: 'eval-1', queenId: 'queen-1', colonyStrength: 4 };
@@ -7,7 +7,9 @@ import {
 import { PrismaService } from '../prisma/prisma.service';
 import { AuditLogService } from '../audit/audit-log.service';
 import { WebhooksService } from '../webhooks/webhooks.service';
-import { CalendarSyncService } from '../calendar-sync/calendar-sync.service';
+import { DomainEventsService } from '../domain-events/domain-events.service';
+import { utcDay } from '../domain-events/domain-event-types';
+import { lockRowForUpdate, lockRowsForUpdate } from '../prisma/row-lock';
 import { Queen, QueenStatus } from '../../generated/prisma';
 import { CreateQueenDto } from './dto/create-queen.dto';
 import { UpdateQueenDto } from './dto/update-queen.dto';
@@ -64,7 +66,7 @@ export class QueensService {
    private readonly prisma: PrismaService,
    private readonly auditLog: AuditLogService,
    private readonly webhooks: WebhooksService,
-    private readonly calendarSync: CalendarSyncService,
+    private readonly events: DomainEventsService,
  ) {}

  async findAll(filter: QueenFilter = {}): Promise<Queen[]> {
@@ -158,28 +160,55 @@ export class QueensService {
    dto: AssignColonyDto,
    actingUserId: string,
  ): Promise<Queen> {
-    const queen = await this.findById(id);
-    if (QUEEN_NO_LONGER_AVAILABLE_STATUSES.includes(queen.status)) {
-      throw new ConflictException(
-        'Die Königin ist nicht mehr verfügbar für eine Zuweisung - korrigiere zuerst ihren Status, falls dies ein Versehen war',
-      );
-    }
-    const colony = await this.prisma.colony.findUnique({
-      where: { id: dto.colonyId },
+    await this.findById(id);
+    // Only to know which colony to lock; the transaction re-reads it.
+    const preRead = await this.prisma.colony.findUnique({
+      where: { currentQueenId: id },
+      select: { id: true },
    });
-    if (!colony) {
-      throw new NotFoundException('Colony not found');
-    }
-    if (!isLivingColonyStatus(colony.status)) {
-      throw new ConflictException(
-        `Das Volk besteht nicht mehr (Status: ${colony.status}) und kann keine Königin aufnehmen`,
-      );
-    }
-    if (colony.currentQueenId === queen.id) {
-      throw new ConflictException('Die Königin sitzt bereits in diesem Volk');
-    }
+    return this.prisma.$transaction(async (tx) => {
+      // Colonies first (one statement, ascending ids), then the queen: the
+      // order every writer of both uses.
+      await lockRowsForUpdate(tx, 'Colony', [
+        dto.colonyId,
+        ...(preRead ? [preRead.id] : []),
+      ]);
+      await lockRowForUpdate(tx, 'Queen', id);
+      const queen = await tx.queen.findUnique({ where: { id } });
+      if (!queen) {
+        throw new NotFoundException('Queen not found');
+      }
+      if (QUEEN_NO_LONGER_AVAILABLE_STATUSES.includes(queen.status)) {
+        throw new ConflictException(
+          'Die Königin ist nicht mehr verfügbar für eine Zuweisung - korrigiere zuerst ihren Status, falls dies ein Versehen war',
+        );
+      }
+      const colony = await tx.colony.findUnique({
+        where: { id: dto.colonyId },
+      });
+      if (!colony) {
+        throw new NotFoundException('Colony not found');
+      }
+      if (!isLivingColonyStatus(colony.status)) {
+        throw new ConflictException(
+          `Das Volk besteht nicht mehr (Status: ${colony.status}) und kann keine Königin aufnehmen`,
+        );
+      }
+      if (colony.currentQueenId === queen.id) {
+        throw new ConflictException('Die Königin sitzt bereits in diesem Volk');
+      }
+      // The colony the queen leaves must let go of her - currentQueenId is
+      // unique, and a stale pointer would both block this assignment and
+      // keep showing her in a colony she is no longer in.
+      const previousColony = await tx.colony.findUnique({
+        where: { currentQueenId: queen.id },
+      });
+      if ((previousColony?.id ?? null) !== (preRead?.id ?? null)) {
+        throw new ConflictException(
+          'Die Königin wurde gerade umgesetzt - bitte erneut versuchen',
+        );
+      }

-    const history = await this.prisma.$transaction(async (tx) => {
      const now = new Date();
      if (colony.currentQueenId) {
        await tx.queenColonyHistory.updateMany({
@@ -191,12 +220,6 @@ export class QueensService {
          data: { endDate: now },
        });
      }
-      // The colony the queen leaves must let go of her - currentQueenId is
-      // unique, and a stale pointer would both block this assignment and
-      // keep showing her in a colony she is no longer in.
-      const previousColony = await tx.colony.findUnique({
-        where: { currentQueenId: queen.id },
-      });
      if (previousColony) {
        await tx.colony.update({
          where: { id: previousColony.id },
@@ -221,29 +244,39 @@ export class QueensService {
          status: colony.status === 'QUEENLESS' ? 'ACTIVE' : undefined,
        },
      });
-      return tx.queenColonyHistory.create({
+      const history = await tx.queenColonyHistory.create({
        data: {
          queenId: queen.id,
          colonyId: colony.id,
-          startDate: new Date(),
+          startDate: now,
          notes: dto.notes,
        },
      });
+      await this.auditLog.record(
+        {
+          entityType: 'Queen',
+          entityId: queen.id,
+          action: 'UPDATE',
+          userId: actingUserId,
+          beforeData: queen,
+          afterData: { ...queen, assignedColonyId: colony.id },
+        },
+        tx,
+      );
+      await this.events.record(tx, {
+        type: 'QueenAssigned',
+        aggregateType: 'Queen',
+        aggregateId: queen.id,
+        actorUserId: actingUserId,
+        payload: {
+          colonyId: colony.id,
+          historyId: history.id,
+          assignedOn: utcDay(now),
+          previousColonyId: previousColony?.id ?? null,
+        },
+      });
+      return tx.queen.findUniqueOrThrow({ where: { id } });
    });
-
-    await this.auditLog.record({
-      entityType: 'Queen',
-      entityId: queen.id,
-      action: 'UPDATE',
-      userId: actingUserId,
-      beforeData: queen,
-      afterData: { ...queen, assignedColonyId: colony.id },
-    });
-
-    const updated = await this.findById(id);
-    await this.webhooks.dispatch('queen.replaced', updated);
-    await this.calendarSync.enqueue('QUEEN_ASSIGNMENT', history.id);
-    return updated;
  }

  async getHistory(id: string) {