KC-APP-Server/src/sync/sync.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { SyncOperation } from '@prisma/client';
import { PrismaClient } from '../prisma/prisma.module';

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];

interface IncomingEntry {
  sequence: number;
  model: string;
  recordId: string;
  operation: SyncOperation;
  payload: Record<string, unknown>;
  originId: string;
}

/// 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).
@Injectable()
export class SyncService {
  private readonly logger = new Logger(SyncService.name);
  readonly serverId: string;

  constructor(
    private readonly prisma: PrismaClient,
    private readonly config: ConfigService,
  ) {
    this.serverId = config.getOrThrow<string>('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,
      },
    });
  }

  async getEntriesSince(sequence: number, limit = 500) {
    return this.prisma.syncLogEntry.findMany({
      where: { sequence: { gt: sequence } },
      orderBy: { sequence: 'asc' },
      take: limit,
    });
  }

  /// Applies entries received from a peer; never re-captures them, which is
  /// what prevents echo loops between the two servers.
  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,
          });
        }
      } catch (err) {
        this.logger.warn(
          `Failed to apply sync entry ${entry.model}/${entry.recordId}: ${(err as Error).message}`,
        );
      }
    }
  }

  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 };

    const res = await fetch(`${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}`);
    }
    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 };

    await this.applyIncoming(entries);
    await this.prisma.syncCursor.update({
      where: { peerId },
      data: { lastPulledSequence: entries[entries.length - 1].sequence },
    });
    return { pulled: entries.length };
  }

  private async getOrCreateCursor(peerId: string) {
    return this.prisma.syncCursor.upsert({
      where: { peerId },
      create: { peerId },
      update: {},
    });
  }

  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;
    // 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<unknown>;
      delete: (args: { where: { id: string } }) => Promise<unknown>;
    };
  }
}