import type { Prisma } from "@prisma/client"; import { emitEvent } from "@/server/events"; import { writeAuditLog } from "@/server/audit"; import { ServiceError, type ServiceCtx } from "@/server/services/context"; import { requireVisibleWorkOrder } from "@/server/services/work-orders/visibility"; import { CONFLICTING_OPS, type SyncOperationInput, type SyncOpResult, type SyncOpType, type SyncResponse } from "@/lib/sync/envelope"; import { OP_PAYLOAD_SCHEMAS, type ParsedOpPayload } from "@/lib/sync/ops"; import { endSession, pauseSession, resumeSession, startSession } from "@/server/services/field/sessions"; import { createNote } from "@/server/services/field/notes"; import { toggleChecklistItem } from "@/server/services/field/checklist"; import { upsertMaterialUsage } from "@/server/services/field/materials"; import { attachPhoto } from "@/server/services/field/photos"; import { attachVoiceNote } from "@/server/services/field/voice"; // TODO(merge L2): replace with "@/server/services/work-orders/transition" import { transitionWorkOrder } from "@/server/services/work-orders/transition"; import { EXTERNAL_OP_OWNERS, EXTERNAL_OPS } from "./external-ops"; /** * Server side of the operation-based sync (ARCHITEKTUR §4.6). Online and offline clients use the * same path. Per op: * 1. idempotency: SyncOperation(tenantId, clientOpId) already stored → `duplicate` (stored result) * 2. payload validation (src/lib/sync/ops.ts) → `rejected invalid` * 3. conflict check for CONFLICTING_OPS: WorkOrder.version ≠ baseVersion → `conflict`, nothing * written, SyncOperation(status=conflict) for the backoffice list, event `sync.failed` * 4. dispatch to the domain services (the same ones the UI would use) * Deterministic outcomes are stored; transient failures (internal errors, ops whose lane is not * deployed yet) are NOT stored so the device can retry with the same clientOpId. */ export type SyncRequest = { deviceId: string; operations: SyncOperationInput[] }; type HandlerResult = { idMap?: Record; entityVersion?: number }; type Handler = (ctx: ServiceCtx, payload: unknown, op: SyncOperationInput) => Promise; const idMap = (clientId: string | undefined, serverId: string) => (clientId ? { [clientId]: serverId } : undefined); function h(fn: (ctx: ServiceCtx, payload: ParsedOpPayload, op: SyncOperationInput) => Promise): Handler { return (ctx, payload, op) => fn(ctx, payload as ParsedOpPayload, op); } const FIELD_HANDLERS: Partial> = { "session.start": h<"session.start">(async (ctx, p) => { const r = await startSession(ctx, p); return { idMap: idMap(p.clientId, r.sessionId), entityVersion: r.workOrderVersion }; }), "session.pause": h<"session.pause">(async (ctx, p) => ({ entityVersion: (await pauseSession(ctx, p)).workOrderVersion })), "session.resume": h<"session.resume">(async (ctx, p) => ({ entityVersion: (await resumeSession(ctx, p)).workOrderVersion })), "session.end": h<"session.end">(async (ctx, p) => ({ entityVersion: (await endSession(ctx, p)).workOrderVersion })), "work_order.transition": h<"work_order.transition">(async (ctx, p, op) => { const r = await transitionWorkOrder(ctx, { workOrderId: p.workOrderId, to: p.to, reason: p.reason, baseVersion: op.baseVersion }); return { entityVersion: r.version }; }), "note.create": h<"note.create">(async (ctx, p) => ({ idMap: idMap(p.clientId, (await createNote(ctx, p)).noteId) })), "checklist.toggle": h<"checklist.toggle">(async (ctx, p) => { await toggleChecklistItem(ctx, p); return {}; }), "material.upsert": h<"material.upsert">(async (ctx, p) => ({ idMap: idMap(p.clientId, (await upsertMaterialUsage(ctx, p)).usageId) })), "photo.attach": h<"photo.attach">(async (ctx, p) => ({ idMap: idMap(p.clientId, (await attachPhoto(ctx, p)).photoId) })), "voice.attach": h<"voice.attach">(async (ctx, p) => ({ idMap: idMap(p.clientId, (await attachVoiceNote(ctx, p)).voiceNoteId) })), }; class NotAvailable extends Error {} /** Route a validated op to the field handler or the registered module of another lane. */ async function dispatch(ctx: ServiceCtx, payload: unknown, op: SyncOperationInput): Promise { const handler = FIELD_HANDLERS[op.opType]; if (handler) return handler(ctx, payload, op); const load = EXTERNAL_OPS[op.opType]; const external = load ? await load() : null; if (!external) throw new NotAvailable(`operation ${op.opType} is not available yet (lane ${EXTERNAL_OP_OWNERS[op.opType] ?? "unknown"})`); return external(ctx, op); } /** * Backoffice „Übernehmen" of a stored conflict (services/work-orders/conflicts.ts, L10b): the op * is dispatched again against the CURRENT state — same payload validation and domain services as * the sync path, but without the baseVersion comparison. Only conflict-prone ops * (`work_order.transition`, `report.submit`) can end up as conflicts. */ export async function reapplyOperation( ctx: ServiceCtx, stored: { opType: string; entityType?: string | null; entityId: string | null; payload: unknown }, ): Promise { const opType = stored.opType as SyncOpType; if (!CONFLICTING_OPS.includes(opType)) throw new ServiceError("invalid", "reapply_unsupported", { opType: stored.opType }); const payload: Record = { ...((stored.payload ?? {}) as Record) }; // stored ops whose work order is only referenced via entityId (accepted by the former L2 stub) if (typeof payload.workOrderId !== "string" && stored.entityId && (!stored.entityType || stored.entityType === "work_order")) { payload.workOrderId = stored.entityId; } const parsed = OP_PAYLOAD_SCHEMAS[opType].safeParse(payload); if (!parsed.success) throw new ServiceError("invalid", "sync_payload_invalid"); const op: SyncOperationInput = { clientOpId: "00000000-0000-4000-8000-000000000000", // not used by handlers; idempotency stays with the stored op opType, entityType: stored.entityType ?? undefined, entityId: stored.entityId ?? undefined, baseVersion: undefined, payload, clientCreatedAt: new Date().toISOString(), }; try { return await dispatch(ctx, parsed.data, op); } catch (err) { if (err instanceof NotAvailable) throw new ServiceError("invalid", "reapply_unsupported", { opType }); throw err; } } function workOrderIdOf(op: SyncOperationInput): string | undefined { const fromPayload = (op.payload as { workOrderId?: unknown }).workOrderId; if (typeof fromPayload === "string") return fromPayload; return op.entityType === "work_order" ? op.entityId : undefined; } async function record( ctx: ServiceCtx, op: SyncOperationInput, deviceId: string, status: "applied" | "conflict" | "rejected", result: Record, errorCode?: string, ): Promise<{ id: string } | "duplicate"> { try { return await ctx.db.syncOperation.create({ data: { tenantId: ctx.tenantId, userId: ctx.userId, clientOpId: op.clientOpId, opType: op.opType, entityType: op.entityType ?? (workOrderIdOf(op) ? "work_order" : null), entityId: op.entityId ?? workOrderIdOf(op) ?? null, baseVersion: op.baseVersion ?? null, payload: op.payload as Prisma.InputJsonValue, status, result: { ...result, deviceId } as Prisma.InputJsonValue, errorCode: errorCode ?? null, clientCreatedAt: new Date(op.clientCreatedAt), }, select: { id: true }, }); } catch (err) { if ((err as { code?: string }).code === "P2002") return "duplicate"; throw err; } } async function applyOne(ctx: ServiceCtx, deviceId: string, op: SyncOperationInput): Promise { const base = { clientOpId: op.clientOpId }; // 1. idempotency const prior = await ctx.db.syncOperation.findFirst({ where: { clientOpId: op.clientOpId } }); if (prior) { if (prior.userId !== ctx.userId) return { ...base, status: "rejected", errorCode: "invalid", message: "clientOpId already used" }; const stored = (prior.result ?? {}) as HandlerResult & { message?: string }; return { ...base, status: "duplicate", idMap: stored.idMap, entityVersion: stored.entityVersion, errorCode: (prior.errorCode as SyncOpResult["errorCode"]) ?? undefined, message: prior.status === "applied" ? undefined : `original status: ${prior.status}`, }; } // 2. payload const parsed = OP_PAYLOAD_SCHEMAS[op.opType].safeParse(op.payload); if (!parsed.success) { const message = parsed.error.issues.map((i) => `${i.path.join(".")}: ${i.message}`).join("; ").slice(0, 500); const rec = await record(ctx, op, deviceId, "rejected", { message }, "invalid"); return rec === "duplicate" ? { ...base, status: "duplicate" } : { ...base, status: "rejected", errorCode: "invalid", message }; } try { // 3. conflict check if (CONFLICTING_OPS.includes(op.opType)) { const workOrderId = workOrderIdOf(op); if (!workOrderId || op.baseVersion === undefined) throw new ServiceError("invalid", "workOrderId and baseVersion are required"); const wo = await requireVisibleWorkOrder(ctx, workOrderId, { id: true, number: true, version: true }); if (wo.version !== op.baseVersion) { const rec = await record(ctx, op, deviceId, "conflict", { currentVersion: wo.version, message: "work order was changed in the meantime" }, "conflict"); if (rec === "duplicate") return { ...base, status: "duplicate" }; await writeAuditLog({ tenantId: ctx.tenantId, actorId: ctx.userId, action: "create", entity: "sync_operation", entityId: rec.id, after: { opType: op.opType, workOrderId, status: "conflict", baseVersion: op.baseVersion, currentVersion: wo.version }, }); await emitEvent(ctx, { type: "sync.failed", entityType: "sync_operation", entityId: rec.id, data: { opType: op.opType, number: wo.number, reason: "conflict" } }); return { ...base, status: "conflict", entityVersion: wo.version, errorCode: "conflict", message: "work order was changed in the meantime" }; } } // 4. dispatch const result = await dispatch(ctx, parsed.data, op); const rec = await record(ctx, op, deviceId, "applied", { ...result }); if (rec === "duplicate") return { ...base, status: "duplicate", ...result }; return { ...base, status: "applied", ...result }; } catch (err) { if (err instanceof NotAvailable) return { ...base, status: "rejected", errorCode: "invalid", message: err.message }; if (err instanceof ServiceError) { // a transition conflict detected inside the service (race after the pre-check) const status = err.code === "conflict" && CONFLICTING_OPS.includes(op.opType) ? "conflict" : "rejected"; const details = err.code === "blocked" ? { blockers: err.details } : {}; const rec = await record(ctx, op, deviceId, status, { message: err.message, ...details }, err.code); if (rec === "duplicate") return { ...base, status: "duplicate" }; if (status === "conflict") { await emitEvent(ctx, { type: "sync.failed", entityType: "sync_operation", entityId: rec.id, data: { opType: op.opType, reason: "conflict" } }); } return { ...base, status, errorCode: err.code, message: err.code === "blocked" ? JSON.stringify(err.details ?? []) : err.message }; } console.error(`[sync] ${op.opType} ${op.clientOpId} failed:`, err); return { ...base, status: "rejected", errorCode: "internal", message: "internal error" }; } } export async function applyOperations(ctx: ServiceCtx, request: SyncRequest): Promise { const results: SyncOpResult[] = []; // sequential on purpose: ops of one device depend on each other (start → pause → end) for (const op of request.operations) results.push(await applyOne(ctx, request.deviceId, op)); return { results, serverTime: new Date().toISOString() }; }