fix(dalpuri): populate locationId and fix closedFlag on opportunities

- Add ownerLevelRecId -> locationId mapping to opportunity translation
- Include soOppStatus in opportunity query and derive closedFlag from
  status.closedFlag (with fallback to legacy oldCloseFlag field)
- Add locationId sanitization guard in both sync.ts and sync-by-table.ts

Note: departmentId is not available in CW SO_Opportunity table and
remains null for synced records.

.github/workflows/deploy.yaml

@@ -130,6 +130,17 @@ jobs:
             ghcr.io/horizonstacksoftware/optima-api-migrate:latest
             ghcr.io/horizonstacksoftware/optima-api-migrate:${{ github.event.release.tag_name }}
 
+      - name: Build and push the dalpuri sync image
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: api/Dockerfile
+          push: true
+          target: dalpuri-sync
+          tags: |
+            ghcr.io/horizonstacksoftware/optima-dalpuri-sync:latest
+            ghcr.io/horizonstacksoftware/optima-dalpuri-sync:${{ github.event.release.tag_name }}
+
   build-worker:
     name: Build - Worker
     needs: [test-api, test-dalpuri, test-ui]
@@ -276,6 +287,55 @@ jobs:
           files: |
             ui/out/make/**/*.exe
 
+  # Runs a full CW → API data sync as a Kubernetes Job (the CW MSSQL and
+  # API Postgres addresses are internal to the cluster and unreachable from
+  # GitHub-hosted runners).  Waits for both images to be built first and
+  # must succeed before either the API or worker deploys.
+  sync-cw-to-api:
+    name: Sync - CW to API
+    needs: [migrate-api, build-worker]
+    runs-on: ubuntu-latest
+    steps:
+      - name: Set the Kubernetes context
+        uses: azure/k8s-set-context@v2
+        with:
+          method: kubeconfig
+          kubeconfig: ${{ secrets.KUBECONFIG }}
+
+      - name: Checkout source code
+        uses: actions/checkout@v4
+
+      - name: Delete previous sync job if exists
+        run: kubectl delete job -n optima -l app=dalpuri-sync --ignore-not-found
+
+      - name: Apply sync job
+        run: |
+          TAG=${{ github.event.release.tag_name }}
+          sed "s/RELEASE_TAG/${TAG}/g" dalpuri/kubernetes/sync-job.yaml | kubectl apply -f -
+
+      - name: Wait for sync to complete
+        run: |
+          TAG=${{ github.event.release.tag_name }}
+          JOB="job/dalpuri-sync-${TAG}"
+
+          kubectl wait --for=condition=complete --timeout=7200s -n optima "$JOB" &
+          WAIT_COMPLETE=$!
+          kubectl wait --for=condition=failed --timeout=7200s -n optima "$JOB" &
+          WAIT_FAILED=$!
+
+          wait -n $WAIT_COMPLETE $WAIT_FAILED
+
+          echo "--- Sync job logs ---"
+          kubectl logs -n optima "$JOB" --tail=500 || true
+
+          if kubectl get -n optima "$JOB" -o jsonpath='{.status.conditions[?(@.type=="Complete")].status}' | grep -q "True"; then
+            echo "Sync completed successfully."
+            exit 0
+          else
+            echo "Sync FAILED."
+            exit 1
+          fi
+
   # ==========================================================================
   # Deploy jobs
   # ==========================================================================
@@ -332,7 +392,7 @@ jobs:
 
   deploy-api:
     name: Deploy - API
-    needs: [migrate-api]
+    needs: [migrate-api, sync-cw-to-api]
     runs-on: ubuntu-latest
     steps:
       - name: Set the Kubernetes context
@@ -402,7 +462,7 @@ jobs:
 
   deploy-worker:
     name: Deploy - Worker
-    needs: [build-worker]
+    needs: [build-worker, sync-cw-to-api]
     runs-on: ubuntu-latest
     steps:
       - name: Set the Kubernetes context

