Semih Yesilyurt
79bbd1f1b3
Some checks failed
QA Gate (P0/P1) / Test affected app (pull_request) Has been cancelled
isLeafLinkPath flagged EVERY pcat node as a leaf (any pcat: linkPath), so the backfill worker fetched a mid-group folder's "parts" — a 1-level drill that only revealed its sub-groups — and never recursed. Deep pcat trees (whose top level is ~100% folders) were left one level shallow: no parts seeded, so the new tree part-counts stayed 0 and the cross-tree catalog search (DB-only) found nothing on a freshly decoded vehicle. Exactly why serkan's Ford Mondeo looked like an empty catalog. Use the captured hasSubgroups flag: a pcat node is a parts leaf only when it is NOT a known parent group. Folders now queue a children job and the existing recursion (queueCategoryJob → processChildren → getChildren) drills to full depth (MAX_DEPTH=5), seeding parts at every leaf. The rolling backfill rescan warms existing shallow vehicles wave by wave; new decodes warm deep via the reactive schedulePrefetch. Rate limiter + PCAT_PACE + backlog guard keep it bounded (all env-tunable to ramp). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
647 lines
24 KiB
TypeScript
647 lines
24 KiB
TypeScript
import {
|
|
Inject,
|
|
Injectable,
|
|
Logger,
|
|
type OnModuleDestroy,
|
|
type OnModuleInit,
|
|
} from "@nestjs/common";
|
|
import { DelayedError, type Job, type Queue, Worker } from "bullmq";
|
|
import { and, asc, eq, gt, inArray, isNull, notExists, sql } from "drizzle-orm";
|
|
import { CategoriesService } from "../categories/categories.service";
|
|
import { DATABASE, type Database } from "../database/database.provider";
|
|
import { categories, parts, vehicles } from "../database/schema/core";
|
|
import { PostHogService } from "../posthog/posthog.service";
|
|
import { RedisService } from "../redis/redis.service";
|
|
import { QUEUE_NAMES, getBullConnection } from "./bull.config";
|
|
import { backfillContext } from "./prefetch-context";
|
|
import {
|
|
RateLimitError,
|
|
checkCooldown,
|
|
checkTimeWindow,
|
|
initProgress,
|
|
isCatalogBackfillEnabled,
|
|
isWithinTimeWindow,
|
|
updateProgress,
|
|
} from "./prefetch-utils";
|
|
import { PrefetchCategoryJobData, PrefetchInitJobData } from "./prefetch.types";
|
|
import { CATALOG_PREFETCH_QUEUE } from "./queues/catalog-prefetch.queue";
|
|
|
|
const MAX_DEPTH = 5;
|
|
|
|
// ── Backfill scan (hourly cron) tuning ──
|
|
/**
|
|
* Vehicles queued per scan wave. Was 20 while pcat needed business-hours
|
|
* gating; doubled to 40 once the warm JWT pool + Redis hydration made the
|
|
* scrape stack 24/7. The MAX_BACKLOG guard below still gates run-away
|
|
* pileups, and per-source cooldown still yields to live users.
|
|
*/
|
|
const BACKFILL_BATCH_SIZE = 40;
|
|
/** Skip the wave entirely if the queue already has more than this many jobs pending. */
|
|
const BACKFILL_MAX_BACKLOG = 1000;
|
|
/** In-flight guard TTL (seconds) — safety net if a run dies without clearing. */
|
|
const BACKFILL_SCHEDULED_TTL = 6 * 60 * 60;
|
|
/** Only these decode sources have catalogs worth prefetching. */
|
|
const BACKFILL_SOURCES = ["pl24", "emex", "parts-catalogs"];
|
|
/** Redis key holding the rolling rescan cursor (last createdAt seen). */
|
|
const BACKFILL_CURSOR_KEY = "prefetch:backfill:cursor";
|
|
|
|
// ── Throughput knobs (env-tunable so prod can ramp without a redeploy) ──
|
|
/**
|
|
* Concurrent jobs the worker runs. emex/pl24 fetches parallelise well, so >1
|
|
* lets a slow parts-catalogs job stop head-of-line-blocking emex. A user-active
|
|
* cooldown still pauses the whole worker, so this only speeds up idle periods.
|
|
*/
|
|
const WORKER_CONCURRENCY = Number(process.env.PREFETCH_CONCURRENCY) || 3;
|
|
/** Global queue rate ceiling (jobs per minute). Was a hard 5/min. */
|
|
const WORKER_RATE_MAX = Number(process.env.PREFETCH_RATE_MAX) || 20;
|
|
/**
|
|
* Per-job pacing for parts-catalogs only (its browser/JWT capture is heavy).
|
|
* Set 0 to disable. Other sources are paced by the limiter + cooldown alone.
|
|
*/
|
|
const PCAT_PACE_MS = Number(process.env.PREFETCH_PCAT_DELAY_MS) || 15_000;
|
|
|
|
// ── Phase-1 residue exclusion ──
|
|
/**
|
|
* Skip a zero-parts vehicle once this many backfill attempts have completed with
|
|
* still-zero parts (genuine upstream-data gaps — e.g. a VIN in no catalog).
|
|
* Re-picking these every wave wastes capacity and starves the Phase-2 rescan.
|
|
*/
|
|
const NORESULT_MAX_ATTEMPTS = Number(process.env.PREFETCH_NORESULT_MAX) || 2;
|
|
/** TTL for the no-result counter — excluded vehicles retry after this, so a later
|
|
* catalog fix eventually re-fills them (default 7 days). */
|
|
const NORESULT_TTL_S = (Number(process.env.PREFETCH_NORESULT_TTL_DAYS) || 7) * 86_400;
|
|
|
|
@Injectable()
|
|
export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
|
|
private readonly logger = new Logger(PrefetchWorkerService.name);
|
|
private worker: Worker | null = null;
|
|
|
|
constructor(
|
|
@Inject(CATALOG_PREFETCH_QUEUE) private queue: Queue,
|
|
private categoriesService: CategoriesService,
|
|
private redis: RedisService,
|
|
private posthog: PostHogService,
|
|
@Inject(DATABASE) private db: Database,
|
|
) {}
|
|
|
|
onModuleInit() {
|
|
this.worker = new Worker(
|
|
QUEUE_NAMES.CATALOG_PREFETCH,
|
|
// Run every job inside the backfill context so source services' touchActivity()
|
|
// skips the user-activity cooldown key — the worker must not throttle itself.
|
|
(job, token) => backfillContext.run(true, () => this.process(job, token)),
|
|
{
|
|
connection: getBullConnection(),
|
|
concurrency: WORKER_CONCURRENCY,
|
|
limiter: { max: WORKER_RATE_MAX, duration: 60_000 },
|
|
},
|
|
);
|
|
|
|
this.worker.on("failed", (job, err) => {
|
|
// RateLimitError is handled in process() via moveToDelayed (a DelayedError,
|
|
// not a failure), so anything reaching here is a genuine failure.
|
|
this.logger.warn(`[prefetch] Job ${job?.name} failed: ${err.message}`);
|
|
});
|
|
|
|
this.worker.on("error", (err) => {
|
|
this.logger.error(`[prefetch] Worker error: ${err.message}`);
|
|
});
|
|
|
|
this.logger.log(
|
|
`[prefetch] Worker started (concurrency=${WORKER_CONCURRENCY}, ${WORKER_RATE_MAX} jobs/min, pcatPace=${PCAT_PACE_MS}ms)`,
|
|
);
|
|
}
|
|
|
|
async onModuleDestroy() {
|
|
if (this.worker) {
|
|
await this.worker.close();
|
|
this.worker = null;
|
|
}
|
|
}
|
|
|
|
private async process(job: Job, token?: string): Promise<void> {
|
|
try {
|
|
const data = job.data as { source?: string };
|
|
if (data.source === "parts-catalogs" && PCAT_PACE_MS > 0) {
|
|
await new Promise((r) => setTimeout(r, PCAT_PACE_MS));
|
|
}
|
|
|
|
if (job.name === "backfill-scan") {
|
|
return await this.processBackfillScan();
|
|
}
|
|
if (job.name === "prefetch-init") {
|
|
return await this.processInit(job as Job<PrefetchInitJobData>);
|
|
}
|
|
if (job.name === "prefetch-children") {
|
|
return await this.processChildren(job as Job<PrefetchCategoryJobData>);
|
|
}
|
|
if (job.name === "prefetch-parts") {
|
|
return await this.processParts(job as Job<PrefetchCategoryJobData>);
|
|
}
|
|
this.logger.warn(`[prefetch] Unknown job name: ${job.name}`);
|
|
} catch (err) {
|
|
if (err instanceof RateLimitError) {
|
|
const delayMs = Math.max(1000, err.retryAfterMs);
|
|
if (err.cause === "cooldown") {
|
|
// Pause the WHOLE worker for the cooldown window: the per-source
|
|
// activity key is refreshed by every user request, so per-job defer
|
|
// livelocks (job comes back, key is still set, defer again, …). One
|
|
// global wait is correct: when it ends and cooldown is still active,
|
|
// we'll pause again — at most ~1 cycle per retryAfterMs instead of
|
|
// cycling every queued job through.
|
|
await this.worker?.rateLimit(delayMs);
|
|
this.logger.debug(`[prefetch] Worker paused ${delayMs}ms (cooldown ${job.data.source})`);
|
|
throw Worker.RateLimitError();
|
|
}
|
|
// Time-window deferral can be hours (msUntilNext9AM). Per-job defer so
|
|
// EMEX (no window) keeps flowing while PL24/pcat jobs sleep till 09:00.
|
|
await job.moveToDelayed(Date.now() + delayMs, token);
|
|
this.logger.debug(`[prefetch] Job ${job.name} deferred ${delayMs}ms (off-hours)`);
|
|
throw new DelayedError();
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Init job: walk the category tree for a vehicle and queue sub-jobs.
|
|
*/
|
|
private async processInit(job: Job<PrefetchInitJobData>): Promise<void> {
|
|
const { vehicleId, source } = job.data;
|
|
this.logger.log(`[prefetch] Init for vehicle=${vehicleId}, source=${source}`);
|
|
|
|
await checkCooldown(this.redis, source);
|
|
checkTimeWindow(source);
|
|
|
|
// Verify vehicle still exists
|
|
const [vehicle] = await this.db
|
|
.select({ id: vehicles.id })
|
|
.from(vehicles)
|
|
.where(eq(vehicles.id, vehicleId))
|
|
.limit(1);
|
|
|
|
if (!vehicle) {
|
|
this.logger.warn(`[prefetch] Vehicle ${vehicleId} not found, skipping`);
|
|
return;
|
|
}
|
|
|
|
await initProgress(this.redis, vehicleId);
|
|
|
|
// Get all top-level categories for this vehicle
|
|
let topCategories = await this.db
|
|
.select()
|
|
.from(categories)
|
|
.where(and(eq(categories.vehicleId, vehicleId), isNull(categories.parentId)));
|
|
|
|
// Self-seed: a vehicle that was decoded but never viewed has no categories in
|
|
// DB yet. getCategoryTree fetches & inserts the top-level groups from upstream
|
|
// (PL24/PSA/EMEX) so backfill can proceed without a user opening the page.
|
|
// Cooldown/time-window were already enforced above, so this only runs in-window.
|
|
if (topCategories.length === 0) {
|
|
try {
|
|
await this.categoriesService.getCategoryTree(vehicleId);
|
|
topCategories = await this.db
|
|
.select()
|
|
.from(categories)
|
|
.where(and(eq(categories.vehicleId, vehicleId), isNull(categories.parentId)));
|
|
} catch (err) {
|
|
this.logger.warn(
|
|
`[prefetch] Top-category seed failed for ${vehicleId}: ${(err as Error).message}`,
|
|
);
|
|
}
|
|
}
|
|
|
|
if (topCategories.length === 0) {
|
|
this.logger.log(`[prefetch] No categories for vehicle=${vehicleId}`);
|
|
// No categories after a self-seed attempt = no catalog data for this VIN.
|
|
// Count it as residue so Phase-1 stops re-picking it every wave.
|
|
await this.markNoResult(vehicleId);
|
|
await updateProgress(this.redis, vehicleId, {
|
|
status: "completed",
|
|
total: 0,
|
|
});
|
|
return;
|
|
}
|
|
|
|
let queued = 0;
|
|
|
|
for (const cat of topCategories) {
|
|
if (cat.unavailable) continue;
|
|
|
|
// Check if this category already has children in DB
|
|
const [childCheck] = await this.db
|
|
.select({ id: categories.id })
|
|
.from(categories)
|
|
.where(eq(categories.parentId, cat.id))
|
|
.limit(1);
|
|
|
|
if (childCheck) {
|
|
// Has children — queue recursive exploration of children
|
|
const children = await this.db
|
|
.select()
|
|
.from(categories)
|
|
.where(eq(categories.parentId, cat.id));
|
|
|
|
for (const child of children) {
|
|
if (child.unavailable) continue;
|
|
await this.queueCategoryJob(child, vehicleId, source, 1);
|
|
queued++;
|
|
}
|
|
} else if (this.isLeafLinkPath(cat.linkPath, cat.source, cat.hasSubgroups)) {
|
|
// Leaf — check if parts already fetched
|
|
const [partCheck] = await this.db
|
|
.select({ id: parts.id })
|
|
.from(parts)
|
|
.where(eq(parts.categoryId, cat.id))
|
|
.limit(1);
|
|
|
|
if (!partCheck && cat.linkPath) {
|
|
await this.addJob("prefetch-parts", {
|
|
vehicleId,
|
|
categoryId: cat.id,
|
|
source,
|
|
action: "parts" as const,
|
|
depth: 0,
|
|
});
|
|
queued++;
|
|
}
|
|
} else if (cat.linkPath) {
|
|
// Non-leaf without children — needs children fetch
|
|
await this.addJob("prefetch-children", {
|
|
vehicleId,
|
|
categoryId: cat.id,
|
|
source,
|
|
action: "children" as const,
|
|
depth: 0,
|
|
});
|
|
queued++;
|
|
}
|
|
}
|
|
|
|
await updateProgress(this.redis, vehicleId, { total: queued });
|
|
this.logger.log(`[prefetch] Queued ${queued} sub-jobs for vehicle=${vehicleId}`);
|
|
}
|
|
|
|
/**
|
|
* Fetch children (sub-categories) for a category.
|
|
*/
|
|
private async processChildren(job: Job<PrefetchCategoryJobData>): Promise<void> {
|
|
const { vehicleId, categoryId, source, depth } = job.data;
|
|
this.logger.log(`[prefetch] Children for category=${categoryId}, depth=${depth}`);
|
|
|
|
await checkCooldown(this.redis, source);
|
|
checkTimeWindow(source);
|
|
|
|
if (depth >= MAX_DEPTH) {
|
|
this.logger.warn(`[prefetch] Max depth reached for category=${categoryId}`);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
const children = await this.categoriesService.getChildren(categoryId);
|
|
|
|
let queued = 0;
|
|
for (const child of children) {
|
|
if (child.unavailable) continue;
|
|
await this.queueCategoryJob(child, vehicleId, source, depth + 1);
|
|
queued++;
|
|
}
|
|
|
|
if (queued > 0) {
|
|
await updateProgress(this.redis, vehicleId, {
|
|
total:
|
|
((await this.redis.getJson<{ total: number }>(`prefetch:progress:${vehicleId}`))
|
|
?.total || 0) + queued,
|
|
});
|
|
}
|
|
|
|
await this.incrementCompleted(vehicleId);
|
|
} catch (err) {
|
|
if (err instanceof RateLimitError) throw err;
|
|
this.logger.error(
|
|
`[prefetch] Children fetch failed for ${categoryId}: ${(err as Error).message}`,
|
|
);
|
|
await this.incrementErrors(vehicleId);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Fetch parts + schema for a leaf category.
|
|
*/
|
|
private async processParts(job: Job<PrefetchCategoryJobData>): Promise<void> {
|
|
const { vehicleId, categoryId, source } = job.data;
|
|
this.logger.log(`[prefetch] Parts for category=${categoryId}`);
|
|
|
|
await checkCooldown(this.redis, source);
|
|
checkTimeWindow(source);
|
|
|
|
try {
|
|
await this.categoriesService.getCategoryWithParts(categoryId);
|
|
await this.incrementCompleted(vehicleId);
|
|
} catch (err) {
|
|
if (err instanceof RateLimitError) throw err;
|
|
this.logger.error(
|
|
`[prefetch] Parts fetch failed for ${categoryId}: ${(err as Error).message}`,
|
|
);
|
|
await this.incrementErrors(vehicleId);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Backfill scan (hourly cron): find decoded vehicles whose catalog isn't fully
|
|
* prefetched and queue prefetch-init for them. prefetch-init is idempotent — it
|
|
* only fetches missing categories/parts, so re-running over a complete vehicle is
|
|
* cheap. Self-throttling via a queue-depth guard, per-source cooldown, and the
|
|
* business-hours window. Goal: no decoded vehicle is left without catalog data.
|
|
*/
|
|
private async processBackfillScan(): Promise<void> {
|
|
// Production (sase.tr) only — dev.sase.tr uses a separate DB and must never
|
|
// sweep/scrape. Defense-in-depth in case a scan job lands here via a stale
|
|
// scheduler. NODE_ENV is "production" on both envs, so gate on the prod host.
|
|
if (!isCatalogBackfillEnabled()) {
|
|
this.logger.log("[backfill] Skip — not prod host");
|
|
return;
|
|
}
|
|
|
|
// Live-tunable knobs via the `cfg-backfill-tuning` remote-config flag — change
|
|
// batch size / backlog ceiling / business-hours gating from the PostHog UI
|
|
// without a redeploy. Falls back to the compiled-in defaults on any miss.
|
|
const cfg = await this.posthog.payload<{
|
|
batchSize?: number;
|
|
maxBacklog?: number;
|
|
businessHoursOnly?: boolean;
|
|
}>("cfg-backfill-tuning", {});
|
|
const batchSize =
|
|
typeof cfg.batchSize === "number" && cfg.batchSize > 0 ? cfg.batchSize : BACKFILL_BATCH_SIZE;
|
|
const maxBacklog =
|
|
typeof cfg.maxBacklog === "number" && cfg.maxBacklog > 0
|
|
? cfg.maxBacklog
|
|
: BACKFILL_MAX_BACKLOG;
|
|
|
|
// Self-throttle: don't pile on if the queue is already deep — let it drain.
|
|
const counts = await this.queue.getJobCounts("waiting", "delayed", "active");
|
|
const backlog = (counts.waiting ?? 0) + (counts.delayed ?? 0) + (counts.active ?? 0);
|
|
if (backlog > maxBacklog) {
|
|
this.logger.log(`[backfill] Skip — queue backlog ${backlog} > ${maxBacklog}`);
|
|
return;
|
|
}
|
|
|
|
// Only target sources eligible right now: not in cooldown (user active) and
|
|
// inside their scrape window (PL24/parts-catalogs office hours; EMEX always).
|
|
const eligible: string[] = [];
|
|
for (const s of BACKFILL_SOURCES) {
|
|
if (await this.redis.exists(`prefetch:activity:${s}`)) continue;
|
|
if (cfg.businessHoursOnly !== false && !isWithinTimeWindow(s)) continue;
|
|
eligible.push(s);
|
|
}
|
|
if (eligible.length === 0) {
|
|
this.logger.log("[backfill] Skip — no eligible sources (cooldown / off-hours)");
|
|
return;
|
|
}
|
|
|
|
const picked: Array<{ id: string; source: string }> = [];
|
|
const seen = new Set<string>();
|
|
const overfetch = batchSize * 4; // headroom for in-flight skips
|
|
|
|
const tryPick = async (v: { id: string; source: string | null }): Promise<void> => {
|
|
if (picked.length >= batchSize || seen.has(v.id) || !v.source) return;
|
|
if (await this.redis.exists(`prefetch:scheduled:${v.id}`)) return; // already in flight
|
|
// Skip exhausted residue: vehicles whose prefetch keeps finishing with zero
|
|
// parts (no catalog data). They'd otherwise be re-picked every wave forever.
|
|
const noResult = await this.redis.get(this.noResultKey(v.id));
|
|
if (noResult && Number(noResult) >= NORESULT_MAX_ATTEMPTS) return;
|
|
seen.add(v.id);
|
|
picked.push({ id: v.id, source: v.source });
|
|
};
|
|
|
|
// Phase 1 — clear the obvious backlog first: decoded vehicles with zero parts.
|
|
const noParts = await this.db
|
|
.select({ id: vehicles.id, source: vehicles.source })
|
|
.from(vehicles)
|
|
.where(
|
|
and(
|
|
inArray(vehicles.source, eligible),
|
|
notExists(
|
|
this.db.select({ one: sql`1` }).from(parts).where(eq(parts.vehicleId, vehicles.id)),
|
|
),
|
|
),
|
|
)
|
|
.orderBy(asc(vehicles.createdAt))
|
|
.limit(overfetch);
|
|
|
|
for (const v of noParts) await tryPick(v);
|
|
|
|
// Phase 2 — rolling rescan of ALL decoded vehicles to gap-fill partially-fetched
|
|
// ones. A createdAt cursor walks forward and wraps around at the end.
|
|
if (picked.length < batchSize) {
|
|
const cursorObj = await this.redis.getJson<{ ts: string }>(BACKFILL_CURSOR_KEY);
|
|
const cursor = cursorObj?.ts ? new Date(cursorObj.ts) : new Date(0);
|
|
|
|
const rolling = await this.db
|
|
.select({ id: vehicles.id, source: vehicles.source, createdAt: vehicles.createdAt })
|
|
.from(vehicles)
|
|
.where(and(inArray(vehicles.source, eligible), gt(vehicles.createdAt, cursor)))
|
|
.orderBy(asc(vehicles.createdAt))
|
|
.limit(overfetch);
|
|
|
|
if (rolling.length === 0 && cursor.getTime() > 0) {
|
|
// Reached the end — restart the rescan from the beginning next wave.
|
|
await this.redis.setJson(
|
|
BACKFILL_CURSOR_KEY,
|
|
{ ts: new Date(0).toISOString() },
|
|
30 * 86400,
|
|
);
|
|
}
|
|
|
|
let lastTs: Date | null = null;
|
|
for (const v of rolling) {
|
|
lastTs = v.createdAt;
|
|
await tryPick(v);
|
|
}
|
|
if (lastTs) {
|
|
await this.redis.setJson(BACKFILL_CURSOR_KEY, { ts: lastTs.toISOString() }, 30 * 86400);
|
|
}
|
|
}
|
|
|
|
if (picked.length === 0) {
|
|
this.logger.log("[backfill] No candidates this wave");
|
|
return;
|
|
}
|
|
|
|
for (const v of picked) await this.enqueueInit(v.id, v.source);
|
|
this.logger.log(
|
|
`[backfill] Queued ${picked.length} vehicle(s) (sources=${eligible.join(",")}, backlog=${backlog})`,
|
|
);
|
|
}
|
|
|
|
/** Queue a prefetch-init for a vehicle and set the in-flight guard. */
|
|
private async enqueueInit(vehicleId: string, source: string): Promise<void> {
|
|
await this.queue.add(
|
|
"prefetch-init",
|
|
{ vehicleId, source: source as PrefetchInitJobData["source"] },
|
|
{ removeOnComplete: { count: 1000 }, removeOnFail: { count: 5000 } },
|
|
);
|
|
// Guard cleared on completion (incrementCompleted) or by TTL if the run dies.
|
|
await this.redis.set(`prefetch:scheduled:${vehicleId}`, "1", BACKFILL_SCHEDULED_TTL);
|
|
}
|
|
|
|
// ==================== Helpers ====================
|
|
|
|
private async queueCategoryJob(
|
|
cat: {
|
|
id: string;
|
|
linkPath: string | null;
|
|
source: string;
|
|
unavailable: boolean;
|
|
hasSubgroups?: boolean | null;
|
|
},
|
|
vehicleId: string,
|
|
source: string,
|
|
depth: number,
|
|
): Promise<void> {
|
|
if (cat.unavailable) return;
|
|
|
|
if (this.isLeafLinkPath(cat.linkPath, cat.source, cat.hasSubgroups)) {
|
|
// Leaf — check if already has parts
|
|
const [partCheck] = await this.db
|
|
.select({ id: parts.id })
|
|
.from(parts)
|
|
.where(eq(parts.categoryId, cat.id))
|
|
.limit(1);
|
|
|
|
if (!partCheck && cat.linkPath) {
|
|
await this.addJob("prefetch-parts", {
|
|
vehicleId,
|
|
categoryId: cat.id,
|
|
source: source as "pl24" | "emex",
|
|
action: "parts" as const,
|
|
depth,
|
|
});
|
|
}
|
|
} else if (cat.linkPath) {
|
|
// Check if children already exist
|
|
const [childCheck] = await this.db
|
|
.select({ id: categories.id })
|
|
.from(categories)
|
|
.where(eq(categories.parentId, cat.id))
|
|
.limit(1);
|
|
|
|
if (childCheck) {
|
|
// Already has children — explore them recursively
|
|
const children = await this.db
|
|
.select()
|
|
.from(categories)
|
|
.where(eq(categories.parentId, cat.id));
|
|
|
|
for (const child of children) {
|
|
if (child.unavailable) continue;
|
|
await this.queueCategoryJob(child, vehicleId, source, depth + 1);
|
|
}
|
|
} else {
|
|
await this.addJob("prefetch-children", {
|
|
vehicleId,
|
|
categoryId: cat.id,
|
|
source: source as "pl24" | "emex",
|
|
action: "children" as const,
|
|
depth,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
private isLeafLinkPath(
|
|
linkPath: string | null,
|
|
source: string,
|
|
hasSubgroups?: boolean | null,
|
|
): boolean {
|
|
if (!linkPath) return false;
|
|
// EMEX: Vehicle.aspx group nodes are parents to drill; Unit.aspx (hierarchical
|
|
// tree) and QuickDetails.aspx (legacy flat) leaves carry parts.
|
|
if (source === "emex") return !linkPath.includes("Vehicle.aspx");
|
|
// parts-catalogs: EVERY node has a pcat: linkPath, so the old "pcat: prefix =
|
|
// leaf" rule mis-flagged every mid-group folder as a leaf — the worker fetched
|
|
// its "parts" (a 1-level drill that just revealed sub-groups) and never
|
|
// recursed, leaving deep pcat trees a single level shallow (no parts → empty
|
|
// part counts, dead cross-tree search). Use the captured hasSubgroups flag
|
|
// instead: a node is a parts leaf only when it is NOT a known parent group, so
|
|
// folders queue a children job and the recursion drills to full depth. Unknown
|
|
// flag (null, rare pre-migration rows) → treated as leaf, preserving the old
|
|
// 1-level behaviour for those.
|
|
if (source === "parts-catalogs") return hasSubgroups !== true;
|
|
// PL24 leaf indicators
|
|
return (
|
|
linkPath.includes("/bom/") ||
|
|
linkPath.includes("/bomdetails") ||
|
|
linkPath.includes("/partinfo/") ||
|
|
linkPath.includes("/servicepart/vin_items")
|
|
);
|
|
}
|
|
|
|
private async addJob(name: string, data: PrefetchCategoryJobData): Promise<void> {
|
|
const opts: Record<string, unknown> = {
|
|
// BullMQ rejects custom job IDs containing ":" (its key separator), so use
|
|
// "-" instead. The values are UUIDs — the ID only needs to be deterministic
|
|
// (for dedup), not parseable.
|
|
jobId: `prefetch-${data.vehicleId}-${data.categoryId}-${data.action}`,
|
|
};
|
|
|
|
// parts-catalogs pacing is handled per-job in process() (PCAT_PACE_MS) + the
|
|
// limiter. The old cumulative `index * 20s` delay was pathological (the Nth
|
|
// leaf of a vehicle waited N*20s) and is gone.
|
|
|
|
await this.queue.add(name, data, opts);
|
|
}
|
|
|
|
private async incrementCompleted(vehicleId: string): Promise<void> {
|
|
const progress = await this.redis.getJson<{
|
|
completed: number;
|
|
total: number;
|
|
}>(`prefetch:progress:${vehicleId}`);
|
|
if (!progress) return;
|
|
|
|
const completed = (progress.completed || 0) + 1;
|
|
const isFinished = completed >= progress.total;
|
|
|
|
await updateProgress(this.redis, vehicleId, {
|
|
completed,
|
|
...(isFinished ? { status: "completed" } : {}),
|
|
});
|
|
|
|
if (isFinished) {
|
|
this.logger.log(`[prefetch] Completed all jobs for vehicle=${vehicleId}`);
|
|
// Genuine residue: the whole chain finished but the vehicle still has no
|
|
// parts (all leaves empty / no catalog data). Count it so Phase-1 stops
|
|
// re-picking it every wave.
|
|
const [hasPart] = await this.db
|
|
.select({ id: parts.id })
|
|
.from(parts)
|
|
.where(eq(parts.vehicleId, vehicleId))
|
|
.limit(1);
|
|
if (!hasPart) await this.markNoResult(vehicleId);
|
|
// Clean up Redis keys — data is in PostgreSQL now
|
|
await this.redis.del(`prefetch:scheduled:${vehicleId}`);
|
|
await this.redis.del(`prefetch:progress:${vehicleId}`);
|
|
}
|
|
}
|
|
|
|
private async incrementErrors(vehicleId: string): Promise<void> {
|
|
const progress = await this.redis.getJson<{ errors: number }>(`prefetch:progress:${vehicleId}`);
|
|
if (!progress) return;
|
|
await updateProgress(this.redis, vehicleId, {
|
|
errors: (progress.errors || 0) + 1,
|
|
});
|
|
}
|
|
|
|
private noResultKey(vehicleId: string): string {
|
|
return `prefetch:noresult:${vehicleId}`;
|
|
}
|
|
|
|
/** Record that a backfill attempt finished with the vehicle still at zero parts. */
|
|
private async markNoResult(vehicleId: string): Promise<void> {
|
|
const key = this.noResultKey(vehicleId);
|
|
const n = await this.redis.incr(key);
|
|
await this.redis.expire(key, NORESULT_TTL_S);
|
|
this.logger.debug(`[prefetch] no-result #${n} for vehicle=${vehicleId}`);
|
|
}
|
|
}
|