From 76bc0e4269bcccc7202228ba16852b2b521d1b5b Mon Sep 17 00:00:00 2001 From: Florian Wilde Date: Mon, 12 Oct 2026 10:54:41 +0200 Subject: [PATCH] feat(auftraege): worker, Phasen und Live-Fortschritt per SSE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Aufträge laufen als pg-boss-Jobs im worker. Der Phasen-Runner der Engine kennt weder Proxmox noch Datenbank; eine Senke schreibt Phasen und Log in die DB und meldet per NOTIFY, die app streamt per SSE an Web und CLI (Last-Event-ID, Herzschlag, Polling-Fallback im Browser). VMIDs vergibt eine Advisory-Sperre je Server zusammen mit einer Reservierungstabelle; zehn parallele Reservierungen im Test bekommen zehn verschiedene IDs. Der Proxmox-Client hat einen Trockenmodus für den Dry-Run: Lesendes echt, Schreibendes nur aufgezeichnet. Live-Ansicht im Aufbau der CLI-Ausgabe. Im echten Betrieb aufgefallen: ssh2 und pg sind CommonJS und brauchen im worker Default-Importe. Closes #006 Co-Authored-By: Claude Opus 5.5 --- docs/issues/006-auftraege-sse.md | 27 +- docs/issues/README.md | 2 +- package.json | 5 +- pnpm-lock.yaml | 91 +++++- .../20261012082805_auftraege/migration.sql | 118 ++++++++ .../20261012084349_templates/migration.sql | 24 ++ prisma/schema.prisma | 138 +++++++++ .../(app)/auftraege/[kurzId]/live-ansicht.tsx | 194 +++++++++++++ src/app/(app)/auftraege/[kurzId]/page.tsx | 34 +++ src/app/(app)/auftraege/page.tsx | 55 ++++ .../api/v1/auftraege/[id]/abbrechen/route.ts | 13 + .../api/v1/auftraege/[id]/ereignisse/route.ts | 101 +++++++ src/app/api/v1/auftraege/[id]/log/route.ts | 14 + src/app/api/v1/auftraege/[id]/route.ts | 11 + src/app/api/v1/auftraege/route.ts | 10 + src/components/terminal-karte.tsx | 179 ++++++++++++ src/engine/ablauf/ablauf.ts | 86 ++++++ src/engine/ablauf/fehler.ts | 46 +++ src/engine/proxmox/aufgaben.ts | 4 +- src/engine/proxmox/client.ts | 30 +- src/engine/ssh/verbindung.ts | 5 +- src/lib/auftrag-text.ts | 18 ++ src/lib/navigation.ts | 2 +- src/lib/schemas/auftrag.ts | 35 +++ src/server/ereignisse/hoeren.ts | 49 ++++ src/server/ereignisse/senke.ts | 83 ++++++ src/server/services/auftraege.ts | 191 +++++++++++++ src/server/services/server.ts | 4 +- src/server/services/vmid.ts | 72 +++++ src/server/warteschlange.ts | 43 +++ src/worker/ablaeufe.ts | 4 + src/worker/aufruf-log.ts | 20 ++ src/worker/ausfuehren.ts | 127 ++++++++ src/worker/main.ts | 39 +++ src/worker/umgebung.ts | 11 + tests/auftraege.test.ts | 270 ++++++++++++++++++ tests/engine/ablauf.test.ts | 130 +++++++++ tests/zugang.test.ts | 11 + 38 files changed, 2274 insertions(+), 22 deletions(-) create mode 100644 prisma/migrations/20261012082805_auftraege/migration.sql create mode 100644 prisma/migrations/20261012084349_templates/migration.sql create mode 100644 src/app/(app)/auftraege/[kurzId]/live-ansicht.tsx create mode 100644 src/app/(app)/auftraege/[kurzId]/page.tsx create mode 100644 src/app/(app)/auftraege/page.tsx create mode 100644 src/app/api/v1/auftraege/[id]/abbrechen/route.ts create mode 100644 src/app/api/v1/auftraege/[id]/ereignisse/route.ts create mode 100644 src/app/api/v1/auftraege/[id]/log/route.ts create mode 100644 src/app/api/v1/auftraege/[id]/route.ts create mode 100644 src/app/api/v1/auftraege/route.ts create mode 100644 src/components/terminal-karte.tsx create mode 100644 src/engine/ablauf/ablauf.ts create mode 100644 src/engine/ablauf/fehler.ts create mode 100644 src/lib/auftrag-text.ts create mode 100644 src/lib/schemas/auftrag.ts create mode 100644 src/server/ereignisse/hoeren.ts create mode 100644 src/server/ereignisse/senke.ts create mode 100644 src/server/services/auftraege.ts create mode 100644 src/server/services/vmid.ts create mode 100644 src/server/warteschlange.ts create mode 100644 src/worker/ablaeufe.ts create mode 100644 src/worker/aufruf-log.ts create mode 100644 src/worker/ausfuehren.ts create mode 100644 src/worker/main.ts create mode 100644 src/worker/umgebung.ts create mode 100644 tests/auftraege.test.ts create mode 100644 tests/engine/ablauf.test.ts diff --git a/docs/issues/006-auftraege-sse.md b/docs/issues/006-auftraege-sse.md index 6c09f63..41ed31f 100644 --- a/docs/issues/006-auftraege-sse.md +++ b/docs/issues/006-auftraege-sse.md @@ -1,25 +1,32 @@ # #006 Aufträge, Phasen, worker und Live-Fortschritt -**Status:** offen +**Status:** erledigt **Meilenstein:** v0.2.0 **Labels:** feature, engine -**Geschätzt:** 8 h **Tatsächlich:** – +**Geschätzt:** 8 h **Tatsächlich:** 0 h 15 min ## Ziel Hintergrund-Jobs mit Phasen in der DB, live per SSE in der Web-GUI. ## Aufgaben -- [ ] pg-boss im worker, Modelle `Auftrag`, `Phase`, `Logeintrag`, `VmidReservierung` -- [ ] Phasen-Runner mit Ereignis-Senke (DB + NOTIFY), Fehler mit Ursache + Hinweis -- [ ] SSE-Endpunkt mit Last-Event-ID, Polling-Fallback -- [ ] VMID-Vergabe mit Advisory-Lock, Test mit parallelen Aufträgen -- [ ] Live-Ansicht als Terminal-Karte, Detail-Log, Ergebnis-Karte +- [x] pg-boss im worker, Modelle `Auftrag`, `Phase`, `Logeintrag`, `VmidReservierung` +- [x] Phasen-Runner (frameworkfrei) mit Ereignis-Senke (DB + NOTIFY), Fehler mit Ursache + Hinweis, Abbruch zwischen Phasen +- [x] SSE-Endpunkt mit Last-Event-ID, Herzschlag, `X-Accel-Buffering: no`; Polling-Fallback im Client +- [x] VMID-Vergabe mit Advisory-Lock, Test mit zehn parallelen Reservierungen +- [x] Live-Ansicht als Terminal-Karte (Aufbau der CLI-Ausgabe), Detail-Log, Ergebnis-Box; Verlaufsseite +- [x] Verwaiste Aufträge nach worker-Neustart als fehlgeschlagen markieren; doppelt gelieferte Jobs laufen nur einmal ## Akzeptanzkriterien -- [ ] Zwei parallele Aufträge bekommen nie dieselbe VMID (Test) -- [ ] Neuladen der Seite mitten im Auftrag zeigt den vollständigen Stand +- [x] Zwei parallele Aufträge bekommen nie dieselbe VMID (Test) +- [x] Neuladen der Seite mitten im Auftrag zeigt den vollständigen Stand (Stand kommt immer ganz, Log ab Last-Event-ID) +- [x] Live geprüft mit einem echten Template-Bau auf pve-demo (#007) ## Abhängigkeiten #003 ## Notizen / Entscheidungen -– +- **ssh2 und pg sind CommonJS:** benannte Importe gehen im Next-Bundle, aber nicht im worker unter reinem Node-ESM – jetzt Default-Importe. +- **Der worker lädt `.env` selbst** (`process.loadEnvFile`, falls vorhanden); im Container kommt die Umgebung von außen. +- **Kein automatisches Wiederholen** (`retryLimit: 0`): halb angelegte Gäste soll ein Mensch ansehen. +- **Abfrageschleifen** (Task-Status, VM-Status) erscheinen im Detail-Log nur bei Fehlern. +- **Kopfzeile** (`Auftrag.kopf`) setzt die Engine – Web und CLI zeigen denselben Text. +- Client-Komponenten importieren nichts aus `src/server` (struktureller Test in `tests/zugang.test.ts`). diff --git a/docs/issues/README.md b/docs/issues/README.md index bf12b2b..f41e528 100644 --- a/docs/issues/README.md +++ b/docs/issues/README.md @@ -9,7 +9,7 @@ Details und Architektur: [../architektur.md](../architektur.md). Zeiten werden a | [#003](003-verbindungstest.md) | Verbindungstest und Fähigkeitserkennung | erledigt | v0.1.0 | 6 h | 0 h 25 min | | [#004](004-katalog.md) | Katalog: Distributionen und Profile als Daten | erledigt | v0.2.0 | 5 h | 0 h 09 min | | [#005](005-adapter-profile.md) | Adapter debian und Profil-Engine | erledigt | v0.2.0 | 6 h | 0 h 20 min | -| [#006](006-auftraege-sse.md) | Aufträge, Phasen, worker und Live-Fortschritt | offen | v0.2.0 | 8 h | – | +| [#006](006-auftraege-sse.md) | Aufträge, Phasen, worker und Live-Fortschritt | erledigt | v0.2.0 | 8 h | 0 h 15 min | | [#007](007-template-bau.md) | Template-Bau für VMs | offen | v0.2.0 | 8 h | – | | [#008](008-vm-ausrollen.md) | Deploy-Assistent und VM ausrollen | offen | v0.2.0 | 10 h | – | | [#009](009-lxc.md) | LXC-Container ausrollen | offen | v0.3.0 | 8 h | – | diff --git a/package.json b/package.json index 53ee96b..7e6b05e 100644 --- a/package.json +++ b/package.json @@ -24,7 +24,9 @@ "test": "vitest run", "test:watch": "vitest", "test:e2e": "playwright test", - "typecheck": "tsc --noEmit" + "test:container": "tests/container/profile-pruefen.sh", + "typecheck": "tsc --noEmit", + "worker": "tsx src/worker/main.ts" }, "dependencies": { "@fontsource-variable/jetbrains-mono": "^5.3.0", @@ -37,6 +39,7 @@ "lucide-react": "1.47.0", "next": "16.4.0", "pg": "^8.23.0", + "pg-boss": "12.37.1", "pino": "10.4.0", "react": "19.3.0", "react-dom": "19.3.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3574d7d..31c7f06 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -192,10 +192,13 @@ importers: version: 1.47.0(react@19.3.0) next: specifier: 16.4.0 - version: 16.4.0(@babel/core@7.29.7(supports-color@7.2.0))(@playwright/test@1.63.0)(@types/node@24.19.2)(react-dom@19.3.0(react@19.3.0))(react@19.3.0) + version: 16.4.0(@babel/core@7.29.7(supports-color@7.2.0))(@opentelemetry/api@1.9.1)(@playwright/test@1.63.0)(@types/node@24.19.2)(react-dom@19.3.0(react@19.3.0))(react@19.3.0) pg: specifier: ^8.23.0 version: 8.23.1 + pg-boss: + specifier: 12.37.1 + version: 12.37.1(@opentelemetry/api@1.9.1) pino: specifier: 10.4.0 version: 10.4.0 @@ -286,7 +289,7 @@ importers: version: 8.3.0(@types/node@24.19.2)(esbuild@0.27.7)(jiti@2.7.0)(tsx@4.23.15)(yaml@2.9.1) vitest: specifier: 5.0.1 - version: 5.0.1(@types/node@24.19.2)(vite@8.3.0(@types/node@24.19.2)(esbuild@0.27.7)(jiti@2.7.0)(tsx@4.23.15)(yaml@2.9.1)) + version: 5.0.1(@opentelemetry/api@1.9.1)(@types/node@24.19.2)(vite@8.3.0(@types/node@24.19.2)(esbuild@0.27.7)(jiti@2.7.0)(tsx@4.23.15)(yaml@2.9.1)) packages: @@ -1199,6 +1202,10 @@ packages: resolution: {integrity: sha512-nn5ozdjYQpUCZlWGuxcJY/KpxkWQs4DcbMCmKojjyrYDEAGy4Ce19NN4v5MduafTwJlbKc99UA8YhSVqq9yPZA==} engines: {node: '>=12.4.0'} + '@opentelemetry/api@1.9.1': + resolution: {integrity: sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==} + engines: {node: '>=8.0.0'} + '@oxc-project/types@0.153.0': resolution: {integrity: sha512-jSNrRIyvPVLL5E1z/tS5zLAprosGz/FuI1PWwEOz59j3E4TTTzSpqKb/RDGqJhhW0QHYISGXxq582wGx6T11Tw==} @@ -2109,6 +2116,10 @@ packages: resolution: {integrity: sha512-9IkYqtX3YHPCzoVg1Py+o9057a3i0fp7S530UWokCSaFVTc7CwXPRiOjRjBQQ18ZCNafx78YfnG+HALxtVmOGA==} engines: {node: '>=10.0.0'} + cron-parser@5.10.2: + resolution: {integrity: sha512-JfAYMY0LZaWQTuict8PhtF/ksSnq8xefSDmQUnnoqvkPYkU/y3UP0DTWzJKFBz0P1IXsDG8q0vY3zHkoFZOhvw==} + engines: {node: '>=18'} + cross-spawn@7.0.6: resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==} engines: {node: '>= 8'} @@ -3080,6 +3091,10 @@ packages: peerDependencies: react: ^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0 + luxon@3.7.2: + resolution: {integrity: sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==} + engines: {node: '>=12'} + magic-string@0.30.21: resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==} @@ -3164,6 +3179,10 @@ packages: resolution: {integrity: sha512-BdiKAgbfRlvfMq/XzO/AF9NPmaA8a+YpLAyAFTVFBj3gARrAIfWvNq7Q7NE4se3KgFu0KkrvQMxTS0XVnsGzHg==} engines: {node: '>=18'} + non-error@0.1.0: + resolution: {integrity: sha512-TMB1uHiGsHRGv1uYclfhivcnf0/PdFp2pNqRxXjncaAsjYMoisaQJI+SSZCqRq+VliwRTC8tsMQfmrWjDMhkPQ==} + engines: {node: '>=20'} + object-assign@4.1.1: resolution: {integrity: sha512-rJgTQnkUnH1sFw8yT6VSU3zD3sWmu6sZhIseY8VX+GRu3P6F7Fu+JNDoXfklElbLJSnc3FUQHVe4cU5hj+BcUg==} engines: {node: '>=0.10.0'} @@ -3248,6 +3267,13 @@ packages: perfect-debounce@2.1.0: resolution: {integrity: sha512-LjgdTytVFXeUgtHZr9WYViYSM/g8MkcTPYDlPa3cDqMirHjKiSZPYd6DoL7pK8AJQr+uWkQvCjHNdiMqsrJs+g==} + pg-boss@12.37.1: + resolution: {integrity: sha512-LghWeOIbMCiLcTY3+8MInXaJ0wk0AOkbeUZ8oE+cGO0uoheGSW3GK57eDqEbrDGpwTIpX9BVphvdvWEDYTcJ9A==} + engines: {node: '>=22.12.0'} + hasBin: true + peerDependencies: + '@opentelemetry/api': ^1.9.0 + pg-cloudflare@1.4.1: resolution: {integrity: sha512-6PQbsFWZcp9EmJEwy5cGQ2La+AMWpP46lgbb8X+U/XsHIUweYDNCpeuKck5RxL2MdVFi7krbbEi5nX4Zh7JhrQ==} @@ -3469,6 +3495,9 @@ packages: engines: {node: ^20.19.0 || >=22.12.0} hasBin: true + rrule-temporal@2.2.8: + resolution: {integrity: sha512-+wWUe0J02oMh1UWKIt1RTu2H0Cz+gP11O/iweK0MN2UP6JLGeepTUkbSBa90Hxq0W+ywSxM9IlJNouIUdMOwhg==} + run-parallel@1.2.0: resolution: {integrity: sha512-5l4VyZR86LZ/lDxZTR6jqL8AFE2S0IFLMP26AbjsLVADxHdhB/c0GUsH+y39UfCi3dzz8OlQuPmnaJOMoDHQBA==} @@ -3510,6 +3539,10 @@ packages: seq-queue@0.0.5: resolution: {integrity: sha512-hr3Wtp/GZIc/6DAGPDcV4/9WoZhjrkXsi5B/07QgX8tsdc6ilr7BFM6PM6rbdAX1kFSDYeZGLipIZZKyQP0O5Q==} + serialize-error@13.0.2: + resolution: {integrity: sha512-G+YKwp9I6IvWMe5RlwqxdmTPIcq/G8HK/sU51gdV1XAJznYkRSTB0ph+zB9AfSxG0fx6xrqYA6i9HeIcHFTXow==} + engines: {node: '>=20'} + set-function-length@1.2.2: resolution: {integrity: sha512-pgRc4hJ4/sNjWCSS9AmnS40x3bNMDTknHgL5UaMBTMyJnU90EgWh1Rz+MC9eFu4BuN/UwZjKQuY/1v3rM7HMfg==} engines: {node: '>= 0.4'} @@ -3668,6 +3701,10 @@ packages: resolution: {integrity: sha512-ot0WnXS9fgdkgIcePe6RHNk1WA8+muPa6cSjeR3V8K27q9BB1rTE3R1p7Hv0z1ZyAc8s6Vvv8DIyWf681MAt0w==} engines: {node: '>= 0.4'} + tagged-tag@1.0.0: + resolution: {integrity: sha512-yEFYrVhod+hdNyx7g5Bnkkb0G6si8HJurOoOEgC8B/O0uXLHlaey/65KRv6cuWBNhBgHKAROVpc7QyYqE5gFng==} + engines: {node: '>=20'} + tailwindcss@4.3.3: resolution: {integrity: sha512-gOhV3P7ufE62QDGg1zVaTgCR+EtPv92k2nIhVcVKcLmxT1sUBsQGhnZj175j+MqRt4zLF7ic+sCYjfhxMxj7YQ==} @@ -3675,6 +3712,9 @@ packages: resolution: {integrity: sha512-uxc/zpqFg6x7C8vOE7lh6Lbda8eEL9zmVm/PLeTPBRhh1xCgdWaQ+J1CUieGpIfm2HdtsUpRv+HshiasBMcc6A==} engines: {node: '>=6'} + temporal-spec@1.0.1: + resolution: {integrity: sha512-wxVoanmDeavXie1vu2JaQ3WIc3JZnWAOYFBsJyATaVsXsycKYUflGsyBmrRSnoCpZJpwPyr38VpgSUlQ8CbFxg==} + thread-stream@4.2.0: resolution: {integrity: sha512-e2zZ96wSChazBsbENf/Pcm/4swHt2cEKQ92rhUjkL9GCKiTDJIaTBenjE/m9DXi0QBmTMDkFDdOomUy20A1tDQ==} engines: {node: '>=20'} @@ -3723,6 +3763,10 @@ packages: resolution: {integrity: sha512-XleUoc9uwGXqjWwXaUTZAmzMcFZ5858QA2vvx1Ur5xIcixXIP+8LnFDgRplU30us6teqdlskFfu+ae4K79Ooew==} engines: {node: '>= 0.8.0'} + type-fest@5.10.0: + resolution: {integrity: sha512-NoSdpq/WEiAg5sjmBkmV/hfxv6HJH4NqPNrqjtSO5CwRmpsDfaf4begxW34KdJykH/l1yHtwBWQkCRdoXO8mPA==} + engines: {node: '>=20'} + typed-array-buffer@1.0.3: resolution: {integrity: sha512-nAYYwfY3qnzX30IkA6AQZjVbtK6duGontcQm1WSG1MD94YLqK0515GNApXkoxKOWMusVssAHWLh9SeaoefYFGw==} engines: {node: '>= 0.4'} @@ -4663,6 +4707,8 @@ snapshots: '@nolyfill/is-core-module@1.0.39': {} + '@opentelemetry/api@1.9.1': {} + '@oxc-project/types@0.153.0': {} '@pinojs/redact@0.4.0': {} @@ -5557,6 +5603,10 @@ snapshots: nan: 2.29.0 optional: true + cron-parser@5.10.2: + dependencies: + luxon: 3.7.2 + cross-spawn@7.0.6: dependencies: path-key: 3.1.1 @@ -6640,6 +6690,8 @@ snapshots: dependencies: react: 19.3.0 + luxon@3.7.2: {} + magic-string@0.30.21: dependencies: '@jridgewell/sourcemap-codec': 1.6.0 @@ -6694,7 +6746,7 @@ snapshots: natural-compare@1.4.0: {} - next@16.4.0(@babel/core@7.29.7(supports-color@7.2.0))(@playwright/test@1.63.0)(@types/node@24.19.2)(react-dom@19.3.0(react@19.3.0))(react@19.3.0): + next@16.4.0(@babel/core@7.29.7(supports-color@7.2.0))(@opentelemetry/api@1.9.1)(@playwright/test@1.63.0)(@types/node@24.19.2)(react-dom@19.3.0(react@19.3.0))(react@19.3.0): dependencies: '@next/env': 16.4.0 '@swc/helpers': 0.5.23 @@ -6713,6 +6765,7 @@ snapshots: '@next/swc-linux-x64-musl': 16.4.0 '@next/swc-win32-arm64-msvc': 16.4.0 '@next/swc-win32-x64-msvc': 16.4.0 + '@opentelemetry/api': 1.9.1 '@playwright/test': 1.63.0 sharp: 0.35.5(@types/node@24.19.2) transitivePeerDependencies: @@ -6729,6 +6782,8 @@ snapshots: node-releases@2.0.58: {} + non-error@0.1.0: {} + object-assign@4.1.1: {} object-inspect@1.13.4: {} @@ -6822,6 +6877,16 @@ snapshots: perfect-debounce@2.1.0: {} + pg-boss@12.37.1(@opentelemetry/api@1.9.1): + dependencies: + '@opentelemetry/api': 1.9.1 + cron-parser: 5.10.2 + pg: 8.23.1 + rrule-temporal: 2.2.8 + serialize-error: 13.0.2 + transitivePeerDependencies: + - pg-native + pg-cloudflare@1.4.1: optional: true @@ -7052,6 +7117,10 @@ snapshots: '@rolldown/binding-win32-arm64-msvc': 1.2.13 '@rolldown/binding-win32-x64-msvc': 1.2.13 + rrule-temporal@2.2.8: + dependencies: + temporal-spec: 1.0.1 + run-parallel@1.2.0: dependencies: queue-microtask: 1.2.3 @@ -7091,6 +7160,11 @@ snapshots: seq-queue@0.0.5: {} + serialize-error@13.0.2: + dependencies: + non-error: 0.1.0 + type-fest: 5.10.0 + set-function-length@1.2.2: dependencies: define-data-property: 1.1.4 @@ -7303,10 +7377,14 @@ snapshots: supports-preserve-symlinks-flag@1.0.0: {} + tagged-tag@1.0.0: {} + tailwindcss@4.3.3: {} tapable@2.3.3: {} + temporal-spec@1.0.1: {} + thread-stream@4.2.0: dependencies: real-require: 1.0.0 @@ -7351,6 +7429,10 @@ snapshots: dependencies: prelude-ls: 1.2.1 + type-fest@5.10.0: + dependencies: + tagged-tag: 1.0.0 + typed-array-buffer@1.0.3: dependencies: call-bound: 1.0.4 @@ -7467,7 +7549,7 @@ snapshots: tsx: 4.23.15 yaml: 2.9.1 - vitest@5.0.1(@types/node@24.19.2)(vite@8.3.0(@types/node@24.19.2)(esbuild@0.27.7)(jiti@2.7.0)(tsx@4.23.15)(yaml@2.9.1)): + vitest@5.0.1(@opentelemetry/api@1.9.1)(@types/node@24.19.2)(vite@8.3.0(@types/node@24.19.2)(esbuild@0.27.7)(jiti@2.7.0)(tsx@4.23.15)(yaml@2.9.1)): dependencies: '@types/chai': 5.2.3 '@vitest/mocker': 5.0.1(vite@8.3.0(@types/node@24.19.2)(esbuild@0.27.7)(jiti@2.7.0)(tsx@4.23.15)(yaml@2.9.1)) @@ -7484,6 +7566,7 @@ snapshots: vite: 8.3.0(@types/node@24.19.2)(esbuild@0.27.7)(jiti@2.7.0)(tsx@4.23.15)(yaml@2.9.1) why-is-node-running: 2.3.0 optionalDependencies: + '@opentelemetry/api': 1.9.1 '@types/node': 24.19.2 transitivePeerDependencies: - msw diff --git a/prisma/migrations/20261012082805_auftraege/migration.sql b/prisma/migrations/20261012082805_auftraege/migration.sql new file mode 100644 index 0000000..9ce7d71 --- /dev/null +++ b/prisma/migrations/20261012082805_auftraege/migration.sql @@ -0,0 +1,118 @@ +-- CreateEnum +CREATE TYPE "GastTyp" AS ENUM ('VM', 'LXC'); + +-- CreateEnum +CREATE TYPE "AuftragsArt" AS ENUM ('AUSROLLEN', 'TEMPLATE_BAUEN', 'AUFRAEUMEN'); + +-- CreateEnum +CREATE TYPE "AuftragsStatus" AS ENUM ('WARTEND', 'LAEUFT', 'ERFOLGREICH', 'FEHLGESCHLAGEN', 'ABGEBROCHEN', 'AUFGERAEUMT'); + +-- CreateEnum +CREATE TYPE "PhasenStatus" AS ENUM ('OFFEN', 'LAEUFT', 'ERLEDIGT', 'FEHLER', 'UEBERSPRUNGEN'); + +-- CreateTable +CREATE TABLE "Auftrag" ( + "id" TEXT NOT NULL, + "kurzId" TEXT NOT NULL, + "art" "AuftragsArt" NOT NULL, + "status" "AuftragsStatus" NOT NULL DEFAULT 'WARTEND', + "serverId" TEXT NOT NULL, + "knoten" TEXT, + "gastTyp" "GastTyp", + "vmid" INTEGER, + "name" TEXT, + "distributionId" TEXT, + "profile" TEXT[], + "anfrage" JSONB NOT NULL, + "plan" JSONB, + "kopf" TEXT, + "ergebnis" JSONB, + "angelegt" JSONB, + "fehler" JSONB, + "quelle" "Quelle" NOT NULL, + "erstelltVonId" TEXT NOT NULL, + "wiederholtVonId" TEXT, + "bezugId" TEXT, + "abbrechen" BOOLEAN NOT NULL DEFAULT false, + "erstelltAm" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "gestartetAm" TIMESTAMP(3), + "beendetAm" TIMESTAMP(3), + + CONSTRAINT "Auftrag_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "Phase" ( + "id" TEXT NOT NULL, + "auftragId" TEXT NOT NULL, + "reihenfolge" INTEGER NOT NULL, + "schluessel" TEXT NOT NULL, + "titel" TEXT NOT NULL, + "status" "PhasenStatus" NOT NULL DEFAULT 'OFFEN', + "detail" TEXT, + "ursache" TEXT, + "hinweis" TEXT, + "gestartetAm" TIMESTAMP(3), + "beendetAm" TIMESTAMP(3), + + CONSTRAINT "Phase_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "Logeintrag" ( + "id" BIGSERIAL NOT NULL, + "auftragId" TEXT NOT NULL, + "zeit" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "stufe" TEXT NOT NULL, + "phase" TEXT, + "nachricht" TEXT NOT NULL, + "daten" JSONB, + + CONSTRAINT "Logeintrag_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "VmidReservierung" ( + "serverId" TEXT NOT NULL, + "vmid" INTEGER NOT NULL, + "auftragId" TEXT NOT NULL, + "ablaufAm" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "VmidReservierung_pkey" PRIMARY KEY ("serverId","vmid") +); + +-- CreateIndex +CREATE UNIQUE INDEX "Auftrag_kurzId_key" ON "Auftrag"("kurzId"); + +-- CreateIndex +CREATE INDEX "Auftrag_serverId_erstelltAm_idx" ON "Auftrag"("serverId", "erstelltAm"); + +-- CreateIndex +CREATE INDEX "Auftrag_status_idx" ON "Auftrag"("status"); + +-- CreateIndex +CREATE INDEX "Auftrag_erstelltAm_idx" ON "Auftrag"("erstelltAm"); + +-- CreateIndex +CREATE INDEX "Phase_auftragId_reihenfolge_idx" ON "Phase"("auftragId", "reihenfolge"); + +-- CreateIndex +CREATE UNIQUE INDEX "Phase_auftragId_schluessel_key" ON "Phase"("auftragId", "schluessel"); + +-- CreateIndex +CREATE INDEX "Logeintrag_auftragId_id_idx" ON "Logeintrag"("auftragId", "id"); + +-- CreateIndex +CREATE INDEX "VmidReservierung_auftragId_idx" ON "VmidReservierung"("auftragId"); + +-- AddForeignKey +ALTER TABLE "Auftrag" ADD CONSTRAINT "Auftrag_serverId_fkey" FOREIGN KEY ("serverId") REFERENCES "Server"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "Auftrag" ADD CONSTRAINT "Auftrag_erstelltVonId_fkey" FOREIGN KEY ("erstelltVonId") REFERENCES "Benutzer"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "Phase" ADD CONSTRAINT "Phase_auftragId_fkey" FOREIGN KEY ("auftragId") REFERENCES "Auftrag"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "Logeintrag" ADD CONSTRAINT "Logeintrag_auftragId_fkey" FOREIGN KEY ("auftragId") REFERENCES "Auftrag"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/migrations/20261012084349_templates/migration.sql b/prisma/migrations/20261012084349_templates/migration.sql new file mode 100644 index 0000000..2f01f30 --- /dev/null +++ b/prisma/migrations/20261012084349_templates/migration.sql @@ -0,0 +1,24 @@ +-- CreateTable +CREATE TABLE "Template" ( + "id" TEXT NOT NULL, + "serverId" TEXT NOT NULL, + "knoten" TEXT NOT NULL, + "vmid" INTEGER NOT NULL, + "distributionId" TEXT NOT NULL, + "storage" TEXT NOT NULL, + "imageDatei" TEXT NOT NULL, + "pruefsumme" TEXT NOT NULL, + "gebautVonAuftragId" TEXT, + "gebautAm" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "Template_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "Template_serverId_distributionId_idx" ON "Template"("serverId", "distributionId"); + +-- CreateIndex +CREATE UNIQUE INDEX "Template_serverId_vmid_key" ON "Template"("serverId", "vmid"); + +-- AddForeignKey +ALTER TABLE "Template" ADD CONSTRAINT "Template_serverId_fkey" FOREIGN KEY ("serverId") REFERENCES "Server"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 3c76596..298b846 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -38,6 +38,7 @@ model Benutzer { erstelltAm DateTime @default(now()) geaendertAm DateTime @updatedAt protokoll Protokoll[] + auftraege Auftrag[] } /// Audit-Log. Nur anhängen – ein Trigger verweigert UPDATE und DELETE. @@ -99,6 +100,8 @@ model Server { geprueftAm DateTime? erstelltAm DateTime @default(now()) geaendertAm DateTime @updatedAt + auftraege Auftrag[] + templates Template[] } enum Herkunft { @@ -139,3 +142,138 @@ model Profil { geaendertVonId String? geaendertAm DateTime @updatedAt } + +enum GastTyp { + VM + LXC +} + +enum AuftragsArt { + AUSROLLEN + TEMPLATE_BAUEN + AUFRAEUMEN +} + +enum AuftragsStatus { + WARTEND + LAEUFT + ERFOLGREICH + FEHLGESCHLAGEN + ABGEBROCHEN + AUFGERAEUMT +} + +enum PhasenStatus { + OFFEN + LAEUFT + ERLEDIGT + FEHLER + UEBERSPRUNGEN +} + +/// Ein Hintergrundjob: Ausrollen, Template bauen oder Aufräumen. Läuft im worker. +model Auftrag { + id String @id @default(cuid()) + /// Kurz und tippbar für CLI und URLs: „d-7f3k2“. + kurzId String @unique + art AuftragsArt + status AuftragsStatus @default(WARTEND) + serverId String + server Server @relation(fields: [serverId], references: [id], onDelete: Restrict) + knoten String? + gastTyp GastTyp? + vmid Int? + name String? + distributionId String? + profile String[] + /// Validierte Anfrage – Grundlage für „Wiederholen“. + anfrage Json + /// Plan zum Zeitpunkt des Starts (Phasen und geplante API-Aufrufe). + plan Json? + /// Kopfzeile wie in der CLI: „Server rz1 · Node pve01 · Debian 13 (Template 9013)“ – von der Engine gesetzt. + kopf String? + /// IP, SSH-Befehl, Dauer … + ergebnis Json? + /// Was ProxDeploy angelegt hat – Grundlage fürs Aufräumen. + angelegt Json? + /// { phase, ursache, hinweis } + fehler Json? + quelle Quelle + erstelltVonId String + erstelltVon Benutzer @relation(fields: [erstelltVonId], references: [id], onDelete: Restrict) + wiederholtVonId String? + /// Bei AUFRAEUMEN: der Auftrag, dessen Reste entfernt werden. + bezugId String? + abbrechen Boolean @default(false) + erstelltAm DateTime @default(now()) + gestartetAm DateTime? + beendetAm DateTime? + phasen Phase[] + log Logeintrag[] + + @@index([serverId, erstelltAm]) + @@index([status]) + @@index([erstelltAm]) +} + +model Phase { + id String @id @default(cuid()) + auftragId String + auftrag Auftrag @relation(fields: [auftragId], references: [id], onDelete: Cascade) + reihenfolge Int + schluessel String + titel String + status PhasenStatus @default(OFFEN) + /// Der graue Detailtext hinter dem Phasennamen – in Web und CLI derselbe. + detail String? + ursache String? + hinweis String? + gestartetAm DateTime? + beendetAm DateTime? + + @@unique([auftragId, schluessel]) + @@index([auftragId, reihenfolge]) +} + +model Logeintrag { + /// Zugleich die Event-ID im SSE-Strom (Last-Event-ID). + id BigInt @id @default(autoincrement()) + auftragId String + auftrag Auftrag @relation(fields: [auftragId], references: [id], onDelete: Cascade) + zeit DateTime @default(now()) + stufe String + phase String? + nachricht String + daten Json? + + @@index([auftragId, id]) +} + +/// Vergebene VMIDs, solange ein Auftrag läuft – verhindert Doppelvergabe zwischen parallelen Aufträgen. +model VmidReservierung { + serverId String + vmid Int + auftragId String + ablaufAm DateTime + + @@id([serverId, vmid]) + @@index([auftragId]) +} + +/// Von ProxDeploy gebaute Templates. Maßgeblich ist Proxmox (Tag proxdeploy-template); hier steht, woraus gebaut wurde. +model Template { + id String @id @default(cuid()) + serverId String + server Server @relation(fields: [serverId], references: [id], onDelete: Cascade) + knoten String + vmid Int + distributionId String + storage String + imageDatei String + pruefsumme String + gebautVonAuftragId String? + gebautAm DateTime @default(now()) + + @@unique([serverId, vmid]) + @@index([serverId, distributionId]) +} diff --git a/src/app/(app)/auftraege/[kurzId]/live-ansicht.tsx b/src/app/(app)/auftraege/[kurzId]/live-ansicht.tsx new file mode 100644 index 0000000..9785bf4 --- /dev/null +++ b/src/app/(app)/auftraege/[kurzId]/live-ansicht.tsx @@ -0,0 +1,194 @@ +"use client"; + +import { useEffect, useRef, useState } from "react"; +import { Code } from "@/components/formular"; +import { Knopf } from "@/components/knopf"; +import { Status } from "@/components/status"; +import { STATUS_ANZEIGE } from "@/lib/auftrag-text"; +import { TerminalKarte, type TerminalErgebnis } from "@/components/terminal-karte"; +import { apiAufruf } from "@/lib/api-client"; +import { dauer } from "@/lib/formate"; +import type { AuftragSicht, LogZeile } from "@/lib/schemas/auftrag"; + +const ENDE = new Set(["ERFOLGREICH", "FEHLGESCHLAGEN", "ABGEBROCHEN", "AUFGERAEUMT"]); + +function ergebnisAus(a: AuftragSicht): TerminalErgebnis | null { + if (!ENDE.has(a.status)) return null; + const e = (a.ergebnis ?? {}) as Record; + const gesamt = + a.gestartetAm && a.beendetAm + ? dauer(new Date(a.beendetAm).getTime() - new Date(a.gestartetAm).getTime()) + : null; + const name = a.name ?? a.kurzId; + if (a.status !== "ERFOLGREICH") { + return { + ok: false, + titel: `${name} ${a.status === "ABGEBROCHEN" ? "abgebrochen" : "fehlgeschlagen"}`, + dauer: gesamt, + zeilen: [], + }; + } + const zeilen: Array<[string, string]> = []; + if (a.vmid) zeilen.push([a.gastTyp === "LXC" ? "CTID" : "VMID", String(a.vmid)]); + if (typeof e["ip"] === "string") zeilen.push(["IP", e["ip"]]); + if (a.profile.length > 0) zeilen.push(["Profile", a.profile.join(", ")]); + if (typeof e["ssh"] === "string") zeilen.push(["SSH", e["ssh"]]); + return { ok: true, titel: `${name} bereit`, dauer: gesamt, zeilen }; +} + +/** + * Hängt sich per SSE an den Auftrag. Fällt SSE aus (z. B. ein Proxy, der Streams puffert), wird + * alle 3 s abgefragt – die Anzeige bleibt dieselbe. + */ +export function LiveAnsicht({ + anfang, + anfangsLog, + version, + abbrechenErlaubt, +}: { + anfang: AuftragSicht; + anfangsLog: LogZeile[]; + version: string; + abbrechenErlaubt: boolean; +}) { + const [auftrag, setAuftrag] = useState(anfang); + const [log, setLog] = useState(anfangsLog); + const [verbindung, setVerbindung] = useState<"live" | "abfrage" | "beendet">( + ENDE.has(anfang.status) ? "beendet" : "live", + ); + const logBis = useRef(anfangsLog.at(-1)?.id ?? "0"); + + useEffect(() => { + if (ENDE.has(anfang.status)) return; + let fehlschlaege = 0; + let abfrage: ReturnType | undefined; + const quelle = new EventSource(`/api/v1/auftraege/${anfang.kurzId}/ereignisse?ab=${logBis.current}`); + + const zeilenAnhaengen = (neu: LogZeile[]) => { + const bekannt = BigInt(logBis.current); + const frisch = neu.filter((z) => BigInt(z.id) > bekannt); + if (frisch.length === 0) return; + logBis.current = frisch.at(-1)!.id; + setLog((alt) => [...alt, ...frisch]); + }; + + quelle.addEventListener("stand", (e) => { + fehlschlaege = 0; + setAuftrag(JSON.parse((e as MessageEvent).data)); + }); + quelle.addEventListener("log", (e) => zeilenAnhaengen(JSON.parse((e as MessageEvent).data))); + quelle.addEventListener("ende", () => { + quelle.close(); + setVerbindung("beendet"); + }); + quelle.onerror = () => { + fehlschlaege += 1; + if (fehlschlaege < 3 || abfrage) return; + quelle.close(); + setVerbindung("abfrage"); + abfrage = setInterval(async () => { + try { + const stand = await apiAufruf("GET", `/api/v1/auftraege/${anfang.kurzId}`); + setAuftrag(stand); + zeilenAnhaengen( + await apiAufruf("GET", `/api/v1/auftraege/${anfang.kurzId}/log?ab=${logBis.current}`), + ); + if (ENDE.has(stand.status)) { + clearInterval(abfrage); + setVerbindung("beendet"); + } + } catch { + // nächster Versuch in 3 s + } + }, 3000); + }; + return () => { + quelle.close(); + if (abfrage) clearInterval(abfrage); + }; + }, [anfang.kurzId, anfang.status]); + + const [art, text] = + auftrag.status === "WARTEND" + ? (["offen", "wartet auf den worker"] as const) + : STATUS_ANZEIGE[auftrag.status]; + const ergebnis = ergebnisAus(auftrag); + const ssh = (auftrag.ergebnis as Record | null)?.["ssh"]; + + return ( +
+
+
+