api/Dockerfile

@@ -134,3 +134,33 @@ RUN chmod +x /app/api/prisma/migrate-entrypoint.sh
 
 WORKDIR /app/api
 CMD ["sh", "prisma/migrate-entrypoint.sh"]
+
+# ---- Stage 7: Dalpuri CW-to-API sync runner ----
+FROM oven/bun:1.3.11 AS dalpuri-sync
+
+WORKDIR /app
+
+COPY package.json bun.lock ./
+COPY api/package.json ./api/package.json
+COPY dalpuri/package.json ./dalpuri/package.json
+COPY ui/package.json ./ui/package.json
+COPY patches ./patches
+
+RUN bun install --frozen-lockfile
+
+COPY dalpuri/src/ ./dalpuri/src/
+COPY dalpuri/prisma/ ./dalpuri/prisma/
+COPY dalpuri/prisma.config.ts ./dalpuri/prisma.config.ts
+
+COPY api/prisma/ ./api/prisma/
+COPY api/prisma.config.ts ./api/prisma.config.ts
+
+WORKDIR /app/dalpuri
+RUN DATABASE_URL="sqlserver://localhost:1433;database=dummy;user=dummy;password=dummy;trustServerCertificate=true" \
+    bunx prisma generate
+
+WORKDIR /app/api
+RUN DATABASE_URL="postgresql://dummy:dummy@localhost:5432/dummy" bunx prisma generate
+
+WORKDIR /app/dalpuri
+CMD ["bun", "run", "src/sync.ts"]

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");
 

dalpuri/kubernetes/sync-job.yaml

@@ -0,0 +1,33 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: dalpuri-sync-RELEASE_TAG
+  namespace: optima
+  labels:
+    app: dalpuri-sync
+spec:
+  backoffLimit: 0
+  ttlSecondsAfterFinished: 86400
+  activeDeadlineSeconds: 7200
+  template:
+    metadata:
+      labels:
+        app: dalpuri-sync
+    spec:
+      containers:
+        - name: sync
+          image: ghcr.io/horizonstacksoftware/optima-dalpuri-sync:RELEASE_TAG
+          env:
+            - name: CW_DATABASE_URL
+              valueFrom:
+                secretKeyRef:
+                  name: api-env-secret
+                  key: CW_DATABASE_URL
+            - name: API_DATABASE_URL
+              valueFrom:
+                secretKeyRef:
+                  name: api-env-secret
+                  key: DATABASE_URL
+      restartPolicy: Never
+      imagePullSecrets:
+        - name: github-container-registry

dalpuri/src/sync-by-table.ts

@@ -294,6 +294,22 @@ const refreshContextFromApi = async (
     }
   }
 
+  const cwMembers = await apiPrisma.cwMember.findMany({
+    select: { cwMemberId: true, identifier: true },
+  });
+  for (const member of cwMembers) {
+    if (
+      member.cwMemberId != null &&
+      member.identifier &&
+      !context.userIdentifiersByMemberRecId.has(member.cwMemberId)
+    ) {
+      context.userIdentifiersByMemberRecId.set(
+        member.cwMemberId,
+        member.identifier
+      );
+    }
+  }
+
   for (const board of boards) {
     context.serviceTicketBoardUidsById.set(board.id, board.uid);
   }
@@ -426,6 +442,12 @@ const sanitizeModelData = (
     ) {
       sanitized.statusId = null;
     }
+    if (
+      sanitized.locationId != null &&
+      !context.corporateLocationIds.has(sanitized.locationId as number)
+    ) {
+      sanitized.locationId = null;
+    }
   }
 
   if (targetModel === "schedule") {
@@ -734,6 +756,11 @@ const getConfigForTable = (table: string): SyncTableConfig | null => {
               secondarySalesFlag: true,
             },
           },
+          soOppStatus: {
+            select: {
+              closedFlag: true,
+            },
+          },
         },
       },
     },

dalpuri/src/sync.ts

