From 5fd2d47a6499413b643f18318bdb47c75c58c618 Mon Sep 17 00:00:00 2001 From: linus Date: Sat, 12 Sep 2026 15:40:42 +0200 Subject: [PATCH] Add code resolver, sync conflict handling, and user isolation Introduce a CodeResolverService to classify user login codes, complete with detailed resolution logic and usability checks. Extend the sync system to handle conflicts via last-write-wins arbitration, with detailed conflict tracking for review. Update file permissions and runtime isolation in Docker to enhance security. --- Dockerfile | 9 +- .../migration.sql | 31 ++ prisma/schema.prisma | 52 ++- src/auth/auth.controller.ts | 13 + src/auth/auth.module.ts | 4 +- src/auth/code-resolver.service.spec.ts | 186 +++++++++ src/auth/code-resolver.service.ts | 135 ++++++ src/auth/dto/resolve-code.dto.ts | 7 + src/auth/guest-auth.service.ts | 9 +- src/main.ts | 50 ++- src/onboarding/onboarding.controller.ts | 11 + src/onboarding/onboarding.service.spec.ts | 19 +- src/onboarding/onboarding.service.ts | 27 +- src/sync/dto/ingest-entries.dto.ts | 54 ++- src/sync/sync-scheduler.service.ts | 19 +- src/sync/sync-secret.guard.ts | 20 +- src/sync/sync.controller.ts | 25 +- src/sync/sync.service.ts | 389 +++++++++++++----- src/sync/synced-models.ts | 21 + 19 files changed, 950 insertions(+), 131 deletions(-) create mode 100644 prisma/migrations/20260912122142_sync_conflict_resolution/migration.sql create mode 100644 src/auth/code-resolver.service.spec.ts create mode 100644 src/auth/code-resolver.service.ts create mode 100644 src/auth/dto/resolve-code.dto.ts create mode 100644 src/sync/synced-models.ts diff --git a/Dockerfile b/Dockerfile index d3b9b56..1c50fff 100644 --- a/Dockerfile +++ b/Dockerfile @@ -24,9 +24,12 @@ WORKDIR /app # Prisma needs OpenSSL at runtime. RUN apt-get update && apt-get install -y --no-install-recommends openssl \ && rm -rf /var/lib/apt/lists/* -COPY --from=api-build /src/node_modules ./node_modules -COPY --from=api-build /src/dist ./dist -COPY --from=api-build /src/prisma ./prisma +RUN groupadd -g 1001 kc-user && useradd -u 1001 -g kc-user -m -d /home/kc-user kc-user +COPY --from=api-build --chown=kc-user:kc-user /src/node_modules ./node_modules +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; # app.module reads WEB_CLIENT_DIR. See docker-compose.yml. EXPOSE 3000 diff --git a/prisma/migrations/20260912122142_sync_conflict_resolution/migration.sql b/prisma/migrations/20260912122142_sync_conflict_resolution/migration.sql new file mode 100644 index 0000000..beb6190 --- /dev/null +++ b/prisma/migrations/20260912122142_sync_conflict_resolution/migration.sql @@ -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"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4b05f3d..b80376b 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -348,15 +348,19 @@ enum SyncOperation { /// Append-only log of local mutations, replicated to the peer server (local /// <-> 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). +/// `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 { - id String @id @default(cuid()) - sequence Int @default(autoincrement()) - model String - recordId String - operation SyncOperation - payload Json - originId String - createdAt DateTime @default(now()) + id String @id @default(cuid()) + sequence Int @default(autoincrement()) + model String + recordId String + operation SyncOperation + payload Json + originId String + occurredAt DateTime @default(now()) + createdAt DateTime @default(now()) } /// Per-peer replication progress, kept on the side that initiates sync @@ -368,3 +372,35 @@ model SyncCursor { lastPushedSequence 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()) +} diff --git a/src/auth/auth.controller.ts b/src/auth/auth.controller.ts index d0dd059..a946014 100644 --- a/src/auth/auth.controller.ts +++ b/src/auth/auth.controller.ts @@ -5,6 +5,8 @@ import { TeamAuthService } from './team-auth.service'; import { CreateGuestDto } from './dto/create-guest.dto'; import { TeamLoginDto } from './dto/team-login.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 { GuestJwtPayload } from './guest-auth.service'; @@ -13,8 +15,19 @@ export class AuthController { constructor( private readonly guestAuth: GuestAuthService, 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 /// client can render a role-aware UI. `kind` is "guest" for a Konfi token, /// "user" for an Authentik or local Teamer token. diff --git a/src/auth/auth.module.ts b/src/auth/auth.module.ts index 20d2733..8d49e92 100644 --- a/src/auth/auth.module.ts +++ b/src/auth/auth.module.ts @@ -9,6 +9,7 @@ import { AuthentikStrategy } from './authentik.strategy'; import { GuestJwtStrategy } from './guest-jwt.strategy'; import { TeamJwtStrategy } from './team-jwt.strategy'; import { TokenVerificationService } from './token-verification.service'; +import { CodeResolverService } from './code-resolver.service'; @Module({ imports: [ @@ -29,7 +30,8 @@ import { TokenVerificationService } from './token-verification.service'; GuestJwtStrategy, TeamJwtStrategy, TokenVerificationService, + CodeResolverService, ], - exports: [TokenVerificationService, TeamAuthService], + exports: [TokenVerificationService, TeamAuthService, CodeResolverService], }) export class AuthModule {} diff --git a/src/auth/code-resolver.service.spec.ts b/src/auth/code-resolver.service.spec.ts new file mode 100644 index 0000000..105e3a2 --- /dev/null +++ b/src/auth/code-resolver.service.spec.ts @@ -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); + }); +}); diff --git a/src/auth/code-resolver.service.ts b/src/auth/code-resolver.service.ts new file mode 100644 index 0000000..52bf9eb --- /dev/null +++ b/src/auth/code-resolver.service.ts @@ -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 { + 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; + } +} diff --git a/src/auth/dto/resolve-code.dto.ts b/src/auth/dto/resolve-code.dto.ts new file mode 100644 index 0000000..6211116 --- /dev/null +++ b/src/auth/dto/resolve-code.dto.ts @@ -0,0 +1,7 @@ +import { IsNotEmpty, IsString } from 'class-validator'; + +export class ResolveCodeDto { + @IsString() + @IsNotEmpty() + code!: string; +} diff --git a/src/auth/guest-auth.service.ts b/src/auth/guest-auth.service.ts index 4131c4b..12ca131 100644 --- a/src/auth/guest-auth.service.ts +++ b/src/auth/guest-auth.service.ts @@ -30,7 +30,14 @@ export class GuestAuthService { firstName: string, lastName: 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) { throw new NotFoundException('Unknown or inactive KC invite code'); } diff --git a/src/main.ts b/src/main.ts index 15e4af3..54291c9 100644 --- a/src/main.ts +++ b/src/main.ts @@ -3,11 +3,59 @@ import { ValidationPipe } from '@nestjs/common'; import { WsAdapter } from '@nestjs/platform-ws'; 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() { const app = await NestFactory.create(AppModule); app.setGlobalPrefix('api'); 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)); await app.listen(process.env.PORT ?? 3000); } diff --git a/src/onboarding/onboarding.controller.ts b/src/onboarding/onboarding.controller.ts index dabbffe..b0e4162 100644 --- a/src/onboarding/onboarding.controller.ts +++ b/src/onboarding/onboarding.controller.ts @@ -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. @Get('requests') @UseGuards(AuthGuard(['authentik', 'team']), RolesGuard) diff --git a/src/onboarding/onboarding.service.spec.ts b/src/onboarding/onboarding.service.spec.ts index 7e1ee35..0005820 100644 --- a/src/onboarding/onboarding.service.spec.ts +++ b/src/onboarding/onboarding.service.spec.ts @@ -39,6 +39,11 @@ function makeService(seed: { ? { id: 'kc-1', name: 'KC 2026', inviteCode: 'code-1', isActive: true, gemeinden: [] } : seed.kc, ), + findFirst: jest.fn().mockResolvedValue( + seed.kc === undefined + ? { id: 'kc-1', name: 'KC 2026', inviteCode: 'code-1', isActive: true, gemeinden: [] } + : seed.kc, + ), }, gemeinde: { findUnique: jest.fn().mockResolvedValue( @@ -90,12 +95,14 @@ describe('OnboardingService.resolveInvite', () => { }); it('returns the KC name and its Gemeinden', async () => { - const { service, prisma } = makeService({}); - prisma.kc.findUnique = jest.fn().mockResolvedValue({ - id: 'kc-1', - name: 'KC 2026', - isActive: true, - gemeinden: [{ id: 'gem-1', name: 'Nord' }], + const { service } = makeService({ + kc: { + id: 'kc-1', + name: 'KC 2026', + inviteCode: 'code-1', + isActive: true, + gemeinden: [{ id: 'gem-1', name: 'Nord' }], + } as never, }); await expect(service.resolveInvite('code-1')).resolves.toEqual({ kcId: 'kc-1', diff --git a/src/onboarding/onboarding.service.ts b/src/onboarding/onboarding.service.ts index 87158ff..f9d7423 100644 --- a/src/onboarding/onboarding.service.ts +++ b/src/onboarding/onboarding.service.ts @@ -25,12 +25,20 @@ export class OnboardingService { /// 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. async resolveInvite(inviteCode: string) { - const kc = await this.prisma.kc.findUnique({ - where: { inviteCode }, - include: { - gemeinden: { select: { id: true, name: true }, orderBy: { name: 'asc' } }, - }, - }); + const trimmed = inviteCode.trim(); + const kc = + (await this.prisma.kc.findFirst({ + where: { inviteCode: { equals: trimmed, mode: 'insensitive' } }, + include: { + 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) { 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 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) { throw new NotFoundException('Unknown or inactive KC invite code'); } diff --git a/src/sync/dto/ingest-entries.dto.ts b/src/sync/dto/ingest-entries.dto.ts index 54cc55a..d3a11a0 100644 --- a/src/sync/dto/ingest-entries.dto.ts +++ b/src/sync/dto/ingest-entries.dto.ts @@ -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; + + @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 { @IsArray() - @IsNotEmpty() - entries!: unknown[]; + @ValidateNested({ each: true }) + @Type(() => SyncEntryDto) + entries!: SyncEntryDto[]; } + diff --git a/src/sync/sync-scheduler.service.ts b/src/sync/sync-scheduler.service.ts index ea56c4f..21680fb 100644 --- a/src/sync/sync-scheduler.service.ts +++ b/src/sync/sync-scheduler.service.ts @@ -9,6 +9,11 @@ import { SyncService } from './sync.service'; @Injectable() export class SyncSchedulerService { 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( private readonly sync: SyncService, @@ -21,12 +26,22 @@ export class SyncSchedulerService { const peerUrl = this.config.get('SYNC_PEER_URL'); const peerSecret = this.config.get('SYNC_SHARED_SECRET'); if (!peerUrl || !peerSecret) return; + if (this.running) { + this.logger.debug('Previous sync tick still running, skipping this tick'); + return; + } + this.running = true; try { - await this.sync.pushToPeer(peerUrl, peerSecret); - await this.sync.pullFromPeer(peerUrl, peerSecret); + const pushResult = await this.sync.pushToPeer(peerUrl, peerSecret); + const pullResult = await this.sync.pullFromPeer(peerUrl, peerSecret); + this.logger.debug( + `Sync tick ok: pushed=${pushResult.pushed} pulled=${pullResult.pulled}`, + ); } catch (err) { this.logger.debug(`Sync with peer skipped: ${(err as Error).message}`); + } finally { + this.running = false; } } } diff --git a/src/sync/sync-secret.guard.ts b/src/sync/sync-secret.guard.ts index 3db94ce..6580e45 100644 --- a/src/sync/sync-secret.guard.ts +++ b/src/sync/sync-secret.guard.ts @@ -1,6 +1,7 @@ import { CanActivate, ExecutionContext, ForbiddenException, Injectable } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { Request } from 'express'; +import { timingSafeEqual } from 'crypto'; /// Server-to-server auth for /sync/*: a shared secret header, not a user token. @Injectable() @@ -10,9 +11,26 @@ export class SyncSecretGuard implements CanActivate { canActivate(context: ExecutionContext): boolean { const request = context.switchToHttp().getRequest(); const expected = this.config.getOrThrow('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'); } 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); + } } diff --git a/src/sync/sync.controller.ts b/src/sync/sync.controller.ts index c84107f..32c4f44 100644 --- a/src/sync/sync.controller.ts +++ b/src/sync/sync.controller.ts @@ -8,6 +8,8 @@ import { Roles } from '../common/roles.decorator'; import { RolesGuard } from '../common/roles.guard'; import { Role } from '../common/role.enum'; +const EXPORT_QUERY_MAX = 1_000_000_000; + @Controller('sync') export class SyncController { constructor( @@ -19,15 +21,20 @@ export class SyncController { @Post('ingest') @UseGuards(SyncSecretGuard) async ingest(@Body() dto: IngestEntriesDto) { - await this.sync.applyIncoming(dto.entries as never); - return { applied: dto.entries.length }; + return this.sync.applyIncoming(dto.entries); } /// Peer pulls our new entries since their last known sequence. @Get('export') @UseGuards(SyncSecretGuard) 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 }; } @@ -42,4 +49,16 @@ export class SyncController { const pulled = await this.sync.pullFromPeer(peerUrl, peerSecret); 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('SYNC_PEER_URL'); + return this.sync.getStatus(peerUrl); + } } diff --git a/src/sync/sync.service.ts b/src/sync/sync.service.ts index 1c0609d..b6fa58e 100644 --- a/src/sync/sync.service.ts +++ b/src/sync/sync.service.ts @@ -2,46 +2,58 @@ import { Injectable, Logger } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { SyncOperation } from '@prisma/client'; import { PrismaClient } from '../prisma/prisma.module'; +import { SYNCED_MODELS, SyncedModel } from './synced-models'; -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]; +const PAGE_SIZE = 500; +/// Hard ceiling per push/pull call so a huge backlog (e.g. days offline) +/// can't turn one tick into an unbounded, memory-hungry transfer. The +/// scheduler just picks it back up on the next tick. +const MAX_ENTRIES_PER_CALL = 10 * PAGE_SIZE; -interface IncomingEntry { +export interface IncomingEntry { sequence: number; - model: string; + model: SyncedModel; recordId: string; operation: SyncOperation; payload: Record; originId: string; + occurredAt: string | Date; } -/// Replicates mutations between the local (on-site) and cloud server. The -/// local server is the sole source of truth while an event is live, so -/// incoming entries are applied with simple upserts - no conflict resolution -/// is needed by design (see plan doc). +export interface SyncPeerStatus { + peerId: string; + lastPushedSequence: number; + 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() export class SyncService { private readonly logger = new Logger(SyncService.name); 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(); + private readonly peerLocks = new Set(); + constructor( private readonly prisma: PrismaClient, private readonly config: ConfigService, @@ -49,94 +61,274 @@ export class SyncService { this.serverId = config.getOrThrow('SERVER_ID'); } - /// Called by feature services right after a mutation to append it to the replication log. - async capture(model: SyncedModel, operation: SyncOperation, recordId: string, payload: object) { - await this.prisma.syncLogEntry.create({ - data: { - model, - recordId, - operation, - payload: payload as never, - originId: this.serverId, - }, - }); + /// Called by feature services right after a mutation to append it to the + /// replication log and mark this server as the latest writer of record. + async capture( + model: SyncedModel, + operation: SyncOperation, + recordId: string, + payload: object, + occurredAt: Date = new Date(), + ) { + await this.prisma.$transaction([ + this.prisma.syncLogEntry.create({ + data: { + model, + recordId, + operation, + payload: payload as never, + 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({ where: { sequence: { gt: sequence } }, 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 - /// what prevents echo loops between the two servers. + /// Applies entries received from a peer inside one transaction, so a + /// 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[]) { - for (const entry of entries) { - if (entry.originId === this.serverId) continue; - const delegate = this.delegateFor(entry.model); - if (!delegate) { - this.logger.warn(`Skipping sync entry for unknown model "${entry.model}"`); - continue; - } - try { - if (entry.operation === SyncOperation.DELETE) { - await delegate.delete({ where: { id: entry.recordId } }); - } else { - await delegate.upsert({ - where: { id: entry.recordId }, - create: entry.payload, - update: entry.payload, + let appliedCount = 0; + let conflictCount = 0; + await this.prisma.$transaction(async (tx) => { + for (const entry of entries) { + if (entry.originId === this.serverId) continue; + if (!SYNCED_MODELS.includes(entry.model)) { + this.logger.warn(`Rejecting sync entry for unknown model "${entry.model}"`); + 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 { + if (entry.operation === SyncOperation.DELETE) { + await delegate.delete({ where: { id: entry.recordId } }); + } 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({ + where: { id: entry.recordId }, + create: { id: entry.recordId, ...rest }, + 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) { + this.logger.warn( + `Failed to apply sync entry ${entry.model}/${entry.recordId}: ${(err as Error).message}`, + ); } - } catch (err) { - this.logger.warn( - `Failed to apply sync entry ${entry.model}/${entry.recordId}: ${(err as Error).message}`, - ); } - } + }); + return { applied: appliedCount, conflicts: conflictCount }; } async pushToPeer(peerUrl: string, peerSecret: string) { const peerId = new URL(peerUrl).host; - const cursor = await this.getOrCreateCursor(peerId); - const entries = await this.getEntriesSince(cursor.lastPushedSequence); - if (entries.length === 0) return { pushed: 0 }; + return this.withPeerLock(peerId, async () => { + const cursor = await this.getOrCreateCursor(peerId); + let pushed = 0; + let lastSequence = cursor.lastPushedSequence; - const res = await fetch(`${peerUrl}/sync/ingest`, { - method: 'POST', - headers: { 'Content-Type': 'application/json', 'x-sync-secret': peerSecret }, - body: JSON.stringify({ entries }), + // 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', + headers: { 'Content-Type': 'application/json', 'x-sync-secret': peerSecret }, + body: JSON.stringify({ entries }), + }); + if (!res.ok) { + throw new Error(`Peer rejected sync push: ${res.status}`); + } + + lastSequence = entries[entries.length - 1].sequence; + pushed += entries.length; + await this.prisma.syncCursor.update({ + where: { peerId }, + data: { lastPushedSequence: lastSequence }, + }); + + if (entries.length < PAGE_SIZE || pushed >= MAX_ENTRIES_PER_CALL) break; + } + + this.recordSuccess(peerId, { lastPushAt: new Date() }); + return { pushed }; }); - if (!res.ok) { - throw new Error(`Peer rejected sync push: ${res.status}`); - } - await this.prisma.syncCursor.update({ - where: { peerId }, - data: { lastPushedSequence: entries[entries.length - 1].sequence }, - }); - return { pushed: entries.length }; } async pullFromPeer(peerUrl: string, peerSecret: string) { const peerId = new URL(peerUrl).host; - const cursor = await this.getOrCreateCursor(peerId); - const res = await fetch(`${peerUrl}/sync/export?since=${cursor.lastPulledSequence}`, { - headers: { 'x-sync-secret': peerSecret }, - }); - if (!res.ok) { - throw new Error(`Peer rejected sync pull: ${res.status}`); - } - const { entries } = (await res.json()) as { entries: IncomingEntry[] }; - if (entries.length === 0) return { pulled: 0 }; + return this.withPeerLock(peerId, async () => { + const cursor = await this.getOrCreateCursor(peerId); + let pulled = 0; + let conflicts = 0; + let lastSequence = cursor.lastPulledSequence; - await this.applyIncoming(entries); - await this.prisma.syncCursor.update({ - where: { peerId }, - data: { lastPulledSequence: entries[entries.length - 1].sequence }, + for (;;) { + const res = await this.fetchWithTimeout( + `${peerUrl}/sync/export?since=${lastSequence}`, + { headers: { 'x-sync-secret': peerSecret } }, + ); + if (!res.ok) { + throw new Error(`Peer rejected sync pull: ${res.status}`); + } + const { entries } = (await res.json()) as { entries: IncomingEntry[] }; + if (entries.length === 0) break; + + const result = await this.applyIncoming(entries); + conflicts += result.conflicts; + lastSequence = entries[entries.length - 1].sequence; + pulled += entries.length; + await this.prisma.syncCursor.update({ + where: { peerId }, + data: { lastPulledSequence: lastSequence }, + }); + + if (entries.length < PAGE_SIZE || pulled >= MAX_ENTRIES_PER_CALL) break; + } + + this.recordSuccess(peerId, { lastPullAt: new Date(), recentConflicts: conflicts }); + return { pulled, conflicts }; }); - return { pulled: entries.length }; + } + + /// Snapshot of replication health for a peer, for the /sync/status endpoint. + async getStatus(peerUrl: string): Promise { + 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(peerId: string, fn: () => Promise): Promise { + 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) { + 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) { @@ -147,14 +339,21 @@ export class SyncService { }); } - private delegateFor(model: string) { - if (!SYNCED_MODELS.includes(model as SyncedModel)) return null; - const key = (model.charAt(0).toLowerCase() + model.slice(1)) as keyof PrismaClient; + private delegateFor( + tx: Parameters[0]>[0], + 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 // delegate exposes the same upsert/delete shape we need here. - return this.prisma[key] as unknown as { - upsert: (args: { where: { id: string }; create: object; update: object }) => Promise; + return tx[key] as unknown as { + upsert: (args: { + where: { id: string }; + create: object; + update: object; + }) => Promise; delete: (args: { where: { id: string } }) => Promise; + findUnique: (args: { where: { id: string } }) => Promise; }; } } diff --git a/src/sync/synced-models.ts b/src/sync/synced-models.ts new file mode 100644 index 0000000..9914e11 --- /dev/null +++ b/src/sync/synced-models.ts @@ -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];