Compare commits

...

9 Commits

Author SHA1 Message Date
HoloPanio b2cd26af30 fix(release): unblock deploy workflow image build and desktop rebuild 2026-04-10 03:44:33 +00:00
HoloPanio 0594816ea4 fix(api): include pdfmake Roboto fonts in runtime image 2026-04-10 03:00:31 +00:00
HoloPanio 71fe36c0b8 fix(worker): restore reliable 5s incremental sync cadence 2026-04-10 01:00:04 +00:00
HoloPanio e0d575454e fix(dalpuri): sync CW Members before Users to resolve FK ordering issue
User rows have a FK constraint to CwMember (User_cwMemberId_fkey). Syncing
Users first caused all 140 User upserts to fail since the CwMember table was
empty. This cascade failure then caused all Opportunity upserts to fail because
Opportunity.primarySalesRepId is FK-constrained to User.cwIdentifier.

Fix: reorder steps so CW Members syncs first, then Users.
2026-04-09 01:04:00 +00:00
HoloPanio 32bba31e72 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.
2026-04-09 00:22:41 +00:00
HoloPanio 1233535b20 fix(dalpuri): populate userIdentifiersByMemberRecId from CwMember table
When no User accounts have cwMemberId linked, the context map was empty
and all opportunities got primarySalesRepId = null. Now also populate
the map from CwMember rows directly (User-linked entries take precedence),
so rep identifiers resolve correctly regardless of user account linkage.
2026-04-08 23:23:51 +00:00
HoloPanio 2c737b22f1 fix(dalpuri): exit(0) after sync completes to release k8s job
Prisma MSSQL adapter keeps connections open after the sync finishes,
preventing the process from exiting naturally. The k8s job was staying
in Running state indefinitely. Call process.exit(0) on success so the
job completes and the GH workflow step passes.
2026-04-08 21:50:52 +00:00
HoloPanio a3bfe9f374 fix(ci): increase dalpuri sync timeout from 30min to 2h
Full initial sync has 500k+ rows across all tables and exceeded the
30-minute activeDeadlineSeconds. Bump both the k8s job deadline and
the kubectl wait timeout to 7200s (2 hours).
2026-04-08 21:19:43 +00:00
HoloPanio a106bb15a8 fix(ci): explicit env vars in dalpuri sync job; add CW_DATABASE_URL to secret
envFrom was loading api-env-secret but CW_DATABASE_URL was absent from the
deployed secret, causing sync.ts to fall back to DATABASE_URL (Postgres) as
the MSSQL connection string -> 'Invalid port number: //optima'.

- Replaced envFrom with explicit CW_DATABASE_URL and API_DATABASE_URL env
  entries so the mapping is unambiguous
