# Plattform-Rückgrat Implementation Plan **Status (2026-10-16):** Alle 26 Tasks umgesetzt und gemergt (Commits `c01a388`, `cdd30cf`, Merge `7097fcd`). Nicht ausgerollt; das Rollout-Runbook unten ist noch bei keinem Schritt angewendet. Issues: [#13](https://git.encry.de/Nick.Rosenloecher/ApiCore/issues/13) (Paket), [#28](https://git.encry.de/Nick.Rosenloecher/ApiCore/issues/28) (Rollout). > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** ApiCore läuft aus einem Image in drei Rollen (`api`, `worker`, `job`). Fachliche Folgen hängen nicht mehr an direkten Service-Aufrufen nach dem Commit, sondern an einer dauerhaften Outbox (`DomainEvent`), die der Worker genau einmal je Handler zustellt. Zeitgesteuerte Arbeit läuft als Kubernetes-CronJob. Rate-Limits zählen über alle Pods. Schlüssel sind je Zweck getrennt. Die öffentliche Herkunftsseite verrät keinen Standort. Die Manifeste in `deploy/k8s` werden wieder zur Quelle der Wahrheit. **Architecture:** NestJS-Backend mit drei Einstiegspunkten: `dist/main.js` (`api`/`all`), `dist/worker.js` (`worker`) und `dist/cli.js ` (`job`). Fachservices schreiben Audit-Eintrag und `DomainEvent` in derselben Prisma-Transaktion und wecken den Worker per `pg_notify`. Der Worker übernimmt Ereignisse mit `FOR UPDATE SKIP LOCKED` und ruft per `@OnDomainEvent` registrierte Handler auf. Jeder Handler läuft in einer eigenen Transaktion, die seine `DomainEventDelivery`-Zeile sperrt. Handler schreiben nur in die DB; Netzwerkarbeit liegt in eigenen Outboxen (`CalendarSyncOutboxEntry`, `WebhookDelivery`), die Worker-Schleifen abarbeiten. Die Jobs schreiben `JobRun`. Die Seite „Verwaltung > System“ zeigt Läufe, Warteschlange und Fehler. Kustomize mit `base`, Komponenten je Rollout-Schritt und Overlay `encry`. **Tech Stack:** NestJS 11, Prisma 7.9 (`@prisma/adapter-pg`, `pg` 8), PostgreSQL 18, `@nestjs/throttler` 6.5, `cron` 4.4, Jest 30 (Unit + e2e gegen Test-DB), Next.js App Router + TanStack Query + Vitest/RTL, Kustomize, kubeconform, Gitea Actions. **Spec:** `docs/superpowers/specs/2026-10-12-plattform-rueckgrat-und-fahrplan-design.md`, Abschnitte 2–9. Abschnitt 10 ist nicht Teil dieses Plans. --- ## Global Constraints **Rollen und Prozesse** - `APICORE_ROLE` ∈ `api | worker | job | all`. Ohne Wert gilt außerhalb von Produktion je Einstieg: `main` → `all`, `worker` → `worker`, `cli` → `job`. - Bei `NODE_ENV=production`: `APICORE_ROLE` ist Pflicht, `all` wird abgelehnt, eine Rolle, die nicht zum Einstiegspunkt passt, wird abgelehnt. - Einstiege: `node dist/main.js` (`api`/`all`), `node dist/worker.js` (`worker`), `node dist/cli.js [args]` (`job`). Exit-Codes von cli: `0` erfolgreich, `1` Job gescheitert, `2` falscher Aufruf. - Jobs: `daily-evaluation`, `task-templates`, `cleanup`, `app-backup`, `rekey` (dazu `rekey --dry-run`). **Worker** - Health: `GET /health` auf `WORKER_HEALTH_PORT` (Standard `3002`). `200 {"status":"ok"}`, wenn `SELECT 1` in ≤ 5 s antwortet und der letzte Dispatcher-Durchlauf ≤ 120 000 ms her ist. Sonst `503 {"status":"unhealthy","reason":"database"|"dispatcher-stale"}`. - Schleifen: - Ereignisse: alle 5 000 ms, Weckkanal `domain_event`. - Webhook-Zustellung: alle 5 000 ms, Weckkanal `webhook_delivery`. - Kalender-Outbox: alle 60 000 ms, Weckkanal `calendar_sync`. - `pg_notify`-Payload ist immer `''`, also keine Daten im Kanal. - Wiederverbinden des LISTEN-Clients: 1 s, verdoppelt, höchstens 30 s. **Dispatcher und Outbox** - Batch 50, Claim per `UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP LOCKED LIMIT 50) RETURNING *`. - Wiederholabstand nach Fehlversuch n: 30 s, 2 min, 10 min, 1 h, ab n=5 jeweils 6 h (`RETRY_DELAYS_MS = [30_000, 120_000, 600_000, 3_600_000, 21_600_000]`). - Nach 8 Versuchen (`MAX_ATTEMPTS = 8`) → `FAILED`. - Liegengebliebener Claim: `claimedAt` älter als 5 min (`STALE_CLAIM_MS = 300_000`). Er wird mit `attempts + 1` sofort wieder `PENDING`, ab 8 Versuchen `FAILED`. - Handler-Transaktion: Timeout 10 000 ms. `drain()` macht höchstens 20 Runden je Takt. - Alle Zeitvergleiche und Zeitrechnungen für Claims, Backoff und Throttler laufen **in SQL mit `now()`** (DB-Uhr), nie mit einem JS-`Date` gegen DB-Zeitstempel. - Handler sind **reine DB-Schreiber**: nur über den übergebenen `tx` schreiben, keine HTTP- oder Graph-Aufrufe. Lesen über `tx` oder `PrismaService` ist erlaubt. - Zeitbasierte Ereignisse: `dedupeKey = "::"` (UTC-Tag des fachlichen Ereignisses), eindeutig. Nachholfenster 7 Tage. **Webhooks** - Gleiche Wiederholregel wie der Dispatcher, Batch 50, höchstens 5 gleichzeitig, 5 s Timeout (wie bisher). - Nur `2xx` gilt als zugestellt. - `X-ApiCore-Delivery` ist die `WebhookDelivery.id` und bleibt über alle Wiederholungen gleich. - Signatur wie bisher über `".."`. **Kalender-Outbox:** unverändert `MAX_ATTEMPTS = 5`, `BATCH_SIZE = 20`. Sie läuft im Worker. **Throttler:** UNLOGGED-Tabelle `ThrottlerHit`, PK `(key, throttlerName)`, eine einzige Upsert-Anweisung mit `now()`. Grenzen unverändert: Standard 100/min, sensible Aktionen 10/min, öffentliche Chargen 30/min. **Schlüssel** - `MFA_ENCRYPTION_KEY`, `WEBHOOK_ENCRYPTION_KEY`, `BACKUP_ENCRYPTION_KEY` sind in Produktion Pflicht: je ≥ 32 Zeichen, paarweise verschieden und verschieden von `JWT_ACCESS_SECRET`/`JWT_REFRESH_SECRET`. - Optional der vorige Schlüssel als `*_PREVIOUS`. - Chiffrat-Format `k<8 hex>:..` (Base64-Teile), AES-256-GCM, Schlüssel per HKDF-SHA256 je Zweck. **Öffentliche Herkunftsseite:** `Cache-Control: public, max-age=300, s-maxage=3600` nur auf `200` von `GET /api/v1/public/batches/:lotNumber` (Route `public/batches/:lotNumber`, siehe `public.controller.ts`). Alle anderen Antworten `no-store`. **Ressourcen (Spec §7)** | Pod | CPU request | CPU limit | RAM request | RAM limit | |---|---|---|---|---| | api | 50m | 500m | 192Mi | 512Mi | | worker | 20m | 200m | 128Mi | 256Mi | | frontend | 30m | 300m | 128Mi | 256Mi | | CronJob-Pods und initContainer | 50m | 500m | 192Mi | 512Mi | - HPA `api`: 1–2 Pods bei 70 % CPU. **CronJobs** - Alle mit `timeZone: Europe/Berlin`, `concurrencyPolicy: Forbid`, `startingDeadlineSeconds: 600`, `successfulJobsHistoryLimit: 3`, `failedJobsHistoryLimit: 3`, `ttlSecondsAfterFinished: 86400`. - Zeitpläne: - `cleanup`: `0 4 * * *` - `daily-evaluation`: `30 4 * * *` - `task-templates`: `0 5 * * *` - `app-backup`: `0 * * * *` (stündliche Prüfung, ob fällig) - `rekey` und `rekey-dry-run`: `suspend: true`, `0 0 31 2 *` - `backoffLimit: 1` für `cleanup`, `daily-evaluation`, `task-templates`; `0` für `app-backup` und `rekey`. `activeDeadlineSeconds` 1800, für `app-backup` 3600. **Aufbewahrung (`cleanup`):** `DONE`-Ereignisse und `DELIVERED`-Zustellungen 30 Tage, `FAILED` 90 Tage, `JobRun` 90 Tage. `RUNNING`-Läufe älter als 6 h → `FAILED` („abgebrochen“). **Cluster** - Namespace `apicore`, `ingressClassName: traefik`, Annotation `traefik.ingress.kubernetes.io/router.middlewares: apicore-cloudflare-only@kubernetescrd`. - Labels und Selektoren entsprechen dem **Live-Stand** (read-only geprüft; z. B. Postgres `app.kubernetes.io/name: apicore-postgres`, Service-Port `postgres`) – Ruling 2026-10-13 nach Review Task 26, StatefulSet-Selektoren sind unveränderlich. Neu: `app: apicore-worker`, `app: apicore-job` + `apicore.io/job: `. - DB-Zugriff nur für Pods mit `db-client: "true"`. **Projektregeln** - Sicherheit hat Priorität 1. - Konsistenter Code ist oberstes Gesetz: ein Muster, gemeinsame Helfer: - `retry-policy.ts`, `outbox-claims.ts`, `notifyWorker`, `withTimeout`, `loggableError`, `DbClient`. - Pflicht-`tx` steht als erster Parameter, optionaler `tx` als letzter. - Jedes Feld ist bearbeitbar (Null-Regel, ARCHITECTURE Annahme 20, `update-dto-null-handling.spec.ts`). - TDD: zuerst der fehlschlagende Test. - UI-Texte auf Deutsch; Code-Bezeichner, Log- und Fehlermeldungen auf Englisch wie im Bestand. - `lastError` und Logs nur über `loggableError()`, nie Payload, Secret oder Antwortkörper. - Kein `prisma db push`. Migrationen als SQL in `prisma/migrations/_/migration.sql`. Erzeugt werden sie mit `prisma migrate diff --from-config-datasource --to-schema prisma/schema.prisma --script` gegen die Test-DB und danach von Hand nachgearbeitet. `migrate dev` nicht gegen geteilte DBs. - **Agents committen nicht.** Der Orchestrator macht am Ende einen zentralen Commit. Darum gibt es in den Tasks keinen Commit-Schritt, sondern am Ende jedes Tasks einen Schritt „Verifikation“. - e2e-Tests laufen gegen die lokale Test-DB aus `.env.test` (localhost:5433). - Test-DB migrieren (Git Bash, in `ApiCore-Backend/`): `export DATABASE_URL="$(grep '^DATABASE_URL=' .env.test | cut -d= -f2- | tr -d '"')" && npx prisma migrate deploy && npx prisma generate` - Einzelne e2e-Datei: `npx jest --config ./test/jest-e2e.json test/.e2e-spec.ts --runInBand` - Unit: `npx jest `. Lint wie im Image-Gate: `npx eslint "{src,test}/**/*.ts"`. Typecheck: `npm run typecheck`. - Frontend: vor Next.js-Code `node_modules/next/dist/docs/` lesen (siehe `ApiCore-Frontend/AGENTS.md`). Tests mit `npx vitest run ""`, dazu `npm run lint` und `npm run typecheck`. - Neue Abhängigkeiten nur: `pg` + `@types/pg` (Task 5) und `cron` direkt statt `@nestjs/schedule` (Task 17). - `ARCHITECTURE.md` ist eine geteilte Datei. Laufen mehrere Tasks einer Welle parallel, führt der Orchestrator deren Doku-Schritte nacheinander aus. --- ## Review Focus 1. **Worker fällt stundenlang aus → Ereignisse stauen sich.** Die API arbeitet weiter, der Rückstau muss in Reihenfolge und ohne Verhungern abfließen und in „Verwaltung > System“ sichtbar sein. - Tests: - Task 4 e2e `drains a backlog of 120 events in batches of at most 50` - Task 24 unit `reports the oldest pending age from the database clock` - Task 18 unit `catches up occurrences of the last 7 days exactly once` - Task 25 vitest `warns when the oldest open event is older than 10 minutes` 2. **Handler nicht idempotent bei erneuter Zustellung** (Stale-Release, zweiter Worker, Retry nach Teilfehler). - Gegenmittel: Delivery-Zeile `FOR UPDATE` plus Handler in derselben Transaktion, und Handler schreiben nur in die DB. - Tests: - Task 4 e2e `two dispatchers in parallel produce exactly one effect per event and handler` - Task 4 e2e `a stale-released event does not repeat a handler that already committed` - Task 4 Wächter `handler files never import network clients` - Task 11 e2e `creates exactly one automation task when the handler fails once and is retried` 3. **`rekey` bricht mittendrin ab.** Der Job muss wiederaufnehmbar sein, darf parallele Änderungen (neue MFA-Einrichtung) nicht überschreiben, und Altchiffrate bleiben lesbar, bis die JWT-Secrets rotiert sind. - Tests: - Task 21 unit `resumes after an interruption: current rows are skipped, the rest are rekeyed` - Task 21 unit `skips a row changed since it was read and reports it` - Task 21 e2e `a second run rekeys nothing; dry-run writes nothing` 4. **NetworkPolicy sperrt DNS, Probes oder Postgres-Zugriffe.** - Tests: Task 26 `manifest-policy.test.mjs`: - jeder Pod hat DNS-Egress zu `k8s-app: kube-dns` UDP/TCP 53 - Postgres-Ingress nur von `db-client: "true"` - Frontend ohne Weg zu 5432 - 443-Egress schließt private Bereiche aus - Default-Deny vorhanden - Dazu Prüfbefehle im Runbook, Schritt 2: Probes, `getent`, Frontend→Postgres blockiert. 5. **Zwei API-Pods plus Throttler bei Uhrenabweichung.** Die Zähler müssen über beide Instanzen gelten, nur mit der DB-Uhr rechnen, und parallele Treffer dürfen nicht verloren gehen. e2e-Dateien dürfen sich nicht gegenseitig drosseln. - Tests: - Task 19 unit `uses only the database clock (no JS dates in the statement)` - Task 19 e2e `counts across two app instances sharing one namespace` - Task 19 e2e `40 parallel requests over two instances yield exactly 10 × 429` - Task 19 e2e `separate namespaces never share counters` --- ## Abhängigkeiten und Parallelisierung | Welle | Tasks (parallel, verschiedene Dateien) | wartet auf | |---|---|---| | A | 1, 2, 23, 26 | – | | B | 3, 5, 20, 22 | 3 → 2; 20 → 1; 22 → 2 | | C | 4 | 3 | | D | 6, 19 | 6 → 1, 4, 5; 19 → 2, 4 | | E | 7, 8 | 6 | | F | 9 | 3, 7, 8 | | G | 10 | 9 | | H | 11, 12, 13, 14, 15 | 10 (15 → 3, 9) | | I | 16, 18, 21, 24 | 15 (21 → 20; 24 → 4, 7) | | J | 17, 25 | 17 → 11–16; 25 → 24, 7 | Bekannte geteilte Dateien, deshalb jeweils nur ein Task pro Welle: - `src/app.module.ts`: 3 (B), 19 (D), 15 (H), 17 (J) - `package.json`: 5 (B), 6 (D), 17 (J) - `test/utils/create-test-app.ts`: nur 19 - `src/worker/worker-runtime.service.ts`: 6 (D), 7 (E) - `src/automation/automation.service.ts`: 10 (G), 17 (J). Die Producer-Tasks 11–14 fassen die Datei **nicht** an; ihre Alt-Wrapper entfernt Task 17. --- ## Tasks ### Task 1: Laufrollen und gemeinsame Startprüfungen **Files** - Create: `ApiCore-Backend/src/runtime/role.ts`, `ApiCore-Backend/src/runtime/startup-checks.ts` - Test: `ApiCore-Backend/src/runtime/role.spec.ts`, `ApiCore-Backend/src/runtime/startup-checks.spec.ts` - Modify: - `ApiCore-Backend/src/main.ts`: `REQUIRED_ENV_VARS`, `MIN_JWT_SECRET_LENGTH`, `assertRequiredEnvVars`, `assertProductionSettings` wandern nach `startup-checks.ts`. - `ApiCore-Backend/.env.example`: `APICORE_ROLE` und `WORKER_HEALTH_PORT` dokumentiert, auskommentiert. **Interfaces** - Consumes: – - Produces: ```ts export const APICORE_ROLES = ['api', 'worker', 'job', 'all'] as const; export type ApicoreRole = (typeof APICORE_ROLES)[number]; export type EntryPoint = 'main' | 'worker' | 'cli'; export function resolveRole(entry: EntryPoint, env?: NodeJS.ProcessEnv): ApicoreRole; export function assertStartupEnvironment(role: ApicoreRole, env?: NodeJS.ProcessEnv): void; ``` - [ ] **Schritt 1: Failing Tests schreiben** (`role.spec.ts`, `startup-checks.spec.ts`) ```ts it('defaults per entry point outside production', () => { expect(resolveRole('main', {})).toBe('all'); expect(resolveRole('worker', {})).toBe('worker'); expect(resolveRole('cli', {})).toBe('job'); }); it('requires APICORE_ROLE in production', () => expect(() => resolveRole('main', { NODE_ENV: 'production' })).toThrow('APICORE_ROLE is required in production')); it('rejects all in production', () => expect(() => resolveRole('main', { NODE_ENV: 'production', APICORE_ROLE: 'all' })) .toThrow('APICORE_ROLE=all is for local development and tests only')); it('rejects an unknown role', () => expect(() => resolveRole('main', { APICORE_ROLE: 'admin' })).toThrow('APICORE_ROLE must be one of api, worker, job, all')); it('rejects a role that does not belong to the entry point', () => expect(() => resolveRole('main', { APICORE_ROLE: 'worker' })).toThrow('APICORE_ROLE=worker cannot start dist/main.js')); it('accepts api for main in production', () => expect(resolveRole('main', { NODE_ENV: 'production', APICORE_ROLE: 'api' })).toBe('api')); // startup-checks.spec.ts – Verhalten 1:1 aus main.ts übernommen it('lists missing DATABASE_URL', () => expect(() => assertStartupEnvironment('api', { JWT_ACCESS_SECRET: 'x'.repeat(32) })).toThrow('Missing required environment variable(s): DATABASE_URL')); it('rejects a JWT secret shorter than 32 characters', ...); it('requires an https FRONTEND_ORIGIN in production', ...); it('rejects COOKIE_SECURE=false in production', ...); ``` - [ ] **Schritt 2:** `npx jest src/runtime` → FAIL („Cannot find module './role'“). - [ ] **Schritt 3: Implementieren** - Erlaubte Rollen je Einstieg: `main: ['api','all']`, `worker: ['worker']`, `cli: ['job']`. - Fehlermeldungen exakt wie in den Tests. - `main.ts`: `const role = resolveRole('main'); assertStartupEnvironment(role);` vor `NestFactory.create`. Die Modulwahl je Rolle folgt in Task 6. - [ ] **Schritt 4:** `npx jest src/runtime` → PASS. - [ ] **Schritt 5: Verifikation** - `npm run typecheck && npx eslint "{src,test}/**/*.ts" && npm test -- --ci` - Kein Commit. Geänderte Dateien an den Orchestrator melden. --- ### Task 2: Schema und Migration des Rückgrats **Files** - Modify: `ApiCore-Backend/prisma/schema.prisma` - Create: `ApiCore-Backend/prisma/migrations/20261013100000_platform_backbone/migration.sql` - Modify: `ApiCore-Backend/src/backup/backup.service.ts` (`MODELS_EXCLUDED_FROM_BACKUP`) - Test: `ApiCore-Backend/src/backup/backup.service.spec.ts` (bestehender Wächter), neu `ApiCore-Backend/test/platform-schema.e2e-spec.ts` - Docs: `ARCHITECTURE.md` §8, Bullet „Export/Import“: Ausnahmeliste um die fünf Modelle ergänzen. **Interfaces** - Produces (Prisma, ladungstragend): ```prisma enum DomainEventStatus { PENDING PROCESSING DONE FAILED } model DomainEvent { id String @id @default(cuid()) type String aggregateType String aggregateId String payload Json actorUserId String? correlationId String dedupeKey String? @unique occurredAt DateTime @default(now()) status DomainEventStatus @default(PENDING) attempts Int @default(0) availableAt DateTime @default(now()) claimedAt DateTime? processedAt DateTime? lastError String? deliveries DomainEventDelivery[] @@index([status, availableAt]) @@index([correlationId]) @@index([aggregateType, aggregateId]) } enum DomainEventDeliveryStatus { PENDING DONE FAILED } model DomainEventDelivery { id String @id @default(cuid()) eventId String event DomainEvent @relation(fields: [eventId], references: [id], onDelete: Cascade) handler String status DomainEventDeliveryStatus @default(PENDING) attempts Int @default(0) lastError String? completedAt DateTime? createdAt DateTime @default(now()) updatedAt DateTime @updatedAt @@unique([eventId, handler]) } enum WebhookDeliveryStatus { PENDING PROCESSING DELIVERED FAILED } model WebhookDelivery { id String @id @default(cuid()) subscriptionId String subscription WebhookSubscription @relation(fields: [subscriptionId], references: [id], onDelete: Cascade) event String body String status WebhookDeliveryStatus @default(PENDING) attempts Int @default(0) availableAt DateTime @default(now()) claimedAt DateTime? lastStatusCode Int? lastError String? deliveredAt DateTime? createdAt DateTime @default(now()) @@index([status, availableAt]) @@index([subscriptionId, createdAt]) } enum JobRunStatus { RUNNING SUCCEEDED FAILED } model JobRun { id String @id @default(cuid()) job String status JobRunStatus @default(RUNNING) startedAt DateTime @default(now()) finishedAt DateTime? result Json? error String? @@index([job, startedAt]) @@index([status]) } // UNLOGGED in der Migration: kein WAL, nach Absturz geleert – nur Zähler gehen verloren. model ThrottlerHit { key String throttlerName String hits Int expiresAt DateTime blockedUntil DateTime? @@id([key, throttlerName]) @@index([expiresAt]) } ``` - Dazu: - `WebhookSubscription.deliveries WebhookDelivery[]` - `Apiary.publicLocationLabel String?` - `SiteSettings.originCountry String @default("Deutschland")` - `dedupeKey` als nullbares `@unique` statt partiellem Index. Das umgeht die Drift, vor der Spec §3.5 warnt. - [ ] **Schritt 1: Failing Tests** - Erst die Modelle in `schema.prisma` eintragen und `npx prisma generate` laufen lassen. Danach muss der bestehende Wächter scheitern. - Neu `test/platform-schema.e2e-spec.ts`: ```ts it('creates ThrottlerHit as an UNLOGGED table', async () => { const [row] = await prisma.$queryRaw<{ relpersistence: string }[]>`SELECT relpersistence FROM pg_class WHERE relname = 'ThrottlerHit'`; expect(row.relpersistence).toBe('u'); }); it('keeps at most one event per dedupe key', async () => { const data = { type: 'TaskDue', aggregateType: 'Task', aggregateId: `t-${suffix}`, payload: {}, correlationId: 'c', dedupeKey: `TaskDue:t-${suffix}:2026-10-13` }; expect((await prisma.domainEvent.createMany({ data: [data], skipDuplicates: true })).count).toBe(1); expect((await prisma.domainEvent.createMany({ data: [data], skipDuplicates: true })).count).toBe(0); }); ``` - [ ] **Schritt 2:** - `npx jest src/backup/backup.service.spec.ts` → FAIL: der Wächter listet DomainEvent, DomainEventDelivery, WebhookDelivery, JobRun und ThrottlerHit als „weder gesichert noch ausgenommen“. - e2e → FAIL: Tabelle fehlt. - [ ] **Schritt 3: Implementieren** - Migration erzeugen (Test-DB, siehe Global Constraints): `npx prisma migrate deploy`, dann `npx prisma migrate diff --from-config-datasource --to-schema prisma/schema.prisma --script` in die neue `migration.sql`. - Von Hand nachbearbeiten: - `CREATE TABLE "ThrottlerHit"` → `CREATE UNLOGGED TABLE "ThrottlerHit"` - Kopfkommentar im Stil von `20261012100000_calendar_sync_claims` - Danach `npx prisma migrate deploy && npx prisma generate`. - `MODELS_EXCLUDED_FROM_BACKUP` ergänzen (englische Begründungen): - `DomainEvent`: 'Operational outbox - side effects of restored data must never be replayed after a restore.' - `DomainEventDelivery`: 'Per-handler progress of DomainEvent rows, excluded with them.' - `WebhookDelivery`: 'Delivery log and retry queue - restoring it would resend old events to receivers.' - `JobRun`: 'Run history of this deployment\'s CronJobs, meaningless elsewhere.' - `ThrottlerHit`: 'Rate-limit counters (UNLOGGED), valid for seconds to minutes.' - [ ] **Schritt 4:** - Backup-Spec → PASS. - `npx jest --config ./test/jest-e2e.json test/platform-schema.e2e-spec.ts --runInBand` → PASS. - Drift-Check `npx prisma migrate diff --from-config-datasource --to-schema prisma/schema.prisma --exit-code` → Exit 0. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - ARCHITECTURE §8 ergänzen - kein Commit --- ### Task 3: Schreibbausteine in der Transaktion – Audit, Benachrichtigungen, Ereignis-Kern **Files** - Create: - `src/prisma/db-client.ts` - `src/common/worker-wakeup.ts` - `src/domain-events/domain-event-types.ts` - `src/domain-events/domain-event-context.ts` - `src/domain-events/domain-events.service.ts` - `src/domain-events/on-domain-event.decorator.ts` - `src/domain-events/domain-event-registry.ts` - `src/domain-events/domain-events.module.ts` (`@Global`, importiert `DiscoveryModule`) - Modify: - `src/audit/audit-log.service.ts` - `src/notifications/notifications.service.ts` - `src/winter-checks/winter-checks.service.ts`: `tx.auditLog.create` → `auditLog.record(…, tx)` - `src/app.module.ts`: Import `DomainEventsModule` - Test: - `src/audit/audit-log.service.spec.ts` - `src/notifications/notifications.service.spec.ts` - `src/domain-events/domain-events.service.spec.ts` - `src/domain-events/domain-event-registry.spec.ts` - `src/winter-checks/winter-checks.service.spec.ts` Alle Pfade relativ zu `ApiCore-Backend/`. **Interfaces** - Consumes: Prisma-Modelle aus Task 2. - Produces: ```ts // src/prisma/db-client.ts export type DbClient = PrismaService | Prisma.TransactionClient; // src/common/worker-wakeup.ts export const WAKE_CHANNELS = { domainEvents: 'domain_event', calendarSync: 'calendar_sync', webhookDeliveries: 'webhook_delivery' } as const; export type WakeChannel = (typeof WAKE_CHANNELS)[keyof typeof WAKE_CHANNELS]; export function notifyWorker(db: DbClient, channel: WakeChannel): Promise; // SELECT pg_notify(channel, '') // Audit / Benachrichtigungen AuditLogService.record(input: RecordAuditLogInput, tx?: Prisma.TransactionClient): Promise; NotificationsService.record(input: CreateNotificationInput, tx?: Prisma.TransactionClient): Promise; NotificationsService.recordForPermission(permission: PermissionKey, input: Omit, tx?: Prisma.TransactionClient): Promise; // Ereignis-Katalog (vollständig, wird später nicht mehr erweitert) export interface ColonyStateSnapshot { status: ColonyStatus; healthStatus: ColonyCondition | null } export interface DomainEventPayloads { InspectionRecorded: { colonyId: string }; InspectionCorrected: { colonyId: string }; VarroaMeasured: { colonyId: string }; VarroaCorrected: { colonyId: string }; TreatmentRecorded: { colonyId: string }; TreatmentCorrected: { colonyId: string }; FeedingRecorded: { colonyId: string }; FeedingCorrected: { colonyId: string }; HarvestRecorded: { apiaryId: string; colonyIds: string[] }; HarvestCorrected: { apiaryId: string; colonyIds: string[] }; ColonyStatusChanged: { from: ColonyStateSnapshot; to: ColonyStateSnapshot }; ColonyEnded: { status: ColonyStatus }; BatchBottled: { honeyBatchId: string }; BottlingCorrected: { honeyBatchId: string; removed: boolean }; StockChanged: { bottlingBatchIds: string[]; productIds: string[]; rawMaterialKinds: RawMaterialKind[] }; TaskCreated: { relatedType: TaskRelatedType; relatedId: string | null; dueDate: string | null }; TaskUpdated: { status: TaskStatus; dueDate: string | null }; TaskCompleted: { relatedType: TaskRelatedType; relatedId: string | null }; WithdrawalEnded: { colonyId: string; endedOn: string }; TaskDue: { dueOn: string }; QueenAged: { hatchYear: number; colonyId: string | null }; VarroaMeasurementDue: { dueOn: string; lastMeasuredOn: string | null }; InspectionOverdue: { dueOn: string; lastInspectedOn: string | null }; } export type DomainEventType = keyof DomainEventPayloads; export const DOMAIN_EVENT_TYPES: readonly DomainEventType[]; export type AggregateType = 'Inspection' | 'VarroaMeasurement' | 'Treatment' | 'Feeding' | 'Harvest' | 'Colony' | 'BottlingBatch' | 'ProductionBatch' | 'BatchMovement' | 'ShopStockRequest' | 'Task' | 'Queen'; export interface NewDomainEvent { type: T; aggregateType: AggregateType; aggregateId: string; payload: DomainEventPayloads[T]; actorUserId?: string | null; dedupeKey?: string; } export interface DomainEventEnvelope { id: string; type: T; aggregateType: AggregateType; aggregateId: string; payload: DomainEventPayloads[T]; actorUserId: string | null; correlationId: string; occurredAt: Date; attempts: number; } export type DomainEventHandlerFn = (event: DomainEventEnvelope, tx: Prisma.TransactionClient) => Promise; export function dailyDedupeKey(type: DomainEventType, aggregateId: string, day: string): string; // `${type}:${aggregateId}:${day}` export const domainEventContext: AsyncLocalStorage<{ correlationId: string }>; @Injectable() class DomainEventsService { record(tx: Prisma.TransactionClient, event: NewDomainEvent): Promise; recordAll(tx: Prisma.TransactionClient, events: readonly NewDomainEvent[]): Promise; } export function OnDomainEvent(types: DomainEventType | readonly DomainEventType[], handlerId: string): MethodDecorator; export const HANDLER_ID_PATTERN = /^[a-z0-9-]+\.[a-z0-9-]+$/; export interface RegisteredDomainEventHandler { id: string; invoke: DomainEventHandlerFn } @Injectable() class DomainEventRegistry implements OnModuleInit { handlersFor(type: DomainEventType): readonly RegisteredDomainEventHandler[]; handlerIds(): string[]; } ``` - [ ] **Schritt 1: Failing Tests** ```ts // audit-log.service.spec.ts it('writes through the transaction client when one is given', async () => { const tx = { auditLog: { create: jest.fn() } }; await service.record({ entityType: 'Inspection', entityId: 'i1', action: 'CREATE' }, tx as never); expect(tx.auditLog.create).toHaveBeenCalledTimes(1); expect(prisma.auditLog.create).not.toHaveBeenCalled(); }); // notifications.service.spec.ts it('notifies every active user whose active role holds the permission', async () => { prisma.user.findMany.mockResolvedValue([{ id: 'u1' }, { id: 'u2' }]); await expect(service.recordForPermission('backup:configure', { priority: 'WARNUNG', message: 'Job „App-Backup“ fehlgeschlagen' })).resolves.toBe(2); expect(prisma.notification.createMany.mock.calls[0][0].data.map((n) => n.userId)).toEqual(['u1', 'u2']); }); // domain-events.service.spec.ts it('records the event and the wake-up on the same transaction client', async () => { await service.record(tx, { type: 'InspectionRecorded', aggregateType: 'Inspection', aggregateId: 'i1', payload: { colonyId: 'c1' }, actorUserId: 'u1' }); expect(tx.domainEvent.createMany).toHaveBeenCalledWith({ data: [expect.objectContaining({ type: 'InspectionRecorded', aggregateId: 'i1', actorUserId: 'u1', correlationId: expect.any(String) })], skipDuplicates: true }); expect(tx.$executeRaw).toHaveBeenCalledTimes(1); // pg_notify('domain_event','') }); it('refuses the root PrismaService (must run inside the business transaction)', () => expect(service.record(prismaServiceMock as never, event)).rejects.toThrow('DomainEventsService.record needs the transaction client')); it('inherits the correlation id of the event being handled', async () => { await domainEventContext.run({ correlationId: 'corr-1' }, () => service.record(tx, event)); expect(tx.domainEvent.createMany.mock.calls[0][0].data[0].correlationId).toBe('corr-1'); }); it('wakes the worker only when something was inserted', async () => { tx.domainEvent.createMany.mockResolvedValue({ count: 0 }); await expect(service.recordAll(tx, [daily])).resolves.toBe(0); expect(tx.$executeRaw).not.toHaveBeenCalled(); }); // domain-event-registry.spec.ts it('finds handlers by type, also when one method serves several types', ...); it('refuses two handlers with the same id for one type', () => expect(() => registry.onModuleInit()).toThrow('Duplicate domain event handler')); it('rejects a handler id that is not "."', ...); ``` - [ ] **Schritt 2:** `npx jest src/audit src/notifications src/domain-events src/winter-checks` → FAIL (Module fehlen, `record` ignoriert `tx`). - [ ] **Schritt 3: Implementieren** - `record` delegiert an `recordAll([event])`. `recordAll` macht `tx.domainEvent.createMany({ data, skipDuplicates: true })` und bei `count > 0` `notifyWorker(tx, 'domain_event')`. - `correlationId = domainEventContext.getStore()?.correlationId ?? randomUUID()`. - Prüfung „nicht der Root-Client“ über `'$connect' in tx` (Prisma 7.9: der Transaktions-Client hat ebenfalls `$transaction`; geteilter Helfer `assertTransactionClient` in `src/prisma/db-client.ts`). - Payload-JSON enthält nur IDs und Werte aus `DomainEventPayloads`, Datumswerte als `YYYY-MM-DD`. - Registry: `DiscoveryService.getProviders()` + `MetadataScanner`. Der Decorator hängt Metadaten an ein Array an, damit Stapeln möglich ist. Bei doppelten IDs oder einem ungültigen `handlerId` bricht der Boot ab. - `recordForPermission`: `user.findMany({ where: { isActive: true, role: { status: 'ACTIVE', permissions: { some: { permission: { key } } } } } })`, danach `createMany`. - [ ] **Schritt 4:** gleiche Befehle → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - `npx jest --config ./test/jest-e2e.json test/winter-checks.e2e-spec.ts --runInBand` → PASS (Audit jetzt über den Service) - kein Commit --- ### Task 4: Dispatcher **Files** - Create: - `src/common/retry-policy.ts` (+ `.spec.ts`) - `src/common/outbox/outbox-claims.ts` - `src/domain-events/domain-event-dispatcher.ts` (+ `.spec.ts`) - `src/domain-events/handler-contract.spec.ts` (Wächter) - `test/utils/domain-events.ts` - `test/domain-event-dispatcher.e2e-spec.ts` - Modify: `src/domain-events/domain-events.module.ts` (Provider `DomainEventDispatcher`) - Docs: `ARCHITECTURE.md`, neuer Abschnitt „## 10. Ereignis-Rückgrat und Laufrollen“, Unterabschnitt „Ereignisse“: Outbox, Claim, Backoff, `FAILED`, Delivery-Sperre, Handler-Vertrag. **Interfaces** - Consumes: Task 3 (`DomainEventRegistry`, `domainEventContext`, Typen). - Produces: ```ts // retry-policy.ts export const RETRY_DELAYS_MS = [30_000, 120_000, 600_000, 3_600_000, 21_600_000] as const; export const MAX_ATTEMPTS = 8; export const STALE_CLAIM_MS = 300_000; export function retryDelayMs(attempts: number): number; // outbox-claims.ts (Tabellenname nur aus geschlossener Union -> Prisma.raw sicher) export type OutboxTable = 'DomainEvent' | 'WebhookDelivery'; export function claimBatch(db: PrismaService, table: OutboxTable, options: { limit: number; filter?: Prisma.Sql; orderBy: 'occurredAt' | 'createdAt' }): Promise; export function releaseStaleClaims(db: PrismaService, table: OutboxTable): Promise; export function settleSuccess(db: DbClient, table: OutboxTable, id: string, claimedAt: Date, status: 'DONE' | 'DELIVERED'): Promise; export function settleFailure(db: DbClient, table: OutboxTable, id: string, claimedAt: Date, attemptsBefore: number, lastError: string): Promise<'retry' | 'failed'>; // domain-event-dispatcher.ts export const DOMAIN_EVENT_BATCH_SIZE = 50; export const HANDLER_TX_TIMEOUT_MS = 10_000; export const DRAIN_MAX_ROUNDS = 20; export interface DispatchFilter { correlationIds?: readonly string[] } @Injectable() class DomainEventDispatcher { get lastTickAt(): Date | null; runOnce(filter?: DispatchFilter): Promise; // Anzahl übernommener Ereignisse drain(filter?: DispatchFilter, maxRounds?: number): Promise; releaseStaleClaims(): Promise; } // test/utils/domain-events.ts export function drainDomainEvents(app: INestApplicationContext, rootAggregateIds: readonly string[]): Promise; ``` - Gepinnte SQL-Kerne: - Claim: `UPDATE "" SET "status"='PROCESSING', "claimedAt"=now() WHERE "id" IN (SELECT "id" FROM "" WHERE "status"='PENDING' AND "availableAt" <= now() [AND filter] ORDER BY "" LIMIT ${limit} FOR UPDATE SKIP LOCKED) RETURNING *` - Stale: `… SET "status" = CASE WHEN "attempts"+1 >= 8 THEN 'FAILED' ELSE 'PENDING' END, "attempts"="attempts"+1, "claimedAt"=NULL, "availableAt"=now(), "lastError"='claim expired (worker restarted?)' WHERE "status"='PROCESSING' AND "claimedAt" < now() - interval '5 minutes'` - Fehler: `"availableAt" = now() + (${retryDelayMs(n)} * interval '1 millisecond')`, CAS über `id` + `claimedAt`. - Ablauf je Ereignis und Handler: 1. `prisma.$transaction(tx => …, { timeout: 10_000 })` 2. `domainEventDelivery.createMany({ skipDuplicates })` 3. `SELECT "status" … FOR UPDATE`, bei `DONE` überspringen 4. Handler in `domainEventContext.run({ correlationId })` 5. Delivery `DONE`, `attempts + 1` - Fehler (außerhalb der Transaktion): Delivery `attempts + 1`, `lastError = loggableError(e).message`. - Abschluss je Ereignis: - alle `DONE` → Ereignis `DONE` (`processedAt = now()`) - sonst `settleFailure` mit `lastError = ": "`; bei `failed` werden offene Deliveries `FAILED` - Ereignisse ohne Handler → `DONE`. - [ ] **Schritt 1: Failing Tests** ```ts // retry-policy.spec.ts expect([1, 2, 3, 4, 5, 6, 7].map(retryDelayMs)).toEqual([30_000, 120_000, 600_000, 3_600_000, 21_600_000, 21_600_000, 21_600_000]); expect(MAX_ATTEMPTS).toBe(8); // domain-event-dispatcher.spec.ts (Prisma/Registry gemockt) it('marks an event DONE when every handler succeeded', ...); it('retries only the failed handler and schedules the event 30 s ahead after the first failure', ...); // SQL-Parameter enthält 30000 it('marks the event and its open deliveries FAILED after the 8th failed attempt', ...); it('skips a handler whose delivery is already DONE', ...); it('runs the handler inside the delivery transaction with the event correlation id', ...); it('updates lastTickAt even when nothing was claimed', ...); it('stores lastError without the Prisma message', ...); // PrismaClientKnownRequestError mit "secret-host" -> nicht im lastError // handler-contract.spec.ts (Wächter, Review Focus 2) it('handler files never import network clients', () => { for (const file of glob('src/**/*.handlers.ts')) { expect(read(file)).not.toMatch(/graph-client|webhook-delivery(?!\.worker)|node:https?|\bfetch\(/); } }); // test/domain-event-dispatcher.e2e-spec.ts – eigenes Testmodul mit CountingHandler // @OnDomainEvent('TaskDue', 'test.counter') -> tx.auditLog.create({ entityType: 'DispatcherE2E', entityId: e.aggregateId, action: 'CREATE' }) // @OnDomainEvent('QueenAged', 'test.failing') -> throw new Error('boom') (beide Typen haben im Plan keine echten Handler) it('two dispatchers in parallel produce exactly one effect per event and handler', async () => { await recordTaskDue(20); await Promise.all([appA.get(DomainEventDispatcher).drain(filter), appB.get(DomainEventDispatcher).drain(filter)]); expect(await prisma.auditLog.count({ where: { entityType: 'DispatcherE2E', entityId: { startsWith: suffix } } })).toBe(20); }); it('drains a backlog of 120 events in batches of at most 50', ...); // runOnce-Ergebnisse [50, 50, 20, 0] it('a stale-released event does not repeat a handler that already committed', ...); // claimedAt per SQL auf now()-6 min it('schedules a failed event about 30 s ahead by the database clock', async () => { const [{ s }] = await prisma.$queryRaw<{ s: number }[]>`SELECT EXTRACT(EPOCH FROM "availableAt" - now())::float AS s FROM "DomainEvent" WHERE id = ${id}`; expect(s).toBeGreaterThan(25); expect(s).toBeLessThanOrEqual(30.5); }); ``` - [ ] **Schritt 2:** `npx jest src/common/retry-policy.spec.ts src/domain-events` und die e2e-Datei → FAIL. - [ ] **Schritt 3: Implementieren** wie oben beschrieben. - `drainDomainEvents` sucht die `correlationId`s der Wurzel-Aggregate und ruft `dispatcher.drain({ correlationIds })` auf. - Der Filter verhindert, dass parallele e2e-Dateien fremde Ereignisse abarbeiten. - [ ] **Schritt 4:** gleiche Befehle → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - ARCHITECTURE §10 „Ereignisse“ - kein Commit --- ### Task 5: Worker-Bausteine (Takt, LISTEN, Health) **Files** - Create: - `src/common/with-timeout.ts` (+ `.spec.ts`) - `src/worker/polling-loop.ts` (+ `.spec.ts`) - `src/worker/pg-listener.ts` (+ `.spec.ts`) - `src/worker/worker-health.ts` (+ `.spec.ts`) - Modify: - `src/health/health.controller.ts`: lokales `withTimeout` durch den gemeinsamen Helfer ersetzen - `package.json`: `npm install pg@^8.23.0 && npm install -D @types/pg` **Interfaces** - Produces: ```ts export function withTimeout(promise: Promise, ms: number, message: string): Promise; export class PollingLoop { constructor(options: { name: string; intervalMs: number; tick: () => Promise; logger: Logger }); start(): void; wake(): void; stop(): Promise; } export interface ListenClient { connect(): Promise; query(sql: string): Promise; on(event: 'notification' | 'error' | 'end', cb: (arg: unknown) => void): void; end(): Promise } export class PgListener { constructor(options: { connectionString: string; channels: readonly WakeChannel[]; onNotify: (channel: WakeChannel) => void; logger: Logger; clientFactory?: (connectionString: string) => ListenClient }); start(): Promise; stop(): Promise; } export const HEALTH_STALE_MS = 120_000; export const DEFAULT_WORKER_HEALTH_PORT = 3002; export interface WorkerHealth { ok: boolean; reason?: 'database' | 'dispatcher-stale' } export function workerHealthPort(env?: NodeJS.ProcessEnv): number; export function startWorkerHealthServer(options: { port: number; check: () => Promise }): Promise; ``` - [ ] **Schritt 1: Failing Tests** (Jest Fake Timers) ```ts // polling-loop.spec.ts it('ticks every intervalMs', ...); it('never runs two ticks at once', ...); it('wake() during a tick runs exactly one more tick right after', ...); it('logs a throwing tick via loggableError and keeps going', ...); it('stop() waits for the running tick and schedules nothing more', ...); // pg-listener.spec.ts (Fake-Client) it('LISTENs on every channel after connecting', () => expect(client.query.mock.calls.map((c) => c[0])).toEqual(['LISTEN "domain_event"', 'LISTEN "calendar_sync"', 'LISTEN "webhook_delivery"'])); it('reconnects after an error with 1 s, 2 s, 4 s … at most 30 s', ...); it('stop() ends the client and does not reconnect', ...); // worker-health.spec.ts (echter Server auf Port 0) it('answers 200 {"status":"ok"} on GET /health when healthy', ...); it('answers 503 with only status and reason when unhealthy', async () => { expect(await getJson('/health')).toEqual({ status: 503, body: { status: 'unhealthy', reason: 'dispatcher-stale' } }); }); it('answers 404 for any other path or method', ...); it('treats a throwing check as 503 database', ...); ``` - [ ] **Schritt 2:** `npx jest src/worker src/common/with-timeout.spec.ts src/health` → FAIL. - [ ] **Schritt 3: Implementieren** - `PollingLoop` mit `setTimeout`, kein `setInterval`, damit der Wächter aus Task 17 greift. - Single-Flight mit Flag `rerunRequested`. - Kanalnamen in `LISTEN` mit doppelten Anführungszeichen; Werte kommen nur aus `WAKE_CHANNELS`. - Health-Server mit `node:http`, ohne Express. - [ ] **Schritt 4:** → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - `npx jest --config ./test/jest-e2e.json test/health.e2e-spec.ts --runInBand` → PASS - kein Commit --- ### Task 6: Worker-Prozess und Einstiegspunkte **Files** - Create: - `src/worker/worker-runtime.service.ts` (+ `.spec.ts`) - `src/worker/worker-runtime.module.ts` - `src/worker/worker-health-server.service.ts` - `src/worker/worker.module.ts` - `src/all-in-one.module.ts` - `src/runtime/root-module.ts` (+ `.spec.ts`) - `src/worker.ts` - `test/worker-runtime.e2e-spec.ts` - Modify: - `src/main.ts`: Modul je Rolle - `src/calendar-sync/calendar-sync.service.ts`: `@Cron` `scheduledRun`, `startRun`, `activeRun` und `onModuleDestroy` entfernen; neu `requestRun` - `src/calendar-sync/calendar-sync.controller.ts` - `src/calendar-sync/calendar-sync.service.spec.ts` - `test/calendar-sync.e2e-spec.ts`: 202-Test - `package.json`: Skripte `"worker": "node dist/worker.js"`, `"job": "node dist/cli.js"` - Docs: `ARCHITECTURE.md` §10, Unterabschnitt „Laufrollen“ (Tabelle aus Spec §2). Annahme 9: Outbox-Takt im Worker statt `@Cron`, „Jetzt synchronisieren“ weckt den Worker. **Interfaces** - Consumes: Task 1, 4, 5. - Produces: ```ts export const DOMAIN_EVENT_POLL_MS = 5_000; export const CALENDAR_OUTBOX_POLL_MS = 60_000; @Injectable() class WorkerRuntime implements OnApplicationBootstrap, OnApplicationShutdown { health(): Promise } @Module({ providers: [WorkerRuntime], exports: [WorkerRuntime] }) class WorkerRuntimeModule {} // WorkerModule imports: ConfigModule.forRoot({ isGlobal: true }), PrismaModule, AuditModule, NotificationsModule, // DomainEventsModule, WebhooksModule, CalendarSyncModule, TasksModule, AutomationModule, WorkerRuntimeModule // WorkerModule providers: WorkerHealthServer (nur hier, nicht in AllInOneModule) @Module({ imports: [AppModule, WorkerRuntimeModule] }) class AllInOneModule {} export function rootModuleFor(role: 'api' | 'all'): Type; CalendarSyncService.requestRun(): Promise; // FAILED -> PENDING, notifyWorker(prisma, 'calendar_sync') // getStatus().running = Anzahl gültiger Claims > 0 ``` - Schleifen in `WorkerRuntime` (Tabelle im Code, Task 7 ergänzt eine Zeile): - `domain-events`: 5 000 ms, `domain_event`, `releaseStaleClaims()` und dann `drain()` - `calendar-outbox`: 60 000 ms, `calendar_sync`, `processOutbox()` - [ ] **Schritt 1: Failing Tests** ```ts // root-module.spec.ts – „API-Rolle startet nie Hintergrundschleifen“ it('the api module never imports the worker runtime', () => expect(Reflect.getMetadata('imports', AppModule)).not.toContain(WorkerRuntimeModule)); it('all adds the worker runtime to the API', () => expect(rootModuleFor('all')).toBe(AllInOneModule)); // worker-runtime.service.spec.ts it('starts the listener and every loop at bootstrap', ...); it('wakes the domain-events loop on a domain_event notification', ...); it('is healthy after a recent tick and dispatcher-stale after 120 s without one', ...); it('reports database when SELECT 1 fails or exceeds 5 s', ...); it('stops loops and listener on shutdown', ...); // calendar-sync.service.spec.ts it('requestRun resets FAILED entries and wakes the worker without calling Graph', async () => { await service.requestRun(); expect(graph.createEvent).not.toHaveBeenCalled(); expect(prisma.$executeRaw).toHaveBeenCalled(); // pg_notify('calendar_sync') }); // test/worker-runtime.e2e-spec.ts (WorkerModule + CountingHandler auf 'TaskDue') it('wakes up on NOTIFY and processes a new event well before the 5 s poll', async () => { await recordTaskDueInTx(); await expectWithin(1_500, () => countAuditRows('DispatcherE2E'), 1); }); // test/calendar-sync.e2e-spec.ts: „Jetzt synchronisieren“ -> 202, FAILED wird PENDING, kein Graph-Aufruf aus der API ``` - [ ] **Schritt 2:** `npx jest src/runtime src/worker src/calendar-sync` und beide e2e-Dateien → FAIL. - [ ] **Schritt 3: Implementieren** - `worker.ts`: `resolveRole('worker')` → `assertStartupEnvironment` → `NestFactory.createApplicationContext(WorkerModule)` → `enableShutdownHooks()`. - `main.ts`: `NestFactory.create(rootModuleFor(role))`. - `WorkerHealthServer` startet in `onApplicationBootstrap` auf `workerHealthPort()` und schließt in `onApplicationShutdown`. - [ ] **Schritt 4:** → PASS. - `npm run build && ls dist/main.js dist/worker.js` - Optional, mit lokaler Dev-DB: `APICORE_ROLE=worker node dist/worker.js` und `curl -s localhost:3002/health` → `{"status":"ok"}` - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - Docs - kein Commit --- **Carry-forward (Orchestrator):** `worker.ts` ruft wie `main.ts` zuerst `resolveRole('worker')` und `assertStartupEnvironment`; Health-Check = `withTimeout(SELECT 1, 5000)` → `database`, Alter des letzten Dispatcher-Durchlaufs (`lastTickAt`, wird je Ereignis aktualisiert) ≤ `HEALTH_STALE_MS` → sonst `dispatcher-stale`, gemessen mit monotoner Uhr (`performance.now()`). ### Task 7: Webhook-Zustellung mit Wiederholung und Zustellprotokoll **Files** - Create: - `src/webhooks/webhook-delivery.worker.ts` (+ `.spec.ts`) - `src/webhooks/webhook.handlers.ts` (+ `.spec.ts`) - `src/webhooks/dto/list-webhook-deliveries.dto.ts` - `test/utils/webhooks.ts` - Modify: - `src/webhooks/webhooks.service.ts`: `enqueueDeliveries`; `dispatch` (Altpfad) nutzt sie; `deliver()` entfällt - `src/webhooks/webhook-events.ts`: + `varroa.measured`, `treatment.created`, `feeding.created` - `src/webhooks/webhooks.controller.ts`, `src/webhooks/webhooks.module.ts`, `src/worker/worker-runtime.service.ts` (Schleife) - `src/webhooks/webhooks.service.spec.ts`, `test/webhooks.e2e-spec.ts` - Frontend `ApiCore-Frontend/src/lib/webhooks/types.ts` und `labels.ts`: + `stock.changed` „Bestand einer Charge geändert“, `varroa.measured` „Varroamessung erfasst“, `treatment.created` „Behandlung erfasst“, `feeding.created` „Fütterung erfasst“ - Docs: `ARCHITECTURE.md` §9, Zeile „Ausgehende Aufrufe“: Zustelltabelle, Wiederholung, stabile Delivery-ID, nur 2xx. §10: Webhook-Zustellung. **Interfaces** - Consumes: Task 3, 4, 6. - Produces: ```ts WebhooksService.enqueueDeliveries(db: DbClient, event: WebhookEvent, payload: unknown): Promise; // + notifyWorker('webhook_delivery') WebhooksService.dispatch(event: WebhookEvent, payload: unknown): Promise; // Altpfad, runAfterCommit WebhooksService.findDeliveries(subscriptionId: string, limit: number): Promise; export interface WebhookDeliveryView { id: string; event: string; status: WebhookDeliveryStatus; attempts: number; lastStatusCode: number | null; lastError: string | null; createdAt: Date; deliveredAt: Date | null } export const WEBHOOK_DELIVERY_POLL_MS = 5_000; export const WEBHOOK_DELIVERY_CONCURRENCY = 5; @Injectable() class WebhookDeliveryWorker { runOnce(filter?: { subscriptionIds?: readonly string[] }): Promise; drain(filter?: { subscriptionIds?: readonly string[] }, maxRounds?: number): Promise; releaseStaleClaims(): Promise; } export const DOMAIN_EVENT_WEBHOOKS: Readonly>> = { InspectionRecorded: 'inspection.created', VarroaMeasured: 'varroa.measured', TreatmentRecorded: 'treatment.created', FeedingRecorded: 'feeding.created', HarvestRecorded: 'harvest.created', BatchBottled: 'batch.updated', BottlingCorrected: 'batch.updated', StockChanged: 'stock.changed', TaskCreated: 'task.created', TaskCompleted: 'task.completed', }; class WebhookEventHandlers { @OnDomainEvent(Object.keys(DOMAIN_EVENT_WEBHOOKS) as DomainEventType[], 'webhooks.forward') forward(event: DomainEventEnvelope, tx: Prisma.TransactionClient): Promise; } // GET /webhooks/:id/deliveries?limit=1..100 (Standard 50), @RequirePermission('webhooks:read'), ohne body ``` - Inhalt (`data`) wie heute: der Datensatz, geladen über `tx`. - Harvest mit `colonies: { include: { colony: true } }` - `batch.updated` → `honeyBatch` - `stock.changed` → `{ bottlingBatchIds, productIds }`, und nur wenn eines davon nicht leer ist - Fehlt der Datensatz → nichts einreihen. - `body = JSON.stringify({ event, data })` wird beim Einreihen festgehalten, damit jede Wiederholung dieselben Bytes sendet. - [ ] **Schritt 1: Failing Tests** ```ts // webhook-delivery.worker.spec.ts it('signs ".." and sends the delivery id as X-ApiCore-Delivery', ...); it('counts only 2xx as delivered and stores the status code', ...); it('schedules a retry 30 s ahead after a 500 and keeps the same id', ...); it('marks FAILED after the 8th attempt', ...); it('fails a delivery for an inactive subscription without sending ("subscription inactive")', ...); it('never stores the response body or the secret in lastError', ...); it('sends at most 5 at once', ...); // webhook.handlers.spec.ts it('enqueues inspection.created with the inspection as data', ...); it('enqueues nothing for StockChanged that only touches raw material', ...); it('skips a record that no longer exists', ...); // webhooks.service.spec.ts it('dispatch never throws when enqueuing fails (legacy path)', ...); it('enqueueDeliveries writes one row per active subscription for the event and wakes the worker', ...); // test/webhooks.e2e-spec.ts (Altpfad colony.created, die Producer folgen später) it('delivers colony.created through the queue with a stable delivery id', async () => { await post('/api/v1/colonies', body); await deliverWebhooks(app, [subId]); expect(received[0].headers['x-apicore-delivery']).toBe((await prisma.webhookDelivery.findFirstOrThrow({ where: { subscriptionId: subId } })).id); }); it('retries a 500 with the same delivery id and then marks it DELIVERED', ...); // availableAt per SQL auf now() it('lists deliveries for admins and refuses staff', ...); // 200 / 403 ``` - [ ] **Schritt 2:** `npx jest src/webhooks` und `test/webhooks.e2e-spec.ts` → FAIL. - [ ] **Schritt 3: Implementieren** - Claims und Settle über `outbox-claims.ts` (`orderBy: 'createdAt'`). - `postWebhook` bleibt. - Worker-Schleife `webhook-deliveries`: 5 000 ms, Kanal `webhook_delivery`, `releaseStaleClaims()` und dann `drain()`. - `test/utils/webhooks.ts`: `deliverWebhooks(app, subscriptionIds)`. - [ ] **Schritt 4:** → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - Frontend: `npm run lint && npm run typecheck` - Docs - kein Commit --- **Carry-forward (Orchestrator, aus Review Task 6):** Ein Test stellt sicher, dass die unter `WorkerModule` entdeckten Handler-IDs denen unter `AppModule` entsprechen (sonst läuft ein Handler, der in einem nicht vom Worker importierten Modul liegt, in Produktion nie). Neue Schleifen nur als Zeile in der Loop-Tabelle von `WorkerRuntime`; sie respektieren das Stop-Signal aus `onModuleDestroy`. ### Task 8: Kalender-Handler und „Quelle weg“-Semantik **Files** - Create: `src/calendar-sync/calendar-sync.handlers.ts` (+ `.spec.ts`) - Modify: - `src/calendar-sync/calendar-sync.service.ts`: - `enqueueIn` als Kern; `enqueue` als Altpfad für Königinnen - `cancel` delegiert vorläufig an `enqueue`; Task 9 entfernt es - `buildEventInput('TASK')` liefert `null` bei `DONE`/`CANCELLED` oder ohne `dueDate` - `processEntry`: bei `null` und vorhandenem `graphEventId` das Graph-Event löschen, dann `SKIPPED` - `src/calendar-sync/calendar-sync.module.ts`, `src/calendar-sync/calendar-sync.service.spec.ts`, `test/calendar-sync.e2e-spec.ts` - Docs: `ARCHITECTURE.md` Annahme 9: Abmelden über erneutes Einreihen, der Worker löscht das Event. **Interfaces** - Consumes: Task 3, 6. - Produces: ```ts CalendarSyncService.enqueueIn(db: DbClient, sourceType: CalendarSyncSourceType, sourceId: string): Promise; // wirft weiter CalendarSyncService.enqueue(sourceType: CalendarSyncSourceType, sourceId: string): Promise; // Altpfad, wirft nie class CalendarSyncEventHandlers { @OnDomainEvent(['InspectionRecorded', 'InspectionCorrected'], 'calendar-sync.inspection') inspection(e, tx): Promise; @OnDomainEvent(['TreatmentRecorded', 'TreatmentCorrected'], 'calendar-sync.treatment') treatment(e, tx): Promise; @OnDomainEvent(['FeedingRecorded', 'FeedingCorrected'], 'calendar-sync.feeding') feeding(e, tx): Promise; @OnDomainEvent(['HarvestRecorded', 'HarvestCorrected'], 'calendar-sync.harvest') harvest(e, tx): Promise; @OnDomainEvent(['TaskCreated', 'TaskUpdated', 'TaskCompleted'], 'calendar-sync.task') task(e, tx): Promise; // TaskCreated ohne dueDate: nichts } ``` - [ ] **Schritt 1: Failing Tests** ```ts it('enqueueIn writes through the given client and lets errors propagate (handler retry)', ...); it('enqueue (legacy) never throws', ...); it('deletes the Graph event of a task that is DONE and settles the entry SKIPPED', ...); it('treats 404/410 on delete as already deleted', ...); it('counts any other delete error as a failed attempt', ...); it('TaskCreated without a due date enqueues nothing', ...); // handlers.spec // e2e: Aufgabe mit Event erledigen -> Eintrag neu eingereiht -> processOutbox() ruft deleteEvent und setzt SKIPPED ``` - [ ] **Schritt 2:** `npx jest src/calendar-sync` und `test/calendar-sync.e2e-spec.ts` → FAIL. - [ ] **Schritt 3: Implementieren** wie in Files beschrieben. - [ ] **Schritt 4:** → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - Docs - kein Commit --- ### Task 9: Aufgaben über TasksService mit Ereignissen **Files** - Modify: - `src/tasks/tasks.service.ts` (+ `.spec.ts`) - `src/tasks/tasks.module.ts`: Import `TaskTemplatesModule` entfällt - `src/task-templates/task-templates.service.ts` (+ `.spec.ts`): `generateDueTasks` und die `WebhooksService`-Abhängigkeit entfallen - `src/inspections/inspections.service.ts`: `createTaskFromInspection` über `createInTx` - `src/inspections/inspections.module.ts`: `TasksModule` importieren - `src/calendar-sync/calendar-sync.service.ts`: `cancel` entfernen - Test: - `test/tasks.e2e-spec.ts`, `test/task-templates.e2e-spec.ts` - `test/data-integrity.e2e-spec.ts` („gives a task generated from a recurring template its due date“) - `test/webhooks.e2e-spec.ts`: `task.created`/`task.completed` über Drain + Zustellung **Interfaces** - Consumes: Task 3, 7, 8. - Produces: ```ts export interface NewTaskInput { title: string; description?: string | null; priority?: TaskPriority; dueDate?: Date | null; assignedUserId?: string | null; relatedType?: TaskRelatedType; relatedId?: string | null; sourceInspectionId?: string | null; sourceTemplateId?: string | null; sourceAutomationTrigger?: AutomationTriggerType | null; } TasksService.createInTx(tx: Prisma.TransactionClient, input: NewTaskInput, actingUserId: string | null): Promise; // Task + Audit + TaskCreated TasksService.create(dto: CreateTaskDto, actingUserId: string): Promise; TasksService.update(id: string, dto: UpdateTaskDto, actingUserId: string): Promise; // + TaskUpdated TasksService.complete(id: string, actingUserId: string): Promise; // + TaskCompleted, bei Wiederholung nichts TasksService.generateFromTemplates(now?: Date): Promise; // je Vorlage: CAS-Claim + createInTx in EINER Transaktion ``` - [ ] **Schritt 1: Failing Tests** ```ts it('createInTx writes task, audit entry and TaskCreated on the same client', async () => { await service.createInTx(tx, { title: 'T', dueDate: new Date('2027-05-01T00:00:00Z') }, 'u1'); expect(audit.record).toHaveBeenCalledWith(expect.objectContaining({ entityType: 'Task', action: 'CREATE' }), tx); expect(events.record).toHaveBeenCalledWith(tx, expect.objectContaining({ type: 'TaskCreated', payload: { relatedType: 'GENERAL', relatedId: null, dueDate: '2027-05-01' } })); }); it('complete records TaskCompleted once and nothing on a repeated complete', ...); it('generateFromTemplates creates no task when another replica already claimed the occurrence', ...); it('a template task gets an audit entry without a user', ...); // e2e it('POST /tasks leaves a TaskCreated event whose calendar and webhook deliveries are DONE after draining', ...); ``` - [ ] **Schritt 2:** `npx jest src/tasks src/task-templates src/inspections` und die e2e-Dateien → FAIL. - [ ] **Schritt 3: Implementieren** - `findAll` ruft weiter lazy `generateFromTemplates()` auf (Spec §4). - `createdByUserId: actingUserId ?? undefined`. - [ ] **Schritt 4:** → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - kein Commit --- **Carry-forward (Orchestrator, aus Review Task 8):** Sobald die direkten `calendarSync.enqueue`/`cancel`-Aufrufe entfernt sind, deckt ein e2e-Fall den echten Weg ab: Aufgabe erledigen → `TaskCompleted` → Dispatcher → Handler `calendar-sync.task` → Outbox-Eintrag → `processOutbox()` löscht das Graph-Event und setzt `SKIPPED`. ### Task 10: Automationen als Handler **Files** - Create: `src/automation/automation.handlers.ts` (+ `.spec.ts`) - Modify: - `src/automation/automation.service.ts` (+ `.spec.ts`) - `src/automation/automation.module.ts`: importiert `TasksModule`, `NotificationsModule`, `StockModule`; `WebhooksService` entfällt - Docs: `ARCHITECTURE.md` Annahme 12: Automatik-Aufgaben laufen über `TasksService` mit Audit, Kalender und Webhook. **Interfaces** - Consumes: Task 3, 9. - Produces: ```ts AutomationService.evaluateInspectionIn(tx: Prisma.TransactionClient, inspection: Inspection): Promise; AutomationService.evaluateVarroaMeasurementIn(tx, measurement: VarroaMeasurement): Promise; AutomationService.evaluateTreatmentIn(tx, treatment: Treatment): Promise; AutomationService.evaluateFeedingIn(tx, feeding: Feeding): Promise; AutomationService.evaluateColonyHealthIn(tx, colonyId: string, from: ColonyStateSnapshot, to: ColonyStateSnapshot): Promise; AutomationService.evaluateRawMaterialLowIn(tx, kind: RawMaterialKind, remainingGrams: number): Promise; AutomationService.evaluateSwarmControlCheckIn(tx, check: SwarmControlCheck): Promise; AutomationService.evaluateStockItemIn(tx, item: StockItem): Promise; // Altpfade bleiben bis Task 17: evaluateInspection, evaluateVarroaMeasurement, evaluateTreatment, evaluateFeeding, // evaluateColonyHealth, evaluateRawMaterialLow, evaluateSwarmControlCheck, evaluateStockItem // = runAfterCommit(() => prisma.$transaction((tx) => xIn(tx, …))) // privat: triggerActionIn(tx, rule, relatedType, relatedId, sourceInspectionId?, options?) – Sperre auf die Regel, // Dublettenprüfung, tasks.createInTx(tx, …, null), notifications.record(…, tx), auditLog.record(AutomationRule CREATE, tx) class AutomationEventHandlers { @OnDomainEvent('InspectionRecorded', 'automation.inspection-rules') @OnDomainEvent(['VarroaMeasured', 'VarroaCorrected'], 'automation.varroa-threshold') @OnDomainEvent('TreatmentRecorded', 'automation.treatment-follow-up') @OnDomainEvent('FeedingRecorded', 'automation.feeding-follow-up') @OnDomainEvent('ColonyStatusChanged', 'automation.colony-health') @OnDomainEvent('StockChanged', 'automation.raw-material-low') // je rawMaterialKind: StockService.remainingByRawMaterialKind } ``` - [ ] **Schritt 1: Failing Tests** ```ts it('triggerActionIn creates the task through TasksService.createInTx on the handler transaction', ...); it('leaves no task, notification or audit entry when the notification write fails (one rollback)', ...); it('still creates at most one open task per rule and object', ...); it('handlers load the record via tx and skip a missing one', ...); it('colony-health fires only on the transition into CRITICAL', ...); it('legacy wrappers never throw', ...); ``` - [ ] **Schritt 2:** `npx jest src/automation` → FAIL. - [ ] **Schritt 3: Implementieren** wie beschrieben. - Gemeinsamer Kern: Altpfade und Handler rufen dieselben `…In`-Funktionen auf. - `ensureDefaults` läuft über den übergebenen Client. - [ ] **Schritt 4:** - `npx jest src/automation` → PASS - `npx jest --config ./test/jest-e2e.json test/automation.e2e-spec.ts test/follow-up-controls.e2e-spec.ts --runInBand` → PASS (Altpfade noch aktiv) - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - Docs - kein Commit --- **Carry-forward (Orchestrator, aus T3 Task 2):** Der VARROA_THRESHOLD-Handler nutzt `varroaThresholdOf` aus `src/automation/varroa-threshold.ts` (einzige Quelle für den Standardwert 10); das inline `?? 10` in `automation.service.ts` entfällt. Das ist T3-Plan Task 2 Schritt 9. ### Task 11: Erzeuger Durchsichten und Varroa **Files** - Modify: - `src/inspections/inspections.service.ts` (+ `.spec.ts`), `src/inspections/inspections.module.ts` (`AutomationModule`-Import entfällt) - `src/varroa/varroa.service.ts` (+ `.spec.ts`), `src/varroa/varroa.module.ts` - Test: `test/post-commit-side-effects.e2e-spec.ts` (alleiniger Besitzer dieser Datei), `test/automation.e2e-spec.ts`, `test/inspections.e2e-spec.ts`, `test/varroa.e2e-spec.ts` **Interfaces** - Consumes: Task 3, 10. - Produces, je Schreibmethode `prisma.$transaction(async (tx) => { write; auditLog.record(…, tx); events.record(tx, …) })`: - `InspectionsService.create` → `InspectionRecorded { colonyId }` - `InspectionsService.update` → `InspectionCorrected { colonyId }`. Bewusst wie heute ohne Automatik: eine Korrektur löst keine Regel aus. - `VarroaService.create` → `VarroaMeasured { colonyId }`, `update` → `VarroaCorrected { colonyId }` - Direkte Aufrufe `webhooks.dispatch`, `automation.evaluate*` und `calendarSync.enqueue` entfallen. - [ ] **Schritt 1: Failing Tests** ```ts // unit it('create writes inspection, audit entry and InspectionRecorded in one transaction and calls no side effect directly', ...); // test/post-commit-side-effects.e2e-spec.ts (Review Focus 2, Spec §9) it('creates exactly one automation task when the handler fails once and is retried', async () => { jest.spyOn(app.get(NotificationsService), 'record').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); await drainDomainEvents(app, [body.id]); expect(await taskCount('QUEENLESS_NO_EGGS', colonyId)).toBe(0); await prisma.$executeRaw`UPDATE "DomainEvent" SET "availableAt" = now() WHERE "aggregateId" = ${body.id}`; await drainDomainEvents(app, [body.id]); expect(await taskCount('QUEENLESS_NO_EGGS', colonyId)).toBe(1); 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 }); expect(deliveries.find((d) => d.handler === 'webhooks.forward')).toMatchObject({ status: 'DONE', attempts: 1 }); }); it('still answers 201 and replays on retry with the same Idempotency-Key (no side effect in the request)', ...); it('leaves one open task when several triggers for the same colony arrive at once', ...); // nach Drain ``` - [ ] **Schritt 2:** → FAIL. - [ ] **Schritt 3: Implementieren** - `automation.e2e`: die Assertions auf Aufgaben und Benachrichtigungen kommen nach `drainDomainEvents`. - [ ] **Schritt 4:** `npx jest src/inspections src/varroa` und die vier e2e-Dateien → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - kein Commit --- ### Task 12: Erzeuger Behandlungen, Fütterungen und Ernten **Files** - Modify: `src/treatments/treatments.service.ts`, `src/feedings/feedings.service.ts`, `src/harvests/harvests.service.ts` mit Specs und Modulen - Test: `test/follow-up-controls.e2e-spec.ts`, `test/treatments.e2e-spec.ts`, `test/feedings.e2e-spec.ts`, `test/harvests.e2e-spec.ts` **Interfaces** - Consumes: Task 3, 8, 10. - Produces: - `TreatmentRecorded`/`TreatmentCorrected { colonyId }` - `FeedingRecorded`/`FeedingCorrected { colonyId }` - `HarvestRecorded`/`HarvestCorrected { apiaryId, colonyIds }` - jeweils mit Audit in der Transaktion - Webhook `harvest.created` und der Kalender kommen jetzt aus den Handlern. `HarvestsService.create` braucht kein `findById` mehr für den Webhook. - [ ] **Schritt 1: Failing Tests** ```ts it('create writes treatment, audit entry and TreatmentRecorded in one transaction', ...); it('update records TreatmentCorrected and no automation', ...); it('harvest create records HarvestRecorded with the colony ids', ...); // follow-up-controls.e2e: Folgekontrolle erst nach drainDomainEvents, Fälligkeit wie bisher ab Maßnahme-Datum ``` - [ ] **Schritt 2:** → FAIL. - [ ] **Schritt 3: Implementieren.** - [ ] **Schritt 4:** `npx jest src/treatments src/feedings src/harvests` und die e2e-Dateien → PASS. - `test/post-commit-side-effects.e2e-spec.ts` bleibt ohne Änderung grün; der Behandlungs-Test prüft nur 201 und Replay. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - kein Commit --- ### Task 13: Erzeuger Völker und Überwinterung **Files** - Modify: - `src/colonies/colonies.service.ts` (+ `.spec.ts`), `src/colonies/colonies.module.ts` - `src/winter-checks/winter-checks.service.ts` (+ `.spec.ts`) - Test: `test/colonies.e2e-spec.ts`, `test/winter-checks.e2e-spec.ts` **Interfaces** - Consumes: Task 3, 10. - Produces: - `update`, `merge` (für jedes vereinigte Volk), `dissolve`, `reactivate` und `markDeceasedIfRequested` schreiben `ColonyStatusChanged { from, to }`, sobald sich `status` oder `healthStatus` ändert. - Beim Übergang von lebend zu beendet (`DISSOLVED`, `SOLD`, `DECEASED`, `MERGED`, siehe `isLivingColonyStatus`) zusätzlich `ColonyEnded { status }`. - Audit überall in der Transaktion, auch bei `create`, `move` und `split`. - `colony.created` bleibt im Altpfad `webhooks.dispatch`; dafür gibt es in dieser Stufe kein Ereignis. - [ ] **Schritt 1: Failing Tests** ```ts it('update into CRITICAL health records ColonyStatusChanged with from/to snapshots', ...); it('dissolve records ColonyStatusChanged and ColonyEnded { status: "DISSOLVED" }', ...); it('merge records ColonyEnded "MERGED" for every merged colony', ...); it('a winter loss with markColonyDeceased records ColonyEnded "DECEASED" in the same transaction', ...); // e2e: COLONY_HEALTH_CRITICAL-Aufgabe erst nach drainDomainEvents ``` - [ ] **Schritt 2:** → FAIL. - [ ] **Schritt 3: Implementieren.** - [ ] **Schritt 4:** → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - kein Commit --- ### Task 14: Erzeuger Bestand und Abfüllung (mit Herstellung und Rohstoffen) **Files** - Modify: - `src/stock/stock.service.ts` (+ `.spec.ts`): `notify()` entfällt, neu `recordChange`. `consumeForProduction` und `returnProductionInputs` geben die verbrauchten Sorten zurück. - `src/bottling/bottling.service.ts` (+ `.spec.ts`) - `src/production-batches/production-batches.service.ts` (+ `.spec.ts`): `checkRawMaterialLow` entfällt - `src/raw-materials/raw-materials.service.ts` (+ `.spec.ts`): `checkLowStock` entfällt - die zugehörigen Module (`AutomationModule`- und `WebhooksService`-Abhängigkeiten entfallen) - Test: `test/stock.e2e-spec.ts`, `test/bottling.e2e-spec.ts`, `test/shop-sync.e2e-spec.ts`, `test/raw-materials.e2e-spec.ts`, `test/data-integrity.e2e-spec.ts` (Abschnitt „Bottling quantities“ unverändert grün) **Interfaces** - Consumes: Task 3, 7, 10. - Produces: ```ts export interface StockChange { bottlingBatchIds: string[]; productIds: string[]; rawMaterialKinds: RawMaterialKind[] } StockService.recordChange(tx: Prisma.TransactionClient, change: StockChange, actingUserId: string | null, aggregate: { type: 'BatchMovement' | 'BottlingBatch' | 'ProductionBatch' | 'ShopStockRequest'; id: string }): Promise; // nichts bei leerer Änderung StockService.consumeForProduction(tx, productionBatchId, lotNumber, inputs, actingUserId): Promise; StockService.returnProductionInputs(tx, productionBatchId, lotNumber, actingUserId): Promise; ``` - Wer welches Ereignis schreibt: - `withdraw` und `correct` (inkl. Rohstoff-Ziel mit der Sorte der Charge) → `StockChanged` in ihrer Transaktion. - Shop-Buchung → ein `StockChanged` je angenommener Anfrage, in deren Transaktion. - Abfüllung: `create` → `BatchBottled { honeyBatchId }` + `StockChanged`; `update`/`remove` → `BottlingCorrected { honeyBatchId, removed }` + `StockChanged`. - Herstellung `create`/`update`/`remove` → `StockChanged` mit `productIds` und den verbrauchten Sorten. - [ ] **Schritt 1: Failing Tests** ```ts it('withdraw books the movement, the audit entry and StockChanged in one transaction', ...); it('a raw material withdrawal records StockChanged with the lot kind and no shop ids', ...); it('a shop request records one StockChanged per accepted booking', ...); it('bottling create records BatchBottled and StockChanged together', ...); // raw-materials.e2e: RAW_MATERIAL_LOW-Aufgabe erst nach drainDomainEvents (minGrams gesetzt) ``` - [ ] **Schritt 2:** → FAIL. - [ ] **Schritt 3: Implementieren.** - [ ] **Schritt 4:** `npx jest src/stock src/bottling src/production-batches src/raw-materials` und die e2e-Dateien → PASS. - [ ] **Schritt 5: Verifikation** - typecheck, lint, `npm test -- --ci` - kein Commit --- ### Task 15: Job-Grundgerüst (JobRun, cli.ts, cleanup, task-templates) **Files** - Create: - `src/system/system.module.ts` - `src/system/job-run.service.ts` (+ `.spec.ts`) - `src/jobs/job-names.ts` - `src/jobs/job.ts` - `src/jobs/job-registry.ts` (+ `.spec.ts`) - `src/jobs/jobs.module.ts` - `src/jobs/run-cli.ts` (+ `.spec.ts`) - `src/jobs/cleanup.job.ts` (+ `.spec.ts`) - `src/jobs/task-templates.job.ts` - Platzhalter `src/jobs/daily-evaluation.job.ts`, `src/jobs/app-backup.job.ts`, `src/jobs/rekey.job.ts` (werfen `Error('job not implemented yet')`, werden in 16/18/21 ersetzt) - `src/cli.ts` - `test/jobs.e2e-spec.ts` - Modify: - `src/common/idempotency/idempotency-cleanup.service.ts` (+ `.spec.ts`): `@Cron` und `scheduledCleanup` entfallen - `src/app.module.ts`: `IdempotencyCleanupService`-Provider entfällt, `SystemModule` wird importiert - Docs: `ARCHITECTURE.md` §8 „Automatischer Zeitplan“ (Idempotenz-Aufräumen per CronJob) und §10 „Jobs“. **Interfaces** - Consumes: Task 2, 3, 9. - Produces: ```ts export const JOB_NAMES = ['daily-evaluation', 'task-templates', 'cleanup', 'app-backup', 'rekey'] as const; export type JobName = (typeof JOB_NAMES)[number]; export const JOB_LABELS_DE: Record = { 'daily-evaluation': 'Tägliche Auswertung', 'task-templates': 'Aufgaben aus Vorlagen', cleanup: 'Aufräumen', 'app-backup': 'App-Backup', rekey: 'Schlüssel umstellen' }; export const JOB_FAILURE_PERMISSION: Record = { 'daily-evaluation': 'automation:update', 'task-templates': 'automation:update', cleanup: 'system:manage', 'app-backup': 'backup:configure', rekey: 'system:manage' }; export type JobResult = Record; export interface Job { readonly name: JobName; run(args: readonly string[]): Promise } @Injectable() class JobRunService { run(job: JobName, work: () => Promise): Promise; // RUNNING -> SUCCEEDED|FAILED; bei FAILED: recordForPermission(…, 'Job „