fix(worker): pass socket to enqueueDalpuriFullSync

The socket retrieved from ensureManagerSocketReady() was never passed to
enqueueDalpuriFullSync(), so inside createWorkerJob the socket.emit('requestId')
call crashed with 'TypeError: undefined is not an object (evaluating A.emit)'.

This caused every full sync job to fail immediately, leaving the DB empty.
The 5s incremental sync interval then flooded the queue with 4700+ jobs that
all failed too since there was no data.

Also manually cleared the backlog of 4720 failed/pending incremental jobs and
2 failed full sync jobs from the production queue.

api/prisma/migrate-entrypoint.sh

@@ -26,8 +26,11 @@ while [ $ATTEMPT -lt $MAX_RETRIES ]; do
   # P3009: a previously-failed migration is blocking deploy.
   # The error message contains the migration name in backticks:
   #   The `20260402000000_fix_severity_typo` migration started at ... failed
-  if echo "$DEPLOY_OUTPUT" | grep -q "P3009"; then
+  # Strip ANSI escape codes first (Prisma may colorize output even without TTY),
-    FAILED=$(echo "$DEPLOY_OUTPUT" | grep -oE '\`[0-9]{14}(_[a-zA-Z_]+)?\`' | tr -d '\`' | head -1)
+  # then use a simple backtick-content regex rather than a rigid format match.
+  CLEAN_OUTPUT=$(printf '%s\n' "$DEPLOY_OUTPUT" | sed 's/\x1b\[[0-9;]*[mGKHFJr]//g')
+  if printf '%s\n' "$CLEAN_OUTPUT" | grep -q "P3009"; then
+    FAILED=$(printf '%s\n' "$CLEAN_OUTPUT" | grep -o '`[^`]*`' | grep '[0-9]' | tr -d '`' | head -1)
     if [ -n "$FAILED" ]; then
       echo "[migrate] Resolving failed migration as rolled-back: $FAILED"
       RESOLVE_OUTPUT=""

api/prisma/migrations/20260408100000_sync_schema_from_db_push/migration.sql

@@ -210,6 +210,8 @@ CREATE UNIQUE INDEX IF NOT EXISTS "CatalogItem_id_key" ON "CatalogItem"("id");
 
 -- =============================================================================
 -- SECTION 4: Company — change id TEXT→INTEGER, add uid PK, add columns
+-- Production has ~4500 rows with CUID text PKs and cw_CompanyId integers
+-- that must be preserved as uid and id respectively.
 -- =============================================================================
 
 -- Drop FKs that reference Company by old id
@@ -229,12 +231,18 @@ DROP INDEX IF EXISTS "Company_cw_CompanyId_key";
 DROP INDEX IF EXISTS "Company_cw_Identifier_key";
 
 DO $$ BEGIN
-  -- Add uid PK column if missing
+  -- Step 1: Add uid as NULLABLE (no default) so existing rows stay NULL temporarily
   IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'Company' AND column_name = 'uid') THEN
-    ALTER TABLE "Company" ADD COLUMN "uid" TEXT NOT NULL DEFAULT '';
+    ALTER TABLE "Company" ADD COLUMN "uid" TEXT;
   END IF;
 
-  -- Swap PK from id to uid
+  -- Step 2: Populate uid from the old text PK (old id was a CUID — it becomes uid)
+  UPDATE "Company" SET "uid" = "id" WHERE "uid" IS NULL;
+
+  -- Step 3: Now make uid NOT NULL (all rows are populated)
+  ALTER TABLE "Company" ALTER COLUMN "uid" SET NOT NULL;
+
+  -- Step 4: Swap PK from id (text) to uid (text)
   IF EXISTS (
     SELECT 1 FROM information_schema.table_constraints tc
     JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name
@@ -244,7 +252,8 @@ DO $$ BEGIN
     ALTER TABLE "Company" ADD CONSTRAINT "Company_pkey" PRIMARY KEY ("uid");
   END IF;
 
-  -- Change id from TEXT to INTEGER
+  -- Step 5: Change id from TEXT to INTEGER
+  -- NOTE: do this BEFORE dropping cw_CompanyId so we can populate from it below
   IF EXISTS (
     SELECT 1 FROM information_schema.columns
     WHERE table_name = 'Company' AND column_name = 'id' AND data_type = 'text'
@@ -253,7 +262,12 @@ DO $$ BEGIN
     ALTER TABLE "Company" ADD COLUMN "id" INTEGER;
   END IF;
 
-  -- Drop old CW-specific columns
+  -- Step 6: Populate new integer id from cw_CompanyId (CW integer company id)
+  IF EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'Company' AND column_name = 'cw_CompanyId') THEN
+    UPDATE "Company" SET "id" = "cw_CompanyId" WHERE "id" IS NULL;
+  END IF;
+
+  -- Step 7: Drop old CW-specific columns (data now in id and uid)
   IF EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'Company' AND column_name = 'cw_CompanyId') THEN
     ALTER TABLE "Company" DROP COLUMN "cw_CompanyId";
   END IF;