@@ -323,6 +323,22 @@ const refreshContextFromApi = async (
     }
   }
 
+  const cwMembers = await apiPrisma.cwMember.findMany({
+    select: { cwMemberId: true, identifier: true },
+  });
+  for (const member of cwMembers) {
+    if (
+      member.cwMemberId != null &&
+      member.identifier &&
+      !context.userIdentifiersByMemberRecId.has(member.cwMemberId)
+    ) {
+      context.userIdentifiersByMemberRecId.set(
+        member.cwMemberId,
+        member.identifier
+      );
+    }
+  }
+
   for (const board of boards) {
     context.serviceTicketBoardUidsById.set(board.id, board.uid);
   }
@@ -636,6 +652,13 @@ const sanitizeModelData = (
     ) {
       sanitized.stageId = null;
     }
+    // Nullify locationId if the corporate location doesn't exist
+    if (
+      sanitized.locationId != null &&
+      !context.corporateLocationIds.has(sanitized.locationId as number)
+    ) {
+      sanitized.locationId = null;
+    }
     // Nullify taxCodeId if the tax code hasn't synced yet
     if (
       sanitized.taxCodeId != null &&
@@ -1585,6 +1608,11 @@ export const executeFullDalpuriSync = async (options?: {
               secondarySalesFlag: true,
             },
           },
+          soOppStatus: {
+            select: {
+              closedFlag: true,
+            },
+          },
         },
       },
     },
@@ -1860,7 +1888,11 @@ export const executeForcedIncrementalDalpuriSync = async (options?: {
 };
 
 if (import.meta.main) {
-  executeFullDalpuriSync().catch((error) => {
+  executeFullDalpuriSync()
+    .then(() => {
+      process.exit(0);
+    })
+    .catch((error) => {
       console.error("CW -> API sync failed:", error);
       process.exit(1);
     });

dalpuri/src/translations/opportunity.ts

@@ -1,6 +1,7 @@
 import {
   Opportunity as CwOpportunity,
   OpportunityMember as CwOpportunityMember,
+  SoOppStatus as CwSoOppStatus,
 } from "../../generated/prisma/client";
 import { OpportunityInterest } from "../../../api/generated/prisma/client";
 import { Translation, skipRow } from "./types";
@@ -30,6 +31,7 @@ type ApiOpportunityRecord = {
   dateBecameLead?: Date | null;
   closedDate?: Date | null;
   closedFlag: boolean;
+  locationId?: number | null;
   closedById?: string | null;
   updatedBy: string;
   eneteredBy: string;
@@ -42,6 +44,7 @@ type CwOpportunityWithMembers = CwOpportunity & {
     CwOpportunityMember,
     "memberRecId" | "primarySalesFlag" | "secondarySalesFlag"
   >[];
+  soOppStatus?: Pick<CwSoOppStatus, "closedFlag"> | null;
 };
 
 const toInterest = (value: number | null): OpportunityInterest | null => {
@@ -119,13 +122,19 @@ export const opportunityTranslation: Translation<
     },
     { from: "companyRecId", to: "companyId" },
     { from: "contactRecId", to: "contactId" },
+    { from: "ownerLevelRecId", to: "locationId" },
     { from: "companyAddressRecId", to: "siteId" },
     { from: "poNumber", to: "customerPO" },
     { from: "dateCloseExpected", to: "expectedCloseDate" },
     { from: "datePipelineChange", to: "pipelineChangeDate" },
     { from: "dateBecameLead", to: "dateBecameLead" },
     { from: "dateClosed", to: "closedDate" },
-    { from: "oldCloseFlag", to: "closedFlag" },
+    {
+      from: "oldCloseFlag",
+      to: "closedFlag",
+      process: (_value, _context, row) =>
+        row.soOppStatus?.closedFlag ?? row.oldCloseFlag ?? false,
+    },
     { from: "closedBy", to: "closedById" },
     {
       from: "updatedBy",