Basic new MVP with control tower fully functional

2025-12-31 01:17:13 +00:00
parent a369a69978
commit 363c9fbf4f
11 changed files with 284 additions and 577 deletions
--- a/app/api/ingest/cycle/route.ts
+++ b/app/api/ingest/cycle/route.ts
@@ -1,11 +1,37 @@
 import { NextResponse } from "next/server";
 import { prisma } from "@/lib/prisma";

+function unwrapEnvelope(raw: any) {
+  if (!raw || typeof raw !== "object") return raw;
+  const payload = raw.payload;
+  if (!payload || typeof payload !== "object") return raw;
+
+  const hasMeta =
+    raw.schemaVersion !== undefined ||
+    raw.machineId !== undefined ||
+    raw.tsMs !== undefined ||
+    raw.tsDevice !== undefined ||
+    raw.seq !== undefined ||
+    raw.type !== undefined;
+  if (!hasMeta) return raw;
+
+  return {
+    ...payload,
+    machineId: raw.machineId ?? payload.machineId,
+    tsMs: raw.tsMs ?? payload.tsMs,
+    tsDevice: raw.tsDevice ?? payload.tsDevice,
+    schemaVersion: raw.schemaVersion ?? payload.schemaVersion,
+    seq: raw.seq ?? payload.seq,
+  };
+}
+
 export async function POST(req: Request) {
  const apiKey = req.headers.get("x-api-key");
  if (!apiKey) return NextResponse.json({ ok: false, error: "Missing api key" }, { status: 401 });

-  const body = await req.json().catch(() => null);
+  let body = await req.json().catch(() => null);
+  body = unwrapEnvelope(body);
+
  if (!body?.machineId || !body?.cycle) {
    return NextResponse.json({ ok: false, error: "Invalid payload" }, { status: 400 });
  }
@@ -22,6 +48,8 @@ export async function POST(req: Request) {
    (typeof c.timestamp === "number" && c.timestamp) ||
    (typeof c.ts === "number" && c.ts) ||
    (typeof c.event_timestamp === "number" && c.event_timestamp) ||
+    (typeof body.tsMs === "number" && body.tsMs) ||
+    (typeof body.tsDevice === "number" && body.tsDevice) ||
    undefined;

  const ts = tsMs ? new Date(tsMs) : new Date();
--- a/app/api/ingest/event/route.ts
+++ b/app/api/ingest/event/route.ts
@@ -1,22 +1,5 @@
 import { NextResponse } from "next/server";
 import { prisma } from "@/lib/prisma";
-import { normalizeEventV1 } from "@/lib/contracts/v1";
-
-function getClientIp(req: Request) {
-  const xf = req.headers.get("x-forwarded-for");
-  if (xf) return xf.split(",")[0]?.trim() || null;
-  return req.headers.get("x-real-ip") || null;
-}
-
-function parseSeqToBigInt(seq: unknown): bigint | null {
-  if (seq === null || seq === undefined) return null;
-  if (typeof seq === "number") {
-    if (!Number.isInteger(seq) || seq < 0) return null;
-    return BigInt(seq);
-  }
-  if (typeof seq === "string" && /^\d+$/.test(seq)) return BigInt(seq);
-  return null;
-}

 const normalizeType = (t: any) =>
  String(t ?? "")
@@ -55,124 +38,73 @@ const MICROSTOP_SEC = 60;
 const MACROSTOP_SEC = 300;

 export async function POST(req: Request) {
-  const endpoint = "/api/ingest/event";
-  const ip = getClientIp(req);
-  const userAgent = req.headers.get("user-agent");
+  const apiKey = req.headers.get("x-api-key");
+  if (!apiKey) return NextResponse.json({ ok: false, error: "Missing api key" }, { status: 401 });

-  let rawBody: any = null;
-  let orgId: string | null = null;
-  let machineId: string | null = null;
-  let schemaVersion: string | null = null;
-  let seq: bigint | null = null;
-  let tsDeviceDate: Date | null = null;
+  let body: any = await req.json().catch(() => null);

-  try {
-    // 1) Auth header exists
-    const apiKey = req.headers.get("x-api-key");
-    if (!apiKey) {
-      await prisma.ingestLog.create({
-        data: {
-          endpoint,
-          ok: false,
-          status: 401,
-          errorCode: "MISSING_API_KEY",
-          errorMsg: "Missing api key",
-          ip,
-          userAgent,
-        },
-      });
-      return NextResponse.json({ ok: false, error: "Missing api key" }, { status: 401 });
+  // ✅ if Node-RED sent an array as the whole body, unwrap it
+  if (Array.isArray(body)) body = body[0];
+
+  // ✅ accept multiple common keys
+  const machineId = body?.machineId ?? body?.machine_id ?? body?.machine?.id;
+  let rawEvent =
+    body?.event ??
+    body?.events ??
+    body?.anomalies ??
+    body?.payload?.event ??
+    body?.payload?.events ??
+    body?.payload?.anomalies ??
+    body?.payload ??
+    body?.data;           // sometimes "data"
+
+    if (rawEvent?.event && typeof rawEvent.event === "object") rawEvent = rawEvent.event;
+    if (Array.isArray(rawEvent?.events)) rawEvent = rawEvent.events;
+
+  if (!machineId || !rawEvent) {
+    return NextResponse.json(
+      { ok: false, error: "Invalid payload", got: { hasMachineId: !!machineId, keys: Object.keys(body ?? {}) } },
+      { status: 400 }
+    );
+  }
+
+  const machine = await prisma.machine.findFirst({
+    where: { id: String(machineId), apiKey },
+    select: { id: true, orgId: true },
+  });
+  if (!machine) return NextResponse.json({ ok: false, error: "Unauthorized" }, { status: 401 });
+
+  // ✅ normalize to array no matter what
+  const events = Array.isArray(rawEvent) ? rawEvent : [rawEvent];
+
+  const created: { id: string; ts: Date; eventType: string }[] = [];
+  const skipped: any[] = [];
+
+  for (const ev of events) {
+    if (!ev || typeof ev !== "object") {
+      skipped.push({ reason: "invalid_event_object" });
+      continue;
    }

-    // 2) Parse JSON
-    rawBody = await req.json().catch(() => null);
-
-    // 3) Reject arrays at the contract boundary (Phase 0 rule)
-    // Edge MUST split arrays into one event per POST.
-    if (rawBody?.event && Array.isArray(rawBody.event)) {
-      await prisma.ingestLog.create({
-        data: {
-          endpoint,
-          ok: false,
-          status: 400,
-          errorCode: "EVENT_ARRAY_NOT_ALLOWED",
-          errorMsg: "Edge must split arrays; send one event per request.",
-          body: rawBody,
-          machineId: rawBody?.machineId ? String(rawBody.machineId) : null,
-          ip,
-          userAgent,
-        },
-      });
-      return NextResponse.json(
-        { ok: false, error: "Invalid payload", detail: "event array not allowed; split on edge" },
-        { status: 400 }
-      );
-    }
-
-    // 4) Normalize to v1 (legacy tolerated)
-    const normalized = normalizeEventV1(rawBody);
-    if (!normalized.ok) {
-      await prisma.ingestLog.create({
-        data: {
-          endpoint,
-          ok: false,
-          status: 400,
-          errorCode: "INVALID_PAYLOAD",
-          errorMsg: normalized.error,
-          body: rawBody,
-          ip,
-          userAgent,
-        },
-      });
-      return NextResponse.json({ ok: false, error: "Invalid payload", detail: normalized.error }, { status: 400 });
-    }
-
-    const body = normalized.value;
-
-    schemaVersion = body.schemaVersion;
-    machineId = body.machineId;
-    seq = parseSeqToBigInt(body.seq);
-    tsDeviceDate = new Date(body.tsDevice);
-
-    // 5) Authorize machineId + apiKey
-    const machine = await prisma.machine.findFirst({
-      where: { id: machineId, apiKey },
-      select: { id: true, orgId: true },
-    });
-
-    if (!machine) {
-      await prisma.ingestLog.create({
-        data: {
-          endpoint,
-          ok: false,
-          status: 401,
-          errorCode: "UNAUTHORIZED",
-          errorMsg: "Unauthorized (machineId/apiKey mismatch)",
-          body: rawBody,
-          machineId,
-          schemaVersion,
-          seq,
-          tsDevice: tsDeviceDate,
-          ip,
-          userAgent,
-        },
-      });
-      return NextResponse.json({ ok: false, error: "Unauthorized" }, { status: 401 });
-    }
-
-    orgId = machine.orgId;
-
-    // 6) Canonicalize + classify type (keep for now; later move to edge in A1)
-    const ev = body.event;
-
-    const rawType =
-      (ev as any).eventType ?? (ev as any).anomaly_type ?? (ev as any).topic ?? (body as any).topic ?? "";
+    const rawType = (ev as any).eventType ?? (ev as any).anomaly_type ?? (ev as any).topic ?? body.topic ?? "";
    const typ0 = normalizeType(rawType);
    const typ = CANON_TYPE[typ0] ?? typ0;

-    let finalType = typ;
+    // Determine timestamp
+    const tsMs =
+      (typeof (ev as any)?.timestamp === "number" && (ev as any).timestamp) ||
+      (typeof (ev as any)?.data?.timestamp === "number" && (ev as any).data.timestamp) ||
+      (typeof (ev as any)?.data?.event_timestamp === "number" && (ev as any).data.event_timestamp) ||
+      null;
+
+    const ts = tsMs ? new Date(tsMs) : new Date();
+
+    // Severity defaulting (do not skip on severity — store for audit)
+    let sev = String((ev as any).severity ?? "").trim().toLowerCase();
+    if (!sev) sev = "warning";

    // Stop classification -> microstop/macrostop
+    let finalType = typ;
    if (typ === "stop") {
      const stopSec =
        (typeof (ev as any)?.data?.stoppage_duration_seconds === "number" && (ev as any).data.stoppage_duration_seconds) ||
@@ -182,85 +114,36 @@ export async function POST(req: Request) {
      if (stopSec != null) {
        finalType = stopSec >= MACROSTOP_SEC ? "macrostop" : "microstop";
      } else {
+        // missing duration -> conservative
        finalType = "microstop";
      }
    }

    if (!ALLOWED_TYPES.has(finalType)) {
-      await prisma.ingestLog.create({
-        data: {
-          orgId,
-          machineId: machine.id,
-          endpoint,
-          ok: false,
-          status: 400,
-          errorCode: "TYPE_NOT_ALLOWED",
-          errorMsg: `Event type not allowed: ${finalType}`,
-          schemaVersion,
-          seq,
-          tsDevice: tsDeviceDate,
-          body: rawBody,
-          ip,
-          userAgent,
-        },
-      });
-      return NextResponse.json(
-        { ok: false, error: "Invalid event type", detail: finalType },
-        { status: 400 }
-      );
+      skipped.push({ reason: "type_not_allowed", typ: finalType, sev });
+      continue;
    }

-    // Determine severity
-    let sev = String((ev as any).severity ?? "").trim().toLowerCase();
-    if (!sev) sev = "warning";
-
    const title =
      String((ev as any).title ?? "").trim() ||
-      (finalType === "slow-cycle"
-        ? "Slow Cycle Detected"
-        : finalType === "macrostop"
-        ? "Macrostop Detected"
-        : finalType === "microstop"
-        ? "Microstop Detected"
-        : "Event");
+      (finalType === "slow-cycle" ? "Slow Cycle Detected" :
+       finalType === "macrostop" ? "Macrostop Detected" :
+       finalType === "microstop" ? "Microstop Detected" :
+       "Event");

    const description = (ev as any).description ? String((ev as any).description) : null;

-    // store full blob
+    // store full blob, ensure object
    const rawData = (ev as any).data ?? ev;
-    const dataObj =
-      typeof rawData === "string"
-        ? (() => {
-            try {
-              return JSON.parse(rawData);
-            } catch {
-              return { raw: rawData };
-            }
-          })()
-        : rawData;
+    const dataObj = typeof rawData === "string" ? (() => {
+      try { return JSON.parse(rawData); } catch { return { raw: rawData }; }
+    })() : rawData;

-    // Prefer work_order_id always
-    const workOrderId =
-      (ev as any)?.work_order_id ? String((ev as any).work_order_id)
-      : (ev as any)?.data?.work_order_id ? String((ev as any).data.work_order_id)
-      : null;
-
-    const sku =
-      (ev as any)?.sku ? String((ev as any).sku)
-      : (ev as any)?.data?.sku ? String((ev as any).data.sku)
-      : null;
-
-    // 7) Store event with Phase 0 meta
    const row = await prisma.machineEvent.create({
      data: {
-        orgId,
+        orgId: machine.orgId,
        machineId: machine.id,
-
-        // Phase 0 meta
-        schemaVersion,
-        seq,
-        ts: tsDeviceDate,
-
+        ts,
        topic: String((ev as any).topic ?? finalType),
        eventType: finalType,
        severity: sev,
@@ -268,69 +151,19 @@ export async function POST(req: Request) {
        title,
        description,
        data: dataObj,
-        workOrderId,
-        sku,
+        workOrderId:
+          (ev as any)?.work_order_id ? String((ev as any).work_order_id)
+          : (ev as any)?.data?.work_order_id ? String((ev as any).data.work_order_id)
+          : null,
+        sku:
+          (ev as any)?.sku ? String((ev as any).sku)
+          : (ev as any)?.data?.sku ? String((ev as any).data.sku)
+          : null,
      },
    });

-    // Optional: update machine last seen
-    await prisma.machine.update({
-      where: { id: machine.id },
-      data: {
-        schemaVersion,
-        seq,
-        tsDevice: tsDeviceDate,
-        tsServer: new Date(),
-      },
-    });
-
-    // 8) Ingest log success
-    await prisma.ingestLog.create({
-      data: {
-        orgId,
-        machineId: machine.id,
-        endpoint,
-        ok: true,
-        status: 200,
-        schemaVersion,
-        seq,
-        tsDevice: tsDeviceDate,
-        body: rawBody,
-        ip,
-        userAgent,
-      },
-    });
-
-    return NextResponse.json({
-      ok: true,
-      createdCount: 1,
-      created: [{ id: row.id, ts: row.ts, eventType: row.eventType }],
-      skippedCount: 0,
-      skipped: [],
-    });
-  } catch (err: any) {
-    const msg = err?.message ? String(err.message) : "Unknown error";
-
-    try {
-      await prisma.ingestLog.create({
-        data: {
-          orgId,
-          machineId,
-          endpoint,
-          ok: false,
-          status: 500,
-          errorCode: "SERVER_ERROR",
-          errorMsg: msg,
-          schemaVersion,
-          seq,
-          tsDevice: tsDeviceDate ?? undefined,
-          body: rawBody,
-          ip,
-          userAgent,
-        },
-      });
-    } catch {}
-
-    return NextResponse.json({ ok: false, error: "Server error", detail: msg }, { status: 500 });
+    created.push({ id: row.id, ts: row.ts, eventType: row.eventType });
  }
+
+  return NextResponse.json({ ok: true, createdCount: created.length, created, skippedCount: skipped.length, skipped });
 }
--- a/app/api/ingest/kpi/route.ts
+++ b/app/api/ingest/kpi/route.ts
@@ -103,6 +103,8 @@ export async function POST(req: Request) {
    orgId = machine.orgId;

    const wo = body.activeWorkOrder ?? {};
+    const good = typeof wo.good === "number" ? wo.good : (typeof wo.goodParts === "number" ? wo.goodParts : null);
+    const scrap = typeof wo.scrap === "number" ? wo.scrap : (typeof wo.scrapParts === "number" ? wo.scrapParts : null)
    const k = body.kpis ?? {};
    const safeCycleTime =
    typeof body.cycleTime === "number" && body.cycleTime > 0
@@ -128,8 +130,8 @@ export async function POST(req: Request) {
        workOrderId: wo.id ? String(wo.id) : null,
        sku: wo.sku ? String(wo.sku) : null,
        target: typeof wo.target === "number" ? Math.trunc(wo.target) : null,
-        good: typeof wo.good === "number" ? Math.trunc(wo.good) : null,
-        scrap: typeof wo.scrap === "number" ? Math.trunc(wo.scrap) : null,
+        good: good != null ? Math.trunc(good) : null,
+        scrap: scrap != null ? Math.trunc(scrap) : null,

        // Counters
        cycleCount: typeof body.cycle_count === "number" ? body.cycle_count : null,
--- a/app/api/machines/[machineId]/route.ts
+++ b/app/api/machines/[machineId]/route.ts
@@ -238,11 +238,7 @@ const ALLOWED_TYPES = new Set([

 const events = normalized
  .filter((e) => ALLOWED_TYPES.has(e.eventType))
-  // keep slow-cycle even if severity is info, otherwise require warning/critical/error
-  .filter((e) =>
-    ["slow-cycle", "microstop", "macrostop"].includes(e.eventType) ||
-    ["warning", "critical", "error"].includes(e.severity)
-  )
+  // drop severity gating so recent info events appear
  .slice(0, 30);


--- a/lib/contracts/v1.ts
+++ b/lib/contracts/v1.ts
@@ -10,6 +10,53 @@ export const SCHEMA_VERSION = "1.0";
 // KPI scale is frozen as 0..100 (you confirmed)
 const KPI_0_100 = z.number().min(0).max(100);

+function unwrapCanonicalEnvelope(raw: unknown) {
+  if (!raw || typeof raw !== "object") return raw;
+  const obj: any = raw;
+  const payload = obj.payload;
+  if (!payload || typeof payload !== "object") return raw;
+
+  const hasMeta =
+    obj.schemaVersion !== undefined ||
+    obj.machineId !== undefined ||
+    obj.tsMs !== undefined ||
+    obj.tsDevice !== undefined ||
+    obj.seq !== undefined ||
+    obj.type !== undefined;
+  if (!hasMeta) return raw;
+
+  const tsDevice =
+    typeof obj.tsDevice === "number"
+      ? obj.tsDevice
+      : typeof obj.tsMs === "number"
+        ? obj.tsMs
+        : typeof payload.tsDevice === "number"
+          ? payload.tsDevice
+          : typeof payload.tsMs === "number"
+            ? payload.tsMs
+            : undefined;
+
+  return {
+    ...payload,
+    schemaVersion: obj.schemaVersion ?? payload.schemaVersion,
+    machineId: obj.machineId ?? payload.machineId,
+    tsDevice: tsDevice ?? payload.tsDevice,
+    seq: obj.seq ?? payload.seq,
+  };
+}
+
+function normalizeTsDevice(raw: unknown) {
+  if (!raw || typeof raw !== "object") return raw;
+  const obj: any = raw;
+  if (typeof obj.tsDevice === "number") return obj;
+  if (typeof obj.tsMs === "number") return { ...obj, tsDevice: obj.tsMs };
+  return obj;
+}
+
+function preprocessPayload(raw: unknown) {
+  return normalizeTsDevice(unwrapCanonicalEnvelope(raw));
+}
+
 export const SnapshotV1 = z
  .object({
    schemaVersion: z.literal(SCHEMA_VERSION),
@@ -20,15 +67,26 @@ export const SnapshotV1 = z

    // current shape (keep it flat so Node-RED changes are minimal)
    activeWorkOrder: z
-      .object({
-        id: z.string(),
-        sku: z.string().optional(),
-        target: z.number().optional(),
-        good: z.number().optional(),
-        scrap: z.number().optional(),
-      })
-      .partial()
-      .optional(),
+    .object({
+      id: z.string(),
+      sku: z.string().optional(),
+      target: z.number().optional(),
+      good: z.number().optional(),
+      scrap: z.number().optional(),
+
+      // add the ones you actually rely on
+      cycleTime: z.number().optional(),
+      cavities: z.number().optional(),
+      progressPercent: z.number().optional(),
+      status: z.string().optional(),
+      lastUpdateIso: z.string().optional(),
+      cycle_count: z.number().optional(),
+      good_parts: z.number().optional(),
+      scrap_parts: z.number().optional(),
+    })
+    .partial()
+    .passthrough()
+    .optional(),

    cycle_count: z.number().int().nonnegative().optional(),
    good_parts: z.number().int().nonnegative().optional(),
@@ -64,15 +122,16 @@ const SnapshotLegacy = z
 export type SnapshotV1Type = z.infer<typeof SnapshotV1>;

 export function normalizeSnapshotV1(raw: unknown): { ok: true; value: SnapshotV1Type } | { ok: false; error: string } {
-  const strict = SnapshotV1.safeParse(raw);
+  const candidate = preprocessPayload(raw);
+  const strict = SnapshotV1.safeParse(candidate);
  if (strict.success) return { ok: true, value: strict.data };

  // Legacy fallback (temporary)
-  const legacy = SnapshotLegacy.safeParse(raw);
+  const legacy = SnapshotLegacy.safeParse(candidate);
  if (!legacy.success) {
    return { ok: false, error: strict.error.message };
  }
-
+/*
  const b: any = legacy.data;

  // Build a "best effort" SnapshotV1 so ingest works during transition.
@@ -89,6 +148,93 @@ export function normalizeSnapshotV1(raw: unknown): { ok: true; value: SnapshotV1
  const recheck = SnapshotV1.safeParse(migrated);
  if (!recheck.success) return { ok: false, error: recheck.error.message };
  return { ok: true, value: recheck.data };
+  */
+    const b: any = legacy.data;
+
+    const legacyCycleTime =
+      b.cycleTime ??
+      b.cycle_time ??
+      b.theoretical_cycle_time ??
+      b.theoreticalCycleTime ??
+      b.standard_cycle_time ??
+      b.kpi_snapshot?.cycleTime ??
+      b.kpi_snapshot?.cycle_time ??
+      undefined;
+
+    const legacyActualCycleTime =
+      b.actualCycleTime ??
+      b.actual_cycle_time ??
+      b.actualCycleSeconds ??
+      b.kpi_snapshot?.actualCycleTime ??
+      b.kpi_snapshot?.actual_cycle_time ??
+      undefined;
+
+    const legacyWorkOrderId =
+      b.activeWorkOrder?.id ??
+      b.work_order_id ??
+      b.workOrderId ??
+      b.kpis?.workOrderId ??
+      b.kpi_snapshot?.work_order_id ??
+      undefined;
+
+    const legacySku =
+      b.activeWorkOrder?.sku ??
+      b.sku ??
+      b.kpis?.sku ??
+      b.kpi_snapshot?.sku ??
+      undefined;
+
+    const legacyTarget =
+      b.activeWorkOrder?.target ??
+      b.target ??
+      b.kpis?.target ??
+      b.kpi_snapshot?.target ??
+      undefined;
+
+    const legacyGood =
+      b.activeWorkOrder?.good ??
+      b.good_parts ??
+      b.good ??
+      b.kpis?.good ??
+      b.kpi_snapshot?.good_parts ??
+      undefined;
+
+    const legacyScrap =
+      b.activeWorkOrder?.scrap ??
+      b.scrap_parts ??
+      b.scrap ??
+      b.kpis?.scrap ??
+      b.kpi_snapshot?.scrap_parts ??
+      undefined;
+
+    const migrated: any = {
+      schemaVersion: SCHEMA_VERSION,
+      machineId: String(b.machineId),
+      tsDevice: typeof b.tsDevice === "number" ? b.tsDevice : Date.now(),
+      seq: typeof b.seq === "number" || typeof b.seq === "string" ? b.seq : "0",
+
+      // canonical fields (force them)
+      cycleTime: legacyCycleTime != null ? Number(legacyCycleTime) : undefined,
+      actualCycleTime: legacyActualCycleTime != null ? Number(legacyActualCycleTime) : undefined,
+
+      activeWorkOrder: legacyWorkOrderId
+        ? {
+            id: String(legacyWorkOrderId),
+            sku: legacySku != null ? String(legacySku) : undefined,
+            target: legacyTarget != null ? Number(legacyTarget) : undefined,
+            good: legacyGood != null ? Number(legacyGood) : undefined,
+            scrap: legacyScrap != null ? Number(legacyScrap) : undefined,
+          }
+        : b.activeWorkOrder,
+
+      // keep everything else
+      ...b,
+      
+  };
+  const recheck = SnapshotV1.safeParse(migrated);
+  if (!recheck.success) return { ok: false, error: recheck.error.message };
+  return { ok: true, value: recheck.data };
+
 }

 const HeartbeatV1 = z.object({
@@ -108,11 +254,12 @@ const HeartbeatV1 = z.object({
 }).passthrough();

 export function normalizeHeartbeatV1(raw: unknown) {
-  const strict = HeartbeatV1.safeParse(raw);
+  const candidate = preprocessPayload(raw);
+  const strict = HeartbeatV1.safeParse(candidate);
  if (strict.success) return { ok: true as const, value: strict.data };

  // legacy fallback: allow missing meta
-  const legacy = z.object({ machineId: z.any() }).passthrough().safeParse(raw);
+  const legacy = z.object({ machineId: z.any() }).passthrough().safeParse(candidate);
  if (!legacy.success) return { ok: false as const, error: strict.error.message };

  const b: any = legacy.data;
@@ -149,11 +296,12 @@ const CycleV1 = z.object({
 }).passthrough();

 export function normalizeCycleV1(raw: unknown) {
-  const strict = CycleV1.safeParse(raw);
+  const candidate = preprocessPayload(raw);
+  const strict = CycleV1.safeParse(candidate);
  if (strict.success) return { ok: true as const, value: strict.data };

  // legacy fallback: { machineId, cycle }
-  const legacy = z.object({ machineId: z.any(), cycle: z.any() }).passthrough().safeParse(raw);
+  const legacy = z.object({ machineId: z.any(), cycle: z.any() }).passthrough().safeParse(candidate);
  if (!legacy.success) return { ok: false as const, error: strict.error.message };

  const b: any = legacy.data;
@@ -187,11 +335,12 @@ const EventV1 = z.object({
 }).passthrough();

 export function normalizeEventV1(raw: unknown) {
-  const strict = EventV1.safeParse(raw);
+  const candidate = preprocessPayload(raw);
+  const strict = EventV1.safeParse(candidate);
  if (strict.success) return { ok: true as const, value: strict.data };

  // legacy fallback: allow missing meta, but STILL reject arrays later
-  const legacy = z.object({ machineId: z.any(), event: z.any() }).passthrough().safeParse(raw);
+  const legacy = z.object({ machineId: z.any(), event: z.any() }).passthrough().safeParse(candidate);
  if (!legacy.success) return { ok: false as const, error: strict.error.message };

  const b: any = legacy.data;
--- a/prisma/migrations/20251216173800_init_auth/migration.sql
+++ b/prisma/migrations/20251216173800_init_auth/migration.sql
@@ -1,103 +0,0 @@
-- CreateTable
-CREATE TABLE "Org" (
-    "id" TEXT NOT NULL,
-    "name" TEXT NOT NULL,
-    "slug" TEXT NOT NULL,
-    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-
-    CONSTRAINT "Org_pkey" PRIMARY KEY ("id")
-);
-
-- CreateTable
-CREATE TABLE "User" (
-    "id" TEXT NOT NULL,
-    "email" TEXT NOT NULL,
-    "name" TEXT,
-    "passwordHash" TEXT NOT NULL,
-    "isActive" BOOLEAN NOT NULL DEFAULT true,
-    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-
-    CONSTRAINT "User_pkey" PRIMARY KEY ("id")
-);
-
-- CreateTable
-CREATE TABLE "OrgUser" (
-    "id" TEXT NOT NULL,
-    "orgId" TEXT NOT NULL,
-    "userId" TEXT NOT NULL,
-    "role" TEXT NOT NULL DEFAULT 'MEMBER',
-    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-
-    CONSTRAINT "OrgUser_pkey" PRIMARY KEY ("id")
-);
-
-- CreateTable
-CREATE TABLE "Session" (
-    "id" TEXT NOT NULL,
-    "orgId" TEXT NOT NULL,
-    "userId" TEXT NOT NULL,
-    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "lastSeenAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "expiresAt" TIMESTAMP(3) NOT NULL,
-    "revokedAt" TIMESTAMP(3),
-    "ip" TEXT,
-    "userAgent" TEXT,
-
-    CONSTRAINT "Session_pkey" PRIMARY KEY ("id")
-);
-
-- CreateTable
-CREATE TABLE "Machine" (
-    "id" TEXT NOT NULL,
-    "orgId" TEXT NOT NULL,
-    "name" TEXT NOT NULL,
-    "code" TEXT,
-    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-
-    CONSTRAINT "Machine_pkey" PRIMARY KEY ("id")
-);
-
-- CreateIndex
-CREATE UNIQUE INDEX "Org_slug_key" ON "Org"("slug");
-
-- CreateIndex
-CREATE UNIQUE INDEX "User_email_key" ON "User"("email");
-
-- CreateIndex
-CREATE INDEX "OrgUser_userId_idx" ON "OrgUser"("userId");
-
-- CreateIndex
-CREATE INDEX "OrgUser_orgId_idx" ON "OrgUser"("orgId");
-
-- CreateIndex
-CREATE UNIQUE INDEX "OrgUser_orgId_userId_key" ON "OrgUser"("orgId", "userId");
-
-- CreateIndex
-CREATE INDEX "Session_userId_idx" ON "Session"("userId");
-
-- CreateIndex
-CREATE INDEX "Session_orgId_idx" ON "Session"("orgId");
-
-- CreateIndex
-CREATE INDEX "Session_expiresAt_idx" ON "Session"("expiresAt");
-
-- CreateIndex
-CREATE INDEX "Machine_orgId_idx" ON "Machine"("orgId");
-
-- CreateIndex
-CREATE UNIQUE INDEX "Machine_orgId_code_key" ON "Machine"("orgId", "code");
-
-- AddForeignKey
-ALTER TABLE "OrgUser" ADD CONSTRAINT "OrgUser_orgId_fkey" FOREIGN KEY ("orgId") REFERENCES "Org"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-
-- AddForeignKey
-ALTER TABLE "OrgUser" ADD CONSTRAINT "OrgUser_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-
-- AddForeignKey
-ALTER TABLE "Session" ADD CONSTRAINT "Session_orgId_fkey" FOREIGN KEY ("orgId") REFERENCES "Org"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-
-- AddForeignKey
-ALTER TABLE "Session" ADD CONSTRAINT "Session_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-
-- AddForeignKey
-ALTER TABLE "Machine" ADD CONSTRAINT "Machine_orgId_fkey" FOREIGN KEY ("orgId") REFERENCES "Org"("id") ON DELETE CASCADE ON UPDATE CASCADE;
--- a/prisma/migrations/20251217000852_machines_and_heartbeats/migration.sql
+++ b/prisma/migrations/20251217000852_machines_and_heartbeats/migration.sql
@@ -1,39 +0,0 @@
-/*
-  Warnings:
-
-  - A unique constraint covering the columns `[orgId,name]` on the table `Machine` will be added. If there are existing duplicate values, this will fail.
-  - Added the required column `updatedAt` to the `Machine` table without a default value. This is not possible if the table is not empty.
-
-*/
-- DropIndex
-DROP INDEX "Machine_orgId_code_key";
-
-- AlterTable
-ALTER TABLE "Machine" ADD COLUMN     "location" TEXT,
-ADD COLUMN     "updatedAt" TIMESTAMP(3) NOT NULL;
-
-- CreateTable
-CREATE TABLE "MachineHeartbeat" (
-    "id" TEXT NOT NULL,
-    "orgId" TEXT NOT NULL,
-    "machineId" TEXT NOT NULL,
-    "ts" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "status" TEXT NOT NULL,
-    "message" TEXT,
-    "ip" TEXT,
-    "fwVersion" TEXT,
-
-    CONSTRAINT "MachineHeartbeat_pkey" PRIMARY KEY ("id")
-);
-
-- CreateIndex
-CREATE INDEX "MachineHeartbeat_orgId_machineId_ts_idx" ON "MachineHeartbeat"("orgId", "machineId", "ts");
-
-- CreateIndex
-CREATE UNIQUE INDEX "Machine_orgId_name_key" ON "Machine"("orgId", "name");
-
-- AddForeignKey
-ALTER TABLE "MachineHeartbeat" ADD CONSTRAINT "MachineHeartbeat_orgId_fkey" FOREIGN KEY ("orgId") REFERENCES "Org"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-
-- AddForeignKey
-ALTER TABLE "MachineHeartbeat" ADD CONSTRAINT "MachineHeartbeat_machineId_fkey" FOREIGN KEY ("machineId") REFERENCES "Machine"("id") ON DELETE CASCADE ON UPDATE CASCADE;
--- a/prisma/migrations/20251217001526_machine_api_key/migration.sql
+++ b/prisma/migrations/20251217001526_machine_api_key/migration.sql
@@ -1,11 +0,0 @@
-/*
-  Warnings:
-
-  - A unique constraint covering the columns `[apiKey]` on the table `Machine` will be added. If there are existing duplicate values, this will fail.
-
-*/
-- AlterTable
-ALTER TABLE "Machine" ADD COLUMN     "apiKey" TEXT;
-
-- CreateIndex
-CREATE UNIQUE INDEX "Machine_apiKey_key" ON "Machine"("apiKey");
--- a/prisma/migrations/20251217012912_kpi_snapshots_and_events/migration.sql
+++ b/prisma/migrations/20251217012912_kpi_snapshots_and_events/migration.sql
@@ -1,66 +0,0 @@
-- CreateTable
-CREATE TABLE "MachineKpiSnapshot" (
-    "id" TEXT NOT NULL,
-    "orgId" TEXT NOT NULL,
-    "machineId" TEXT NOT NULL,
-    "ts" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "workOrderId" TEXT,
-    "sku" TEXT,
-    "target" INTEGER,
-    "good" INTEGER,
-    "scrap" INTEGER,
-    "cycleCount" INTEGER,
-    "goodParts" INTEGER,
-    "scrapParts" INTEGER,
-    "cavities" INTEGER,
-    "cycleTime" DOUBLE PRECISION,
-    "actualCycle" DOUBLE PRECISION,
-    "availability" DOUBLE PRECISION,
-    "performance" DOUBLE PRECISION,
-    "quality" DOUBLE PRECISION,
-    "oee" DOUBLE PRECISION,
-    "trackingEnabled" BOOLEAN,
-    "productionStarted" BOOLEAN,
-
-    CONSTRAINT "MachineKpiSnapshot_pkey" PRIMARY KEY ("id")
-);
-
-- CreateTable
-CREATE TABLE "MachineEvent" (
-    "id" TEXT NOT NULL,
-    "orgId" TEXT NOT NULL,
-    "machineId" TEXT NOT NULL,
-    "ts" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "topic" TEXT NOT NULL,
-    "eventType" TEXT NOT NULL,
-    "severity" TEXT NOT NULL,
-    "requiresAck" BOOLEAN NOT NULL DEFAULT false,
-    "title" TEXT NOT NULL,
-    "description" TEXT,
-    "data" JSONB,
-    "workOrderId" TEXT,
-    "sku" TEXT,
-
-    CONSTRAINT "MachineEvent_pkey" PRIMARY KEY ("id")
-);
-
-- CreateIndex
-CREATE INDEX "MachineKpiSnapshot_orgId_machineId_ts_idx" ON "MachineKpiSnapshot"("orgId", "machineId", "ts");
-
-- CreateIndex
-CREATE INDEX "MachineEvent_orgId_machineId_ts_idx" ON "MachineEvent"("orgId", "machineId", "ts");
-
-- CreateIndex
-CREATE INDEX "MachineEvent_orgId_machineId_eventType_ts_idx" ON "MachineEvent"("orgId", "machineId", "eventType", "ts");
-
-- AddForeignKey
-ALTER TABLE "MachineKpiSnapshot" ADD CONSTRAINT "MachineKpiSnapshot_orgId_fkey" FOREIGN KEY ("orgId") REFERENCES "Org"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-
-- AddForeignKey
-ALTER TABLE "MachineKpiSnapshot" ADD CONSTRAINT "MachineKpiSnapshot_machineId_fkey" FOREIGN KEY ("machineId") REFERENCES "Machine"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-
-- AddForeignKey
-ALTER TABLE "MachineEvent" ADD CONSTRAINT "MachineEvent_orgId_fkey" FOREIGN KEY ("orgId") REFERENCES "Org"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-
-- AddForeignKey
-ALTER TABLE "MachineEvent" ADD CONSTRAINT "MachineEvent_machineId_fkey" FOREIGN KEY ("machineId") REFERENCES "Machine"("id") ON DELETE CASCADE ON UPDATE CASCADE;
--- a/prisma/migrations/20251218153109_add_machine_cycles/migration.sql
+++ b/prisma/migrations/20251218153109_add_machine_cycles/migration.sql
@@ -1,27 +0,0 @@
-- CreateTable
-CREATE TABLE "MachineCycle" (
-    "id" TEXT NOT NULL,
-    "orgId" TEXT NOT NULL,
-    "machineId" TEXT NOT NULL,
-    "ts" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "cycleCount" INTEGER,
-    "actualCycleTime" DOUBLE PRECISION NOT NULL,
-    "theoreticalCycleTime" DOUBLE PRECISION,
-    "workOrderId" TEXT,
-    "sku" TEXT,
-    "cavities" INTEGER,
-    "goodDelta" INTEGER,
-    "scrapDelta" INTEGER,
-    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-
-    CONSTRAINT "MachineCycle_pkey" PRIMARY KEY ("id")
-);
-
-- CreateIndex
-CREATE INDEX "MachineCycle_orgId_machineId_ts_idx" ON "MachineCycle"("orgId", "machineId", "ts");
-
-- CreateIndex
-CREATE INDEX "MachineCycle_orgId_machineId_cycleCount_idx" ON "MachineCycle"("orgId", "machineId", "cycleCount");
-
-- AddForeignKey
-ALTER TABLE "MachineCycle" ADD CONSTRAINT "MachineCycle_machineId_fkey" FOREIGN KEY ("machineId") REFERENCES "Machine"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
--- a/prisma/migrations/20251222235834_ingest_log/migration.sql
+++ b/prisma/migrations/20251222235834_ingest_log/migration.sql
@@ -1,55 +0,0 @@
-- AlterTable
-ALTER TABLE "Machine" ADD COLUMN     "schema_version" TEXT,
-ADD COLUMN     "seq" BIGINT,
-ADD COLUMN     "ts" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-ADD COLUMN     "ts_server" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;
-
-- AlterTable
-ALTER TABLE "MachineCycle" ADD COLUMN     "schema_version" TEXT,
-ADD COLUMN     "seq" BIGINT,
-ADD COLUMN     "ts_server" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;
-
-- AlterTable
-ALTER TABLE "MachineEvent" ADD COLUMN     "schema_version" TEXT,
-ADD COLUMN     "seq" BIGINT,
-ADD COLUMN     "ts_server" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;
-
-- AlterTable
-ALTER TABLE "MachineHeartbeat" ADD COLUMN     "schema_version" TEXT,
-ADD COLUMN     "seq" BIGINT,
-ADD COLUMN     "ts_server" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;
-
-- AlterTable
-ALTER TABLE "MachineKpiSnapshot" ADD COLUMN     "schema_version" TEXT,
-ADD COLUMN     "seq" BIGINT,
-ADD COLUMN     "ts_server" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;
-
-- CreateTable
-CREATE TABLE "IngestLog" (
-    "id" TEXT NOT NULL,
-    "orgId" TEXT,
-    "machineId" TEXT,
-    "endpoint" TEXT NOT NULL,
-    "schemaVersion" TEXT,
-    "seq" BIGINT,
-    "tsDevice" TIMESTAMP(3),
-    "tsServer" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    "ok" BOOLEAN NOT NULL,
-    "status" INTEGER NOT NULL,
-    "errorCode" TEXT,
-    "errorMsg" TEXT,
-    "body" JSONB,
-    "ip" TEXT,
-    "userAgent" TEXT,
-
-    CONSTRAINT "IngestLog_pkey" PRIMARY KEY ("id")
-);
-
-- CreateIndex
-CREATE INDEX "IngestLog_endpoint_tsServer_idx" ON "IngestLog"("endpoint", "tsServer");
-
-- CreateIndex
-CREATE INDEX "IngestLog_machineId_tsServer_idx" ON "IngestLog"("machineId", "tsServer");
-
-- CreateIndex
-CREATE INDEX "IngestLog_machineId_seq_idx" ON "IngestLog"("machineId", "seq");