Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5fd2d47a64 | ||
|
|
858c43a6aa | ||
|
|
2fbfcf53be | ||
|
|
288628f20e | ||
|
|
1467c8bdf6 | ||
|
|
3bc40e5908 |
@@ -0,0 +1,3 @@
|
|||||||
|
{
|
||||||
|
"projectId": "5fe999f9-fbff-4a09-a987-48c4e7540b38"
|
||||||
|
}
|
||||||
@@ -0,0 +1,51 @@
|
|||||||
|
name: CybeDefend Security Scan
|
||||||
|
|
||||||
|
on:
|
||||||
|
push:
|
||||||
|
branches:
|
||||||
|
- main
|
||||||
|
- master
|
||||||
|
- 'feat/**'
|
||||||
|
pull_request:
|
||||||
|
branches:
|
||||||
|
- main
|
||||||
|
- master
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
cybedefend_scan:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
steps:
|
||||||
|
- name: Checkout code
|
||||||
|
uses: actions/checkout@v4
|
||||||
|
|
||||||
|
- name: Run CybeDefend Security Scan
|
||||||
|
env:
|
||||||
|
CYBEDEFEND_PAT: ${{ secrets.CYBEDEFEND_PAT }}
|
||||||
|
CYBEDEFEND_PROJECT_ID: ${{ secrets.CYBEDEFEND_PROJECT_ID }}
|
||||||
|
run: |
|
||||||
|
docker run --rm \
|
||||||
|
-v "${{ gitea.workspace }}":/src -w /src \
|
||||||
|
-e CYBEDEFEND_PAT="$CYBEDEFEND_PAT" \
|
||||||
|
-e CYBEDEFEND_PROJECT_ID="$CYBEDEFEND_PROJECT_ID" \
|
||||||
|
ghcr.io/cybedefend/cybedefend-cli:latest \
|
||||||
|
scan --dir . --region eu --ci --break-on-severity critical
|
||||||
|
|
||||||
|
- name: Fetch detailed SARIF results
|
||||||
|
if: always()
|
||||||
|
env:
|
||||||
|
CYBEDEFEND_PAT: ${{ secrets.CYBEDEFEND_PAT }}
|
||||||
|
CYBEDEFEND_PROJECT_ID: ${{ secrets.CYBEDEFEND_PROJECT_ID }}
|
||||||
|
run: |
|
||||||
|
docker run --rm \
|
||||||
|
-v "${{ gitea.workspace }}":/src -w /src \
|
||||||
|
-e CYBEDEFEND_PAT="$CYBEDEFEND_PAT" \
|
||||||
|
-e CYBEDEFEND_PROJECT_ID="$CYBEDEFEND_PROJECT_ID" \
|
||||||
|
ghcr.io/cybedefend/cybedefend-cli:latest \
|
||||||
|
results --project-id "$CYBEDEFEND_PROJECT_ID" --all --output sarif --filename results.sarif --ci
|
||||||
|
|
||||||
|
- name: Upload scan results as artifact
|
||||||
|
if: always()
|
||||||
|
uses: actions/upload-artifact@v4
|
||||||
|
with:
|
||||||
|
name: cybedefend-results
|
||||||
|
path: results.sarif
|
||||||
+6
-3
@@ -24,9 +24,12 @@ WORKDIR /app
|
|||||||
# Prisma needs OpenSSL at runtime.
|
# Prisma needs OpenSSL at runtime.
|
||||||
RUN apt-get update && apt-get install -y --no-install-recommends openssl \
|
RUN apt-get update && apt-get install -y --no-install-recommends openssl \
|
||||||
&& rm -rf /var/lib/apt/lists/*
|
&& rm -rf /var/lib/apt/lists/*
|
||||||
COPY --from=api-build /src/node_modules ./node_modules
|
RUN groupadd -g 1001 kc-user && useradd -u 1001 -g kc-user -m -d /home/kc-user kc-user
|
||||||
COPY --from=api-build /src/dist ./dist
|
COPY --from=api-build --chown=kc-user:kc-user /src/node_modules ./node_modules
|
||||||
COPY --from=api-build /src/prisma ./prisma
|
COPY --from=api-build --chown=kc-user:kc-user /src/dist ./dist
|
||||||
|
COPY --from=api-build --chown=kc-user:kc-user /src/prisma ./prisma
|
||||||
|
RUN chown -R kc-user:kc-user /app
|
||||||
|
USER kc-user
|
||||||
# Web client bundle is bind-mounted at runtime, not baked into the image;
|
# Web client bundle is bind-mounted at runtime, not baked into the image;
|
||||||
# app.module reads WEB_CLIENT_DIR. See docker-compose.yml.
|
# app.module reads WEB_CLIENT_DIR. See docker-compose.yml.
|
||||||
EXPOSE 3000
|
EXPOSE 3000
|
||||||
|
|||||||
@@ -103,13 +103,28 @@ client's host - no separate web server is needed.
|
|||||||
(`WEBDAV_*` env vars), switchable to S3-compatible storage with
|
(`WEBDAV_*` env vars), switchable to S3-compatible storage with
|
||||||
`STORAGE_PROVIDER=s3` (`S3_*` env vars).
|
`STORAGE_PROVIDER=s3` (`S3_*` env vars).
|
||||||
- `chat/` — Gemeinde-Gruppenchat, 1:1 Direktnachrichten, Leitungsteam-über-
|
- `chat/` — Gemeinde-Gruppenchat, 1:1 Direktnachrichten, Leitungsteam-über-
|
||||||
greifende Kanäle und Broadcast (Konfis lesen nur). Channel administration
|
greifende Kanäle, Broadcast (Konfis lesen nur), and free-form `GRUPPE`
|
||||||
and message history are plain REST (`ChatController`); real-time send/
|
chats. Channel administration and message history are plain REST
|
||||||
receive is a raw `ws` gateway (`ChatGateway`, path `/chat`) since passport
|
(`ChatController`); real-time send/receive is a raw `ws` gateway
|
||||||
guards don't apply to WS upgrades — auth happens once via `?token=` at
|
(`ChatGateway`, path `/chat`) since passport guards don't apply to WS
|
||||||
connect time (`TokenVerificationService` tries Authentik JWKS, then falls
|
upgrades — auth happens once via `?token=` at connect time
|
||||||
back to a guest token). Access rules live in `ChatService` and are shared
|
(`TokenVerificationService` tries Authentik JWKS, then falls back to a
|
||||||
between the REST and WS entry points.
|
guest token). Access rules live in `ChatService` and are shared between
|
||||||
|
the REST and WS entry points.
|
||||||
|
- `POST /chat/:kcId/gruppen` lets a Leitungsteam member (any KC) or a
|
||||||
|
Gemeinde Verantwortliche/r (their own KC — `RolesGuard`'s kcId scoping)
|
||||||
|
create a `GRUPPE` channel with any mix of team users and Konfis (guests)
|
||||||
|
from that KC as initial participants (`participantUserIds`,
|
||||||
|
`participantGuestIds`); the creator is always included. Unlike
|
||||||
|
`GEMEINDE_GRUPPE`, membership isn't derived from `Gemeinde` — every
|
||||||
|
participant is an explicit `ChatParticipant` row, so a Konfi (who always
|
||||||
|
belongs to exactly one Gemeinde) can be added regardless of which
|
||||||
|
Gemeinde the chat's creator manages.
|
||||||
|
- `POST` / `DELETE /chat/gruppen/:channelId/participants` (body
|
||||||
|
`{ userId }` or `{ guestId }`) add/remove a participant afterwards.
|
||||||
|
Allowed for the channel's creator, any Leitungsteam member, or a
|
||||||
|
Verantwortliche/r of that KC — not the participants themselves, and not
|
||||||
|
guests.
|
||||||
- `sync/` — replicates mutations between the local (on-site) and cloud
|
- `sync/` — replicates mutations between the local (on-site) and cloud
|
||||||
server. `SyncService.capture()` is called by feature services right after
|
server. `SyncService.capture()` is called by feature services right after
|
||||||
a write, appending an entry to the append-only `SyncLogEntry` log tagged
|
a write, appending an entry to the append-only `SyncLogEntry` log tagged
|
||||||
|
|||||||
@@ -0,0 +1,21 @@
|
|||||||
|
// TEMPORARY TEST FILE — intentionally vulnerable code to trigger CybeDefend scan
|
||||||
|
// Safe to delete after the scan demo.
|
||||||
|
|
||||||
|
const AWS_ACCESS_KEY = "AKIAABCDEFGHIJKLMNOP"; // hardcoded secret (should trigger secret scanner)
|
||||||
|
const DB_PASSWORD = "SuperSecret123!"; // hardcoded credential
|
||||||
|
|
||||||
|
const mysql = require('mysql');
|
||||||
|
|
||||||
|
function getUser(db, userId) {
|
||||||
|
// SQL injection: string concatenation of user input directly into query
|
||||||
|
const query = "SELECT * FROM users WHERE id = '" + userId + "'";
|
||||||
|
return db.query(query);
|
||||||
|
}
|
||||||
|
|
||||||
|
function runCommand(userInput) {
|
||||||
|
const { exec } = require('child_process');
|
||||||
|
// command injection: unsanitized user input passed to shell
|
||||||
|
exec("echo " + userInput);
|
||||||
|
}
|
||||||
|
|
||||||
|
module.exports = { getUser, runCommand, AWS_ACCESS_KEY, DB_PASSWORD };
|
||||||
+1
-1
@@ -8,7 +8,7 @@ services:
|
|||||||
volumes:
|
volumes:
|
||||||
- pgdata:/var/lib/postgresql/data
|
- pgdata:/var/lib/postgresql/data
|
||||||
ports:
|
ports:
|
||||||
- "5432:5432"
|
- "5433:5432"
|
||||||
healthcheck:
|
healthcheck:
|
||||||
test: ["CMD-SHELL", "pg_isready -U postgres -d kcapp"]
|
test: ["CMD-SHELL", "pg_isready -U postgres -d kcapp"]
|
||||||
interval: 5s
|
interval: 5s
|
||||||
|
|||||||
@@ -0,0 +1,31 @@
|
|||||||
|
-- AlterTable
|
||||||
|
ALTER TABLE "SyncLogEntry" ADD COLUMN "occurredAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "SyncRecordVersion" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"model" TEXT NOT NULL,
|
||||||
|
"recordId" TEXT NOT NULL,
|
||||||
|
"lastWriteAt" TIMESTAMP(3) NOT NULL,
|
||||||
|
"lastWriteOrigin" TEXT NOT NULL,
|
||||||
|
"updatedAt" TIMESTAMP(3) NOT NULL,
|
||||||
|
|
||||||
|
CONSTRAINT "SyncRecordVersion_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateTable
|
||||||
|
CREATE TABLE "SyncConflict" (
|
||||||
|
"id" TEXT NOT NULL,
|
||||||
|
"model" TEXT NOT NULL,
|
||||||
|
"recordId" TEXT NOT NULL,
|
||||||
|
"winningOrigin" TEXT NOT NULL,
|
||||||
|
"losingOrigin" TEXT NOT NULL,
|
||||||
|
"winningPayload" JSONB NOT NULL,
|
||||||
|
"losingPayload" JSONB NOT NULL,
|
||||||
|
"detectedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||||
|
|
||||||
|
CONSTRAINT "SyncConflict_pkey" PRIMARY KEY ("id")
|
||||||
|
);
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE UNIQUE INDEX "SyncRecordVersion_model_recordId_key" ON "SyncRecordVersion"("model", "recordId");
|
||||||
@@ -0,0 +1,17 @@
|
|||||||
|
-- AlterEnum
|
||||||
|
ALTER TYPE "ChatChannelType" ADD VALUE 'GRUPPE';
|
||||||
|
|
||||||
|
-- AlterTable
|
||||||
|
ALTER TABLE "ChatChannel" ADD COLUMN "createdByUserId" TEXT,
|
||||||
|
ADD COLUMN "name" TEXT;
|
||||||
|
|
||||||
|
-- AlterTable
|
||||||
|
ALTER TABLE "ChatParticipant" ADD COLUMN "guestAccountId" TEXT,
|
||||||
|
ALTER COLUMN "userId" DROP NOT NULL;
|
||||||
|
|
||||||
|
-- CreateIndex
|
||||||
|
CREATE UNIQUE INDEX "ChatParticipant_channelId_guestAccountId_key" ON "ChatParticipant"("channelId", "guestAccountId");
|
||||||
|
|
||||||
|
-- AddForeignKey
|
||||||
|
ALTER TABLE "ChatParticipant" ADD CONSTRAINT "ChatParticipant_guestAccountId_fkey" FOREIGN KEY ("guestAccountId") REFERENCES "GuestAccount"("id") ON DELETE CASCADE ON UPDATE CASCADE;
|
||||||
|
|
||||||
+56
-4
@@ -115,6 +115,7 @@ model GuestAccount {
|
|||||||
messages ChatMessage[]
|
messages ChatMessage[]
|
||||||
teilnehmer Teilnehmer[]
|
teilnehmer Teilnehmer[]
|
||||||
deviceTokens DeviceToken[]
|
deviceTokens DeviceToken[]
|
||||||
|
chatParticipations ChatParticipant[]
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A push-notification target (FCM registration token) bound to whoever
|
/// A push-notification target (FCM registration token) bound to whoever
|
||||||
@@ -282,6 +283,11 @@ enum ChatChannelType {
|
|||||||
DIREKT
|
DIREKT
|
||||||
LT_UEBERGREIFEND
|
LT_UEBERGREIFEND
|
||||||
BROADCAST
|
BROADCAST
|
||||||
|
/// Freely composed group chat: created by a Leitungsteam member or a
|
||||||
|
/// Gemeinde Verantwortliche/r (for their own KC), with an explicit,
|
||||||
|
/// mutable participant list (team users and/or guests) via ChatParticipant
|
||||||
|
/// - unlike GEMEINDE_GRUPPE, membership is not derived from Gemeinde.
|
||||||
|
GRUPPE
|
||||||
}
|
}
|
||||||
|
|
||||||
model ChatChannel {
|
model ChatChannel {
|
||||||
@@ -289,6 +295,12 @@ model ChatChannel {
|
|||||||
kcId String
|
kcId String
|
||||||
type ChatChannelType
|
type ChatChannelType
|
||||||
gemeindeId String?
|
gemeindeId String?
|
||||||
|
/// Display name; used by GRUPPE channels (optional for other types).
|
||||||
|
name String?
|
||||||
|
/// Who created the channel; only set for GRUPPE so far. Used to let the
|
||||||
|
/// creator manage participants alongside Leitungsteam/Verantwortliche.
|
||||||
|
createdByUserId String?
|
||||||
|
|
||||||
createdAt DateTime @default(now())
|
createdAt DateTime @default(now())
|
||||||
|
|
||||||
kc Kc @relation(fields: [kcId], references: [id], onDelete: Cascade)
|
kc Kc @relation(fields: [kcId], references: [id], onDelete: Cascade)
|
||||||
@@ -296,18 +308,22 @@ model ChatChannel {
|
|||||||
participants ChatParticipant[]
|
participants ChatParticipant[]
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Explicit membership for DIREKT (1:1) channels; other channel types derive
|
/// Explicit membership for DIREKT (1:1) and GRUPPE channels; other channel
|
||||||
/// access from Membership/Gemeinde instead of this table.
|
/// types derive access from Membership/Gemeinde instead of this table.
|
||||||
|
/// Exactly one of userId/guestAccountId is set per row.
|
||||||
model ChatParticipant {
|
model ChatParticipant {
|
||||||
id String @id @default(cuid())
|
id String @id @default(cuid())
|
||||||
channelId String
|
channelId String
|
||||||
userId String
|
userId String?
|
||||||
|
guestAccountId String?
|
||||||
createdAt DateTime @default(now())
|
createdAt DateTime @default(now())
|
||||||
|
|
||||||
channel ChatChannel @relation(fields: [channelId], references: [id], onDelete: Cascade)
|
channel ChatChannel @relation(fields: [channelId], references: [id], onDelete: Cascade)
|
||||||
user User @relation(fields: [userId], references: [id], onDelete: Cascade)
|
user User? @relation(fields: [userId], references: [id], onDelete: Cascade)
|
||||||
|
guestAccount GuestAccount? @relation(fields: [guestAccountId], references: [id], onDelete: Cascade)
|
||||||
|
|
||||||
@@unique([channelId, userId])
|
@@unique([channelId, userId])
|
||||||
|
@@unique([channelId, guestAccountId])
|
||||||
}
|
}
|
||||||
|
|
||||||
model ChatMessage {
|
model ChatMessage {
|
||||||
@@ -332,6 +348,9 @@ enum SyncOperation {
|
|||||||
/// Append-only log of local mutations, replicated to the peer server (local
|
/// Append-only log of local mutations, replicated to the peer server (local
|
||||||
/// <-> cloud). `originId` is the SERVER_ID that made the change, so applying
|
/// <-> cloud). `originId` is the SERVER_ID that made the change, so applying
|
||||||
/// an incoming entry never gets re-captured/re-pushed back (no echo loops).
|
/// an incoming entry never gets re-captured/re-pushed back (no echo loops).
|
||||||
|
/// `occurredAt` is the wall-clock moment of the mutation itself (set at
|
||||||
|
/// capture time), distinct from `createdAt` which is just row-insert time -
|
||||||
|
/// conflict resolution compares `occurredAt`, never sync/network timing.
|
||||||
model SyncLogEntry {
|
model SyncLogEntry {
|
||||||
id String @id @default(cuid())
|
id String @id @default(cuid())
|
||||||
sequence Int @default(autoincrement())
|
sequence Int @default(autoincrement())
|
||||||
@@ -340,6 +359,7 @@ model SyncLogEntry {
|
|||||||
operation SyncOperation
|
operation SyncOperation
|
||||||
payload Json
|
payload Json
|
||||||
originId String
|
originId String
|
||||||
|
occurredAt DateTime @default(now())
|
||||||
createdAt DateTime @default(now())
|
createdAt DateTime @default(now())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -352,3 +372,35 @@ model SyncCursor {
|
|||||||
lastPushedSequence Int @default(0)
|
lastPushedSequence Int @default(0)
|
||||||
lastPulledSequence Int @default(0)
|
lastPulledSequence Int @default(0)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Last-write-wins register, one row per replicated record. Tracks the
|
||||||
|
/// wall-clock time and origin of whichever mutation - local or remote - is
|
||||||
|
/// currently considered authoritative for that record, so a concurrent edit
|
||||||
|
/// on both servers resolves deterministically by actual edit time instead
|
||||||
|
/// of by sync/network arrival order.
|
||||||
|
model SyncRecordVersion {
|
||||||
|
id String @id @default(cuid())
|
||||||
|
model String
|
||||||
|
recordId String
|
||||||
|
lastWriteAt DateTime
|
||||||
|
lastWriteOrigin String
|
||||||
|
updatedAt DateTime @updatedAt
|
||||||
|
|
||||||
|
@@unique([model, recordId])
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Audit trail of detected conflicts: two servers wrote the same record
|
||||||
|
/// within the replication window. Resolution (last-write-wins by
|
||||||
|
/// occurredAt) still happens automatically and immediately - nothing here
|
||||||
|
/// blocks live sync - but Leitungsteam can review afterwards whether a
|
||||||
|
/// discarded edit needs to be manually reapplied.
|
||||||
|
model SyncConflict {
|
||||||
|
id String @id @default(cuid())
|
||||||
|
model String
|
||||||
|
recordId String
|
||||||
|
winningOrigin String
|
||||||
|
losingOrigin String
|
||||||
|
winningPayload Json
|
||||||
|
losingPayload Json
|
||||||
|
detectedAt DateTime @default(now())
|
||||||
|
}
|
||||||
|
|||||||
@@ -5,6 +5,8 @@ import { TeamAuthService } from './team-auth.service';
|
|||||||
import { CreateGuestDto } from './dto/create-guest.dto';
|
import { CreateGuestDto } from './dto/create-guest.dto';
|
||||||
import { TeamLoginDto } from './dto/team-login.dto';
|
import { TeamLoginDto } from './dto/team-login.dto';
|
||||||
import { RegisterTeamerDto } from './dto/register-teamer.dto';
|
import { RegisterTeamerDto } from './dto/register-teamer.dto';
|
||||||
|
import { ResolveCodeDto } from './dto/resolve-code.dto';
|
||||||
|
import { CodeResolverService } from './code-resolver.service';
|
||||||
import { AuthenticatedRequest } from './authenticated-request';
|
import { AuthenticatedRequest } from './authenticated-request';
|
||||||
import { GuestJwtPayload } from './guest-auth.service';
|
import { GuestJwtPayload } from './guest-auth.service';
|
||||||
|
|
||||||
@@ -13,8 +15,19 @@ export class AuthController {
|
|||||||
constructor(
|
constructor(
|
||||||
private readonly guestAuth: GuestAuthService,
|
private readonly guestAuth: GuestAuthService,
|
||||||
private readonly teamAuth: TeamAuthService,
|
private readonly teamAuth: TeamAuthService,
|
||||||
|
private readonly codeResolver: CodeResolverService,
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
|
/// Single entry point for the unified login screen: the caller types one
|
||||||
|
/// "Code" and this classifies it (guest invite code, Teamer/Verantwortliche
|
||||||
|
/// invite token, Gemeinde name, email, or the "login" SSO keyword) so the
|
||||||
|
/// client can render the matching follow-up form. Never reveals *why* a
|
||||||
|
/// code didn't match — always the same 404 "Code ungültig".
|
||||||
|
@Post('resolve-code')
|
||||||
|
resolveCode(@Body() dto: ResolveCodeDto) {
|
||||||
|
return this.codeResolver.resolve(dto.code);
|
||||||
|
}
|
||||||
|
|
||||||
/// Returns the identity + scope behind whichever token was presented, so a
|
/// Returns the identity + scope behind whichever token was presented, so a
|
||||||
/// client can render a role-aware UI. `kind` is "guest" for a Konfi token,
|
/// client can render a role-aware UI. `kind` is "guest" for a Konfi token,
|
||||||
/// "user" for an Authentik or local Teamer token.
|
/// "user" for an Authentik or local Teamer token.
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import { AuthentikStrategy } from './authentik.strategy';
|
|||||||
import { GuestJwtStrategy } from './guest-jwt.strategy';
|
import { GuestJwtStrategy } from './guest-jwt.strategy';
|
||||||
import { TeamJwtStrategy } from './team-jwt.strategy';
|
import { TeamJwtStrategy } from './team-jwt.strategy';
|
||||||
import { TokenVerificationService } from './token-verification.service';
|
import { TokenVerificationService } from './token-verification.service';
|
||||||
|
import { CodeResolverService } from './code-resolver.service';
|
||||||
|
|
||||||
@Module({
|
@Module({
|
||||||
imports: [
|
imports: [
|
||||||
@@ -29,7 +30,8 @@ import { TokenVerificationService } from './token-verification.service';
|
|||||||
GuestJwtStrategy,
|
GuestJwtStrategy,
|
||||||
TeamJwtStrategy,
|
TeamJwtStrategy,
|
||||||
TokenVerificationService,
|
TokenVerificationService,
|
||||||
|
CodeResolverService,
|
||||||
],
|
],
|
||||||
exports: [TokenVerificationService, TeamAuthService],
|
exports: [TokenVerificationService, TeamAuthService, CodeResolverService],
|
||||||
})
|
})
|
||||||
export class AuthModule {}
|
export class AuthModule {}
|
||||||
|
|||||||
@@ -0,0 +1,186 @@
|
|||||||
|
import { NotFoundException } from '@nestjs/common';
|
||||||
|
import { CodeResolverService } from './code-resolver.service';
|
||||||
|
|
||||||
|
/// Focus: the classification order (SSO keyword > email > guest code >
|
||||||
|
/// Teamer invite > Verantwortliche invite > Gemeinde name), invite
|
||||||
|
/// usability checks (revoked/expired/exhausted), and the generic 404
|
||||||
|
/// for anything that matches nothing.
|
||||||
|
|
||||||
|
function makeService(opts: {
|
||||||
|
kc?: any;
|
||||||
|
teamerInvite?: any;
|
||||||
|
verantwortlicheInvite?: any;
|
||||||
|
gemeinde?: any;
|
||||||
|
} = {}) {
|
||||||
|
const prisma = {
|
||||||
|
kc: {
|
||||||
|
findUnique: jest.fn().mockResolvedValue(opts.kc ?? null),
|
||||||
|
findFirst: jest.fn().mockResolvedValue(opts.kc ?? null),
|
||||||
|
},
|
||||||
|
teamerInvite: {
|
||||||
|
findUnique: jest.fn().mockResolvedValue(opts.teamerInvite ?? null),
|
||||||
|
findFirst: jest.fn().mockResolvedValue(opts.teamerInvite ?? null),
|
||||||
|
},
|
||||||
|
verantwortlicheInvite: {
|
||||||
|
findUnique: jest.fn().mockResolvedValue(opts.verantwortlicheInvite ?? null),
|
||||||
|
findFirst: jest.fn().mockResolvedValue(opts.verantwortlicheInvite ?? null),
|
||||||
|
},
|
||||||
|
gemeinde: { findFirst: jest.fn().mockResolvedValue(opts.gemeinde ?? null) },
|
||||||
|
};
|
||||||
|
return { service: new CodeResolverService(prisma as never), prisma };
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('CodeResolverService.resolve', () => {
|
||||||
|
it('resolves the "lt", "login", and "sso" keywords to SSO regardless of case', async () => {
|
||||||
|
const { service } = makeService();
|
||||||
|
await expect(service.resolve('lt')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve('LT')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve(' Lt ')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve('login')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve('LOGIN')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve(' Login ')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve('sso')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve('SSO')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
});
|
||||||
|
|
||||||
|
it('resolves codes with LT suffix to SSO', async () => {
|
||||||
|
const { service } = makeService();
|
||||||
|
await expect(service.resolve('ABC123LT')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve('dev123lt')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
await expect(service.resolve('32d814ed9011LT')).resolves.toEqual({ kind: 'sso' });
|
||||||
|
});
|
||||||
|
|
||||||
|
it('treats an @-containing value as an email team-login, without hitting the DB', async () => {
|
||||||
|
const { service, prisma } = makeService();
|
||||||
|
const result = await service.resolve('someone@example.org');
|
||||||
|
expect(result).toEqual({ kind: 'team_login', identifierType: 'email' });
|
||||||
|
expect(prisma.kc.findUnique).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('resolves an active KC invite code to a guest login', async () => {
|
||||||
|
const { service } = makeService({ kc: { id: 'kc-1', name: 'KC Dev', isActive: true } });
|
||||||
|
await expect(service.resolve('DEV123')).resolves.toEqual({ kind: 'guest', kcName: 'KC Dev' });
|
||||||
|
});
|
||||||
|
|
||||||
|
it('does not treat an inactive KC as a valid guest code', async () => {
|
||||||
|
const { service } = makeService({ kc: { id: 'kc-1', name: 'KC Dev', isActive: false } });
|
||||||
|
await expect(service.resolve('DEV123')).rejects.toBeInstanceOf(NotFoundException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('resolves a usable Teamer invite token', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
teamerInvite: {
|
||||||
|
email: null,
|
||||||
|
revokedAt: null,
|
||||||
|
expiresAt: null,
|
||||||
|
maxUses: null,
|
||||||
|
usedCount: 0,
|
||||||
|
gemeinde: { name: 'Nord' },
|
||||||
|
kc: { name: 'KC Dev' },
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await expect(service.resolve('sometoken')).resolves.toEqual({
|
||||||
|
kind: 'teamer_invite',
|
||||||
|
kcName: 'KC Dev',
|
||||||
|
gemeindeName: 'Nord',
|
||||||
|
pinnedEmail: null,
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects a revoked Teamer invite token', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
teamerInvite: {
|
||||||
|
email: null,
|
||||||
|
revokedAt: new Date(),
|
||||||
|
expiresAt: null,
|
||||||
|
maxUses: null,
|
||||||
|
usedCount: 0,
|
||||||
|
gemeinde: { name: 'Nord' },
|
||||||
|
kc: { name: 'KC Dev' },
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await expect(service.resolve('sometoken')).rejects.toBeInstanceOf(NotFoundException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects an expired Teamer invite token', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
teamerInvite: {
|
||||||
|
email: null,
|
||||||
|
revokedAt: null,
|
||||||
|
expiresAt: new Date(Date.now() - 1000),
|
||||||
|
maxUses: null,
|
||||||
|
usedCount: 0,
|
||||||
|
gemeinde: { name: 'Nord' },
|
||||||
|
kc: { name: 'KC Dev' },
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await expect(service.resolve('sometoken')).rejects.toBeInstanceOf(NotFoundException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects a used-up Teamer invite token', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
teamerInvite: {
|
||||||
|
email: null,
|
||||||
|
revokedAt: null,
|
||||||
|
expiresAt: null,
|
||||||
|
maxUses: 1,
|
||||||
|
usedCount: 1,
|
||||||
|
gemeinde: { name: 'Nord' },
|
||||||
|
kc: { name: 'KC Dev' },
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await expect(service.resolve('sometoken')).rejects.toBeInstanceOf(NotFoundException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('resolves a usable Verantwortliche invite token', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
verantwortlicheInvite: {
|
||||||
|
email: 'pinned@example.org',
|
||||||
|
revokedAt: null,
|
||||||
|
expiresAt: null,
|
||||||
|
maxUses: 1,
|
||||||
|
usedCount: 0,
|
||||||
|
gemeinde: { name: 'Süd' },
|
||||||
|
kc: { name: 'KC Dev' },
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await expect(service.resolve('vertoken')).resolves.toEqual({
|
||||||
|
kind: 'verantwortliche_invite',
|
||||||
|
kcName: 'KC Dev',
|
||||||
|
gemeindeName: 'Süd',
|
||||||
|
pinnedEmail: 'pinned@example.org',
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('resolves a Gemeinde name to a Gemeinde-name team-login', async () => {
|
||||||
|
const { service } = makeService({ gemeinde: { id: 'gem-1', name: 'Mustergemeinde' } });
|
||||||
|
await expect(service.resolve('Mustergemeinde')).resolves.toEqual({
|
||||||
|
kind: 'team_login',
|
||||||
|
identifierType: 'gemeindeName',
|
||||||
|
gemeindeName: 'Mustergemeinde',
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('resolves codes from invite/token URLs', async () => {
|
||||||
|
const { service } = makeService({ kc: { id: 'kc-1', name: 'KC Dev', isActive: true } });
|
||||||
|
await expect(service.resolve('https://example.org/join?code=DEV123')).resolves.toEqual({
|
||||||
|
kind: 'guest',
|
||||||
|
kcName: 'KC Dev',
|
||||||
|
});
|
||||||
|
// noinspection HttpUrlsUsage
|
||||||
|
await expect(service.resolve('http://example.org/join?token=DEV123')).resolves.toEqual({
|
||||||
|
kind: 'guest',
|
||||||
|
kcName: 'KC Dev',
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects an empty code', async () => {
|
||||||
|
const { service } = makeService();
|
||||||
|
await expect(service.resolve(' ')).rejects.toBeInstanceOf(NotFoundException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects a code that matches nothing', async () => {
|
||||||
|
const { service } = makeService();
|
||||||
|
await expect(service.resolve('totally-unknown-code')).rejects.toBeInstanceOf(NotFoundException);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -0,0 +1,135 @@
|
|||||||
|
import { Injectable, NotFoundException } from '@nestjs/common';
|
||||||
|
import { PrismaClient } from '../prisma/prisma.module';
|
||||||
|
|
||||||
|
export type ResolvedCodeKind = 'guest' | 'teamer_invite' | 'verantwortliche_invite' | 'team_login' | 'sso';
|
||||||
|
|
||||||
|
export interface ResolvedCode {
|
||||||
|
kind: ResolvedCodeKind;
|
||||||
|
kcName?: string;
|
||||||
|
gemeindeName?: string;
|
||||||
|
/// *_invite only: whether the invite is pinned to a specific email
|
||||||
|
/// (personal invite) — if so the email field should be locked to it.
|
||||||
|
pinnedEmail?: string | null;
|
||||||
|
/// team_login only: which field the entered value should be sent back as.
|
||||||
|
identifierType?: 'email' | 'gemeindeName';
|
||||||
|
}
|
||||||
|
|
||||||
|
const SSO_KEYWORDS = new Set(['lt', 'login', 'sso']);
|
||||||
|
|
||||||
|
/// Classifies whatever the user typed into the single "Code" field on the
|
||||||
|
/// unified login screen, so the frontend can show the right follow-up form
|
||||||
|
/// without the user picking a login type up front.
|
||||||
|
///
|
||||||
|
/// Resolution order:
|
||||||
|
/// 1. the fixed keywords "lt", "login", "sso" or codes ending with "LT" (e.g. "ABC123LT")
|
||||||
|
/// -> Konfi-Castle-ID (Authentik SSO), for Leitungsteam/Verantwortliche
|
||||||
|
/// 2. an '@'-shaped value -> email + password (Teamer legacy login)
|
||||||
|
/// 3. a KC guest invite code (case-insensitive)
|
||||||
|
/// 4. a Teamer-invite token (self-registration link)
|
||||||
|
/// 5. a Verantwortliche-invite token (LT-issued SSO shortcut link)
|
||||||
|
/// 6. a Gemeinde name -> Gemeinde-name + password (normal Teamer login)
|
||||||
|
/// Anything else: "Code ungültig" (never leaks *why* it didn't match).
|
||||||
|
@Injectable()
|
||||||
|
export class CodeResolverService {
|
||||||
|
constructor(private readonly prisma: PrismaClient) {}
|
||||||
|
|
||||||
|
async resolve(rawCode: string): Promise<ResolvedCode> {
|
||||||
|
const code = rawCode.trim();
|
||||||
|
if (!code) {
|
||||||
|
throw new NotFoundException('Code ungültig');
|
||||||
|
}
|
||||||
|
|
||||||
|
const lower = code.toLowerCase();
|
||||||
|
const upper = code.toUpperCase();
|
||||||
|
|
||||||
|
if (SSO_KEYWORDS.has(lower) || (code.length > 2 && upper.endsWith('LT'))) {
|
||||||
|
return { kind: 'sso' };
|
||||||
|
}
|
||||||
|
|
||||||
|
if (code.includes('@')) {
|
||||||
|
return { kind: 'team_login', identifierType: 'email' };
|
||||||
|
}
|
||||||
|
|
||||||
|
let lookupCode = code;
|
||||||
|
try {
|
||||||
|
if (/^https?:\/\//i.test(code)) {
|
||||||
|
const url = new URL(code);
|
||||||
|
lookupCode =
|
||||||
|
url.searchParams.get('token') ||
|
||||||
|
url.searchParams.get('code') ||
|
||||||
|
url.searchParams.get('invite') ||
|
||||||
|
url.pathname.split('/').filter(Boolean).pop() ||
|
||||||
|
code;
|
||||||
|
}
|
||||||
|
} catch {
|
||||||
|
// Ignore URL parsing errors and keep code as is
|
||||||
|
}
|
||||||
|
|
||||||
|
const kc =
|
||||||
|
(await this.prisma.kc.findFirst({
|
||||||
|
where: { inviteCode: { equals: lookupCode, mode: 'insensitive' } },
|
||||||
|
})) ||
|
||||||
|
(await this.prisma.kc.findUnique({
|
||||||
|
where: { inviteCode: lookupCode },
|
||||||
|
}));
|
||||||
|
if (kc && kc.isActive) {
|
||||||
|
return { kind: 'guest', kcName: kc.name };
|
||||||
|
}
|
||||||
|
|
||||||
|
const teamerInvite =
|
||||||
|
(await this.prisma.teamerInvite.findFirst({
|
||||||
|
where: { token: { equals: lookupCode, mode: 'insensitive' } },
|
||||||
|
include: { gemeinde: true, kc: true },
|
||||||
|
})) ||
|
||||||
|
(await this.prisma.teamerInvite.findUnique({
|
||||||
|
where: { token: lookupCode },
|
||||||
|
include: { gemeinde: true, kc: true },
|
||||||
|
}));
|
||||||
|
if (teamerInvite && this.isInviteUsable(teamerInvite)) {
|
||||||
|
return {
|
||||||
|
kind: 'teamer_invite',
|
||||||
|
kcName: teamerInvite.kc.name,
|
||||||
|
gemeindeName: teamerInvite.gemeinde.name,
|
||||||
|
pinnedEmail: teamerInvite.email,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
const verantwortlicheInvite =
|
||||||
|
(await this.prisma.verantwortlicheInvite.findFirst({
|
||||||
|
where: { token: { equals: lookupCode, mode: 'insensitive' } },
|
||||||
|
include: { gemeinde: true, kc: true },
|
||||||
|
})) ||
|
||||||
|
(await this.prisma.verantwortlicheInvite.findUnique({
|
||||||
|
where: { token: lookupCode },
|
||||||
|
include: { gemeinde: true, kc: true },
|
||||||
|
}));
|
||||||
|
if (verantwortlicheInvite && this.isInviteUsable(verantwortlicheInvite)) {
|
||||||
|
return {
|
||||||
|
kind: 'verantwortliche_invite',
|
||||||
|
kcName: verantwortlicheInvite.kc.name,
|
||||||
|
gemeindeName: verantwortlicheInvite.gemeinde.name,
|
||||||
|
pinnedEmail: verantwortlicheInvite.email,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
const gemeinde = await this.prisma.gemeinde.findFirst({
|
||||||
|
where: { name: { equals: lookupCode, mode: 'insensitive' } },
|
||||||
|
});
|
||||||
|
if (gemeinde) {
|
||||||
|
return { kind: 'team_login', identifierType: 'gemeindeName', gemeindeName: gemeinde.name };
|
||||||
|
}
|
||||||
|
|
||||||
|
throw new NotFoundException('Code ungültig');
|
||||||
|
}
|
||||||
|
|
||||||
|
private isInviteUsable(invite: {
|
||||||
|
revokedAt: Date | null;
|
||||||
|
expiresAt: Date | null;
|
||||||
|
maxUses: number | null;
|
||||||
|
usedCount: number;
|
||||||
|
}): boolean {
|
||||||
|
if (invite.revokedAt) return false;
|
||||||
|
if (invite.expiresAt && invite.expiresAt.getTime() < Date.now()) return false;
|
||||||
|
return invite.maxUses === null || invite.usedCount < invite.maxUses;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
import { IsNotEmpty, IsString } from 'class-validator';
|
||||||
|
|
||||||
|
export class ResolveCodeDto {
|
||||||
|
@IsString()
|
||||||
|
@IsNotEmpty()
|
||||||
|
code!: string;
|
||||||
|
}
|
||||||
@@ -30,7 +30,14 @@ export class GuestAuthService {
|
|||||||
firstName: string,
|
firstName: string,
|
||||||
lastName: string,
|
lastName: string,
|
||||||
): Promise<{ accessToken: string }> {
|
): Promise<{ accessToken: string }> {
|
||||||
const kc = await this.prisma.kc.findUnique({ where: { inviteCode } });
|
const trimmedCode = inviteCode.trim();
|
||||||
|
const kc =
|
||||||
|
(await this.prisma.kc.findFirst({
|
||||||
|
where: { inviteCode: { equals: trimmedCode, mode: 'insensitive' } },
|
||||||
|
})) ||
|
||||||
|
(await this.prisma.kc.findUnique({
|
||||||
|
where: { inviteCode: trimmedCode },
|
||||||
|
}));
|
||||||
if (!kc || !kc.isActive) {
|
if (!kc || !kc.isActive) {
|
||||||
throw new NotFoundException('Unknown or inactive KC invite code');
|
throw new NotFoundException('Unknown or inactive KC invite code');
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,8 +1,10 @@
|
|||||||
import { Body, Controller, Get, Param, Post, Req, UseGuards } from '@nestjs/common';
|
import { Body, Controller, Delete, Get, Param, Post, Req, UseGuards } from '@nestjs/common';
|
||||||
import { AuthGuard } from '@nestjs/passport';
|
import { AuthGuard } from '@nestjs/passport';
|
||||||
import { ChatService } from './chat.service';
|
import { ChatService } from './chat.service';
|
||||||
|
import { ChatGateway } from './chat.gateway';
|
||||||
import { CreateChannelDto } from './dto/create-channel.dto';
|
import { CreateChannelDto } from './dto/create-channel.dto';
|
||||||
import { CreateDirectChannelDto } from './dto/create-direct-channel.dto';
|
import { CreateDirectChannelDto } from './dto/create-direct-channel.dto';
|
||||||
|
import { AddParticipantDto } from './dto/add-participant.dto';
|
||||||
import { Roles } from '../common/roles.decorator';
|
import { Roles } from '../common/roles.decorator';
|
||||||
import { RolesGuard } from '../common/roles.guard';
|
import { RolesGuard } from '../common/roles.guard';
|
||||||
import { Role } from '../common/role.enum';
|
import { Role } from '../common/role.enum';
|
||||||
@@ -14,7 +16,10 @@ type ChatRequest = AuthenticatedRequest & { user?: AuthenticatedRequest['user']
|
|||||||
|
|
||||||
@Controller('chat')
|
@Controller('chat')
|
||||||
export class ChatController {
|
export class ChatController {
|
||||||
constructor(private readonly chat: ChatService) {}
|
constructor(
|
||||||
|
private readonly chat: ChatService,
|
||||||
|
private readonly gateway: ChatGateway,
|
||||||
|
) {}
|
||||||
|
|
||||||
/// Channel administration (Gemeinde-Gruppen, LT-Kanäle, Broadcasts) is Leitungsteam-only.
|
/// Channel administration (Gemeinde-Gruppen, LT-Kanäle, Broadcasts) is Leitungsteam-only.
|
||||||
@Post(':kcId/channels')
|
@Post(':kcId/channels')
|
||||||
@@ -24,6 +29,73 @@ export class ChatController {
|
|||||||
return this.chat.createChannel(kcId, dto.type, dto.gemeindeId);
|
return this.chat.createChannel(kcId, dto.type, dto.gemeindeId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Free-form group chat ("Gruppenchat"): a Leitungsteam member (any KC) or
|
||||||
|
/// a Gemeinde Verantwortliche/r (their own KC, enforced by RolesGuard's
|
||||||
|
/// kcId scoping) can create one and pick any mix of team users and Konfis
|
||||||
|
/// (guests) from this KC as initial participants.
|
||||||
|
@Post(':kcId/gruppen')
|
||||||
|
@UseGuards(AuthGuard(['authentik', 'team']), RolesGuard)
|
||||||
|
@Roles(Role.LEITUNGSTEAM, Role.GEMEINDE_VERANTWORTLICHER)
|
||||||
|
createGruppe(
|
||||||
|
@Param('kcId') kcId: string,
|
||||||
|
@Body() dto: CreateChannelDto,
|
||||||
|
@Req() req: AuthenticatedRequest,
|
||||||
|
) {
|
||||||
|
return this.chat.createGruppe(kcId, dto.name, req.user!.userId, {
|
||||||
|
userIds: dto.participantUserIds,
|
||||||
|
guestIds: dto.participantGuestIds,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Candidates (team users + Konfis) a caller may add to a Gruppenchat in
|
||||||
|
/// this KC. Allowed for LT or a Verantwortliche/r of this KC.
|
||||||
|
@Get(':kcId/gruppen/participant-candidates')
|
||||||
|
@UseGuards(AuthGuard(['authentik', 'team']))
|
||||||
|
listPossibleParticipants(@Param('kcId') kcId: string, @Req() req: AuthenticatedRequest) {
|
||||||
|
return this.chat.listPossibleParticipants(kcId, req.user!);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Add a team user or Konfi to a Gruppenchat. Allowed for the channel's
|
||||||
|
/// creator, any Leitungsteam member, or a Verantwortliche/r of that KC.
|
||||||
|
@Post('gruppen/:channelId/participants')
|
||||||
|
@UseGuards(AuthGuard(['authentik', 'team']))
|
||||||
|
addParticipant(
|
||||||
|
@Param('channelId') channelId: string,
|
||||||
|
@Body() dto: AddParticipantDto,
|
||||||
|
@Req() req: AuthenticatedRequest,
|
||||||
|
) {
|
||||||
|
return this.chat
|
||||||
|
.addParticipant(
|
||||||
|
channelId,
|
||||||
|
{ kind: 'user', user: req.user! },
|
||||||
|
{ userId: dto.userId, guestId: dto.guestId },
|
||||||
|
)
|
||||||
|
.then((result) => {
|
||||||
|
this.gateway.notifyParticipantsChanged(channelId);
|
||||||
|
return result;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Remove a team user or Konfi from a Gruppenchat. Same authorization as add.
|
||||||
|
@Delete('gruppen/:channelId/participants')
|
||||||
|
@UseGuards(AuthGuard(['authentik', 'team']))
|
||||||
|
removeParticipant(
|
||||||
|
@Param('channelId') channelId: string,
|
||||||
|
@Body() dto: AddParticipantDto,
|
||||||
|
@Req() req: AuthenticatedRequest,
|
||||||
|
) {
|
||||||
|
return this.chat
|
||||||
|
.removeParticipant(
|
||||||
|
channelId,
|
||||||
|
{ kind: 'user', user: req.user! },
|
||||||
|
{ userId: dto.userId, guestId: dto.guestId },
|
||||||
|
)
|
||||||
|
.then((result) => {
|
||||||
|
this.gateway.notifyParticipantsChanged(channelId);
|
||||||
|
return result;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
/// Any two team members of the same KC can start a direct conversation
|
/// Any two team members of the same KC can start a direct conversation
|
||||||
/// (Authentik-backed members and local Gemeinde Teamer alike).
|
/// (Authentik-backed members and local Gemeinde Teamer alike).
|
||||||
@Post('direct')
|
@Post('direct')
|
||||||
|
|||||||
@@ -106,4 +106,12 @@ export class ChatGateway implements OnGatewayConnection, OnGatewayDisconnect {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Called by ChatController after add/removeParticipant so anyone with the
|
||||||
|
/// channel already open (e.g. the creator's participant-management UI)
|
||||||
|
/// gets a live update. Newly added participants join the room themselves
|
||||||
|
/// via `chat:join` once they open the chat.
|
||||||
|
notifyParticipantsChanged(channelId: string) {
|
||||||
|
this.broadcast(channelId, { event: 'chat:participants-changed', data: { channelId } });
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,280 @@
|
|||||||
|
import { BadRequestException, ForbiddenException, NotFoundException } from '@nestjs/common';
|
||||||
|
import { ChatChannelType, Role } from '@prisma/client';
|
||||||
|
import { ChatService } from './chat.service';
|
||||||
|
import { AuthenticatedUser } from '../auth/authenticated-request';
|
||||||
|
import { GuestJwtPayload } from '../auth/guest-auth.service';
|
||||||
|
|
||||||
|
/// Focus: GRUPPE channel creation + participant management authorization
|
||||||
|
/// (creator / Leitungsteam / Verantwortliche/r of that KC), and read/write
|
||||||
|
/// access for team users and guests. Prisma + Sync + Push faked in memory.
|
||||||
|
|
||||||
|
function userCaller(userId: string, memberships: AuthenticatedUser['memberships']) {
|
||||||
|
return {
|
||||||
|
kind: 'user' as const,
|
||||||
|
user: { userId, authentikSub: `sub-${userId}`, email: `${userId}@example.org`, memberships },
|
||||||
|
};
|
||||||
|
}
|
||||||
|
function guestCaller(guestId: string, kcId: string, gemeindeId: string | null = null) {
|
||||||
|
const guest: GuestJwtPayload = { guestId, kcId, gemeindeId };
|
||||||
|
return { kind: 'guest' as const, guest };
|
||||||
|
}
|
||||||
|
|
||||||
|
const LT = userCaller('lt-1', [{ kcId: 'kc-1', gemeindeId: null, role: Role.LEITUNGSTEAM }]);
|
||||||
|
const VERANTW = userCaller('ver-1', [
|
||||||
|
{ kcId: 'kc-1', gemeindeId: 'gem-1', role: Role.GEMEINDE_VERANTWORTLICHER },
|
||||||
|
]);
|
||||||
|
const TEAMER = userCaller('teamer-1', [
|
||||||
|
{ kcId: 'kc-1', gemeindeId: 'gem-1', role: Role.GEMEINDE_TEAMER },
|
||||||
|
]);
|
||||||
|
|
||||||
|
function makeService(
|
||||||
|
opts: {
|
||||||
|
channels?: Record<string, any>;
|
||||||
|
memberships?: { kcId: string; userId: string }[];
|
||||||
|
guests?: { id: string; kcId: string }[];
|
||||||
|
} = {},
|
||||||
|
) {
|
||||||
|
const channels: Record<string, any> = opts.channels ?? {};
|
||||||
|
const memberships = opts.memberships ?? [];
|
||||||
|
const guests = opts.guests ?? [];
|
||||||
|
const participants: any[] = [];
|
||||||
|
let participantSeq = 0;
|
||||||
|
|
||||||
|
const prisma = {
|
||||||
|
chatChannel: {
|
||||||
|
create: jest.fn(({ data, include }: any) => {
|
||||||
|
const id = `chan-${Object.keys(channels).length + 1}`;
|
||||||
|
const created = { id, ...data, participants: [] };
|
||||||
|
if (data.participants?.create) {
|
||||||
|
for (const p of data.participants.create) {
|
||||||
|
const row = { id: `part-${++participantSeq}`, channelId: id, userId: null, guestAccountId: null, ...p };
|
||||||
|
participants.push(row);
|
||||||
|
created.participants.push(row);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
channels[id] = created;
|
||||||
|
return Promise.resolve(include ? created : { id, ...data });
|
||||||
|
}),
|
||||||
|
findUnique: jest.fn(({ where, include }: any) => {
|
||||||
|
const channel = channels[where.id];
|
||||||
|
if (!channel) return Promise.resolve(null);
|
||||||
|
if (include?.participants) {
|
||||||
|
const seeded = participants.filter((p) => p.channelId === channel.id);
|
||||||
|
const fallback = Array.isArray(channel.participants) ? channel.participants : [];
|
||||||
|
return Promise.resolve({
|
||||||
|
...channel,
|
||||||
|
participants: seeded.length ? seeded : fallback,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
return Promise.resolve(channel);
|
||||||
|
}),
|
||||||
|
findMany: jest.fn().mockResolvedValue([]),
|
||||||
|
},
|
||||||
|
membership: {
|
||||||
|
findMany: jest.fn(({ where }: any) => {
|
||||||
|
const ids: string[] = where.userId.in;
|
||||||
|
const rows = memberships.filter((m) => m.kcId === where.kcId && ids.includes(m.userId));
|
||||||
|
const seen = new Set<string>();
|
||||||
|
const distinct = rows.filter((r) => (seen.has(r.userId) ? false : (seen.add(r.userId), true)));
|
||||||
|
return Promise.resolve(distinct);
|
||||||
|
}),
|
||||||
|
findFirst: jest.fn(({ where }: any) =>
|
||||||
|
Promise.resolve(memberships.find((m) => m.kcId === where.kcId && m.userId === where.userId) ?? null),
|
||||||
|
),
|
||||||
|
count: jest.fn().mockResolvedValue(0),
|
||||||
|
},
|
||||||
|
guestAccount: {
|
||||||
|
count: jest.fn(({ where }: any) =>
|
||||||
|
Promise.resolve(guests.filter((g) => where.id.in.includes(g.id) && g.kcId === where.kcId).length),
|
||||||
|
),
|
||||||
|
findFirst: jest.fn(({ where }: any) =>
|
||||||
|
Promise.resolve(guests.find((g) => g.id === where.id && g.kcId === where.kcId) ?? null),
|
||||||
|
),
|
||||||
|
},
|
||||||
|
chatParticipant: {
|
||||||
|
upsert: jest.fn(({ create }: any) => {
|
||||||
|
const existing = participants.find(
|
||||||
|
(p) =>
|
||||||
|
p.channelId === create.channelId &&
|
||||||
|
p.userId === (create.userId ?? null) &&
|
||||||
|
p.guestAccountId === (create.guestAccountId ?? null),
|
||||||
|
);
|
||||||
|
if (existing) return Promise.resolve(existing);
|
||||||
|
const row = { id: `part-${++participantSeq}`, userId: null, guestAccountId: null, ...create };
|
||||||
|
participants.push(row);
|
||||||
|
return Promise.resolve(row);
|
||||||
|
}),
|
||||||
|
findFirst: jest.fn(({ where }: any) =>
|
||||||
|
Promise.resolve(
|
||||||
|
participants.find(
|
||||||
|
(p) =>
|
||||||
|
p.channelId === where.channelId &&
|
||||||
|
(where.userId === undefined || p.userId === where.userId) &&
|
||||||
|
(where.guestAccountId === undefined || p.guestAccountId === where.guestAccountId),
|
||||||
|
) ?? null,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
delete: jest.fn(({ where }: any) => {
|
||||||
|
const idx = participants.findIndex((p) => p.id === where.id);
|
||||||
|
const [removed] = participants.splice(idx, 1);
|
||||||
|
return Promise.resolve(removed);
|
||||||
|
}),
|
||||||
|
},
|
||||||
|
chatMessage: { create: jest.fn(), findMany: jest.fn() },
|
||||||
|
};
|
||||||
|
const sync = { capture: jest.fn().mockResolvedValue(undefined) };
|
||||||
|
const push = { notifyChannel: jest.fn().mockResolvedValue(undefined) };
|
||||||
|
const service = new ChatService(prisma as never, sync as never, push as never);
|
||||||
|
return { service, prisma, sync, push, channels, participants };
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('ChatService.createGruppe', () => {
|
||||||
|
it('creates a GRUPPE channel with the creator plus given team/guest participants', async () => {
|
||||||
|
const { service, sync } = makeService({
|
||||||
|
memberships: [{ kcId: 'kc-1', userId: 'ver-1' }, { kcId: 'kc-1', userId: 'teamer-1' }],
|
||||||
|
guests: [{ id: 'guest-1', kcId: 'kc-1' }],
|
||||||
|
});
|
||||||
|
const channel = await service.createGruppe('kc-1', 'Ausflugsplanung', 'ver-1', {
|
||||||
|
userIds: ['teamer-1'],
|
||||||
|
guestIds: ['guest-1'],
|
||||||
|
});
|
||||||
|
expect(channel.type).toBe(ChatChannelType.GRUPPE);
|
||||||
|
expect(channel.createdByUserId).toBe('ver-1');
|
||||||
|
const userIds = channel.participants.map((p: any) => p.userId).filter(Boolean);
|
||||||
|
const guestIds = channel.participants.map((p: any) => p.guestAccountId).filter(Boolean);
|
||||||
|
expect(userIds.sort()).toEqual(['teamer-1', 'ver-1']);
|
||||||
|
expect(guestIds).toEqual(['guest-1']);
|
||||||
|
expect(sync.capture).toHaveBeenCalledWith('ChatChannel', 'CREATE', channel.id, expect.anything());
|
||||||
|
});
|
||||||
|
|
||||||
|
it('does not duplicate the creator if already listed as a participant', async () => {
|
||||||
|
const { service } = makeService({ memberships: [{ kcId: 'kc-1', userId: 'ver-1' }] });
|
||||||
|
const channel = await service.createGruppe('kc-1', 'X', 'ver-1', { userIds: ['ver-1'] });
|
||||||
|
const userIds = channel.participants.map((p: any) => p.userId).filter(Boolean);
|
||||||
|
expect(userIds).toEqual(['ver-1']);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects a participant who is not a member of the KC', async () => {
|
||||||
|
const { service } = makeService({ memberships: [] });
|
||||||
|
await expect(
|
||||||
|
service.createGruppe('kc-1', 'X', 'ver-1', { userIds: ['ghost'] }),
|
||||||
|
).rejects.toBeInstanceOf(BadRequestException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects a guest who is not part of the KC', async () => {
|
||||||
|
const { service } = makeService({ guests: [{ id: 'guest-1', kcId: 'kc-2' }] });
|
||||||
|
await expect(
|
||||||
|
service.createGruppe('kc-1', 'X', 'ver-1', { guestIds: ['guest-1'] }),
|
||||||
|
).rejects.toBeInstanceOf(BadRequestException);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('ChatService participant management', () => {
|
||||||
|
function seedGruppe() {
|
||||||
|
const channels = {
|
||||||
|
'chan-1': { id: 'chan-1', kcId: 'kc-1', type: ChatChannelType.GRUPPE, createdByUserId: 'ver-1', gemeindeId: null },
|
||||||
|
};
|
||||||
|
return channels;
|
||||||
|
}
|
||||||
|
|
||||||
|
it('lets the creator add a team user', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
channels: seedGruppe(),
|
||||||
|
memberships: [{ kcId: 'kc-1', userId: 'teamer-1' }],
|
||||||
|
});
|
||||||
|
const p = await service.addParticipant('chan-1', VERANTW, { userId: 'teamer-1' });
|
||||||
|
expect(p.userId).toBe('teamer-1');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('lets a Leitungsteam member add a guest even if not the creator', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
channels: seedGruppe(),
|
||||||
|
guests: [{ id: 'guest-1', kcId: 'kc-1' }],
|
||||||
|
});
|
||||||
|
const p = await service.addParticipant('chan-1', LT, { guestId: 'guest-1' });
|
||||||
|
expect(p.guestAccountId).toBe('guest-1');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('forbids a plain Teamer (not creator, not LT, not Verantwortliche/r) from managing participants', async () => {
|
||||||
|
const { service } = makeService({ channels: seedGruppe() });
|
||||||
|
await expect(
|
||||||
|
service.addParticipant('chan-1', TEAMER, { userId: 'teamer-1' }),
|
||||||
|
).rejects.toBeInstanceOf(ForbiddenException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('forbids guests from managing participants', async () => {
|
||||||
|
const { service } = makeService({ channels: seedGruppe() });
|
||||||
|
await expect(
|
||||||
|
service.addParticipant('chan-1', guestCaller('g-1', 'kc-1') as never, { userId: 'x' }),
|
||||||
|
).rejects.toBeInstanceOf(ForbiddenException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('404s for a non-GRUPPE channel', async () => {
|
||||||
|
const channels = {
|
||||||
|
'chan-2': { id: 'chan-2', kcId: 'kc-1', type: ChatChannelType.GEMEINDE_GRUPPE, createdByUserId: null },
|
||||||
|
};
|
||||||
|
const { service } = makeService({ channels });
|
||||||
|
await expect(
|
||||||
|
service.addParticipant('chan-2', LT, { userId: 'teamer-1' }),
|
||||||
|
).rejects.toBeInstanceOf(NotFoundException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects adding a user not in the KC', async () => {
|
||||||
|
const { service } = makeService({ channels: seedGruppe(), memberships: [] });
|
||||||
|
await expect(
|
||||||
|
service.addParticipant('chan-1', LT, { userId: 'ghost' }),
|
||||||
|
).rejects.toBeInstanceOf(BadRequestException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('removes a participant and is a no-op if already absent', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
channels: seedGruppe(),
|
||||||
|
memberships: [{ kcId: 'kc-1', userId: 'teamer-1' }],
|
||||||
|
});
|
||||||
|
await service.addParticipant('chan-1', VERANTW, { userId: 'teamer-1' });
|
||||||
|
const res = await service.removeParticipant('chan-1', VERANTW, { userId: 'teamer-1' });
|
||||||
|
expect(res).toEqual({ ok: true });
|
||||||
|
const res2 = await service.removeParticipant('chan-1', VERANTW, { userId: 'teamer-1' });
|
||||||
|
expect(res2).toEqual({ ok: true });
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
describe('ChatService GRUPPE read/write access', () => {
|
||||||
|
function seedGruppeWithParticipants(participants: any[]) {
|
||||||
|
return {
|
||||||
|
'chan-1': {
|
||||||
|
id: 'chan-1',
|
||||||
|
kcId: 'kc-1',
|
||||||
|
type: ChatChannelType.GRUPPE,
|
||||||
|
createdByUserId: 'ver-1',
|
||||||
|
gemeindeId: null,
|
||||||
|
participants,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
it('lets a listed guest read messages', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
channels: seedGruppeWithParticipants([{ userId: null, guestAccountId: 'guest-1' }]),
|
||||||
|
});
|
||||||
|
await expect(
|
||||||
|
service.assertCanRead('chan-1', guestCaller('guest-1', 'kc-1') as never),
|
||||||
|
).resolves.toBeDefined();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('forbids a guest not in the participant list', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
channels: seedGruppeWithParticipants([{ userId: null, guestAccountId: 'guest-1' }]),
|
||||||
|
});
|
||||||
|
await expect(
|
||||||
|
service.assertCanRead('chan-1', guestCaller('guest-2', 'kc-1') as never),
|
||||||
|
).rejects.toBeInstanceOf(ForbiddenException);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('forbids a team user not in the participant list', async () => {
|
||||||
|
const { service } = makeService({
|
||||||
|
channels: seedGruppeWithParticipants([{ userId: 'someone-else', guestAccountId: null }]),
|
||||||
|
});
|
||||||
|
await expect(service.assertCanRead('chan-1', TEAMER)).rejects.toBeInstanceOf(ForbiddenException);
|
||||||
|
});
|
||||||
|
});
|
||||||
+234
-6
@@ -1,4 +1,4 @@
|
|||||||
import { ForbiddenException, Injectable, NotFoundException } from '@nestjs/common';
|
import { BadRequestException, ForbiddenException, Injectable, NotFoundException } from '@nestjs/common';
|
||||||
import { ChatChannelType, Role, SyncOperation } from '@prisma/client';
|
import { ChatChannelType, Role, SyncOperation } from '@prisma/client';
|
||||||
import { PrismaClient } from '../prisma/prisma.module';
|
import { PrismaClient } from '../prisma/prisma.module';
|
||||||
import { AuthenticatedUser } from '../auth/authenticated-request';
|
import { AuthenticatedUser } from '../auth/authenticated-request';
|
||||||
@@ -15,8 +15,14 @@ const CHANNEL_TITLES: Record<ChatChannelType, string> = {
|
|||||||
[ChatChannelType.DIREKT]: 'Direktnachricht',
|
[ChatChannelType.DIREKT]: 'Direktnachricht',
|
||||||
[ChatChannelType.LT_UEBERGREIFEND]: 'Leitungsteam',
|
[ChatChannelType.LT_UEBERGREIFEND]: 'Leitungsteam',
|
||||||
[ChatChannelType.BROADCAST]: 'Ankündigung',
|
[ChatChannelType.BROADCAST]: 'Ankündigung',
|
||||||
|
[ChatChannelType.GRUPPE]: 'Gruppenchat',
|
||||||
};
|
};
|
||||||
|
|
||||||
|
export interface CreateGruppeParticipants {
|
||||||
|
userIds?: string[];
|
||||||
|
guestIds?: string[];
|
||||||
|
}
|
||||||
|
|
||||||
@Injectable()
|
@Injectable()
|
||||||
export class ChatService {
|
export class ChatService {
|
||||||
constructor(
|
constructor(
|
||||||
@@ -31,6 +37,206 @@ export class ChatService {
|
|||||||
return channel;
|
return channel;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Free-form group chat: created by a Leitungsteam member (any KC) or a
|
||||||
|
/// Gemeinde Verantwortliche/r (their own KC — enforced by the RolesGuard's
|
||||||
|
/// kcId scoping at the controller level). Konfis (guests) may be included
|
||||||
|
/// directly, unlike DIREKT/GEMEINDE_GRUPPE channels which are team-only.
|
||||||
|
async createGruppe(
|
||||||
|
kcId: string,
|
||||||
|
name: string | undefined,
|
||||||
|
createdByUserId: string,
|
||||||
|
participants: CreateGruppeParticipants,
|
||||||
|
) {
|
||||||
|
const userIds = [...new Set(participants.userIds ?? [])];
|
||||||
|
const guestIds = [...new Set(participants.guestIds ?? [])];
|
||||||
|
|
||||||
|
if (userIds.length) {
|
||||||
|
// A user may show up under more than one Gemeinde membership; just
|
||||||
|
// make sure every requested id resolves to at least one row for this KC.
|
||||||
|
const distinctUsers = await this.prisma.membership.findMany({
|
||||||
|
where: { kcId, userId: { in: userIds } },
|
||||||
|
select: { userId: true },
|
||||||
|
distinct: ['userId'],
|
||||||
|
});
|
||||||
|
if (distinctUsers.length !== userIds.length) {
|
||||||
|
throw new BadRequestException('One or more users are not part of this KC');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (guestIds.length) {
|
||||||
|
const guestCount = await this.prisma.guestAccount.count({
|
||||||
|
where: { id: { in: guestIds }, kcId },
|
||||||
|
});
|
||||||
|
if (guestCount !== guestIds.length) {
|
||||||
|
throw new BadRequestException('One or more guests are not part of this KC');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const channel = await this.prisma.chatChannel.create({
|
||||||
|
data: {
|
||||||
|
kcId,
|
||||||
|
type: ChatChannelType.GRUPPE,
|
||||||
|
name,
|
||||||
|
createdByUserId,
|
||||||
|
participants: {
|
||||||
|
create: [
|
||||||
|
...(userIds.includes(createdByUserId) ? [] : [{ userId: createdByUserId }]),
|
||||||
|
...userIds.map((userId) => ({ userId })),
|
||||||
|
...guestIds.map((guestAccountId) => ({ guestAccountId })),
|
||||||
|
],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
include: { participants: true },
|
||||||
|
});
|
||||||
|
await this.sync.capture('ChatChannel', SyncOperation.CREATE, channel.id, channel);
|
||||||
|
return channel;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Candidates a caller may add to a GRUPPE channel in this KC: every team
|
||||||
|
/// member (any Gemeinde) plus every Konfi/guest, so a Verantwortliche/r can
|
||||||
|
/// pick across Gemeinde boundaries as intended. Same authorization as
|
||||||
|
/// creating a Gruppenchat (LT or Verantwortliche/r of this KC).
|
||||||
|
async listPossibleParticipants(kcId: string, caller: AuthenticatedUser) {
|
||||||
|
const isLt = caller.memberships.some((m) => m.role === Role.LEITUNGSTEAM);
|
||||||
|
const isVerantwortlicherHere = caller.memberships.some(
|
||||||
|
(m) => m.kcId === kcId && m.role === Role.GEMEINDE_VERANTWORTLICHER,
|
||||||
|
);
|
||||||
|
if (!isLt && !isVerantwortlicherHere) {
|
||||||
|
throw new ForbiddenException('Not allowed to list participants for this KC');
|
||||||
|
}
|
||||||
|
|
||||||
|
const [memberships, guests] = await Promise.all([
|
||||||
|
this.prisma.membership.findMany({
|
||||||
|
where: { kcId, status: 'ACTIVE' },
|
||||||
|
include: { user: { select: { id: true, firstName: true, lastName: true, email: true } } },
|
||||||
|
orderBy: { user: { lastName: 'asc' } },
|
||||||
|
}),
|
||||||
|
this.prisma.guestAccount.findMany({
|
||||||
|
where: { kcId },
|
||||||
|
select: { id: true, firstName: true, lastName: true, gemeindeId: true },
|
||||||
|
orderBy: { lastName: 'asc' },
|
||||||
|
}),
|
||||||
|
]);
|
||||||
|
|
||||||
|
const seenUsers = new Set<string>();
|
||||||
|
const users = [];
|
||||||
|
for (const m of memberships) {
|
||||||
|
if (seenUsers.has(m.userId)) continue;
|
||||||
|
seenUsers.add(m.userId);
|
||||||
|
users.push({
|
||||||
|
userId: m.user.id,
|
||||||
|
firstName: m.user.firstName,
|
||||||
|
lastName: m.user.lastName,
|
||||||
|
email: m.user.email,
|
||||||
|
role: m.role,
|
||||||
|
gemeindeId: m.gemeindeId,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
return {
|
||||||
|
users,
|
||||||
|
guests: guests.map((g) => ({
|
||||||
|
guestId: g.id,
|
||||||
|
firstName: g.firstName,
|
||||||
|
lastName: g.lastName,
|
||||||
|
gemeindeId: g.gemeindeId,
|
||||||
|
})),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Adds a team user or a guest/Konfi to an existing GRUPPE channel. Only
|
||||||
|
/// the channel's creator or a Leitungsteam member may manage participants.
|
||||||
|
async addParticipant(
|
||||||
|
channelId: string,
|
||||||
|
caller: ChatCaller,
|
||||||
|
target: { userId?: string; guestId?: string },
|
||||||
|
) {
|
||||||
|
const channel = await this.getGruppeForManagementOrThrow(channelId, caller);
|
||||||
|
|
||||||
|
if (!target.userId && !target.guestId) {
|
||||||
|
throw new BadRequestException('userId or guestId is required');
|
||||||
|
}
|
||||||
|
if (target.userId && target.guestId) {
|
||||||
|
throw new BadRequestException('Provide either userId or guestId, not both');
|
||||||
|
}
|
||||||
|
|
||||||
|
if (target.userId) {
|
||||||
|
const isMember = await this.prisma.membership.findFirst({
|
||||||
|
where: { kcId: channel.kcId, userId: target.userId },
|
||||||
|
});
|
||||||
|
if (!isMember) {
|
||||||
|
throw new BadRequestException('User is not part of this KC');
|
||||||
|
}
|
||||||
|
const participant = await this.prisma.chatParticipant.upsert({
|
||||||
|
where: { channelId_userId: { channelId, userId: target.userId } },
|
||||||
|
create: { channelId, userId: target.userId },
|
||||||
|
update: {},
|
||||||
|
});
|
||||||
|
await this.sync.capture('ChatParticipant', SyncOperation.CREATE, participant.id, participant);
|
||||||
|
return participant;
|
||||||
|
}
|
||||||
|
|
||||||
|
const guest = await this.prisma.guestAccount.findFirst({
|
||||||
|
where: { id: target.guestId, kcId: channel.kcId },
|
||||||
|
});
|
||||||
|
if (!guest) {
|
||||||
|
throw new BadRequestException('Guest is not part of this KC');
|
||||||
|
}
|
||||||
|
const participant = await this.prisma.chatParticipant.upsert({
|
||||||
|
where: { channelId_guestAccountId: { channelId, guestAccountId: target.guestId! } },
|
||||||
|
create: { channelId, guestAccountId: target.guestId },
|
||||||
|
update: {},
|
||||||
|
});
|
||||||
|
await this.sync.capture('ChatParticipant', SyncOperation.CREATE, participant.id, participant);
|
||||||
|
return participant;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Removes a team user or a guest/Konfi from a GRUPPE channel. Same
|
||||||
|
/// authorization as addParticipant.
|
||||||
|
async removeParticipant(
|
||||||
|
channelId: string,
|
||||||
|
caller: ChatCaller,
|
||||||
|
target: { userId?: string; guestId?: string },
|
||||||
|
) {
|
||||||
|
await this.getGruppeForManagementOrThrow(channelId, caller);
|
||||||
|
|
||||||
|
if (!target.userId && !target.guestId) {
|
||||||
|
throw new BadRequestException('userId or guestId is required');
|
||||||
|
}
|
||||||
|
|
||||||
|
const existing = await this.prisma.chatParticipant.findFirst({
|
||||||
|
where: {
|
||||||
|
channelId,
|
||||||
|
userId: target.userId ?? undefined,
|
||||||
|
guestAccountId: target.guestId ?? undefined,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
if (!existing) return { ok: true };
|
||||||
|
|
||||||
|
await this.prisma.chatParticipant.delete({ where: { id: existing.id } });
|
||||||
|
await this.sync.capture('ChatParticipant', SyncOperation.DELETE, existing.id, { id: existing.id });
|
||||||
|
return { ok: true };
|
||||||
|
}
|
||||||
|
|
||||||
|
private async getGruppeForManagementOrThrow(channelId: string, caller: ChatCaller) {
|
||||||
|
if (caller.kind !== 'user') {
|
||||||
|
throw new ForbiddenException('Guests may not manage channel participants');
|
||||||
|
}
|
||||||
|
const channel = await this.prisma.chatChannel.findUnique({ where: { id: channelId } });
|
||||||
|
if (!channel || channel.type !== ChatChannelType.GRUPPE) {
|
||||||
|
throw new NotFoundException('Gruppenchat not found');
|
||||||
|
}
|
||||||
|
const { user } = caller;
|
||||||
|
const isLt = user.memberships.some((m) => m.role === Role.LEITUNGSTEAM);
|
||||||
|
const isCreator = channel.createdByUserId === user.userId;
|
||||||
|
const isVerantwortlicherHere = user.memberships.some(
|
||||||
|
(m) => m.kcId === channel.kcId && m.role === Role.GEMEINDE_VERANTWORTLICHER,
|
||||||
|
);
|
||||||
|
if (!isLt && !isCreator && !isVerantwortlicherHere) {
|
||||||
|
throw new ForbiddenException('Not allowed to manage this Gruppenchat');
|
||||||
|
}
|
||||||
|
return channel;
|
||||||
|
}
|
||||||
|
|
||||||
async getOrCreateDirectChannel(kcId: string, userAId: string, userBId: string) {
|
async getOrCreateDirectChannel(kcId: string, userAId: string, userBId: string) {
|
||||||
const existing = await this.prisma.chatChannel.findFirst({
|
const existing = await this.prisma.chatChannel.findFirst({
|
||||||
where: {
|
where: {
|
||||||
@@ -57,7 +263,13 @@ export class ChatService {
|
|||||||
async listChannelsForCaller(kcId: string, caller: ChatCaller) {
|
async listChannelsForCaller(kcId: string, caller: ChatCaller) {
|
||||||
if (caller.kind === 'guest') {
|
if (caller.kind === 'guest') {
|
||||||
return this.prisma.chatChannel.findMany({
|
return this.prisma.chatChannel.findMany({
|
||||||
where: { kcId, type: ChatChannelType.BROADCAST },
|
where: {
|
||||||
|
kcId,
|
||||||
|
OR: [
|
||||||
|
{ type: ChatChannelType.BROADCAST },
|
||||||
|
{ type: ChatChannelType.GRUPPE, participants: { some: { guestAccountId: caller.guest.guestId } } },
|
||||||
|
],
|
||||||
|
},
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
const { user } = caller;
|
const { user } = caller;
|
||||||
@@ -75,6 +287,7 @@ export class ChatService {
|
|||||||
{ type: ChatChannelType.BROADCAST },
|
{ type: ChatChannelType.BROADCAST },
|
||||||
{ type: ChatChannelType.GEMEINDE_GRUPPE, gemeindeId: { in: gemeindeIds } },
|
{ type: ChatChannelType.GEMEINDE_GRUPPE, gemeindeId: { in: gemeindeIds } },
|
||||||
{ type: ChatChannelType.DIREKT, participants: { some: { userId: user.userId } } },
|
{ type: ChatChannelType.DIREKT, participants: { some: { userId: user.userId } } },
|
||||||
|
{ type: ChatChannelType.GRUPPE, participants: { some: { userId: user.userId } } },
|
||||||
],
|
],
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
@@ -102,15 +315,23 @@ export class ChatService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (caller.kind === 'guest') {
|
if (caller.kind === 'guest') {
|
||||||
const allowed = channel.type === ChatChannelType.BROADCAST && mode === 'read';
|
if (channel.type === ChatChannelType.BROADCAST && mode === 'read') {
|
||||||
if (!allowed) {
|
|
||||||
throw new ForbiddenException('Guests may only read broadcast channels');
|
|
||||||
}
|
|
||||||
if (caller.guest.kcId !== channel.kcId) {
|
if (caller.guest.kcId !== channel.kcId) {
|
||||||
throw new ForbiddenException('Guest does not belong to this KC');
|
throw new ForbiddenException('Guest does not belong to this KC');
|
||||||
}
|
}
|
||||||
return channel;
|
return channel;
|
||||||
}
|
}
|
||||||
|
if (channel.type === ChatChannelType.GRUPPE) {
|
||||||
|
const isParticipant = channel.participants.some(
|
||||||
|
(p) => p.guestAccountId === caller.guest.guestId,
|
||||||
|
);
|
||||||
|
if (!isParticipant) {
|
||||||
|
throw new ForbiddenException('Not a participant of this Gruppenchat');
|
||||||
|
}
|
||||||
|
return channel;
|
||||||
|
}
|
||||||
|
throw new ForbiddenException('Guests may only read broadcast channels or their Gruppenchats');
|
||||||
|
}
|
||||||
|
|
||||||
const { user } = caller;
|
const { user } = caller;
|
||||||
const isLt = user.memberships.some((m) => m.role === Role.LEITUNGSTEAM);
|
const isLt = user.memberships.some((m) => m.role === Role.LEITUNGSTEAM);
|
||||||
@@ -145,6 +366,13 @@ export class ChatService {
|
|||||||
}
|
}
|
||||||
return channel;
|
return channel;
|
||||||
}
|
}
|
||||||
|
case ChatChannelType.GRUPPE: {
|
||||||
|
const isParticipant = channel.participants.some((p) => p.userId === user.userId);
|
||||||
|
if (!isParticipant) {
|
||||||
|
throw new ForbiddenException('Not a participant of this Gruppenchat');
|
||||||
|
}
|
||||||
|
return channel;
|
||||||
|
}
|
||||||
default:
|
default:
|
||||||
throw new ForbiddenException('Unknown channel type');
|
throw new ForbiddenException('Unknown channel type');
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,13 @@
|
|||||||
|
import { IsOptional, IsString } from 'class-validator';
|
||||||
|
|
||||||
|
/// Exactly one of userId/guestId must be set; validated in the service since
|
||||||
|
/// class-validator doesn't express "exactly one of" declaratively.
|
||||||
|
export class AddParticipantDto {
|
||||||
|
@IsOptional()
|
||||||
|
@IsString()
|
||||||
|
userId?: string;
|
||||||
|
|
||||||
|
@IsOptional()
|
||||||
|
@IsString()
|
||||||
|
guestId?: string;
|
||||||
|
}
|
||||||
@@ -1,4 +1,4 @@
|
|||||||
import { IsEnum, IsOptional, IsString } from 'class-validator';
|
import { ArrayUnique, IsArray, IsEnum, IsOptional, IsString } from 'class-validator';
|
||||||
import { ChatChannelType } from '@prisma/client';
|
import { ChatChannelType } from '@prisma/client';
|
||||||
|
|
||||||
export class CreateChannelDto {
|
export class CreateChannelDto {
|
||||||
@@ -8,4 +8,24 @@ export class CreateChannelDto {
|
|||||||
@IsOptional()
|
@IsOptional()
|
||||||
@IsString()
|
@IsString()
|
||||||
gemeindeId?: string;
|
gemeindeId?: string;
|
||||||
|
|
||||||
|
/// Display name; used for GRUPPE channels.
|
||||||
|
@IsOptional()
|
||||||
|
@IsString()
|
||||||
|
name?: string;
|
||||||
|
|
||||||
|
/// Initial participants for a GRUPPE channel (team users). More can be
|
||||||
|
/// added/removed later via the participants endpoints.
|
||||||
|
@IsOptional()
|
||||||
|
@IsArray()
|
||||||
|
@ArrayUnique()
|
||||||
|
@IsString({ each: true })
|
||||||
|
participantUserIds?: string[];
|
||||||
|
|
||||||
|
/// Initial guest/Konfi participants for a GRUPPE channel.
|
||||||
|
@IsOptional()
|
||||||
|
@IsArray()
|
||||||
|
@ArrayUnique()
|
||||||
|
@IsString({ each: true })
|
||||||
|
participantGuestIds?: string[];
|
||||||
}
|
}
|
||||||
|
|||||||
+49
-1
@@ -3,11 +3,59 @@ import { ValidationPipe } from '@nestjs/common';
|
|||||||
import { WsAdapter } from '@nestjs/platform-ws';
|
import { WsAdapter } from '@nestjs/platform-ws';
|
||||||
import { AppModule } from './app.module';
|
import { AppModule } from './app.module';
|
||||||
|
|
||||||
|
function parseAllowedOrigins(): string[] {
|
||||||
|
const envOrigins = process.env.ALLOWED_ORIGINS || process.env.CORS_ORIGIN;
|
||||||
|
if (envOrigins) {
|
||||||
|
return envOrigins
|
||||||
|
.split(',')
|
||||||
|
.map((o) => o.trim())
|
||||||
|
.filter(Boolean);
|
||||||
|
}
|
||||||
|
|
||||||
|
const defaultOrigins: string[] = [
|
||||||
|
'http://localhost:3000',
|
||||||
|
'http://localhost:3010',
|
||||||
|
'http://localhost:8080',
|
||||||
|
'http://127.0.0.1:3000',
|
||||||
|
'http://127.0.0.1:3010',
|
||||||
|
'http://127.0.0.1:8080',
|
||||||
|
];
|
||||||
|
|
||||||
|
if (process.env.APP_BASE_URL) {
|
||||||
|
try {
|
||||||
|
const parsed = new URL(process.env.APP_BASE_URL);
|
||||||
|
if (!defaultOrigins.includes(parsed.origin)) {
|
||||||
|
defaultOrigins.push(parsed.origin);
|
||||||
|
}
|
||||||
|
} catch {
|
||||||
|
const trimmed = process.env.APP_BASE_URL.trim();
|
||||||
|
if (!defaultOrigins.includes(trimmed)) {
|
||||||
|
defaultOrigins.push(trimmed);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return defaultOrigins;
|
||||||
|
}
|
||||||
|
|
||||||
async function bootstrap() {
|
async function bootstrap() {
|
||||||
const app = await NestFactory.create(AppModule);
|
const app = await NestFactory.create(AppModule);
|
||||||
app.setGlobalPrefix('api');
|
app.setGlobalPrefix('api');
|
||||||
app.useGlobalPipes(new ValidationPipe({ whitelist: true, transform: true }));
|
app.useGlobalPipes(new ValidationPipe({ whitelist: true, transform: true }));
|
||||||
app.enableCors();
|
|
||||||
|
const allowedOrigins = parseAllowedOrigins();
|
||||||
|
app.enableCors({
|
||||||
|
origin: (origin, callback) => {
|
||||||
|
// Allow requests with no origin (e.g. mobile apps, curl, same-origin)
|
||||||
|
if (!origin || allowedOrigins.includes(origin)) {
|
||||||
|
callback(null, true);
|
||||||
|
} else {
|
||||||
|
callback(new Error('Not allowed by CORS'));
|
||||||
|
}
|
||||||
|
},
|
||||||
|
credentials: true,
|
||||||
|
});
|
||||||
|
|
||||||
app.useWebSocketAdapter(new WsAdapter(app));
|
app.useWebSocketAdapter(new WsAdapter(app));
|
||||||
await app.listen(process.env.PORT ?? 3000);
|
await app.listen(process.env.PORT ?? 3000);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -43,6 +43,17 @@ export class OnboardingController {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Redeems a Leitungsteam-issued Verantwortliche invite: immediately
|
||||||
|
/// ACTIVE membership, no approval step (unlike self-registration above).
|
||||||
|
/// Authenticated by the caller's raw Authentik bearer token.
|
||||||
|
@Post('verantwortliche-invites/:token/redeem')
|
||||||
|
redeemVerantwortlicheInvite(
|
||||||
|
@Param('token') token: string,
|
||||||
|
@Headers('authorization') authorization?: string,
|
||||||
|
) {
|
||||||
|
return this.onboarding.redeemInvite(bearer(authorization), token);
|
||||||
|
}
|
||||||
|
|
||||||
/// Leitungsteam: review and act on pending self-registrations.
|
/// Leitungsteam: review and act on pending self-registrations.
|
||||||
@Get('requests')
|
@Get('requests')
|
||||||
@UseGuards(AuthGuard(['authentik', 'team']), RolesGuard)
|
@UseGuards(AuthGuard(['authentik', 'team']), RolesGuard)
|
||||||
|
|||||||
@@ -39,6 +39,11 @@ function makeService(seed: {
|
|||||||
? { id: 'kc-1', name: 'KC 2026', inviteCode: 'code-1', isActive: true, gemeinden: [] }
|
? { id: 'kc-1', name: 'KC 2026', inviteCode: 'code-1', isActive: true, gemeinden: [] }
|
||||||
: seed.kc,
|
: seed.kc,
|
||||||
),
|
),
|
||||||
|
findFirst: jest.fn().mockResolvedValue(
|
||||||
|
seed.kc === undefined
|
||||||
|
? { id: 'kc-1', name: 'KC 2026', inviteCode: 'code-1', isActive: true, gemeinden: [] }
|
||||||
|
: seed.kc,
|
||||||
|
),
|
||||||
},
|
},
|
||||||
gemeinde: {
|
gemeinde: {
|
||||||
findUnique: jest.fn().mockResolvedValue(
|
findUnique: jest.fn().mockResolvedValue(
|
||||||
@@ -90,12 +95,14 @@ describe('OnboardingService.resolveInvite', () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
it('returns the KC name and its Gemeinden', async () => {
|
it('returns the KC name and its Gemeinden', async () => {
|
||||||
const { service, prisma } = makeService({});
|
const { service } = makeService({
|
||||||
prisma.kc.findUnique = jest.fn().mockResolvedValue({
|
kc: {
|
||||||
id: 'kc-1',
|
id: 'kc-1',
|
||||||
name: 'KC 2026',
|
name: 'KC 2026',
|
||||||
|
inviteCode: 'code-1',
|
||||||
isActive: true,
|
isActive: true,
|
||||||
gemeinden: [{ id: 'gem-1', name: 'Nord' }],
|
gemeinden: [{ id: 'gem-1', name: 'Nord' }],
|
||||||
|
} as never,
|
||||||
});
|
});
|
||||||
await expect(service.resolveInvite('code-1')).resolves.toEqual({
|
await expect(service.resolveInvite('code-1')).resolves.toEqual({
|
||||||
kcId: 'kc-1',
|
kcId: 'kc-1',
|
||||||
|
|||||||
@@ -25,12 +25,20 @@ export class OnboardingService {
|
|||||||
/// Public: resolves an invite code to the KC name and its Gemeinden so the
|
/// Public: resolves an invite code to the KC name and its Gemeinden so the
|
||||||
/// registrant can pick theirs. The code itself is the shared secret.
|
/// registrant can pick theirs. The code itself is the shared secret.
|
||||||
async resolveInvite(inviteCode: string) {
|
async resolveInvite(inviteCode: string) {
|
||||||
const kc = await this.prisma.kc.findUnique({
|
const trimmed = inviteCode.trim();
|
||||||
where: { inviteCode },
|
const kc =
|
||||||
|
(await this.prisma.kc.findFirst({
|
||||||
|
where: { inviteCode: { equals: trimmed, mode: 'insensitive' } },
|
||||||
include: {
|
include: {
|
||||||
gemeinden: { select: { id: true, name: true }, orderBy: { name: 'asc' } },
|
gemeinden: { select: { id: true, name: true }, orderBy: { name: 'asc' } },
|
||||||
},
|
},
|
||||||
});
|
})) ||
|
||||||
|
(await this.prisma.kc.findUnique({
|
||||||
|
where: { inviteCode: trimmed },
|
||||||
|
include: {
|
||||||
|
gemeinden: { select: { id: true, name: true }, orderBy: { name: 'asc' } },
|
||||||
|
},
|
||||||
|
}));
|
||||||
if (!kc || !kc.isActive) {
|
if (!kc || !kc.isActive) {
|
||||||
throw new NotFoundException('Unknown or inactive KC invite code');
|
throw new NotFoundException('Unknown or inactive KC invite code');
|
||||||
}
|
}
|
||||||
@@ -43,7 +51,12 @@ export class OnboardingService {
|
|||||||
}
|
}
|
||||||
const { isLeitungsteam, ...claims } = await this.tokens.verifyAuthentikClaims(token);
|
const { isLeitungsteam, ...claims } = await this.tokens.verifyAuthentikClaims(token);
|
||||||
|
|
||||||
const kc = await this.prisma.kc.findUnique({ where: { inviteCode } });
|
const trimmed = inviteCode.trim();
|
||||||
|
const kc =
|
||||||
|
(await this.prisma.kc.findFirst({
|
||||||
|
where: { inviteCode: { equals: trimmed, mode: 'insensitive' } },
|
||||||
|
})) ||
|
||||||
|
(await this.prisma.kc.findUnique({ where: { inviteCode: trimmed } }));
|
||||||
if (!kc || !kc.isActive) {
|
if (!kc || !kc.isActive) {
|
||||||
throw new NotFoundException('Unknown or inactive KC invite code');
|
throw new NotFoundException('Unknown or inactive KC invite code');
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -50,7 +50,7 @@ export class PushService {
|
|||||||
try {
|
try {
|
||||||
const channel = await this.prisma.chatChannel.findUnique({
|
const channel = await this.prisma.chatChannel.findUnique({
|
||||||
where: { id: channelId },
|
where: { id: channelId },
|
||||||
include: { participants: { select: { userId: true } } },
|
include: { participants: { select: { userId: true, guestAccountId: true } } },
|
||||||
});
|
});
|
||||||
if (!channel) return;
|
if (!channel) return;
|
||||||
|
|
||||||
@@ -90,10 +90,22 @@ export class PushService {
|
|||||||
kcId: string;
|
kcId: string;
|
||||||
type: ChatChannelType;
|
type: ChatChannelType;
|
||||||
gemeindeId: string | null;
|
gemeindeId: string | null;
|
||||||
participants: { userId: string }[];
|
participants: { userId: string | null; guestAccountId: string | null }[];
|
||||||
}): Promise<{ userIds: string[]; guestIds: string[] }> {
|
}): Promise<{ userIds: string[]; guestIds: string[] }> {
|
||||||
if (channel.type === ChatChannelType.DIREKT) {
|
if (channel.type === ChatChannelType.DIREKT) {
|
||||||
return { userIds: channel.participants.map((p) => p.userId), guestIds: [] };
|
return {
|
||||||
|
userIds: channel.participants.map((p) => p.userId).filter((id): id is string => !!id),
|
||||||
|
guestIds: [],
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
if (channel.type === ChatChannelType.GRUPPE) {
|
||||||
|
return {
|
||||||
|
userIds: channel.participants.map((p) => p.userId).filter((id): id is string => !!id),
|
||||||
|
guestIds: channel.participants
|
||||||
|
.map((p) => p.guestAccountId)
|
||||||
|
.filter((id): id is string => !!id),
|
||||||
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
const ltUsers = await this.prisma.user.findMany({
|
const ltUsers = await this.prisma.user.findMany({
|
||||||
|
|||||||
@@ -1,7 +1,55 @@
|
|||||||
import { IsArray, IsNotEmpty } from 'class-validator';
|
import { Type } from 'class-transformer';
|
||||||
|
import {
|
||||||
|
IsArray,
|
||||||
|
IsDateString,
|
||||||
|
IsIn,
|
||||||
|
IsInt,
|
||||||
|
IsNotEmpty,
|
||||||
|
IsObject,
|
||||||
|
IsString,
|
||||||
|
Min,
|
||||||
|
ValidateNested,
|
||||||
|
} from 'class-validator';
|
||||||
|
import { SyncOperation } from '@prisma/client';
|
||||||
|
import { SYNCED_MODELS, SyncedModel } from '../synced-models';
|
||||||
|
|
||||||
|
/// Strict per-entry validation: only whitelisted models/operations are
|
||||||
|
/// accepted, and the payload must be a plain object. This is the boundary
|
||||||
|
/// where an untrusted peer's JSON becomes typed data - reject anything that
|
||||||
|
/// doesn't match rather than letting it reach Prisma's generic delegate.
|
||||||
|
export class SyncEntryDto {
|
||||||
|
@IsInt()
|
||||||
|
@Min(1)
|
||||||
|
sequence!: number;
|
||||||
|
|
||||||
|
@IsIn(SYNCED_MODELS)
|
||||||
|
model!: SyncedModel;
|
||||||
|
|
||||||
|
@IsString()
|
||||||
|
@IsNotEmpty()
|
||||||
|
recordId!: string;
|
||||||
|
|
||||||
|
@IsIn(Object.values(SyncOperation))
|
||||||
|
operation!: SyncOperation;
|
||||||
|
|
||||||
|
@IsObject()
|
||||||
|
payload!: Record<string, unknown>;
|
||||||
|
|
||||||
|
@IsString()
|
||||||
|
@IsNotEmpty()
|
||||||
|
originId!: string;
|
||||||
|
|
||||||
|
/// Wall-clock time the mutation actually happened, used for last-write-wins
|
||||||
|
/// conflict resolution - required so a peer can't omit it and silently
|
||||||
|
/// win every conflict via a default "now".
|
||||||
|
@IsDateString()
|
||||||
|
occurredAt!: string;
|
||||||
|
}
|
||||||
|
|
||||||
export class IngestEntriesDto {
|
export class IngestEntriesDto {
|
||||||
@IsArray()
|
@IsArray()
|
||||||
@IsNotEmpty()
|
@ValidateNested({ each: true })
|
||||||
entries!: unknown[];
|
@Type(() => SyncEntryDto)
|
||||||
|
entries!: SyncEntryDto[];
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -9,6 +9,11 @@ import { SyncService } from './sync.service';
|
|||||||
@Injectable()
|
@Injectable()
|
||||||
export class SyncSchedulerService {
|
export class SyncSchedulerService {
|
||||||
private readonly logger = new Logger(SyncSchedulerService.name);
|
private readonly logger = new Logger(SyncSchedulerService.name);
|
||||||
|
/// Guards against a tick starting while the previous one is still running
|
||||||
|
/// (e.g. a large backlog push/pull that exceeds the 30s interval).
|
||||||
|
/// SyncService.withPeerLock is the authoritative per-peer guard; this flag
|
||||||
|
/// just avoids logging noisy "already in progress" warnings every tick.
|
||||||
|
private running = false;
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
private readonly sync: SyncService,
|
private readonly sync: SyncService,
|
||||||
@@ -21,12 +26,22 @@ export class SyncSchedulerService {
|
|||||||
const peerUrl = this.config.get<string>('SYNC_PEER_URL');
|
const peerUrl = this.config.get<string>('SYNC_PEER_URL');
|
||||||
const peerSecret = this.config.get<string>('SYNC_SHARED_SECRET');
|
const peerSecret = this.config.get<string>('SYNC_SHARED_SECRET');
|
||||||
if (!peerUrl || !peerSecret) return;
|
if (!peerUrl || !peerSecret) return;
|
||||||
|
if (this.running) {
|
||||||
|
this.logger.debug('Previous sync tick still running, skipping this tick');
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.running = true;
|
||||||
try {
|
try {
|
||||||
await this.sync.pushToPeer(peerUrl, peerSecret);
|
const pushResult = await this.sync.pushToPeer(peerUrl, peerSecret);
|
||||||
await this.sync.pullFromPeer(peerUrl, peerSecret);
|
const pullResult = await this.sync.pullFromPeer(peerUrl, peerSecret);
|
||||||
|
this.logger.debug(
|
||||||
|
`Sync tick ok: pushed=${pushResult.pushed} pulled=${pullResult.pulled}`,
|
||||||
|
);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
this.logger.debug(`Sync with peer skipped: ${(err as Error).message}`);
|
this.logger.debug(`Sync with peer skipped: ${(err as Error).message}`);
|
||||||
|
} finally {
|
||||||
|
this.running = false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import { CanActivate, ExecutionContext, ForbiddenException, Injectable } from '@nestjs/common';
|
import { CanActivate, ExecutionContext, ForbiddenException, Injectable } from '@nestjs/common';
|
||||||
import { ConfigService } from '@nestjs/config';
|
import { ConfigService } from '@nestjs/config';
|
||||||
import { Request } from 'express';
|
import { Request } from 'express';
|
||||||
|
import { timingSafeEqual } from 'crypto';
|
||||||
|
|
||||||
/// Server-to-server auth for /sync/*: a shared secret header, not a user token.
|
/// Server-to-server auth for /sync/*: a shared secret header, not a user token.
|
||||||
@Injectable()
|
@Injectable()
|
||||||
@@ -10,9 +11,26 @@ export class SyncSecretGuard implements CanActivate {
|
|||||||
canActivate(context: ExecutionContext): boolean {
|
canActivate(context: ExecutionContext): boolean {
|
||||||
const request = context.switchToHttp().getRequest<Request>();
|
const request = context.switchToHttp().getRequest<Request>();
|
||||||
const expected = this.config.getOrThrow<string>('SYNC_SHARED_SECRET');
|
const expected = this.config.getOrThrow<string>('SYNC_SHARED_SECRET');
|
||||||
if (request.headers['x-sync-secret'] !== expected) {
|
const provided = request.headers['x-sync-secret'];
|
||||||
|
|
||||||
|
if (typeof provided !== 'string' || !this.secretsMatch(provided, expected)) {
|
||||||
throw new ForbiddenException('Invalid sync secret');
|
throw new ForbiddenException('Invalid sync secret');
|
||||||
}
|
}
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Plain `!==` leaks timing info proportional to the matching prefix
|
||||||
|
/// length, letting an attacker brute-force the secret byte by byte over
|
||||||
|
/// enough requests. Compare as fixed-length buffers instead.
|
||||||
|
private secretsMatch(provided: string, expected: string): boolean {
|
||||||
|
const providedBuf = Buffer.from(provided);
|
||||||
|
const expectedBuf = Buffer.from(expected);
|
||||||
|
if (providedBuf.length !== expectedBuf.length) {
|
||||||
|
// Still run a constant-time compare against a same-length dummy so
|
||||||
|
// the length check itself doesn't introduce a distinct fast path.
|
||||||
|
timingSafeEqual(expectedBuf, expectedBuf);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
return timingSafeEqual(providedBuf, expectedBuf);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,6 +8,8 @@ import { Roles } from '../common/roles.decorator';
|
|||||||
import { RolesGuard } from '../common/roles.guard';
|
import { RolesGuard } from '../common/roles.guard';
|
||||||
import { Role } from '../common/role.enum';
|
import { Role } from '../common/role.enum';
|
||||||
|
|
||||||
|
const EXPORT_QUERY_MAX = 1_000_000_000;
|
||||||
|
|
||||||
@Controller('sync')
|
@Controller('sync')
|
||||||
export class SyncController {
|
export class SyncController {
|
||||||
constructor(
|
constructor(
|
||||||
@@ -19,15 +21,20 @@ export class SyncController {
|
|||||||
@Post('ingest')
|
@Post('ingest')
|
||||||
@UseGuards(SyncSecretGuard)
|
@UseGuards(SyncSecretGuard)
|
||||||
async ingest(@Body() dto: IngestEntriesDto) {
|
async ingest(@Body() dto: IngestEntriesDto) {
|
||||||
await this.sync.applyIncoming(dto.entries as never);
|
return this.sync.applyIncoming(dto.entries);
|
||||||
return { applied: dto.entries.length };
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Peer pulls our new entries since their last known sequence.
|
/// Peer pulls our new entries since their last known sequence.
|
||||||
@Get('export')
|
@Get('export')
|
||||||
@UseGuards(SyncSecretGuard)
|
@UseGuards(SyncSecretGuard)
|
||||||
async export(@Query('since') since: string) {
|
async export(@Query('since') since: string) {
|
||||||
const entries = await this.sync.getEntriesSince(Number(since) || 0);
|
// Reject garbage/negative/absurd cursors outright rather than silently
|
||||||
|
// coercing them to 0 (which would re-export the whole log to a peer
|
||||||
|
// that sent a malformed value).
|
||||||
|
const parsed = Number(since);
|
||||||
|
const sinceSequence =
|
||||||
|
Number.isInteger(parsed) && parsed >= 0 && parsed <= EXPORT_QUERY_MAX ? parsed : 0;
|
||||||
|
const entries = await this.sync.getEntriesSince(sinceSequence);
|
||||||
return { entries };
|
return { entries };
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -42,4 +49,16 @@ export class SyncController {
|
|||||||
const pulled = await this.sync.pullFromPeer(peerUrl, peerSecret);
|
const pulled = await this.sync.pullFromPeer(peerUrl, peerSecret);
|
||||||
return { ...pushed, ...pulled };
|
return { ...pushed, ...pulled };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Leitungsteam-only visibility into replication health: cursors, pending
|
||||||
|
/// backlog size, and the last push/pull timestamps or error, so a stalled
|
||||||
|
/// sync (e.g. bad secret, network down) shows up before anyone notices
|
||||||
|
/// stale data.
|
||||||
|
@Get('status')
|
||||||
|
@UseGuards(AuthGuard(['authentik', 'team']), RolesGuard)
|
||||||
|
@Roles(Role.LEITUNGSTEAM)
|
||||||
|
async status() {
|
||||||
|
const peerUrl = this.config.getOrThrow<string>('SYNC_PEER_URL');
|
||||||
|
return this.sync.getStatus(peerUrl);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+255
-55
@@ -2,45 +2,58 @@ import { Injectable, Logger } from '@nestjs/common';
|
|||||||
import { ConfigService } from '@nestjs/config';
|
import { ConfigService } from '@nestjs/config';
|
||||||
import { SyncOperation } from '@prisma/client';
|
import { SyncOperation } from '@prisma/client';
|
||||||
import { PrismaClient } from '../prisma/prisma.module';
|
import { PrismaClient } from '../prisma/prisma.module';
|
||||||
|
import { SYNCED_MODELS, SyncedModel } from './synced-models';
|
||||||
|
|
||||||
const SYNCED_MODELS = [
|
const PAGE_SIZE = 500;
|
||||||
'Kc',
|
/// Hard ceiling per push/pull call so a huge backlog (e.g. days offline)
|
||||||
'Gemeinde',
|
/// can't turn one tick into an unbounded, memory-hungry transfer. The
|
||||||
'User',
|
/// scheduler just picks it back up on the next tick.
|
||||||
'Membership',
|
const MAX_ENTRIES_PER_CALL = 10 * PAGE_SIZE;
|
||||||
'TeamerInvite',
|
|
||||||
'VerantwortlicheInvite',
|
|
||||||
'GuestAccount',
|
|
||||||
'Wahl',
|
|
||||||
'Workshop',
|
|
||||||
'Teilnehmer',
|
|
||||||
'ForceZuteilung',
|
|
||||||
'Zuteilung',
|
|
||||||
'File',
|
|
||||||
'ChatChannel',
|
|
||||||
'ChatMessage',
|
|
||||||
'DeviceToken',
|
|
||||||
] as const;
|
|
||||||
export type SyncedModel = (typeof SYNCED_MODELS)[number];
|
|
||||||
|
|
||||||
interface IncomingEntry {
|
export interface IncomingEntry {
|
||||||
sequence: number;
|
sequence: number;
|
||||||
model: string;
|
model: SyncedModel;
|
||||||
recordId: string;
|
recordId: string;
|
||||||
operation: SyncOperation;
|
operation: SyncOperation;
|
||||||
payload: Record<string, unknown>;
|
payload: Record<string, unknown>;
|
||||||
originId: string;
|
originId: string;
|
||||||
|
occurredAt: string | Date;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Replicates mutations between the local (on-site) and cloud server. The
|
export interface SyncPeerStatus {
|
||||||
/// local server is the sole source of truth while an event is live, so
|
peerId: string;
|
||||||
/// incoming entries are applied with simple upserts - no conflict resolution
|
lastPushedSequence: number;
|
||||||
/// is needed by design (see plan doc).
|
lastPulledSequence: number;
|
||||||
|
localMaxSequence: number;
|
||||||
|
pendingPush: number;
|
||||||
|
lastPushAt: Date | null;
|
||||||
|
lastPullAt: Date | null;
|
||||||
|
lastError: string | null;
|
||||||
|
recentConflicts: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Replicates mutations between the local (on-site) and cloud server.
|
||||||
|
///
|
||||||
|
/// Concurrent-edit handling: both servers can legitimately write while an
|
||||||
|
/// event is live (e.g. Leitungsteam edits in the cloud admin UI while the
|
||||||
|
/// on-site server is also active), so we resolve conflicts automatically by
|
||||||
|
/// last-write-wins on the mutation's real wall-clock time (`occurredAt`),
|
||||||
|
/// never by sync/network arrival order. `SyncRecordVersion` tracks, per
|
||||||
|
/// record, the most recent writer and timestamp seen by *this* server
|
||||||
|
/// (whether written locally or applied from a peer). A losing write is
|
||||||
|
/// still recorded to `SyncConflict` for after-the-fact review - resolution
|
||||||
|
/// itself never blocks or pauses live sync.
|
||||||
@Injectable()
|
@Injectable()
|
||||||
export class SyncService {
|
export class SyncService {
|
||||||
private readonly logger = new Logger(SyncService.name);
|
private readonly logger = new Logger(SyncService.name);
|
||||||
readonly serverId: string;
|
readonly serverId: string;
|
||||||
|
|
||||||
|
/// In-memory, per-peer status for observability (health endpoint) plus a
|
||||||
|
/// crude mutex so overlapping ticks (a push+pull pair that takes longer
|
||||||
|
/// than 30s) can't race each other's cursor updates.
|
||||||
|
private readonly peerStatus = new Map<string, SyncPeerStatus>();
|
||||||
|
private readonly peerLocks = new Set<string>();
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
private readonly prisma: PrismaClient,
|
private readonly prisma: PrismaClient,
|
||||||
private readonly config: ConfigService,
|
private readonly config: ConfigService,
|
||||||
@@ -48,62 +61,142 @@ export class SyncService {
|
|||||||
this.serverId = config.getOrThrow<string>('SERVER_ID');
|
this.serverId = config.getOrThrow<string>('SERVER_ID');
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Called by feature services right after a mutation to append it to the replication log.
|
/// Called by feature services right after a mutation to append it to the
|
||||||
async capture(model: SyncedModel, operation: SyncOperation, recordId: string, payload: object) {
|
/// replication log and mark this server as the latest writer of record.
|
||||||
await this.prisma.syncLogEntry.create({
|
async capture(
|
||||||
|
model: SyncedModel,
|
||||||
|
operation: SyncOperation,
|
||||||
|
recordId: string,
|
||||||
|
payload: object,
|
||||||
|
occurredAt: Date = new Date(),
|
||||||
|
) {
|
||||||
|
await this.prisma.$transaction([
|
||||||
|
this.prisma.syncLogEntry.create({
|
||||||
data: {
|
data: {
|
||||||
model,
|
model,
|
||||||
recordId,
|
recordId,
|
||||||
operation,
|
operation,
|
||||||
payload: payload as never,
|
payload: payload as never,
|
||||||
originId: this.serverId,
|
originId: this.serverId,
|
||||||
|
occurredAt,
|
||||||
},
|
},
|
||||||
});
|
}),
|
||||||
|
this.prisma.syncRecordVersion.upsert({
|
||||||
|
where: { model_recordId: { model, recordId } },
|
||||||
|
create: { model, recordId, lastWriteAt: occurredAt, lastWriteOrigin: this.serverId },
|
||||||
|
update: { lastWriteAt: occurredAt, lastWriteOrigin: this.serverId },
|
||||||
|
}),
|
||||||
|
]);
|
||||||
}
|
}
|
||||||
|
|
||||||
async getEntriesSince(sequence: number, limit = 500) {
|
async getEntriesSince(sequence: number, limit = PAGE_SIZE) {
|
||||||
return this.prisma.syncLogEntry.findMany({
|
return this.prisma.syncLogEntry.findMany({
|
||||||
where: { sequence: { gt: sequence } },
|
where: { sequence: { gt: sequence } },
|
||||||
orderBy: { sequence: 'asc' },
|
orderBy: { sequence: 'asc' },
|
||||||
take: limit,
|
take: Math.min(limit, MAX_ENTRIES_PER_CALL),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Applies entries received from a peer; never re-captures them, which is
|
/// Applies entries received from a peer inside one transaction, so a
|
||||||
/// what prevents echo loops between the two servers.
|
/// mid-batch failure can't leave the local DB half-updated relative to the
|
||||||
|
/// cursor we're about to advance. Never re-captures them, which is what
|
||||||
|
/// prevents echo loops between the two servers.
|
||||||
|
///
|
||||||
|
/// Per entry: if this server has a newer local write for the same record
|
||||||
|
/// (by occurredAt) from a *different* origin, the incoming entry loses -
|
||||||
|
/// it's recorded as a SyncConflict and skipped, keeping the newer local
|
||||||
|
/// data intact. Otherwise the incoming entry wins and is applied.
|
||||||
async applyIncoming(entries: IncomingEntry[]) {
|
async applyIncoming(entries: IncomingEntry[]) {
|
||||||
|
let appliedCount = 0;
|
||||||
|
let conflictCount = 0;
|
||||||
|
await this.prisma.$transaction(async (tx) => {
|
||||||
for (const entry of entries) {
|
for (const entry of entries) {
|
||||||
if (entry.originId === this.serverId) continue;
|
if (entry.originId === this.serverId) continue;
|
||||||
const delegate = this.delegateFor(entry.model);
|
if (!SYNCED_MODELS.includes(entry.model)) {
|
||||||
if (!delegate) {
|
this.logger.warn(`Rejecting sync entry for unknown model "${entry.model}"`);
|
||||||
this.logger.warn(`Skipping sync entry for unknown model "${entry.model}"`);
|
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const occurredAt = new Date(entry.occurredAt);
|
||||||
|
const existing = await tx.syncRecordVersion.findUnique({
|
||||||
|
where: { model_recordId: { model: entry.model, recordId: entry.recordId } },
|
||||||
|
});
|
||||||
|
|
||||||
|
if (existing && existing.lastWriteOrigin !== entry.originId && existing.lastWriteAt > occurredAt) {
|
||||||
|
// A newer write (by a different origin) already won for this
|
||||||
|
// record - keep it, log the loser for manual review.
|
||||||
|
conflictCount += 1;
|
||||||
|
this.logger.warn(
|
||||||
|
`Sync conflict on ${entry.model}/${entry.recordId}: keeping ${existing.lastWriteOrigin}'s newer write over ${entry.originId}'s`,
|
||||||
|
);
|
||||||
|
const delegate = this.delegateFor(tx, entry.model);
|
||||||
|
const currentRecord = await delegate
|
||||||
|
.findUnique({ where: { id: entry.recordId } })
|
||||||
|
.catch(() => null);
|
||||||
|
await tx.syncConflict.create({
|
||||||
|
data: {
|
||||||
|
model: entry.model,
|
||||||
|
recordId: entry.recordId,
|
||||||
|
winningOrigin: existing.lastWriteOrigin,
|
||||||
|
losingOrigin: entry.originId,
|
||||||
|
winningPayload: (currentRecord ?? {}) as never,
|
||||||
|
losingPayload: entry.payload as never,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
const delegate = this.delegateFor(tx, entry.model);
|
||||||
try {
|
try {
|
||||||
if (entry.operation === SyncOperation.DELETE) {
|
if (entry.operation === SyncOperation.DELETE) {
|
||||||
await delegate.delete({ where: { id: entry.recordId } });
|
await delegate.delete({ where: { id: entry.recordId } });
|
||||||
} else {
|
} else {
|
||||||
|
// Force the record's id from recordId, not from payload, so a
|
||||||
|
// mismatched/forged id in the payload can never redirect the
|
||||||
|
// write onto a different row.
|
||||||
|
const { id: _ignoredId, ...rest } = entry.payload;
|
||||||
await delegate.upsert({
|
await delegate.upsert({
|
||||||
where: { id: entry.recordId },
|
where: { id: entry.recordId },
|
||||||
create: entry.payload,
|
create: { id: entry.recordId, ...rest },
|
||||||
update: entry.payload,
|
update: rest,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
await tx.syncRecordVersion.upsert({
|
||||||
|
where: { model_recordId: { model: entry.model, recordId: entry.recordId } },
|
||||||
|
create: {
|
||||||
|
model: entry.model,
|
||||||
|
recordId: entry.recordId,
|
||||||
|
lastWriteAt: occurredAt,
|
||||||
|
lastWriteOrigin: entry.originId,
|
||||||
|
},
|
||||||
|
update: { lastWriteAt: occurredAt, lastWriteOrigin: entry.originId },
|
||||||
|
});
|
||||||
|
appliedCount += 1;
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
this.logger.warn(
|
this.logger.warn(
|
||||||
`Failed to apply sync entry ${entry.model}/${entry.recordId}: ${(err as Error).message}`,
|
`Failed to apply sync entry ${entry.model}/${entry.recordId}: ${(err as Error).message}`,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
});
|
||||||
|
return { applied: appliedCount, conflicts: conflictCount };
|
||||||
}
|
}
|
||||||
|
|
||||||
async pushToPeer(peerUrl: string, peerSecret: string) {
|
async pushToPeer(peerUrl: string, peerSecret: string) {
|
||||||
const peerId = new URL(peerUrl).host;
|
const peerId = new URL(peerUrl).host;
|
||||||
|
return this.withPeerLock(peerId, async () => {
|
||||||
const cursor = await this.getOrCreateCursor(peerId);
|
const cursor = await this.getOrCreateCursor(peerId);
|
||||||
const entries = await this.getEntriesSince(cursor.lastPushedSequence);
|
let pushed = 0;
|
||||||
if (entries.length === 0) return { pushed: 0 };
|
let lastSequence = cursor.lastPushedSequence;
|
||||||
|
|
||||||
const res = await fetch(`${peerUrl}/sync/ingest`, {
|
// Loop pages so a large backlog (offline event site catching back up)
|
||||||
|
// is fully drained in one tick instead of trickling 500 at a time
|
||||||
|
// across many 30s intervals.
|
||||||
|
for (;;) {
|
||||||
|
const entries = await this.getEntriesSince(lastSequence);
|
||||||
|
if (entries.length === 0) break;
|
||||||
|
|
||||||
|
const res = await this.fetchWithTimeout(`${peerUrl}/sync/ingest`, {
|
||||||
method: 'POST',
|
method: 'POST',
|
||||||
headers: { 'Content-Type': 'application/json', 'x-sync-secret': peerSecret },
|
headers: { 'Content-Type': 'application/json', 'x-sync-secret': peerSecret },
|
||||||
body: JSON.stringify({ entries }),
|
body: JSON.stringify({ entries }),
|
||||||
@@ -111,31 +204,131 @@ export class SyncService {
|
|||||||
if (!res.ok) {
|
if (!res.ok) {
|
||||||
throw new Error(`Peer rejected sync push: ${res.status}`);
|
throw new Error(`Peer rejected sync push: ${res.status}`);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
lastSequence = entries[entries.length - 1].sequence;
|
||||||
|
pushed += entries.length;
|
||||||
await this.prisma.syncCursor.update({
|
await this.prisma.syncCursor.update({
|
||||||
where: { peerId },
|
where: { peerId },
|
||||||
data: { lastPushedSequence: entries[entries.length - 1].sequence },
|
data: { lastPushedSequence: lastSequence },
|
||||||
|
});
|
||||||
|
|
||||||
|
if (entries.length < PAGE_SIZE || pushed >= MAX_ENTRIES_PER_CALL) break;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.recordSuccess(peerId, { lastPushAt: new Date() });
|
||||||
|
return { pushed };
|
||||||
});
|
});
|
||||||
return { pushed: entries.length };
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async pullFromPeer(peerUrl: string, peerSecret: string) {
|
async pullFromPeer(peerUrl: string, peerSecret: string) {
|
||||||
const peerId = new URL(peerUrl).host;
|
const peerId = new URL(peerUrl).host;
|
||||||
|
return this.withPeerLock(peerId, async () => {
|
||||||
const cursor = await this.getOrCreateCursor(peerId);
|
const cursor = await this.getOrCreateCursor(peerId);
|
||||||
const res = await fetch(`${peerUrl}/sync/export?since=${cursor.lastPulledSequence}`, {
|
let pulled = 0;
|
||||||
headers: { 'x-sync-secret': peerSecret },
|
let conflicts = 0;
|
||||||
});
|
let lastSequence = cursor.lastPulledSequence;
|
||||||
|
|
||||||
|
for (;;) {
|
||||||
|
const res = await this.fetchWithTimeout(
|
||||||
|
`${peerUrl}/sync/export?since=${lastSequence}`,
|
||||||
|
{ headers: { 'x-sync-secret': peerSecret } },
|
||||||
|
);
|
||||||
if (!res.ok) {
|
if (!res.ok) {
|
||||||
throw new Error(`Peer rejected sync pull: ${res.status}`);
|
throw new Error(`Peer rejected sync pull: ${res.status}`);
|
||||||
}
|
}
|
||||||
const { entries } = (await res.json()) as { entries: IncomingEntry[] };
|
const { entries } = (await res.json()) as { entries: IncomingEntry[] };
|
||||||
if (entries.length === 0) return { pulled: 0 };
|
if (entries.length === 0) break;
|
||||||
|
|
||||||
await this.applyIncoming(entries);
|
const result = await this.applyIncoming(entries);
|
||||||
|
conflicts += result.conflicts;
|
||||||
|
lastSequence = entries[entries.length - 1].sequence;
|
||||||
|
pulled += entries.length;
|
||||||
await this.prisma.syncCursor.update({
|
await this.prisma.syncCursor.update({
|
||||||
where: { peerId },
|
where: { peerId },
|
||||||
data: { lastPulledSequence: entries[entries.length - 1].sequence },
|
data: { lastPulledSequence: lastSequence },
|
||||||
});
|
});
|
||||||
return { pulled: entries.length };
|
|
||||||
|
if (entries.length < PAGE_SIZE || pulled >= MAX_ENTRIES_PER_CALL) break;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.recordSuccess(peerId, { lastPullAt: new Date(), recentConflicts: conflicts });
|
||||||
|
return { pulled, conflicts };
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Snapshot of replication health for a peer, for the /sync/status endpoint.
|
||||||
|
async getStatus(peerUrl: string): Promise<SyncPeerStatus> {
|
||||||
|
const peerId = new URL(peerUrl).host;
|
||||||
|
const cursor = await this.getOrCreateCursor(peerId);
|
||||||
|
const latest = await this.prisma.syncLogEntry.findFirst({ orderBy: { sequence: 'desc' } });
|
||||||
|
const localMaxSequence = latest?.sequence ?? 0;
|
||||||
|
const recentConflicts = await this.prisma.syncConflict.count({
|
||||||
|
where: { detectedAt: { gt: new Date(Date.now() - 24 * 60 * 60 * 1000) } },
|
||||||
|
});
|
||||||
|
const cached = this.peerStatus.get(peerId);
|
||||||
|
return {
|
||||||
|
peerId,
|
||||||
|
lastPushedSequence: cursor.lastPushedSequence,
|
||||||
|
lastPulledSequence: cursor.lastPulledSequence,
|
||||||
|
localMaxSequence,
|
||||||
|
pendingPush: Math.max(0, localMaxSequence - cursor.lastPushedSequence),
|
||||||
|
lastPushAt: cached?.lastPushAt ?? null,
|
||||||
|
lastPullAt: cached?.lastPullAt ?? null,
|
||||||
|
lastError: cached?.lastError ?? null,
|
||||||
|
recentConflicts,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Serializes push/pull per peer so an overrunning tick (slow network,
|
||||||
|
/// big backlog) can never overlap with the next scheduled tick and race
|
||||||
|
/// the same cursor row.
|
||||||
|
private async withPeerLock<T>(peerId: string, fn: () => Promise<T>): Promise<T> {
|
||||||
|
if (this.peerLocks.has(peerId)) {
|
||||||
|
throw new Error(`Sync with ${peerId} already in progress, skipping`);
|
||||||
|
}
|
||||||
|
this.peerLocks.add(peerId);
|
||||||
|
try {
|
||||||
|
return await fn();
|
||||||
|
} catch (err) {
|
||||||
|
this.recordFailure(peerId, err as Error);
|
||||||
|
throw err;
|
||||||
|
} finally {
|
||||||
|
this.peerLocks.delete(peerId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private blankStatus(peerId: string): SyncPeerStatus {
|
||||||
|
return {
|
||||||
|
peerId,
|
||||||
|
lastPushedSequence: 0,
|
||||||
|
lastPulledSequence: 0,
|
||||||
|
localMaxSequence: 0,
|
||||||
|
pendingPush: 0,
|
||||||
|
lastPushAt: null,
|
||||||
|
lastPullAt: null,
|
||||||
|
lastError: null,
|
||||||
|
recentConflicts: 0,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
private recordSuccess(peerId: string, patch: Partial<SyncPeerStatus>) {
|
||||||
|
const current = this.peerStatus.get(peerId) ?? this.blankStatus(peerId);
|
||||||
|
this.peerStatus.set(peerId, { ...current, ...patch, lastError: null });
|
||||||
|
}
|
||||||
|
|
||||||
|
private recordFailure(peerId: string, err: Error) {
|
||||||
|
const current = this.peerStatus.get(peerId) ?? this.blankStatus(peerId);
|
||||||
|
this.peerStatus.set(peerId, { ...current, lastError: err.message });
|
||||||
|
}
|
||||||
|
|
||||||
|
private async fetchWithTimeout(url: string, init: RequestInit, timeoutMs = 15_000) {
|
||||||
|
const controller = new AbortController();
|
||||||
|
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||||
|
try {
|
||||||
|
return await fetch(url, { ...init, signal: controller.signal });
|
||||||
|
} finally {
|
||||||
|
clearTimeout(timer);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private async getOrCreateCursor(peerId: string) {
|
private async getOrCreateCursor(peerId: string) {
|
||||||
@@ -146,14 +339,21 @@ export class SyncService {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
private delegateFor(model: string) {
|
private delegateFor(
|
||||||
if (!SYNCED_MODELS.includes(model as SyncedModel)) return null;
|
tx: Parameters<Parameters<PrismaClient['$transaction']>[0]>[0],
|
||||||
const key = (model.charAt(0).toLowerCase() + model.slice(1)) as keyof PrismaClient;
|
model: SyncedModel,
|
||||||
|
) {
|
||||||
|
const key = (model.charAt(0).toLowerCase() + model.slice(1)) as keyof typeof tx;
|
||||||
// Generic dispatch across models is inherent to a replication log; each
|
// Generic dispatch across models is inherent to a replication log; each
|
||||||
// delegate exposes the same upsert/delete shape we need here.
|
// delegate exposes the same upsert/delete shape we need here.
|
||||||
return this.prisma[key] as unknown as {
|
return tx[key] as unknown as {
|
||||||
upsert: (args: { where: { id: string }; create: object; update: object }) => Promise<unknown>;
|
upsert: (args: {
|
||||||
|
where: { id: string };
|
||||||
|
create: object;
|
||||||
|
update: object;
|
||||||
|
}) => Promise<unknown>;
|
||||||
delete: (args: { where: { id: string } }) => Promise<unknown>;
|
delete: (args: { where: { id: string } }) => Promise<unknown>;
|
||||||
|
findUnique: (args: { where: { id: string } }) => Promise<unknown>;
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,21 @@
|
|||||||
|
export const SYNCED_MODELS = [
|
||||||
|
'Kc',
|
||||||
|
'Gemeinde',
|
||||||
|
'User',
|
||||||
|
'Membership',
|
||||||
|
'TeamerInvite',
|
||||||
|
'VerantwortlicheInvite',
|
||||||
|
'GuestAccount',
|
||||||
|
'Wahl',
|
||||||
|
'Workshop',
|
||||||
|
'Teilnehmer',
|
||||||
|
'ForceZuteilung',
|
||||||
|
'Zuteilung',
|
||||||
|
'File',
|
||||||
|
'ChatChannel',
|
||||||
|
'ChatParticipant',
|
||||||
|
'ChatMessage',
|
||||||
|
'DeviceToken',
|
||||||
|
] as const;
|
||||||
|
|
||||||
|
export type SyncedModel = (typeof SYNCED_MODELS)[number];
|
||||||
Reference in New Issue
Block a user