@@ -261,7 +275,7 @@ DO $$ BEGIN
     ALTER TABLE "Company" DROP COLUMN "cw_Identifier";
   END IF;
 
-  -- Add new columns
+  -- Step 8: Add new columns
   IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'Company' AND column_name = 'dateDeleted') THEN
     ALTER TABLE "Company" ADD COLUMN "dateDeleted" TIMESTAMP(3);
   END IF;
@@ -291,10 +305,15 @@ DO $$ BEGIN
   END IF;
 END $$;
 
+-- Make Company.id NOT NULL (all rows were populated from cw_CompanyId above)
+ALTER TABLE "Company" ALTER COLUMN "id" SET NOT NULL;
 CREATE UNIQUE INDEX IF NOT EXISTS "Company_id_key" ON "Company"("id");
 
 -- =============================================================================
--- SECTION 5: UnifiSite — change companyId from TEXT to INTEGER
+-- SECTION 5: UnifiSite — change companyId from TEXT to INTEGER (data migration)
+-- Production has ~180 rows where companyId (text) = Company.uid (the old text
+-- PK that was copied into uid in Section 4). We join on Company.uid to get
+-- the new integer Company.id and preserve the relationship.
 -- =============================================================================
 
 DO $$ BEGIN
@@ -302,8 +321,24 @@ DO $$ BEGIN
     SELECT 1 FROM information_schema.columns
     WHERE table_name = 'UnifiSite' AND column_name = 'companyId' AND data_type = 'text'
   ) THEN
+    -- Add a temporary integer column to hold the mapped value
+    IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'UnifiSite' AND column_name = 'companyId_int') THEN
+      ALTER TABLE "UnifiSite" ADD COLUMN "companyId_int" INTEGER;
+    END IF;
+    -- Map old text companyId (= Company.uid) → new integer Company.id
+    UPDATE "UnifiSite" us
+    SET "companyId_int" = c."id"
+    FROM "Company" c
+    WHERE c."uid" = us."companyId";
+    -- Replace old text column with the populated integer column
     ALTER TABLE "UnifiSite" DROP COLUMN "companyId";
-    ALTER TABLE "UnifiSite" ADD COLUMN "companyId" INTEGER;
+    ALTER TABLE "UnifiSite" RENAME COLUMN "companyId_int" TO "companyId";
+  ELSIF EXISTS (
+    SELECT 1 FROM information_schema.columns
+    WHERE table_name = 'UnifiSite' AND column_name = 'companyId_int'
+  ) THEN
+    -- Edge case: int column added but not renamed (interrupted previous run)
+    ALTER TABLE "UnifiSite" RENAME COLUMN "companyId_int" TO "companyId";
   END IF;
 END $$;
 
@@ -1467,7 +1502,8 @@ ALTER TABLE "CatalogItem" ALTER COLUMN "uid" DROP DEFAULT;
 ALTER TABLE "CatalogItem" ALTER COLUMN "id" SET NOT NULL;
 ALTER TABLE "CatalogItem" ALTER COLUMN "subcategoryId" DROP DEFAULT;
 
--- Company: drop defaults, enforce NOT NULL
+-- Company: uid was added nullable (no default), id was made NOT NULL in Section 4.
+-- These are no-ops but kept for safety on fresh DBs.
 ALTER TABLE "Company" ALTER COLUMN "uid" DROP DEFAULT;
 ALTER TABLE "Company" ALTER COLUMN "id" SET NOT NULL;
 

api/src/workert.ts

@@ -165,7 +165,7 @@ if (import.meta.main) {
   const { executeIncrementalSync } = await import("./modules/workers/dalpuri-sync");
   await boss.work(WorkerQueue.DALPURI_FULL_SYNC, async () => {
     const socket = await ensureManagerSocketReady();
-    await enqueueDalpuriFullSync();
+    await enqueueDalpuriFullSync(socket);
   });
   console.log("[worker] Registered DALPURI_FULL_SYNC job handler");