- Patched api-env-secret in cluster to add CW_DATABASE_URL
2026-04-08 20:41:49 +00:00
11 changed files with 164 additions and 27 deletions
+6 -4
View File
@@ -231,9 +231,10 @@ jobs:
run: bun install --frozen-lockfile
- name: Rebuild native modules
run: npm rebuild
run: npm rebuild --ignore-scripts
env:
HUSKY: "0"
HUSKY_SKIP_INSTALL: "1"
- name: Build macOS distributables
run: bun run make:macos
@@ -272,9 +273,10 @@ jobs:
run: bun install --frozen-lockfile
- name: Rebuild native modules
run: npm rebuild
run: npm rebuild --ignore-scripts
env:
HUSKY: "0"
HUSKY_SKIP_INSTALL: "1"
- name: Build Windows distributables
run: bun run make -- --platform win32
@@ -318,9 +320,9 @@ jobs:
TAG=${{ github.event.release.tag_name }}
JOB="job/dalpuri-sync-${TAG}"
kubectl wait --for=condition=complete --timeout=1800s -n optima "$JOB" &
kubectl wait --for=condition=complete --timeout=7200s -n optima "$JOB" &
WAIT_COMPLETE=$!
kubectl wait --for=condition=failed --timeout=1800s -n optima "$JOB" &
kubectl wait --for=condition=failed --timeout=7200s -n optima "$JOB" &
WAIT_FAILED=$!
wait -n $WAIT_COMPLETE $WAIT_FAILED
+3
View File
@@ -90,6 +90,9 @@ COPY --from=build /app/dalpuri/generated/ ./dalpuri/generated/
# Copy production node_modules (Prisma adapter needs native bindings)
COPY --from=deps /app/node_modules/ ./node_modules/
# Ensure pdfmake Roboto fonts are present at runtime for PDF generation.
COPY --from=build /app/api/node_modules/pdfmake/build/fonts/ ./node_modules/pdfmake/build/fonts/
ENV NODE_ENV=production
# ---- Stage 4: API server runtime image ----
+21 -2
View File
@@ -1,5 +1,5 @@
import PdfPrinter from "pdfmake/src/Printer";
import { readFileSync } from "node:fs";
import { existsSync, readFileSync } from "node:fs";
import { join } from "node:path";
export interface QuoteLineItem {
@@ -110,7 +110,26 @@ const COMPANY = {
const DEFAULT_LOGO_PATH = join(process.cwd(), "logo.png");
const fontDir = join(process.cwd(), "node_modules/pdfmake/build/fonts/Roboto");
function resolveRobotoFontDir(): string {
const candidates = [
join(process.cwd(), "node_modules/pdfmake/build/fonts/Roboto"),
join(import.meta.dir, "../../../node_modules/pdfmake/build/fonts/Roboto"),
join("/app/node_modules/pdfmake/build/fonts/Roboto"),
join("/app/api/node_modules/pdfmake/build/fonts/Roboto"),
];
for (const dir of candidates) {
if (existsSync(join(dir, "Roboto-Medium.ttf"))) {
return dir;
}
}
throw new Error(
`[pdf] Could not locate pdfmake Roboto fonts. Checked: ${candidates.join(", ")}`
);
}
const fontDir = resolveRobotoFontDir();
const fonts = {
Roboto: {
normal: join(fontDir, "Roboto-Regular.ttf"),
+1
View File
@@ -1,6 +1,7 @@
import { Server } from "socket.io";
import { events, EventTypes } from "../globalEvents";
import { WorkerQueue } from "./queues";
import { reserveWorkerId } from "../../workert";
function emitGlobalEvent<K extends keyof EventTypes>(
name: K,
+13 -1
View File
@@ -6,5 +6,17 @@ import { WorkerQueue } from "./queues";
* Called on an interval from the main API process so it survives worker restarts.
*/
export async function enqueueIncrementalSync(): Promise<void> {
await getBoss().send(WorkerQueue.DALPURI_INCREMENTAL_SYNC, {});
const jobId = await getBoss().send(
WorkerQueue.DALPURI_INCREMENTAL_SYNC,
{
enqueuedAt: new Date().toISOString(),
},
{
singletonKey: "dalpuri-incremental-sync",
}
);
if (!jobId) {
console.debug("[interval] DALPURI_INCREMENTAL_SYNC already pending or active");
}
}
+3 -1
View File
@@ -32,6 +32,8 @@ export async function createWorkerJob<T>(
queueType: WorkerQueue,
workFn: (workerSocket: Socket) => Promise<T>,
): Promise<T> {
const managerUrl = process.env.MANAGER_SOCKET_URL ?? "http://localhost:8671";
return new Promise((resolve, reject) => {
// Request a worker ID and namespace from the manager
socket.emit(
@@ -53,7 +55,7 @@ export async function createWorkerJob<T>(
}
// Connect to the worker-specific namespace
const workerSocket = io(`http://localhost:8671/worker-${workerId}`, {
const workerSocket = io(`${managerUrl}/worker-${workerId}`, {
reconnection: false,
});
+29 -1
View File
@@ -163,6 +163,7 @@ if (import.meta.main) {
// Register job handler for DALPURI_FULL_SYNC
const { enqueueDalpuriFullSync } = await import("./modules/workers/sync-manager");
const { executeIncrementalSync } = await import("./modules/workers/dalpuri-sync");
const { enqueueIncrementalSync } = await import("./modules/workers/incremental-sync");
await boss.work(WorkerQueue.DALPURI_FULL_SYNC, async () => {
const socket = await ensureManagerSocketReady();
await enqueueDalpuriFullSync(socket);
@@ -170,10 +171,37 @@ if (import.meta.main) {
console.log("[worker] Registered DALPURI_FULL_SYNC job handler");
await boss.work(WorkerQueue.DALPURI_INCREMENTAL_SYNC, async () => {
await executeIncrementalSync();
const startedAt = Date.now();
console.log("[worker] DALPURI_INCREMENTAL_SYNC started");
try {
await executeIncrementalSync();
console.log(
`[worker] DALPURI_INCREMENTAL_SYNC completed in ${Date.now() - startedAt}ms`
);
} catch (err) {
console.error(
`[worker] DALPURI_INCREMENTAL_SYNC failed in ${Date.now() - startedAt}ms`,
err
);
throw err;
}
});
console.log("[worker] Registered DALPURI_INCREMENTAL_SYNC job handler");
const enqueueIncrementalWithLogging = () => {
enqueueIncrementalSync().catch((err) => {
console.error(
`[worker] interval enqueueIncrementalSync failed: ${err?.message ?? err}`
);
});
};
// Keep a worker-local 5s scheduler so incremental sync continues even when
// API interval scheduling is unavailable.
enqueueIncrementalWithLogging();
setInterval(enqueueIncrementalWithLogging, 5_000);
console.log("[worker] Started 5-second incremental enqueue interval");
// Register job handler for REFRESH_SALES_METRICS
const { executeSalesMetricsRefresh } = await import("./modules/workers/sales-metrics");
await boss.work(WorkerQueue.REFRESH_SALES_METRICS, async (jobs) => {
+6 -4
View File
@@ -8,7 +8,7 @@ metadata:
spec:
backoffLimit: 0
ttlSecondsAfterFinished: 86400
activeDeadlineSeconds: 1800
activeDeadlineSeconds: 7200
template:
metadata:
labels:
@@ -17,10 +17,12 @@ spec:
containers:
- name: sync
image: ghcr.io/horizonstacksoftware/optima-dalpuri-sync:RELEASE_TAG
envFrom:
- secretRef:
name: api-env-secret
env:
- name: CW_DATABASE_URL
valueFrom:
secretKeyRef:
name: api-env-secret
key: CW_DATABASE_URL
- name: API_DATABASE_URL
valueFrom:
secretKeyRef:
+27
View File
@@ -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,
},
},
},
},
},
+45 -13
View File
@@ -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 &&
@@ -1328,6 +1351,15 @@ export const executeFullDalpuriSync = async (options?: {
const isTimedOut = () => Date.now() - syncStartTime > timeoutMs;
const steps: Step[] = [
{
name: "CW Members",
sourceModel: "member",
targetModel: "cwMember",
translation: cwMemberTranslation as unknown as AnyTranslation,
uniqueField: "cwMemberId",
sourceIdField: "memberRecId",
sourceUpdatedField: "lastUpdatedUtc",
},
{
name: "Users",
sourceModel: "member",
@@ -1342,15 +1374,6 @@ export const executeFullDalpuriSync = async (options?: {
},
},
},
{
name: "CW Members",
sourceModel: "member",
targetModel: "cwMember",
translation: cwMemberTranslation as unknown as AnyTranslation,
uniqueField: "cwMemberId",
sourceIdField: "memberRecId",
sourceUpdatedField: "lastUpdatedUtc",
},
{
name: "Companies",
sourceModel: "company",
@@ -1585,6 +1608,11 @@ export const executeFullDalpuriSync = async (options?: {
secondarySalesFlag: true,
},
},
soOppStatus: {
select: {
closedFlag: true,
},
},
},
},
},
@@ -1860,8 +1888,12 @@ export const executeForcedIncrementalDalpuriSync = async (options?: {
};
if (import.meta.main) {
executeFullDalpuriSync().catch((error) => {
console.error("CW -> API sync failed:", error);
process.exit(1);
});
executeFullDalpuriSync()
.then(() => {
process.exit(0);
})
.catch((error) => {
console.error("CW -> API sync failed:", error);
process.exit(1);
});
}
+10 -1
View File
@@ -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",