+ {auftrag.kurzId} · gestartet von {auftrag.erstelltVon} +

+

+ {auftrag.art === "TEMPLATE_BAUEN" + ? "Template bauen" + : auftrag.art === "AUFRAEUMEN" + ? "Aufräumen" + : (auftrag.name ?? "Ausrollen")} +

+
+
+ {text} + {abbrechenErlaubt && !ENDE.has(auftrag.status) ? ( + void apiAufruf("POST", `/api/v1/auftraege/${auftrag.kurzId}/abbrechen`)} + > + Abbrechen + + ) : null} +
+
+ + + + {typeof ssh === "string" ? ( +
+ SSH-Befehl + {ssh} +
+ ) : null} + +
+ + Detail-Log ({log.length} Zeilen) + +
    + {log.map((zeile) => ( +
  1. + + + {zeile.phase ? [{zeile.phase}] : null} + {zeile.nachricht} + +
    2. + ))} +
+
+ +

+ {verbindung === "live" + ? "Live verbunden." + : verbindung === "abfrage" + ? "Live-Verbindung nicht möglich – Aktualisierung alle 3 s." + : null} +

+
+ ); +} diff --git a/src/app/(app)/auftraege/[kurzId]/page.tsx b/src/app/(app)/auftraege/[kurzId]/page.tsx new file mode 100644 index 0000000..b812257 --- /dev/null +++ b/src/app/(app)/auftraege/[kurzId]/page.tsx @@ -0,0 +1,34 @@ +import type { Metadata } from "next"; +import { notFound } from "next/navigation"; +import { versionText } from "@/lib/version"; +import { darf } from "@/server/auth/rechte"; +import { seiteSchuetzen } from "@/server/auth/zugang"; +import { AnwendungsFehler } from "@/server/fehler"; +import { auftragHolen, logAb } from "@/server/services/auftraege"; +import { LiveAnsicht } from "./live-ansicht"; + +export async function generateMetadata({ + params, +}: { + params: Promise<{ kurzId: string }>; +}): Promise { + return { title: (await params).kurzId }; +} + +export default async function AuftragSeite({ params }: { params: Promise<{ kurzId: string }> }) { + const { kurzId } = await params; + const benutzer = await seiteSchuetzen("lesen", `/auftraege/${kurzId}`); + const auftrag = await auftragHolen(kurzId).catch((fehler) => { + if (fehler instanceof AnwendungsFehler && fehler.status === 404) notFound(); + throw fehler; + }); + const log = await logAb(auftrag.id, BigInt(0), 2000); + return ( + + ); +} diff --git a/src/app/(app)/auftraege/page.tsx b/src/app/(app)/auftraege/page.tsx new file mode 100644 index 0000000..9df2e7b --- /dev/null +++ b/src/app/(app)/auftraege/page.tsx @@ -0,0 +1,55 @@ +import type { Metadata } from "next"; +import Link from "next/link"; +import { Karte } from "@/components/karte"; +import { Status } from "@/components/status"; +import { ART_TEXT, STATUS_ANZEIGE } from "@/lib/auftrag-text"; +import { zeitpunkt } from "@/lib/formate"; +import { seiteSchuetzen } from "@/server/auth/zugang"; +import { auftraegeListe } from "@/server/services/auftraege"; + +export const metadata: Metadata = { title: "Verlauf" }; + +/** Verlauf aller Aufträge. Filter und Wiederholen kommen mit #014. */ +export default async function Verlauf() { + await seiteSchuetzen("lesen", "/auftraege"); + const auftraege = await auftraegeListe({ grenze: 100 }); + return ( +
+

Verlauf

+ + {auftraege.length === 0 ? ( + Noch keine Aufträge. + ) : ( +
    + {auftraege.map((a) => { + const [art, text] = STATUS_ANZEIGE[a.status]; + return ( +
  • + + {a.kurzId} + + {a.name ?? ART_TEXT[a.art]} + + {" "} + · {ART_TEXT[a.art]} · {a.server.name} + {a.distributionId ? ` · ${a.distributionId}` : ""} + {a.vmid ? ` · ${a.vmid}` : ""} + + + {zeitpunkt(a.erstelltAm)} + + {text} + + +
    • + ); + })} +
+ )} + +
+ ); +} diff --git a/src/app/api/v1/auftraege/[id]/abbrechen/route.ts b/src/app/api/v1/auftraege/[id]/abbrechen/route.ts new file mode 100644 index 0000000..77f744c --- /dev/null +++ b/src/app/api/v1/auftraege/[id]/abbrechen/route.ts @@ -0,0 +1,13 @@ +import { NextResponse } from "next/server"; +import { aufruferAdresse } from "@/lib/adressen"; +import { api } from "@/server/api"; +import { rechtErforderlich } from "@/server/auth/zugang"; +import { auftragAbbrechen } from "@/server/services/auftraege"; + +type Kontext = { params: Promise<{ id: string }> }; + +export const POST = api(async (anfrage: Request, { params }: Kontext) => { + const aufrufer = await rechtErforderlich(anfrage, "ausrollen"); + await auftragAbbrechen((await params).id, aufrufer, aufruferAdresse(anfrage.headers)); + return NextResponse.json({ ok: true }); +}); diff --git a/src/app/api/v1/auftraege/[id]/ereignisse/route.ts b/src/app/api/v1/auftraege/[id]/ereignisse/route.ts new file mode 100644 index 0000000..a71819b --- /dev/null +++ b/src/app/api/v1/auftraege/[id]/ereignisse/route.ts @@ -0,0 +1,101 @@ +import { api } from "@/server/api"; +import { rechtErforderlich } from "@/server/auth/zugang"; +import { abonnieren } from "@/server/ereignisse/hoeren"; +import { auftragHolen, ENDZUSTAENDE, logAb } from "@/server/services/auftraege"; + +type Kontext = { params: Promise<{ id: string }> }; + +export const dynamic = "force-dynamic"; + +/** + * Live-Fortschritt als Server-Sent Events – für Web-GUI und CLI derselbe Strom. + * + * - `stand`: der vollständige Auftrag mit allen Phasen (klein, deshalb immer ganz) + * - `log`: neue Detail-Log-Zeilen; ihre ID ist die Event-ID, damit ein Wiederverbinden mit + * `Last-Event-ID` (oder `?ab=`) genau dort weitermacht + * - `ende`: der Auftrag ist fertig, der Strom schließt + */ +export const GET = api(async (anfrage: Request, { params }: Kontext) => { + await rechtErforderlich(anfrage, "lesen"); + const anfangsStand = await auftragHolen((await params).id); + const auftragId = anfangsStand.id; + const letzte = anfrage.headers.get("last-event-id") ?? new URL(anfrage.url).searchParams.get("ab") ?? "0"; + let logBis = BigInt(/^\d+$/.test(letzte) ? letzte : "0"); + + const kodierer = new TextEncoder(); + let aufraeumen = () => {}; + + const strom = new ReadableStream({ + async start(steuerung) { + let offen = true; + const senden = (text: string) => { + if (offen) steuerung.enqueue(kodierer.encode(text)); + }; + const ereignis = (art: string, daten: unknown, id?: string) => + senden(`${id ? `id: ${id}\n` : ""}event: ${art}\ndata: ${JSON.stringify(daten)}\n\n`); + + let laeuft = false; + let nochmal = false; + const aktualisieren = async () => { + // Signale können dicht kommen; zusammenfassen statt parallel abzufragen. + if (laeuft) { + nochmal = true; + return; + } + laeuft = true; + try { + do { + nochmal = false; + const stand = await auftragHolen(auftragId); + const zeilen = await logAb(auftragId, logBis); + if (zeilen.length > 0) { + logBis = BigInt(zeilen.at(-1)!.id); + ereignis("log", zeilen, zeilen.at(-1)!.id); + } + ereignis("stand", stand); + if (ENDZUSTAENDE.has(stand.status)) { + ereignis("ende", { status: stand.status }); + schliessen(); + return; + } + } while (nochmal); + } finally { + laeuft = false; + } + }; + + const abmelden = await abonnieren(auftragId, () => void aktualisieren()); + // Herzschlag: hält Proxys davon ab, die Verbindung als tot zu schließen. + const herz = setInterval(() => senden(": ping\n\n"), 15_000); + const schliessen = () => { + if (!offen) return; + offen = false; + clearInterval(herz); + abmelden(); + try { + steuerung.close(); + } catch { + // schon geschlossen + } + }; + aufraeumen = schliessen; + anfrage.signal.addEventListener("abort", schliessen); + + senden("retry: 3000\n\n"); + await aktualisieren(); + }, + cancel() { + aufraeumen(); + }, + }); + + return new Response(strom, { + headers: { + "content-type": "text/event-stream; charset=utf-8", + "cache-control": "no-cache, no-transform", + connection: "keep-alive", + // nginx puffert sonst und der Fortschritt käme in Brocken an. + "x-accel-buffering": "no", + }, + }); +}); diff --git a/src/app/api/v1/auftraege/[id]/log/route.ts b/src/app/api/v1/auftraege/[id]/log/route.ts new file mode 100644 index 0000000..8db2930 --- /dev/null +++ b/src/app/api/v1/auftraege/[id]/log/route.ts @@ -0,0 +1,14 @@ +import { NextResponse } from "next/server"; +import { api } from "@/server/api"; +import { rechtErforderlich } from "@/server/auth/zugang"; +import { auftragHolen, logAb } from "@/server/services/auftraege"; + +type Kontext = { params: Promise<{ id: string }> }; + +/** Detail-Log ab einer Zeilen-ID (`?ab=`), für Polling und `proxdeploy logs`. */ +export const GET = api(async (anfrage: Request, { params }: Kontext) => { + await rechtErforderlich(anfrage, "lesen"); + const auftrag = await auftragHolen((await params).id); + const ab = new URL(anfrage.url).searchParams.get("ab"); + return NextResponse.json(await logAb(auftrag.id, BigInt(/^\d+$/.test(ab ?? "") ? ab! : "0"))); +}); diff --git a/src/app/api/v1/auftraege/[id]/route.ts b/src/app/api/v1/auftraege/[id]/route.ts new file mode 100644 index 0000000..0aa5ad2 --- /dev/null +++ b/src/app/api/v1/auftraege/[id]/route.ts @@ -0,0 +1,11 @@ +import { NextResponse } from "next/server"; +import { api } from "@/server/api"; +import { rechtErforderlich } from "@/server/auth/zugang"; +import { auftragHolen } from "@/server/services/auftraege"; + +type Kontext = { params: Promise<{ id: string }> }; + +export const GET = api(async (anfrage: Request, { params }: Kontext) => { + await rechtErforderlich(anfrage, "lesen"); + return NextResponse.json(await auftragHolen((await params).id)); +}); diff --git a/src/app/api/v1/auftraege/route.ts b/src/app/api/v1/auftraege/route.ts new file mode 100644 index 0000000..b9cd44e --- /dev/null +++ b/src/app/api/v1/auftraege/route.ts @@ -0,0 +1,10 @@ +import { NextResponse } from "next/server"; +import { api } from "@/server/api"; +import { rechtErforderlich } from "@/server/auth/zugang"; +import { auftraegeListe } from "@/server/services/auftraege"; + +/** Verlauf. Filter und Seitenblättern folgen mit #014. */ +export const GET = api(async (anfrage: Request) => { + await rechtErforderlich(anfrage, "lesen"); + return NextResponse.json(await auftraegeListe()); +}); diff --git a/src/components/terminal-karte.tsx b/src/components/terminal-karte.tsx new file mode 100644 index 0000000..5ce8ca7 --- /dev/null +++ b/src/components/terminal-karte.tsx @@ -0,0 +1,179 @@ +"use client"; + +import { clsx } from "clsx"; +import { useEffect, useState } from "react"; +import { StatusSymbol, type StatusArt } from "@/components/status"; +import { dauer } from "@/lib/formate"; + +/** + * Die Live-Ansicht im Aufbau der CLI-Ausgabe: Kopfzeile, eine Zeile je Phase (Symbol, Name in + * fester Breite, Detail grau), am Ende die Ergebnis-Box. Alle Texte kommen aus der Engine. + */ +export type TerminalPhase = { + schluessel: string; + titel: string; + status: "OFFEN" | "LAEUFT" | "ERLEDIGT" | "FEHLER" | "UEBERSPRUNGEN"; + detail: string | null; + ursache: string | null; + hinweis: string | null; + gestartetAm: string | null; +}; + +export type TerminalErgebnis = { + ok: boolean; + titel: string; + dauer: string | null; + zeilen: Array<[string, string]>; +}; + +const ART: Record = { + OFFEN: "offen", + LAEUFT: "laeuft", + ERLEDIGT: "erledigt", + FEHLER: "fehler", + UEBERSPRUNGEN: "offen", +}; + +const FARBE: Record = { + erledigt: "text-akzent", + laeuft: "text-laeuft", + fehler: "text-fehler", + offen: "text-muted", +}; + +function Laufzeit({ seit }: { seit: string }) { + const [jetzt, setJetzt] = useState(() => Date.now()); + useEffect(() => { + const uhr = setInterval(() => setJetzt(Date.now()), 1000); + return () => clearInterval(uhr); + }, []); + return {dauer(jetzt - new Date(seit).getTime())}; +} + +export function TerminalKarte({ + version, + kopf, + trockenlauf = false, + phasen, + ergebnis, +}: { + version: string; + kopf: string; + trockenlauf?: boolean; + phasen: TerminalPhase[]; + ergebnis: TerminalErgebnis | null; +}) { + const breite = Math.max(12, ...phasen.map((p) => p.titel.length)) + 2; + + return ( +
+ +
+

+ ⬡ + Prox + Deploy + + {" "} + {version} · {kopf} + + {trockenlauf ? · DRY-RUN : null} +

+ +
    + {phasen.map((phase) => { + const art = ART[phase.status]; + return ( +
  1. + + + + {phase.titel} + + + {" "} + –{" "} + { + { + OFFEN: "offen", + LAEUFT: "läuft", + ERLEDIGT: "erledigt", + FEHLER: "Fehler", + UEBERSPRUNGEN: "übersprungen", + }[phase.status] + } + + + + {phase.status === "FEHLER" ? ( + <> + {phase.ursache} + {phase.hinweis ? → {phase.hinweis} : null} + + ) : ( + <> + {phase.detail} + {phase.status === "LAEUFT" && phase.gestartetAm ? ( + <> + {phase.detail ? " … " : ""} + + + ) : null} + + )} + +
    2. + ); + })} +
+ + {ergebnis ? ( +
+

+ {ergebnis.ok ? "● " : "✗ "} + {ergebnis.titel} + {ergebnis.dauer ? · {ergebnis.dauer} : null} +

+ {ergebnis.zeilen.length > 0 ? ( +
+ {ergebnis.zeilen.map(([name, wert]) => ( +
+
{name}
+
{wert}
+
+ ))} +
+ ) : null} +
+ ) : null} +
+
+ ); +} diff --git a/src/engine/ablauf/ablauf.ts b/src/engine/ablauf/ablauf.ts new file mode 100644 index 0000000..15c444c --- /dev/null +++ b/src/engine/ablauf/ablauf.ts @@ -0,0 +1,86 @@ +import { AbbruchFehler, fehlerBeschreiben, type FehlerBeschreibung } from "./fehler"; + +/** + * Der Phasen-Runner. Er kennt weder Proxmox noch die Datenbank: Phasen sind Funktionen, Ereignisse + * gehen an eine Senke. Der worker hängt eine Senke an, die in die Datenbank schreibt und per + * NOTIFY meldet; Tests hängen eine an, die mitschreibt. + * + * Die Texte (Titel, Detail, Ursache, Hinweis) entstehen nur hier und in den Phasen – Web und CLI + * zeigen sie unverändert an. + */ +export type PhasenStatusWert = "OFFEN" | "LAEUFT" | "ERLEDIGT" | "FEHLER" | "UEBERSPRUNGEN"; + +export type LogStufe = "debug" | "info" | "warn" | "error"; + +export interface EreignisSenke { + phasenAnlegen(phasen: Array<{ schluessel: string; titel: string }>): Promise; + phaseGestartet(schluessel: string, detail?: string): Promise; + /** Zwischenstand, z. B. „Download 43 %“ oder „cloud-init läuft …“. */ + phaseDetail(schluessel: string, detail: string): Promise; + phaseBeendet( + schluessel: string, + status: Exclude, + angaben: { detail?: string; fehler?: FehlerBeschreibung }, + ): Promise; + log(stufe: LogStufe, nachricht: string, angaben?: { phase?: string; daten?: unknown }): Promise; +} + +export type PhasenKontext = { + /** Detailtext der laufenden Phase aktualisieren. */ + detail(text: string): Promise; + log(nachricht: string, stufe?: LogStufe, daten?: unknown): Promise; + /** Wirft AbbruchFehler, wenn der Auftrag abgebrochen werden soll. */ + pruefeAbbruch(): Promise; +}; + +export type PhasenErgebnis = { detail?: string; uebersprungen?: boolean }; + +export type PhasenSchritt = { + schluessel: string; + titel: string; + ausfuehren: (kontext: PhasenKontext) => Promise; +}; + +export type AblaufErgebnis = + { ok: true } | { ok: false; abgebrochen: boolean; phase: string; fehler: FehlerBeschreibung }; + +export async function phasenAusfuehren( + schritte: PhasenSchritt[], + senke: EreignisSenke, + optionen: { abbruchGewuenscht?: () => Promise } = {}, +): Promise { + await senke.phasenAnlegen(schritte.map(({ schluessel, titel }) => ({ schluessel, titel }))); + + for (const schritt of schritte) { + const kontext: PhasenKontext = { + detail: (text) => senke.phaseDetail(schritt.schluessel, text), + log: (nachricht, stufe = "info", daten) => + senke.log(stufe, nachricht, { phase: schritt.schluessel, daten }), + pruefeAbbruch: async () => { + if (await optionen.abbruchGewuenscht?.()) throw new AbbruchFehler(); + }, + }; + try { + await kontext.pruefeAbbruch(); + await senke.phaseGestartet(schritt.schluessel); + const ergebnis = (await schritt.ausfuehren(kontext)) ?? {}; + await senke.phaseBeendet(schritt.schluessel, ergebnis.uebersprungen ? "UEBERSPRUNGEN" : "ERLEDIGT", { + detail: ergebnis.detail, + }); + } catch (fehler) { + const beschreibung = fehlerBeschreiben(fehler); + await senke.phaseBeendet(schritt.schluessel, "FEHLER", { fehler: beschreibung }); + await senke.log("error", `${schritt.titel}: ${beschreibung.ursache}`, { + phase: schritt.schluessel, + daten: fehler instanceof Error ? { name: fehler.name, stack: fehler.stack } : undefined, + }); + return { + ok: false, + abgebrochen: fehler instanceof AbbruchFehler, + phase: schritt.schluessel, + fehler: beschreibung, + }; + } + } + return { ok: true }; +} diff --git a/src/engine/ablauf/fehler.ts b/src/engine/ablauf/fehler.ts new file mode 100644 index 0000000..8f5f686 --- /dev/null +++ b/src/engine/ablauf/fehler.ts @@ -0,0 +1,46 @@ +import { AdapterFehler } from "@/engine/adapter/debian"; +import { ProxmoxFehler } from "@/engine/proxmox/fehler"; +import { VorlagenFehler } from "@/engine/profile/vorlage"; +import { SshFehler } from "@/engine/ssh/verbindung"; + +/** + * Ein Fehler, den eine Phase bewusst wirft – mit Ursache und Lösungshinweis, wie sie unter der + * Phase in Web und CLI stehen. + */ +export class AblaufFehler extends Error { + constructor( + readonly ursache: string, + readonly hinweis: string, + ) { + super(ursache); + this.name = "AblaufFehler"; + } +} + +/** Der Auftrag wurde abgebrochen – kein Fehler im eigentlichen Sinn. */ +export class AbbruchFehler extends Error { + constructor() { + super("Abgebrochen"); + this.name = "AbbruchFehler"; + } +} + +export type FehlerBeschreibung = { ursache: string; hinweis: string }; + +/** Macht aus jedem Fehler Ursache und Hinweis; Unbekanntes wird ehrlich als solches benannt. */ +export function fehlerBeschreiben(fehler: unknown): FehlerBeschreibung { + if (fehler instanceof AblaufFehler || fehler instanceof ProxmoxFehler || fehler instanceof SshFehler) { + return { ursache: fehler.ursache, hinweis: fehler.hinweis }; + } + if (fehler instanceof AdapterFehler || fehler instanceof VorlagenFehler) { + return { ursache: fehler.message, hinweis: "Eingaben bzw. Profil-Variablen prüfen." }; + } + if (fehler instanceof AbbruchFehler) { + return { ursache: "Abgebrochen.", hinweis: "Bereits Angelegtes lässt sich über „Aufräumen“ entfernen." }; + } + const text = fehler instanceof Error ? fehler.message : String(fehler); + return { + ursache: `Unerwarteter Fehler: ${text}`, + hinweis: "Detail-Log prüfen; das ist vermutlich ein Fehler in ProxDeploy.", + }; +} diff --git a/src/engine/proxmox/aufgaben.ts b/src/engine/proxmox/aufgaben.ts index f71c4c7..8c09fa2 100644 --- a/src/engine/proxmox/aufgaben.ts +++ b/src/engine/proxmox/aufgaben.ts @@ -1,4 +1,4 @@ -import { pfad, type ProxmoxClient } from "./client"; +import { istTrockenUpid, pfad, type ProxmoxClient } from "./client"; import { ProxmoxFehler } from "./fehler"; /** @@ -10,6 +10,7 @@ type LogZeile = { n: number; t: string }; /** UPID:node:pid:pstart:starttime:type:id:user: – der Node steht an zweiter Stelle. */ export function nodeAusUpid(upid: string): string { + if (istTrockenUpid(upid)) return upid.split(":")[2] ?? "node"; const teile = upid.split(":"); if (teile[0] !== "UPID" || !teile[1]) { throw new ProxmoxFehler("api", `„${upid}“ ist keine UPID.`, "Antwort von Proxmox prüfen."); @@ -32,6 +33,7 @@ export async function aufgabeAbwarten( upid: string, optionen: AbwartenOptionen = {}, ): Promise { + if (istTrockenUpid(upid)) return; const node = nodeAusUpid(upid); const intervall = optionen.intervallMs ?? 1000; const zeitlimit = optionen.zeitlimitMs ?? 30 * 60 * 1000; diff --git a/src/engine/proxmox/client.ts b/src/engine/proxmox/client.ts index e610ad6..bfe67df 100644 --- a/src/engine/proxmox/client.ts +++ b/src/engine/proxmox/client.ts @@ -26,6 +26,17 @@ export type AufrufBeobachter = ( ergebnis: { dauerMs: number; fehler?: ProxmoxFehler }, ) => void; +/** + * Trockenlauf (Dry-Run): lesende Aufrufe gehen wirklich an Proxmox, schreibende werden nur + * aufgezeichnet und mit einer Platzhalter-UPID beantwortet. Dieselben Phasen laufen so im + * Trockenlauf und im Ernstfall – der angezeigte Plan kann nicht vom tatsächlichen Ablauf abweichen. + */ +export const TROCKEN_UPID_PRAEFIX = "UPID:trocken:"; + +export function istTrockenUpid(upid: unknown): boolean { + return typeof upid === "string" && upid.startsWith(TROCKEN_UPID_PRAEFIX); +} + const STANDARD_ZEITLIMIT_MS = 30_000; /** Formular-Kodierung wie pvesh: Wahrheitswerte als 1/0, Listen als wiederholte Schlüssel. */ @@ -60,7 +71,11 @@ export class ProxmoxClient { constructor( private readonly verbindung: ProxmoxVerbindung, - private readonly optionen: { zeitlimitMs?: number; beobachter?: AufrufBeobachter } = {}, + private readonly optionen: { + zeitlimitMs?: number; + beobachter?: AufrufBeobachter; + trocken?: boolean; + } = {}, ) { if (!TOKEN_ID_MUSTER.test(verbindung.tokenId)) { throw new ProxmoxFehler( @@ -90,7 +105,20 @@ export class ProxmoxClient { return await this.aufrufen({ methode: "DELETE", pfad, parameter }); } + /** Im Trockenlauf aufgezeichnete schreibende Aufrufe, in Reihenfolge. */ + readonly geplant: Aufruf[] = []; + + get trocken(): boolean { + return Boolean(this.optionen.trocken); + } + async aufrufen(aufruf: Aufruf): Promise { + if (this.optionen.trocken && aufruf.methode !== "GET") { + this.geplant.push(aufruf); + this.optionen.beobachter?.(aufruf, { dauerMs: 0 }); + const knoten = /^\/nodes\/([^/]+)/.exec(aufruf.pfad)?.[1] ?? "node"; + return `${TROCKEN_UPID_PRAEFIX}${knoten}:${this.geplant.length}` as T; + } const beginn = Date.now(); try { const ergebnis = await this.senden(aufruf); diff --git a/src/engine/ssh/verbindung.ts b/src/engine/ssh/verbindung.ts index 8710abd..1c111c1 100644 --- a/src/engine/ssh/verbindung.ts +++ b/src/engine/ssh/verbindung.ts @@ -1,5 +1,8 @@ import { createHash } from "node:crypto"; -import { Client } from "ssh2"; +// ssh2 ist CommonJS: unter reinem Node-ESM (worker) gibt es keine benannten Exporte. +import ssh2 from "ssh2"; + +const { Client } = ssh2; /** * SSH zu Proxmox-Nodes oder Gästen. Der Host-Key wird wie das TLS-Zertifikat gepinnt: Ist kein diff --git a/src/lib/auftrag-text.ts b/src/lib/auftrag-text.ts new file mode 100644 index 0000000..75da405 --- /dev/null +++ b/src/lib/auftrag-text.ts @@ -0,0 +1,18 @@ +import type { StatusArt } from "@/components/status"; +import type { AuftragSicht } from "@/lib/schemas/auftrag"; + +/** Beschriftungen für Aufträge – geteilt von Verlauf, Übersicht und Live-Ansicht. */ +export const STATUS_ANZEIGE: Record = { + WARTEND: ["offen", "wartet"], + LAEUFT: ["laeuft", "läuft"], + ERFOLGREICH: ["erledigt", "erfolgreich"], + FEHLGESCHLAGEN: ["fehler", "fehlgeschlagen"], + ABGEBROCHEN: ["fehler", "abgebrochen"], + AUFGERAEUMT: ["offen", "aufgeräumt"], +}; + +export const ART_TEXT: Record = { + AUSROLLEN: "Ausrollen", + TEMPLATE_BAUEN: "Template", + AUFRAEUMEN: "Aufräumen", +}; diff --git a/src/lib/navigation.ts b/src/lib/navigation.ts index f1a5b07..3aeaa9c 100644 --- a/src/lib/navigation.ts +++ b/src/lib/navigation.ts @@ -23,7 +23,7 @@ export type NavigationsEintrag = { export const NAVIGATION: readonly NavigationsEintrag[] = [ { pfad: "/", titel: "Übersicht", symbol: "uebersicht", recht: "lesen" }, { pfad: "/ausrollen", titel: "Ausrollen", symbol: "ausrollen", recht: "ausrollen", ab: "#008" }, - { pfad: "/auftraege", titel: "Verlauf", symbol: "verlauf", recht: "lesen", ab: "#006" }, + { pfad: "/auftraege", titel: "Verlauf", symbol: "verlauf", recht: "lesen" }, { pfad: "/templates", titel: "Templates", symbol: "templates", recht: "lesen", ab: "#007" }, { pfad: "/vorlagen", titel: "Vorlagen", symbol: "vorlagen", recht: "lesen", ab: "#012" }, { pfad: "/katalog", titel: "Katalog", symbol: "katalog", recht: "lesen" }, diff --git a/src/lib/schemas/auftrag.ts b/src/lib/schemas/auftrag.ts new file mode 100644 index 0000000..1207dd6 --- /dev/null +++ b/src/lib/schemas/auftrag.ts @@ -0,0 +1,35 @@ +import type { AuftragsArt, AuftragsStatus, GastTyp, PhasenStatus } from "@/generated/prisma/enums"; + +/** Was die API über einen Auftrag herausgibt – geteilt von Server, Web-GUI und (als JSON) CLI. */ +export type AuftragSicht = { + id: string; + kurzId: string; + art: AuftragsArt; + status: AuftragsStatus; + kopf: string; + server: { name: string; titel: string }; + knoten: string | null; + gastTyp: GastTyp | null; + vmid: number | null; + name: string | null; + distributionId: string | null; + profile: string[]; + ergebnis: unknown; + fehler: unknown; + erstelltVon: string; + erstelltAm: string; + gestartetAm: string | null; + beendetAm: string | null; + phasen: Array<{ + schluessel: string; + titel: string; + status: PhasenStatus; + detail: string | null; + ursache: string | null; + hinweis: string | null; + gestartetAm: string | null; + beendetAm: string | null; + }>; +}; + +export type LogZeile = { id: string; zeit: string; stufe: string; phase: string | null; nachricht: string }; diff --git a/src/server/ereignisse/hoeren.ts b/src/server/ereignisse/hoeren.ts new file mode 100644 index 0000000..2fb60a5 --- /dev/null +++ b/src/server/ereignisse/hoeren.ts @@ -0,0 +1,49 @@ +// pg ist CommonJS – Default-Import, damit es auch im worker (reines Node-ESM) lädt. +import pg from "pg"; +import { logger } from "@/lib/logger"; +import { KANAL } from "./senke"; + +/** + * Eine LISTEN-Verbindung je Prozess, an der beliebig viele SSE-Ströme hängen. Bricht sie ab, wird + * sie neu aufgebaut; die Ströme holen sich beim nächsten Signal ohnehin den vollen Stand. + */ +type Hoerer = (auftragId: string) => void; + +const hoerer = new Set(); +let verbindung: Promise | undefined; + +async function verbinden(): Promise { + const client = new pg.Client({ + connectionString: process.env["DATABASE_URL"], + application_name: "proxdeploy-listen", + }); + client.on("notification", (meldung) => { + if (meldung.channel !== KANAL || !meldung.payload) return; + for (const h of hoerer) h(meldung.payload); + }); + client.on("error", (fehler) => { + logger.warn({ err: fehler }, "LISTEN-Verbindung verloren, baue neu auf"); + verbindung = undefined; + setTimeout(() => void sicherstellen(), 1000); + }); + await client.connect(); + await client.query(`LISTEN ${KANAL}`); + return client; +} + +function sicherstellen(): Promise { + verbindung ??= verbinden().catch((fehler) => { + verbindung = undefined; + throw fehler; + }); + return verbindung; +} + +export async function abonnieren(auftragId: string, rueckruf: () => void): Promise<() => void> { + await sicherstellen(); + const h: Hoerer = (id) => { + if (id === auftragId) rueckruf(); + }; + hoerer.add(h); + return () => hoerer.delete(h); +} diff --git a/src/server/ereignisse/senke.ts b/src/server/ereignisse/senke.ts new file mode 100644 index 0000000..9cc70e1 --- /dev/null +++ b/src/server/ereignisse/senke.ts @@ -0,0 +1,83 @@ +import type { EreignisSenke, LogStufe } from "@/engine/ablauf/ablauf"; +import type { Prisma } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import { logger } from "@/lib/logger"; + +/** + * Schreibt die Ereignisse eines Auftrags in die Datenbank und meldet sie per NOTIFY. Die app hört + * mit (LISTEN) und schiebt sie an alle offenen SSE-Verbindungen – Web und CLI sehen dasselbe. + */ +export const KANAL = "proxdeploy_auftrag"; + +export async function melden(auftragId: string): Promise { + await prisma.$executeRaw`SELECT pg_notify(${KANAL}, ${auftragId})`; +} + +export class DatenbankSenke implements EreignisSenke { + constructor(private readonly auftragId: string) {} + + async phasenAnlegen(phasen: Array<{ schluessel: string; titel: string }>): Promise { + await prisma.$transaction([ + prisma.phase.deleteMany({ where: { auftragId: this.auftragId } }), + prisma.phase.createMany({ + data: phasen.map((p, reihenfolge) => ({ auftragId: this.auftragId, reihenfolge, ...p })), + }), + ]); + await melden(this.auftragId); + } + + async phaseGestartet(schluessel: string, detail?: string): Promise { + await prisma.phase.update({ + where: { auftragId_schluessel: { auftragId: this.auftragId, schluessel } }, + data: { status: "LAEUFT", gestartetAm: new Date(), ...(detail ? { detail } : {}) }, + }); + await melden(this.auftragId); + } + + async phaseDetail(schluessel: string, detail: string): Promise { + await prisma.phase.update({ + where: { auftragId_schluessel: { auftragId: this.auftragId, schluessel } }, + data: { detail }, + }); + await melden(this.auftragId); + } + + async phaseBeendet( + schluessel: string, + status: "ERLEDIGT" | "FEHLER" | "UEBERSPRUNGEN", + angaben: { detail?: string; fehler?: { ursache: string; hinweis: string } }, + ): Promise { + await prisma.phase.update({ + where: { auftragId_schluessel: { auftragId: this.auftragId, schluessel } }, + data: { + status, + beendetAm: new Date(), + ...(angaben.detail !== undefined ? { detail: angaben.detail } : {}), + ...(angaben.fehler ? { ursache: angaben.fehler.ursache, hinweis: angaben.fehler.hinweis } : {}), + }, + }); + await melden(this.auftragId); + } + + async log( + stufe: LogStufe, + nachricht: string, + angaben: { phase?: string; daten?: unknown } = {}, + ): Promise { + await prisma.logeintrag.create({ + data: { + auftragId: this.auftragId, + stufe, + nachricht, + phase: angaben.phase ?? null, + daten: (angaben.daten ?? undefined) as Prisma.InputJsonValue | undefined, + }, + }); + // Ins Betriebslog (JSON) geht jede Zeile mit Auftrags-ID – auch ohne GUI nachvollziehbar. + logger[stufe === "debug" ? "debug" : stufe]( + { auftragId: this.auftragId, phase: angaben.phase }, + nachricht, + ); + await melden(this.auftragId); + } +} diff --git a/src/server/services/auftraege.ts b/src/server/services/auftraege.ts new file mode 100644 index 0000000..46b4d7d --- /dev/null +++ b/src/server/services/auftraege.ts @@ -0,0 +1,191 @@ +import { randomInt } from "node:crypto"; +import type { Auftrag, Prisma } from "@/generated/prisma/client"; +import type { AuftragsArt, GastTyp, Quelle } from "@/generated/prisma/enums"; +import { prisma } from "@/lib/db"; +import type { Aufrufer } from "@/server/auth/zugang"; +import { AnwendungsFehler, nichtGefunden } from "@/server/fehler"; +import { melden } from "@/server/ereignisse/senke"; +import { protokollieren } from "@/server/services/protokoll"; +import { auftragEinstellen } from "@/server/warteschlange"; +import type { AuftragSicht, LogZeile } from "@/lib/schemas/auftrag"; + +export type { AuftragSicht, LogZeile }; + +/** Kurz-IDs ohne Verwechsler (kein 0/o, 1/l/i): „d-7f3k2“ für Ausrollen, „t-…“ für Templates. */ +const ZEICHEN = "23456789abcdefghjkmnpqrstuvwxyz"; +const PRAEFIX: Record = { AUSROLLEN: "d", TEMPLATE_BAUEN: "t", AUFRAEUMEN: "c" }; + +export function kurzIdErzeugen(art: AuftragsArt): string { + let id = ""; + for (let i = 0; i < 5; i += 1) id += ZEICHEN[randomInt(ZEICHEN.length)]; + return `${PRAEFIX[art]}-${id}`; +} + +export type NeuerAuftrag = { + art: AuftragsArt; + serverId: string; + knoten?: string; + gastTyp?: GastTyp; + name?: string; + distributionId?: string; + profile?: string[]; + anfrage: Prisma.InputJsonValue; + plan?: Prisma.InputJsonValue; + bezugId?: string; + wiederholtVonId?: string; +}; + +export async function auftragAnlegen( + neu: NeuerAuftrag, + aufrufer: Aufrufer, + ip: string | null, + einstellen: (id: string) => Promise = auftragEinstellen, +): Promise { + let auftrag: Auftrag | undefined; + for (let versuch = 0; versuch < 5 && !auftrag; versuch += 1) { + try { + auftrag = await prisma.$transaction(async (tx) => { + const angelegt = await tx.auftrag.create({ + data: { + ...neu, + profile: neu.profile ?? [], + kurzId: kurzIdErzeugen(neu.art), + quelle: aufrufer.quelle as Quelle, + erstelltVonId: aufrufer.benutzer.id, + }, + }); + await protokollieren( + { + benutzerId: aufrufer.benutzer.id, + quelle: aufrufer.quelle, + aktion: + neu.art === "AUSROLLEN" + ? "Ausrollen gestartet" + : neu.art === "TEMPLATE_BAUEN" + ? "Template-Bau gestartet" + : "Aufräumen gestartet", + ziel: `auftrag:${angelegt.kurzId}`, + details: { name: neu.name, distribution: neu.distributionId }, + ip, + }, + tx, + ); + return angelegt; + }); + } catch (fehler) { + // Kurz-ID zufällig doppelt – selten, dann einfach neu würfeln. + if ((fehler as { code?: string }).code !== "P2002") throw fehler; + } + } + if (!auftrag) throw new AnwendungsFehler(500, "intern", "Konnte keine freie Auftrags-ID erzeugen."); + await einstellen(auftrag.id); + return auftrag; +} + +const MIT = { + server: { select: { name: true, titel: true } }, + erstelltVon: { select: { benutzername: true } }, + phasen: { orderBy: { reihenfolge: "asc" as const } }, +}; + +type MitBeziehungen = Prisma.AuftragGetPayload<{ include: typeof MIT }>; + +export function auftragSicht(a: MitBeziehungen): AuftragSicht { + const iso = (d: Date | null) => d?.toISOString() ?? null; + return { + id: a.id, + kurzId: a.kurzId, + art: a.art, + status: a.status, + kopf: + a.kopf ?? [`Server ${a.server.name}`, a.knoten ? `Node ${a.knoten}` : null].filter(Boolean).join(" · "), + server: a.server, + knoten: a.knoten, + gastTyp: a.gastTyp, + vmid: a.vmid, + name: a.name, + distributionId: a.distributionId, + profile: a.profile, + ergebnis: a.ergebnis, + fehler: a.fehler, + erstelltVon: a.erstelltVon.benutzername, + erstelltAm: a.erstelltAm.toISOString(), + gestartetAm: iso(a.gestartetAm), + beendetAm: iso(a.beendetAm), + phasen: a.phasen.map((p) => ({ + schluessel: p.schluessel, + titel: p.titel, + status: p.status, + detail: p.detail, + ursache: p.ursache, + hinweis: p.hinweis, + gestartetAm: iso(p.gestartetAm), + beendetAm: iso(p.beendetAm), + })), + }; +} + +/** Sucht per Kurz-ID oder interner ID – die CLI nimmt, was der Benutzer tippt. */ +export async function auftragHolen(kennung: string): Promise { + const auftrag = await prisma.auftrag.findFirst({ + where: { OR: [{ kurzId: kennung }, { id: kennung }] }, + include: MIT, + }); + if (!auftrag) throw nichtGefunden(`Den Auftrag „${kennung}“`); + return auftragSicht(auftrag); +} + +export async function auftraegeListe( + filter: { serverId?: string; grenze?: number } = {}, +): Promise { + const zeilen = await prisma.auftrag.findMany({ + where: filter.serverId ? { serverId: filter.serverId } : {}, + include: MIT, + orderBy: { erstelltAm: "desc" }, + take: filter.grenze ?? 50, + }); + return zeilen.map(auftragSicht); +} + +export async function logAb(auftragId: string, abId: bigint, grenze = 500): Promise { + const zeilen = await prisma.logeintrag.findMany({ + where: { auftragId, id: { gt: abId } }, + orderBy: { id: "asc" }, + take: grenze, + }); + return zeilen.map((z) => ({ + id: z.id.toString(), + zeit: z.zeit.toISOString(), + stufe: z.stufe, + phase: z.phase, + nachricht: z.nachricht, + })); +} + +export const ENDZUSTAENDE = new Set(["ERFOLGREICH", "FEHLGESCHLAGEN", "ABGEBROCHEN", "AUFGERAEUMT"]); + +/** Bittet den worker um Abbruch; er prüft das zwischen den Phasen. */ +export async function auftragAbbrechen( + kennung: string, + aufrufer: Aufrufer, + ip: string | null, +): Promise { + const auftrag = await auftragHolen(kennung); + if (ENDZUSTAENDE.has(auftrag.status)) { + throw new AnwendungsFehler(409, "beendet", "Der Auftrag ist bereits beendet."); + } + await prisma.$transaction(async (tx) => { + await tx.auftrag.update({ where: { id: auftrag.id }, data: { abbrechen: true } }); + await protokollieren( + { + benutzerId: aufrufer.benutzer.id, + quelle: aufrufer.quelle, + aktion: "Abbruch angefordert", + ziel: `auftrag:${auftrag.kurzId}`, + ip, + }, + tx, + ); + }); + await melden(auftrag.id); +} diff --git a/src/server/services/server.ts b/src/server/services/server.ts index 55ebf95..6ec1cd6 100644 --- a/src/server/services/server.ts +++ b/src/server/services/server.ts @@ -1,4 +1,4 @@ -import { utils as sshUtils } from "ssh2"; +import ssh2 from "ssh2"; import type { Server } from "@/generated/prisma/client"; import { ProxmoxClient, type ProxmoxVerbindung } from "@/engine/proxmox/client"; import { erkunden, type VerbindungsBericht } from "@/engine/proxmox/erkundung"; @@ -170,7 +170,7 @@ export async function sshSchluesselErzeugen( ip: string | null, ): Promise { const server = await serverHolen(name); - const paar = sshUtils.generateKeyPairSync("ed25519", { comment: `proxdeploy@${server.name}` }); + const paar = ssh2.utils.generateKeyPairSync("ed25519", { comment: `proxdeploy@${server.name}` }); const neu = await prisma.$transaction(async (tx) => { const aktualisiert = await tx.server.update({ where: { id: server.id }, diff --git a/src/server/services/vmid.ts b/src/server/services/vmid.ts new file mode 100644 index 0000000..f31449b --- /dev/null +++ b/src/server/services/vmid.ts @@ -0,0 +1,72 @@ +import { prisma } from "@/lib/db"; +import { AblaufFehler } from "@/engine/ablauf/fehler"; + +/** + * Vergibt VMIDs ohne Doppelungen zwischen parallelen Aufträgen: unter einer Advisory-Sperre je + * Server wird die kleinste freie ID ≥ `von` gesucht, die weder in Proxmox belegt noch von einem + * anderen laufenden Auftrag reserviert ist, und sofort reserviert. Die Sperre fällt mit der + * Transaktion; die Reservierung bleibt bis zum Auftragsende (Sicherheitsnetz: Ablauf nach 4 h). + * + * Gegen Gäste, die gleichzeitig außerhalb von ProxDeploy entstehen, hilft das nicht – dort + * scheitert `clone`/`create` mit „already exists“, und die Phase wiederholt einmal mit neuer ID. + */ +const RESERVIERUNG_STUNDEN = 4; + +export async function vmidReservieren(eingabe: { + serverId: string; + auftragId: string; + belegt: ReadonlySet; + von: number; + bis?: number; + /** Eine bestimmte ID statt der nächsten freien (vom Benutzer gewählt). */ + wunsch?: number; +}): Promise { + const bis = eingabe.bis ?? 999_999_999; + return await prisma.$transaction(async (tx) => { + await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext('proxdeploy:vmid'), hashtext(${eingabe.serverId}))`; + await tx.vmidReservierung.deleteMany({ where: { ablaufAm: { lt: new Date() } } }); + const reserviert = new Set( + ( + await tx.vmidReservierung.findMany({ where: { serverId: eingabe.serverId }, select: { vmid: true } }) + ).map((r) => r.vmid), + ); + const frei = (id: number) => !eingabe.belegt.has(id) && !reserviert.has(id); + + let vmid: number | null = null; + if (eingabe.wunsch !== undefined) { + if (!frei(eingabe.wunsch)) { + throw new AblaufFehler( + `VMID ${eingabe.wunsch} ist bereits belegt oder reserviert.`, + "Eine andere ID wählen oder das Feld leer lassen – dann nimmt ProxDeploy die nächste freie.", + ); + } + vmid = eingabe.wunsch; + } else { + for (let id = eingabe.von; id <= bis; id += 1) { + if (frei(id)) { + vmid = id; + break; + } + } + } + if (vmid === null) { + throw new AblaufFehler( + `Im Bereich ${eingabe.von}–${bis} ist keine VMID frei.`, + "Bereich in den Server-Einstellungen erweitern.", + ); + } + await tx.vmidReservierung.create({ + data: { + serverId: eingabe.serverId, + vmid, + auftragId: eingabe.auftragId, + ablaufAm: new Date(Date.now() + RESERVIERUNG_STUNDEN * 3600 * 1000), + }, + }); + return vmid; + }); +} + +export async function reservierungenFreigeben(auftragId: string): Promise { + await prisma.vmidReservierung.deleteMany({ where: { auftragId } }); +} diff --git a/src/server/warteschlange.ts b/src/server/warteschlange.ts new file mode 100644 index 0000000..c985cd6 --- /dev/null +++ b/src/server/warteschlange.ts @@ -0,0 +1,43 @@ +import { PgBoss } from "pg-boss"; +import { logger } from "@/lib/logger"; + +/** + * Die Auftragswarteschlange (pg-boss auf derselben PostgreSQL, Schema `pgboss`). Die app stellt + * nur ein, der worker arbeitet ab. Ein Auftrag wird nie automatisch wiederholt: halb angelegte + * Gäste soll ein Mensch ansehen, nicht eine Schleife. + */ +export const WARTESCHLANGE = "auftrag"; + +/** Template-Bau kann bei langsamem Download dauern; danach gilt ein Job als verwaist. */ +const LAUFZEIT_SEKUNDEN = 4 * 3600; + +export type AuftragsJob = { auftragId: string }; + +let instanz: Promise | undefined; + +export function boss(rolle: "app" | "worker"): Promise { + instanz ??= (async () => { + const url = process.env["DATABASE_URL"]; + if (!url) throw new Error("DATABASE_URL ist nicht gesetzt."); + const b = new PgBoss({ + connectionString: url, + application_name: `proxdeploy-${rolle}`, + // Wartung und Zeitpläne macht nur der worker; die app stellt nur ein. + supervise: rolle === "worker", + schedule: false, + }); + b.on("error", (fehler) => logger.error({ err: fehler }, "pg-boss")); + await b.start(); + await b.createQueue(WARTESCHLANGE, { retryLimit: 0, expireInSeconds: LAUFZEIT_SEKUNDEN }); + return b; + })(); + return instanz; +} + +export async function auftragEinstellen(auftragId: string): Promise { + const b = await boss("app"); + await b.send(WARTESCHLANGE, { auftragId } satisfies AuftragsJob, { + retryLimit: 0, + expireInSeconds: LAUFZEIT_SEKUNDEN, + }); +} diff --git a/src/worker/ablaeufe.ts b/src/worker/ablaeufe.ts new file mode 100644 index 0000000..4578ab9 --- /dev/null +++ b/src/worker/ablaeufe.ts @@ -0,0 +1,4 @@ +/** + * Meldet alle Abläufe beim worker an. Template-Bau (#007), Ausrollen (#008) und Aufräumen folgen. + */ +export {}; diff --git a/src/worker/aufruf-log.ts b/src/worker/aufruf-log.ts new file mode 100644 index 0000000..06e4039 --- /dev/null +++ b/src/worker/aufruf-log.ts @@ -0,0 +1,20 @@ +import type { EreignisSenke } from "@/engine/ablauf/ablauf"; +import type { AufrufBeobachter } from "@/engine/proxmox/client"; + +/** + * Schreibt Proxmox-Aufrufe ins Detail-Log. Die Abfrageschleifen (Task-Status, Task-Log, + * VM-Status) erscheinen nur bei Fehlern – sonst stünden bei jedem Download Hunderte gleicher + * Zeilen da, und das Wichtige ginge unter. + */ +const SCHLEIFE = /\/tasks\/[^/]+\/(status|log)$|\/status\/current$/; + +export function aufrufLogger(senke: EreignisSenke): AufrufBeobachter { + return (aufruf, ergebnis) => { + const pfad = decodeURIComponent(aufruf.pfad); + if (!ergebnis.fehler && aufruf.methode === "GET" && SCHLEIFE.test(pfad)) return; + void senke.log( + ergebnis.fehler ? "warn" : "debug", + `${aufruf.methode} ${pfad} · ${ergebnis.dauerMs} ms${ergebnis.fehler ? ` · ${ergebnis.fehler.ursache}` : ""}`, + ); + }; +} diff --git a/src/worker/ausfuehren.ts b/src/worker/ausfuehren.ts new file mode 100644 index 0000000..83d0f2c --- /dev/null +++ b/src/worker/ausfuehren.ts @@ -0,0 +1,127 @@ +import type { Auftrag, Prisma } from "@/generated/prisma/client"; +import type { AuftragsArt } from "@/generated/prisma/enums"; +import type { AblaufErgebnis, EreignisSenke } from "@/engine/ablauf/ablauf"; +import { prisma } from "@/lib/db"; +import { logger } from "@/lib/logger"; +import { DatenbankSenke, melden } from "@/server/ereignisse/senke"; +import { reservierungenFreigeben } from "@/server/services/vmid"; + +/** + * Führt einen Auftrag aus: Status setzen, den passenden Ablauf starten, Ergebnis festhalten. + * Die Abläufe selbst (Template bauen, VM/LXC ausrollen, aufräumen) melden sich hier an. + */ +export type AblaufKontext = { + auftrag: Auftrag; + senke: EreignisSenke; + abbruchGewuenscht: () => Promise; + /** Merkt sich, was angelegt wurde – fürs Aufräumen, auch wenn später etwas scheitert. */ + angelegt: (eintrag: Record) => Promise; +}; + +export type AblaufAusgang = AblaufErgebnis & { ergebnis?: Record }; + +export type AblaufFunktion = (kontext: AblaufKontext) => Promise; + +const ABLAEUFE = new Map(); + +export function ablaufRegistrieren(art: AuftragsArt, funktion: AblaufFunktion): void { + ABLAEUFE.set(art, funktion); +} + +export async function auftragAusfuehren( + auftragId: string, + senke: EreignisSenke = new DatenbankSenke(auftragId), +) { + // Nur einmal starten, auch wenn pg-boss einen Job doppelt liefern sollte. + const gestartet = await prisma.auftrag.updateMany({ + where: { id: auftragId, status: "WARTEND" }, + data: { status: "LAEUFT", gestartetAm: new Date() }, + }); + if (gestartet.count === 0) { + logger.warn({ auftragId }, "Auftrag ist nicht (mehr) wartend – übersprungen"); + return; + } + const auftrag = await prisma.auftrag.findUniqueOrThrow({ where: { id: auftragId } }); + await melden(auftragId); + + let ausgang: AblaufAusgang; + const ablauf = ABLAEUFE.get(auftrag.art); + try { + if (!ablauf) throw new Error(`Für ${auftrag.art} ist kein Ablauf registriert.`); + ausgang = await ablauf({ + auftrag, + senke, + abbruchGewuenscht: async () => + (await prisma.auftrag.findUnique({ where: { id: auftragId }, select: { abbrechen: true } })) + ?.abbrechen ?? false, + angelegt: async (eintrag) => { + const aktuell = ( + await prisma.auftrag.findUnique({ where: { id: auftragId }, select: { angelegt: true } }) + )?.angelegt as Record | null; + await prisma.auftrag.update({ + where: { id: auftragId }, + data: { angelegt: { ...(aktuell ?? {}), ...eintrag } as Prisma.InputJsonValue }, + }); + }, + }); + } catch (fehler) { + // Fehler außerhalb einer Phase (Programmfehler) – trotzdem sauber beenden. + logger.error({ err: fehler, auftragId }, "Ablauf ist abgestürzt"); + const text = fehler instanceof Error ? fehler.message : String(fehler); + ausgang = { + ok: false, + abgebrochen: false, + phase: "", + fehler: { ursache: `Unerwarteter Fehler: ${text}`, hinweis: "Detail-Log prüfen." }, + }; + await senke.log("error", `Unerwarteter Fehler: ${text}`); + } + + await prisma.auftrag.update({ + where: { id: auftragId }, + data: { + status: ausgang.ok ? "ERFOLGREICH" : ausgang.abgebrochen ? "ABGEBROCHEN" : "FEHLGESCHLAGEN", + beendetAm: new Date(), + ...(ausgang.ok + ? { ergebnis: (ausgang.ergebnis ?? {}) as Prisma.InputJsonValue } + : { fehler: { phase: ausgang.phase, ...ausgang.fehler } }), + }, + }); + await reservierungenFreigeben(auftragId); + await melden(auftragId); +} + +/** + * Nach einem Absturz oder Neustart des workers hängen Aufträge auf LAEUFT. Sie werden ehrlich als + * fehlgeschlagen markiert – fortsetzen wäre riskanter als neu starten. + */ +export async function verwaisteAuftraegeBeenden(): Promise { + const verwaist = await prisma.auftrag.findMany({ where: { status: "LAEUFT" }, select: { id: true } }); + for (const { id } of verwaist) { + await prisma.$transaction([ + prisma.phase.updateMany({ + where: { auftragId: id, status: "LAEUFT" }, + data: { + status: "FEHLER", + beendetAm: new Date(), + ursache: "Der worker wurde neu gestartet.", + hinweis: "Auftrag wiederholen; Reste ggf. aufräumen.", + }, + }), + prisma.auftrag.update({ + where: { id }, + data: { + status: "FEHLGESCHLAGEN", + beendetAm: new Date(), + fehler: { + ursache: "Der worker wurde während des Auftrags neu gestartet.", + hinweis: "Auftrag wiederholen; Reste ggf. aufräumen.", + }, + }, + }), + ]); + await reservierungenFreigeben(id); + await melden(id); + } + return verwaist.length; +} diff --git a/src/worker/main.ts b/src/worker/main.ts new file mode 100644 index 0000000..9612d42 --- /dev/null +++ b/src/worker/main.ts @@ -0,0 +1,39 @@ +/** + * Der worker: arbeitet die Auftragswarteschlange ab. Derselbe Codestand wie die app, eigener + * Prozess – ein langer Template-Bau blockiert so nie eine Web-Anfrage. + */ +import "./umgebung"; +import { schluesselPruefen } from "@/lib/geheimnisse"; +import { logger } from "@/lib/logger"; +import { katalogSynchronisieren } from "@/server/services/katalog"; +import { boss, WARTESCHLANGE, type AuftragsJob } from "@/server/warteschlange"; +import { auftragAusfuehren, verwaisteAuftraegeBeenden } from "./ausfuehren"; +import "./ablaeufe"; + +async function starten(): Promise { + schluesselPruefen(); + const sync = await katalogSynchronisieren(); + logger.info({ distributionen: sync.distributionen, profile: sync.profile }, "Katalog abgeglichen"); + const verwaist = await verwaisteAuftraegeBeenden(); + if (verwaist > 0) logger.warn({ anzahl: verwaist }, "Verwaiste Aufträge als fehlgeschlagen markiert"); + + const parallel = Math.max(1, Number(process.env["WORKER_PARALLEL"] ?? 4)); + const b = await boss("worker"); + await b.work(WARTESCHLANGE, { batchSize: 1, localConcurrency: parallel }, async ([job]) => { + if (job) await auftragAusfuehren(job.data.auftragId); + }); + logger.info({ parallel }, "worker bereit"); + + const beenden = async (signal: string) => { + logger.info({ signal }, "worker fährt herunter"); + await b.stop({ graceful: true, timeout: 30_000 }); + process.exit(0); + }; + process.on("SIGTERM", () => void beenden("SIGTERM")); + process.on("SIGINT", () => void beenden("SIGINT")); +} + +starten().catch((fehler) => { + logger.fatal({ err: fehler }, "worker konnte nicht starten"); + process.exit(1); +}); diff --git a/src/worker/umgebung.ts b/src/worker/umgebung.ts new file mode 100644 index 0000000..06ef58a --- /dev/null +++ b/src/worker/umgebung.ts @@ -0,0 +1,11 @@ +/** + * Muss das erste Modul des workers sein: der Logger liest LOG_LEVEL und den Dienstnamen beim + * Import. Next lädt `.env` von selbst, der worker nicht – im Container kommt die Umgebung von + * außen, dann gibt es keine Datei und nichts passiert. + */ +try { + process.loadEnvFile(); +} catch { + // keine .env – Umgebung kommt von außen +} +process.env["PROXDEPLOY_DIENST"] = "worker"; diff --git a/tests/auftraege.test.ts b/tests/auftraege.test.ts new file mode 100644 index 0000000..891c232 --- /dev/null +++ b/tests/auftraege.test.ts @@ -0,0 +1,270 @@ +import { afterAll, beforeEach, describe, expect, it } from "vitest"; +import { GET as ereignisseGet } from "@/app/api/v1/auftraege/[id]/ereignisse/route"; +import { GET as auftragGet } from "@/app/api/v1/auftraege/[id]/route"; +import { POST as abbrechenPost } from "@/app/api/v1/auftraege/[id]/abbrechen/route"; +import { verschluesseln } from "@/lib/geheimnisse"; +import { prisma } from "@/lib/db"; +import { phasenAusfuehren } from "@/engine/ablauf/ablauf"; +import { auftragAnlegen } from "@/server/services/auftraege"; +import { vmidReservieren } from "@/server/services/vmid"; +import { ablaufRegistrieren, auftragAusfuehren, verwaisteAuftraegeBeenden } from "@/worker/ausfuehren"; +import { datenbankLeeren, datenbankVerfuegbar, testPrisma } from "./helpers/db"; +import { angemeldet, anfrage, kontext } from "./helpers/anfragen"; + +async function serverAnlegen() { + return await testPrisma.server.create({ + data: { + name: "rz1", + titel: "RZ 1", + url: "https://127.0.0.1:8006", + tokenId: "a@pve!b", + tokenGeheimnis: verschluesseln("x"), + fingerabdruck: "AA", + }, + }); +} + +/** Liest einen SSE-Strom, bis `ende` kommt, und gibt die Ereignisse zurück. */ +async function stromLesen( + antwort: Response, + nebenbei?: () => Promise, +): Promise> { + const leser = antwort.body!.getReader(); + const dekodierer = new TextDecoder(); + let puffer = ""; + const ereignisse: Array<{ art: string; daten: unknown; id?: string }> = []; + const parallel = nebenbei?.(); + for (;;) { + const { value, done } = await leser.read(); + if (done) break; + puffer += dekodierer.decode(value, { stream: true }); + let grenze: number; + while ((grenze = puffer.indexOf("\n\n")) >= 0) { + const block = puffer.slice(0, grenze); + puffer = puffer.slice(grenze + 2); + const art = /^event: (.+)$/m.exec(block)?.[1]; + const daten = /^data: (.+)$/m.exec(block)?.[1]; + const id = /^id: (.+)$/m.exec(block)?.[1]; + if (art && daten) ereignisse.push({ art, daten: JSON.parse(daten), id }); + } + } + await parallel; + return ereignisse; +} + +describe.runIf(datenbankVerfuegbar)("Aufträge, worker und Live-Strom", () => { + afterAll(async () => { + await prisma.$disconnect(); + }); + beforeEach(async () => { + await datenbankLeeren(); + }); + + it("VMID-Vergabe: zehn parallele Reservierungen bekommen zehn verschiedene IDs, Belegtes wird übersprungen", async () => { + const server = await serverAnlegen(); + const ids = await Promise.all( + Array.from({ length: 10 }, (_, i) => + vmidReservieren({ + serverId: server.id, + auftragId: `a${i}`, + belegt: new Set([100, 101, 103]), + von: 100, + }), + ), + ); + expect(new Set(ids).size).toBe(10); + expect(ids).not.toContain(100); + expect(ids).not.toContain(103); + expect(Math.min(...ids)).toBe(102); + await expect( + vmidReservieren({ serverId: server.id, auftragId: "x", belegt: new Set(), von: 100, wunsch: 102 }), + ).rejects.toThrow(/bereits belegt/); + }); + + it("worker führt einen Auftrag aus; SSE liefert Stand, Log und Ende; Reservierungen werden frei", async () => { + const server = await serverAnlegen(); + const { benutzer, cookie } = await angemeldet("OPERATOR"); + ablaufRegistrieren("AUFRAEUMEN", async ({ auftrag, senke }) => { + await vmidReservieren({ + serverId: auftrag.serverId, + auftragId: auftrag.id, + belegt: new Set(), + von: 100, + }); + const ergebnis = await phasenAusfuehren( + [ + { + schluessel: "eins", + titel: "Eins", + ausfuehren: async (k) => { + await k.log("Zeile 1"); + return { detail: "fertig" }; + }, + }, + { + schluessel: "zwei", + titel: "Zwei", + ausfuehren: async (k) => { + await k.log("Zeile 2"); + return {}; + }, + }, + ], + senke, + ); + return { ...ergebnis, ergebnis: { ip: "10.0.0.5" } }; + }); + const auftrag = await auftragAnlegen( + { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} }, + { benutzer, quelle: "WEB" }, + null, + async () => {}, + ); + expect(auftrag.kurzId).toMatch(/^c-[2-9a-z]{5}$/); + + const strom = await ereignisseGet( + anfrage("GET", `/api/v1/auftraege/${auftrag.kurzId}/ereignisse`, cookie), + kontext({ id: auftrag.kurzId }), + ); + expect(strom.headers.get("content-type")).toContain("text/event-stream"); + const ereignisse = await stromLesen(strom, () => auftragAusfuehren(auftrag.id)); + + const staende = ereignisse + .filter((e) => e.art === "stand") + .map((e) => (e.daten as { status: string }).status); + // Je nach Takt sieht der erste Stand den Auftrag wartend oder schon laufend – nie fertig. + expect(["WARTEND", "LAEUFT"]).toContain(staende[0]); + expect(staende.at(-1)).toBe("ERFOLGREICH"); + const logs = ereignisse + .filter((e) => e.art === "log") + .flatMap((e) => (e.daten as Array<{ nachricht: string }>).map((z) => z.nachricht)); + expect(logs).toEqual(["Zeile 1", "Zeile 2"]); + expect(ereignisse.at(-1)?.art).toBe("ende"); + + const fertig = await ( + await auftragGet( + anfrage("GET", `/api/v1/auftraege/${auftrag.kurzId}`, cookie), + kontext({ id: auftrag.kurzId }), + ) + ).json(); + expect( + fertig.phasen.map((p: { schluessel: string; status: string }) => `${p.schluessel}:${p.status}`), + ).toEqual(["eins:ERLEDIGT", "zwei:ERLEDIGT"]); + expect(fertig.ergebnis).toEqual({ ip: "10.0.0.5" }); + expect(await testPrisma.vmidReservierung.count()).toBe(0); + }); + + it("Wiederverbinden mit Last-Event-ID liefert nur neue Log-Zeilen", async () => { + const server = await serverAnlegen(); + const { benutzer, cookie } = await angemeldet("OPERATOR"); + ablaufRegistrieren("AUFRAEUMEN", async ({ senke }) => + phasenAusfuehren( + [ + { + schluessel: "a", + titel: "A", + ausfuehren: async (k) => { + await k.log("alt"); + await k.log("neu"); + return {}; + }, + }, + ], + senke, + ), + ); + const auftrag = await auftragAnlegen( + { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} }, + { benutzer, quelle: "WEB" }, + null, + async () => {}, + ); + await auftragAusfuehren(auftrag.id); + const erste = await testPrisma.logeintrag.findFirstOrThrow({ + where: { auftragId: auftrag.id, nachricht: "alt" }, + }); + + const strom = await ereignisseGet( + new Request(`http://localhost/api/v1/auftraege/${auftrag.kurzId}/ereignisse`, { + headers: { cookie, "last-event-id": erste.id.toString() }, + }), + kontext({ id: auftrag.kurzId }), + ); + const ereignisse = await stromLesen(strom); + const logs = ereignisse + .filter((e) => e.art === "log") + .flatMap((e) => (e.daten as Array<{ nachricht: string }>).map((z) => z.nachricht)); + expect(logs).toEqual(["neu"]); + }); + + it("Abbruch: Operator fordert an, der Ablauf endet vor der nächsten Phase als ABGEBROCHEN", async () => { + const server = await serverAnlegen(); + const { benutzer, cookie } = await angemeldet("OPERATOR"); + let auftragKurz = ""; + ablaufRegistrieren("AUFRAEUMEN", async ({ senke, abbruchGewuenscht }) => + phasenAusfuehren( + [ + { + schluessel: "a", + titel: "A", + ausfuehren: async () => { + await abbrechenPost( + anfrage("POST", `/api/v1/auftraege/${auftragKurz}/abbrechen`, cookie), + kontext({ id: auftragKurz }), + ); + return {}; + }, + }, + { schluessel: "b", titel: "B", ausfuehren: async () => ({}) }, + ], + senke, + { abbruchGewuenscht }, + ), + ); + const auftrag = await auftragAnlegen( + { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} }, + { benutzer, quelle: "WEB" }, + null, + async () => {}, + ); + auftragKurz = auftrag.kurzId; + await auftragAusfuehren(auftrag.id); + const fertig = await testPrisma.auftrag.findUniqueOrThrow({ + where: { id: auftrag.id }, + include: { phasen: { orderBy: { reihenfolge: "asc" } } }, + }); + expect(fertig.status).toBe("ABGEBROCHEN"); + expect(fertig.phasen.map((p) => p.status)).toEqual(["ERLEDIGT", "FEHLER"]); + expect(await testPrisma.protokoll.count({ where: { aktion: "Abbruch angefordert" } })).toBe(1); + }); + + it("ein doppelt gelieferter Job läuft nur einmal; verwaiste Aufträge werden beim Start beendet", async () => { + const server = await serverAnlegen(); + const { benutzer } = await angemeldet("OPERATOR"); + let laeufe = 0; + ablaufRegistrieren("AUFRAEUMEN", async () => { + laeufe += 1; + return { ok: true }; + }); + const auftrag = await auftragAnlegen( + { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} }, + { benutzer, quelle: "WEB" }, + null, + async () => {}, + ); + await Promise.all([auftragAusfuehren(auftrag.id), auftragAusfuehren(auftrag.id)]); + expect(laeufe).toBe(1); + + const haengt = await auftragAnlegen( + { art: "AUFRAEUMEN", serverId: server.id, anfrage: {} }, + { benutzer, quelle: "WEB" }, + null, + async () => {}, + ); + await testPrisma.auftrag.update({ where: { id: haengt.id }, data: { status: "LAEUFT" } }); + expect(await verwaisteAuftraegeBeenden()).toBe(1); + expect((await testPrisma.auftrag.findUniqueOrThrow({ where: { id: haengt.id } })).status).toBe( + "FEHLGESCHLAGEN", + ); + }); +}); diff --git a/tests/engine/ablauf.test.ts b/tests/engine/ablauf.test.ts new file mode 100644 index 0000000..f0d4f07 --- /dev/null +++ b/tests/engine/ablauf.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, it } from "vitest"; +import { phasenAusfuehren, type EreignisSenke } from "@/engine/ablauf/ablauf"; +import { AblaufFehler } from "@/engine/ablauf/fehler"; +import { ProxmoxFehler } from "@/engine/proxmox/fehler"; + +/** Eine Senke, die mitschreibt – so sieht ein Test genau, was Web und CLI zu sehen bekämen. */ +function mitschreiber() { + const ereignisse: string[] = []; + const senke: EreignisSenke = { + phasenAnlegen: async (p) => void ereignisse.push(`anlegen ${p.map((x) => x.schluessel).join(",")}`), + phaseGestartet: async (s) => void ereignisse.push(`start ${s}`), + phaseDetail: async (s, d) => void ereignisse.push(`detail ${s}: ${d}`), + phaseBeendet: async (s, status, a) => + void ereignisse.push( + `ende ${s} ${status}${a.detail ? `: ${a.detail}` : ""}${a.fehler ? ` – ${a.fehler.ursache} → ${a.fehler.hinweis}` : ""}`, + ), + log: async (stufe, n) => void ereignisse.push(`log ${stufe} ${n}`), + }; + return { senke, ereignisse }; +} + +describe("Phasen-Runner", () => { + it("führt Phasen der Reihe nach aus und meldet Detail, Überspringen und Erfolg", async () => { + const { senke, ereignisse } = mitschreiber(); + const ergebnis = await phasenAusfuehren( + [ + { schluessel: "pruefen", titel: "Prüfungen", ausfuehren: async () => ({ detail: "VMID 120 frei" }) }, + { + schluessel: "template", + titel: "Template", + ausfuehren: async () => ({ uebersprungen: true, detail: "9013 vorhanden" }), + }, + { + schluessel: "klonen", + titel: "Klonen", + ausfuehren: async (k) => { + await k.detail("läuft …"); + await k.log("clone gestartet"); + return { detail: "full clone 9013 → 120" }; + }, + }, + ], + senke, + ); + expect(ergebnis).toEqual({ ok: true }); + expect(ereignisse).toEqual([ + "anlegen pruefen,template,klonen", + "start pruefen", + "ende pruefen ERLEDIGT: VMID 120 frei", + "start template", + "ende template UEBERSPRUNGEN: 9013 vorhanden", + "start klonen", + "detail klonen: läuft …", + "log info clone gestartet", + "ende klonen ERLEDIGT: full clone 9013 → 120", + ]); + }); + + it("bricht bei einem Fehler ab: Ursache und Hinweis, folgende Phasen bleiben offen", async () => { + const { senke, ereignisse } = mitschreiber(); + let danach = false; + const ergebnis = await phasenAusfuehren( + [ + { + schluessel: "klonen", + titel: "Klonen", + ausfuehren: async () => { + throw new ProxmoxFehler( + "recht", + "Permission check failed (/vms/120, VM.Clone)", + "Der Rolle fehlt VM.Clone auf /vms/120.", + ); + }, + }, + { schluessel: "start", titel: "Start", ausfuehren: async () => void (danach = true) }, + ], + senke, + ); + expect(ergebnis).toMatchObject({ ok: false, abgebrochen: false, phase: "klonen" }); + expect(danach).toBe(false); + expect(ereignisse).toContain( + "ende klonen FEHLER – Permission check failed (/vms/120, VM.Clone) → Der Rolle fehlt VM.Clone auf /vms/120.", + ); + expect(ereignisse.some((e) => e.startsWith("start start"))).toBe(false); + }); + + it("prüft vor jeder Phase auf Abbruch", async () => { + const { senke } = mitschreiber(); + let abbrechen = false; + const ergebnis = await phasenAusfuehren( + [ + { schluessel: "a", titel: "A", ausfuehren: async () => void (abbrechen = true) }, + { schluessel: "b", titel: "B", ausfuehren: async () => ({}) }, + ], + senke, + { abbruchGewuenscht: async () => abbrechen }, + ); + expect(ergebnis).toMatchObject({ ok: false, abgebrochen: true, phase: "b" }); + }); + + it("benennt unerwartete Fehler ehrlich", async () => { + const { senke } = mitschreiber(); + const ergebnis = await phasenAusfuehren( + [ + { + schluessel: "x", + titel: "X", + ausfuehren: async () => { + throw new TypeError("kaputt"); + }, + }, + ], + senke, + ); + expect(ergebnis.ok || ergebnis.fehler.ursache).toBe("Unerwarteter Fehler: kaputt"); + const bewusst = await phasenAusfuehren( + [ + { + schluessel: "y", + titel: "Y", + ausfuehren: async () => { + throw new AblaufFehler("U", "H"); + }, + }, + ], + senke, + ); + expect(bewusst.ok || bewusst.fehler).toEqual({ ursache: "U", hinweis: "H" }); + }); +}); diff --git a/tests/zugang.test.ts b/tests/zugang.test.ts index 0ac1919..8565621 100644 --- a/tests/zugang.test.ts +++ b/tests/zugang.test.ts @@ -51,3 +51,14 @@ describe("Zugangsprüfung", () => { expect(istOeffentlich("/server")).toBe(false); }); }); + +describe("Client-Komponenten", () => { + const clientDateien = dateien(join(wurzel, "src")) + .filter((pfad) => /\.tsx?$/.test(pfad) && !pfad.includes("/generated/")) + .filter((pfad) => /^["']use client["']/m.test(readFileSync(pfad, "utf8"))); + + it.each(clientDateien.map((p) => relative(wurzel, p)))("%s importiert nichts aus src/server", (pfad) => { + // Auch reine Typimporte nicht: sonst wandert beim nächsten Umbau doch Servercode ins Browser-Bündel. + expect(readFileSync(join(wurzel, pfad), "utf8")).not.toMatch(/from ["']@\/server\//); + }); +});