Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0594816ea4 | |||
| 71fe36c0b8 | |||
| e0d575454e | |||
| 32bba31e72 | |||
| 1233535b20 | |||
| 2c737b22f1 | |||
| a3bfe9f374 |
@@ -318,9 +318,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
|
||||
|
||||
@@ -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/node_modules/pdfmake/build/fonts/ ./node_modules/pdfmake/build/fonts/
|
||||
|
||||
ENV NODE_ENV=production
|
||||
|
||||
# ---- Stage 4: API server runtime image ----
|
||||
|
||||
@@ -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,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,
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
});
|
||||
|
||||
|
||||
@@ -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 () => {
|
||||
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) => {
|
||||
|
||||
@@ -8,7 +8,7 @@ metadata:
|
||||
spec:
|
||||
backoffLimit: 0
|
||||
ttlSecondsAfterFinished: 86400
|
||||
activeDeadlineSeconds: 1800
|
||||
activeDeadlineSeconds: 7200
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
+42
-10
@@ -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,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);
|
||||
});
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user