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 { NotificationsService } from '../src/notifications/notifications.service'; import { CalendarSyncSettingsService } from '../src/calendar-sync/calendar-sync-settings.service'; import { AutomationService } from '../src/automation/automation.service'; import { AutomationTriggerType, Prisma } from '../generated/prisma'; import { createTestApp } from './utils/create-test-app'; import { login } from './utils/api'; import { drainDomainEvents } from './utils/domain-events'; /** * Side effects of a committed write (automation rules, calendar sync, * webhooks) must never turn that write into an error response: the offline * queue would send the request again and store the record twice - for * treatments that is a duplicate entry in the Bestandsbuch. Producers that * record a domain event run none of them in the request; the handlers do, * and a failed handler is retried without repeating what already worked. */ describe('Post-commit side effects (e2e)', () => { let app: INestApplication; let prisma: PrismaService; const suffix = `${Date.now()}-${Math.round(Math.random() * 1e6)}`; const email = `post-commit-e2e-${suffix}@example.com`; const password = 'SuperSecret123!'; let token: string; let userId: string; let apiaryId: string; const colonyIds: string[] = []; const stockItemIds: string[] = []; const server = () => app.getHttpServer(); const auth = () => ({ Authorization: `Bearer ${token}` }); function post(path: string, body: object, idempotencyKey?: string) { const call = request(server()).post(path).set(auth()); return ( idempotencyKey === undefined ? call : call.set('Idempotency-Key', idempotencyKey) ).send(body); } async function createColony(code: string) { const colony = await prisma.colony.create({ data: { code: `${code}-${suffix}`, apiaryId, colonyType: 'PRODUCTION_COLONY', }, }); colonyIds.push(colony.id); return colony; } /** Open or not: a rule suggestion that exists at all. */ const suggestionCount = (ruleKey: AutomationTriggerType, relatedId: string) => prisma.suggestion.count({ where: { ruleKey, relatedId } }); const openSuggestionCount = ( ruleKey: AutomationTriggerType, relatedId: string, ) => prisma.suggestion.count({ where: { ruleKey, relatedId, status: 'OPEN' }, }); /** The notifications written for the suggestions of the given objects. */ async function suggestionNotifications(relatedIds: string[]) { const ids = ( await prisma.suggestion.findMany({ where: { relatedId: { in: relatedIds } }, select: { id: true }, }) ).map((suggestion) => suggestion.id); return prisma.notification.findMany({ where: { relatedType: 'Suggestion', relatedId: { in: ids } }, select: { userId: true }, }); } /** Everything this file wrote, scoped by its own ids. */ async function cleanUp(): Promise { const byColony = { where: { colonyId: { in: colonyIds } }, select: { id: true }, }; const recordIds = ( await Promise.all([ prisma.inspection.findMany(byColony), prisma.treatment.findMany(byColony), prisma.varroaMeasurement.findMany(byColony), ]) ) .flat() .map((record) => record.id); const relatedIds = [...colonyIds, ...stockItemIds]; const suggestionIds = ( await prisma.suggestion.findMany({ where: { relatedId: { in: relatedIds } }, select: { id: true }, }) ).map((suggestion) => suggestion.id); const aggregateIds = recordIds; // Deliveries go with their event (ON DELETE CASCADE). await prisma.domainEvent.deleteMany({ where: { aggregateId: { in: aggregateIds } }, }); await prisma.calendarSyncOutboxEntry.deleteMany({ where: { sourceId: { in: aggregateIds } }, }); // Rule suggestions and the rules' own entries are written without a user. await prisma.auditLog.deleteMany({ where: { OR: [ { userId }, { entityType: 'Suggestion', entityId: { in: suggestionIds } }, ...relatedIds.map((relatedId) => ({ entityType: 'AutomationRule', afterData: { path: ['relatedId'], equals: relatedId }, })), ], }, }); await prisma.notification.deleteMany({ where: { relatedType: 'Suggestion', relatedId: { in: suggestionIds } }, }); await prisma.suggestion.deleteMany({ where: { id: { in: suggestionIds } }, }); await prisma.inspection.deleteMany({ where: byColony.where }); await prisma.treatment.deleteMany({ where: byColony.where }); await prisma.varroaMeasurement.deleteMany({ where: byColony.where }); await prisma.stockItem.deleteMany({ where: { id: { in: stockItemIds } } }); await prisma.colony.deleteMany({ where: { id: { in: colonyIds } } }); await prisma.apiary.deleteMany({ where: { id: apiaryId } }); await prisma.idempotencyRecord.deleteMany({ where: { userId } }); await prisma.refreshToken.deleteMany({ where: { userId } }); await prisma.user.deleteMany({ where: { id: userId } }); } beforeAll(async () => { app = await createTestApp(); prisma = app.get(PrismaService); userId = ( await prisma.user.create({ data: { email, passwordHash: await bcrypt.hash(password, 12), name: 'Post Commit', role: { connect: { key: 'ADMIN' } }, }, }) ).id; token = (await login(app, email, password)).accessToken; apiaryId = ( await prisma.apiary.create({ data: { name: `Post Commit ${suffix}` } }) ).id; }); afterEach(() => { jest.restoreAllMocks(); }); afterAll(async () => { await cleanUp(); await app.close(); }); describe('when a side effect fails', () => { let errorLog: jest.SpyInstance; /** The error lines the automation's legacy path logged for one record. */ const automationFailuresFor = (entityType: string, entityId: string) => errorLog.mock.calls .map(([message]) => String(message)) .filter((message) => message.startsWith( `Automation rule evaluation failed for ${entityType} ${entityId}:`, ), ); /** * Every transaction still runs for real, but inside it the seeding of * the automation rules (ensureDefaults → tx.automationRule.createMany) * fails: the first write of every rule evaluation. A rule evaluated * inside the request's own transaction would turn it into a 500. */ function failRuleSeedingInTransactions(): void { type Work = (tx: Prisma.TransactionClient) => Promise; const runTransaction = prisma.$transaction.bind(prisma) as ( work: Work, options?: unknown, ) => Promise; const failingRules = (rules: object) => new Proxy(rules, { get: (target, property): unknown => property === 'createMany' ? () => Promise.reject(new Error('automation rules down')) : Reflect.get(target, property), }); const withFailingRuleSeeding = (tx: Prisma.TransactionClient) => new Proxy(tx, { get: (target, property): unknown => property === 'automationRule' ? failingRules(target.automationRule) : Reflect.get(target, property), }); jest .spyOn(prisma, '$transaction') .mockImplementation((work: unknown, options?: unknown) => typeof work === 'function' ? runTransaction( (tx) => (work as Work)(withFailingRuleSeeding(tx)), options, ) : runTransaction(work as Work, options), ); } beforeEach(() => { errorLog = jest .spyOn(Logger.prototype, 'error') .mockImplementation(() => undefined); // Automation: the warning cannot be written. jest .spyOn(app.get(NotificationsService), 'recordForPermission') .mockRejectedValue(new Error('notifications down')); // Calendar sync: enqueue cannot read its settings. jest .spyOn(app.get(CalendarSyncSettingsService), 'get') .mockRejectedValue(new Error('calendar settings down')); // Webhooks: the subscriptions cannot be loaded. jest .spyOn(prisma.webhookSubscription, 'findMany') .mockRejectedValue(new Error('webhook subscriptions down')); }); it('still answers 201 and replays on retry with the same Idempotency-Key (no side effect in the request)', async () => { const colony = await createColony('PC-INSP'); const key = `post-commit-${suffix}-inspection`; const body = { colonyId: colony.id, date: '2026-07-01T00:00:00.000Z', queenSeen: false, eggsPresent: false, }; const first = await post('/api/v1/inspections', body, key).expect(201); const retry = await post('/api/v1/inspections', body, key).expect(201); const inspectionId = (first.body as { id: string }).id; expect((retry.body as { id: string }).id).toBe(inspectionId); expect( await prisma.inspection.count({ where: { colonyId: colony.id } }), ).toBe(1); // The request only recorded its event - the replay not even that. const events = await prisma.domainEvent.findMany({ where: { aggregateId: inspectionId }, }); expect(events).toEqual([ expect.objectContaining({ type: 'InspectionRecorded', status: 'PENDING', attempts: 0, }), ]); // No rule ran in the request, not even one that failed: the suggestion // follows when the handler runs. expect(automationFailuresFor('Inspection', inspectionId)).toEqual([]); expect(await suggestionCount('QUEENLESS_NO_EGGS', colony.id)).toBe(0); expect(await suggestionNotifications([colony.id])).toEqual([]); }); it('still answers 201 for a treatment when the rule evaluation would fail, and replays it', async () => { failRuleSeedingInTransactions(); const colony = await createColony('PC-TREAT'); const key = `post-commit-${suffix}-treatment`; const body = { colonyId: colony.id, date: '2026-08-01T00:00:00.000Z', product: 'Oxalsäure', }; const first = await post('/api/v1/treatments', body, key).expect(201); const retry = await post('/api/v1/treatments', body, key).expect(201); expect((retry.body as { id: string }).id).toBe( (first.body as { id: string }).id, ); expect( await prisma.treatment.count({ where: { colonyId: colony.id } }), ).toBe(1); }); }); describe('when an automation handler fails', () => { it('creates exactly one automation suggestion when the handler fails once and is retried', async () => { jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); const colonyId = (await createColony('PC-RETRY')).id; jest .spyOn(app.get(NotificationsService), 'recordForPermission') .mockRejectedValueOnce(new Error('notifications down')); const { body } = (await post('/api/v1/inspections', { colonyId, date: '2026-07-01T00:00:00.000Z', queenSeen: false, eggsPresent: false, }).expect(201)) as { body: { id: string } }; await drainDomainEvents(app, [body.id]); // The warning failed, so the whole evaluation rolled back. expect(await suggestionCount('QUEENLESS_NO_EGGS', colonyId)).toBe(0); expect(await suggestionNotifications([colonyId])).toEqual([]); // Due again now instead of after the 30 s backoff. await prisma.$executeRaw`UPDATE "DomainEvent" SET "availableAt" = now() WHERE "aggregateId" = ${body.id}`; await drainDomainEvents(app, [body.id]); expect(await suggestionCount('QUEENLESS_NO_EGGS', colonyId)).toBe(1); // One copy per recipient (everybody who may read tasks and // inspections), written once. const warnings = await suggestionNotifications([colonyId]); expect(warnings.length).toBeGreaterThan(0); expect(new Set(warnings.map((w) => w.userId)).size).toBe(warnings.length); const deliveries = await prisma.domainEventDelivery.findMany({ where: { event: { aggregateId: body.id } }, }); expect( deliveries.find((d) => d.handler === 'automation.inspection-rules'), ).toMatchObject({ status: 'DONE', attempts: 2 }); // What already worked did not run again. expect( deliveries.find((d) => d.handler === 'webhooks.forward'), ).toMatchObject({ status: 'DONE', attempts: 1 }); await expect( prisma.domainEvent.findFirstOrThrow({ where: { aggregateId: body.id }, }), ).resolves.toMatchObject({ status: 'DONE', attempts: 1 }); }); }); describe('automation suggestions', () => { it('leaves one open reorder suggestion when several triggers for the same stock item arrive at once', async () => { const automation = app.get(AutomationService); // The same object crossing the threshold on several pods/requests at // once: every evaluation sees "nothing open yet" unless checking and // creating are serialised. const item = await prisma.stockItem.create({ data: { name: `Race ${suffix}`, category: 'CONSUMABLE', unit: 'Stück', quantity: 1, minThreshold: 5, }, }); stockItemIds.push(item.id); await Promise.all( Array.from({ length: 8 }, () => automation.evaluateStockItem(item)), ); expect(await openSuggestionCount('STOCK_ITEM_LOW', item.id)).toBe(1); }); it('leaves one open suggestion when several triggers for the same colony arrive at once', async () => { const colony = await createColony('PC-RACE'); const measurement = { colonyId: colony.id, date: '2026-08-15T00:00:00.000Z', method: 'NATURAL_MITE_FALL', miteCount: 20, // One day on the board: 20 mites per day, above the default 10. durationMinutes: 24 * 60, }; const responses = await Promise.all( Array.from({ length: 6 }, () => post('/api/v1/varroa-measurements', measurement).expect(201), ), ); // One worker each: the handlers evaluate the rule at the same moment. await Promise.all( responses.map(({ body }) => drainDomainEvents(app, [(body as { id: string }).id]), ), ); expect(await openSuggestionCount('VARROA_THRESHOLD', colony.id)).toBe(1); }); }); });