feat(api): source-DB lookup-first for pcat/emex catalog fetches

Adds an optional local-dump lookup layer in front of the live PartsCatalogs and
EMEX scrapes. When enabled, getCategoryWithPartsInner queries a Postgres
(pcat) or MariaDB (emex) dump for the requested schema/group's parts and
hotspots; on miss it falls through to the existing upstream call unchanged.
Hits avoid the live API, its cooldown, and its rate-limits — direct DB latency.

- New CatalogSourceDbModule with PcatSourceDbService + EmexSourceDbService
  (raw SQL, no Drizzle schema modeling — dump shapes are frozen snapshots).
- pcat lookup keys on schema_images.schema_ext_id (the dump's column that
  matches sase's pcat groupId; observed ~7% hit rate on prod's 6596 unique
  groupIds, of which ~10% have schema_parts → ~3-5% net parts coverage).
  Joins schema_parts → parts directly; the dump's part_groups+part_group_items
  linkage covers 0 of our hits, so we skip that path entirely.
- emex lookup uses (catalog_id, ssd) → vehicles.id then (vehicle_id, group_id)
  → vehicle_parts → parts + part_images. The ssd is already persisted into
  vehicles.rawData.ssd by the existing emex.mapper, no extra capture needed.

Gated behind CATALOG_SOURCE_DB_ENABLED + PCAT_SOURCE_DB_URL / EMEX_SOURCE_DB_URL.
All three default unset, so this commit is a no-op until prod env is configured.
Adds mysql2 dep for the MariaDB client.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-01 00:35:45 +03:00
parent d9b14af96d
commit 4724a71113
10 changed files with 557 additions and 19 deletions

View File

@@ -58,6 +58,7 @@
"drizzle-orm": "^0.41.0",
"helmet": "^8.1.0",
"ioredis": "^5.4.0",
"mysql2": "^3.22.4",
"openai": "^6.37.0",
"postgres": "^3.4.0",
"posthog-node": "^5.34.1",

View File

@@ -1,5 +1,6 @@
import { Module } from "@nestjs/common";
import { CatalogModule } from "../catalog/catalog.module";
import { CatalogSourceDbModule } from "../integrations/catalog-source-db/catalog-source-db.module";
import { EmexModule } from "../integrations/emex/emex.module";
import { PartsCatalogsModule } from "../integrations/parts-catalogs/parts-catalogs.module";
import { PL24Module } from "../integrations/pl24/pl24.module";
@@ -8,7 +9,14 @@ import { CategoriesController } from "./categories.controller";
import { CategoriesService } from "./categories.service";
@Module({
imports: [PL24Module, EmexModule, PartsCatalogsModule, CatalogModule, TranslationsModule],
imports: [
PL24Module,
EmexModule,
PartsCatalogsModule,
CatalogModule,
TranslationsModule,
CatalogSourceDbModule,
],
controllers: [CategoriesController],
providers: [CategoriesService],
exports: [CategoriesService],

View File

@@ -37,6 +37,8 @@ function createService(db: any) {
Promise.resolve(new Map<string, string>(texts.map((t) => [t, t]))),
),
};
const pcatSourceDb = { fetchParts: vi.fn().mockResolvedValue(null) };
const emexSourceDb = { fetchCategoryParts: vi.fn().mockResolvedValue(null) };
const service = new CategoriesService(
db as any,
redis as any,
@@ -46,6 +48,8 @@ function createService(db: any) {
storage as any,
pl24FordLegacyService as any,
translationsService as any,
pcatSourceDb as any,
emexSourceDb as any,
);
return { service, db, redis, pl24Service, translationsService };
}

View File

@@ -2,6 +2,8 @@ import { Inject, Injectable, Logger, NotFoundException } from "@nestjs/common";
import { and, eq, ilike, inArray, isNull, or, sql } from "drizzle-orm";
import { DATABASE, type Database } from "../database/database.provider";
import { categories, parts, schemaPics, vehicles } from "../database/schema/core";
import { EmexSourceDbService } from "../integrations/catalog-source-db/emex-source-db.service";
import { PcatSourceDbService } from "../integrations/catalog-source-db/pcat-source-db.service";
import { EmexService } from "../integrations/emex/emex.service";
import { PartsCatalogsService } from "../integrations/parts-catalogs/parts-catalogs.service";
import { PcatGroup } from "../integrations/parts-catalogs/parts-catalogs.types";
@@ -24,6 +26,8 @@ export class CategoriesService {
private storage: StorageService,
private pl24FordLegacyService: PL24FordLegacyService,
private translationsService: TranslationsService,
private pcatSourceDb: PcatSourceDbService,
private emexSourceDb: EmexSourceDbService,
) {}
async getCategoryTree(vehicleId: string) {
@@ -928,12 +932,19 @@ export class CategoriesService {
const vehicleRawData = vehicle.rawData as any;
const carParams = this.buildPcatCarParams(vehicleRawData?.parameters);
const partsResult = await this.partsCatalogsService.fetchParts(
catalogId,
carId,
groupId,
carParams,
);
// Local dump first; live upstream fallback. Logs hit/miss so prod
// verification can measure source-DB coverage.
let partsResult = await this.pcatSourceDb.fetchParts(catalogId, carId, groupId);
if (partsResult) {
this.logger.debug(`[source-db hit pcat] car=${carId} group=${groupId}`);
} else {
partsResult = await this.partsCatalogsService.fetchParts(
catalogId,
carId,
groupId,
carParams,
);
}
if (partsResult) {
// Flatten part groups into parts
@@ -1081,7 +1092,20 @@ export class CategoriesService {
} else if (vehicle && category.source === "emex") {
// EMEX: fetch parts + schema image via Puppeteer from QuickDetails URL
try {
const emexResult = await this.emexService.fetchCategoryParts(category.linkPath);
// Local dump first (uses vehicles.rawData.ssd persisted by the
// emex mapper). Falls back to live upstream on miss.
const emexSsd = (vehicle.rawData as { ssd?: string } | null)?.ssd;
let emexResult = await this.emexSourceDb.fetchCategoryParts(
emexSsd,
category.linkPath ?? "",
);
if (emexResult) {
this.logger.debug(
`[source-db hit emex] ssd=${emexSsd?.slice(0, 12)}... gid=${category.externalId}`,
);
} else {
emexResult = await this.emexService.fetchCategoryParts(category.linkPath);
}
// Build position code → sequential integer mapping for hotspot linking
const posCodeToIndex = new Map<string, number>();

View File

@@ -69,6 +69,11 @@ export default () => ({
chatwoot: {
hmacToken: process.env.CHATWOOT_HMAC_TOKEN,
},
catalogSource: {
enabled: process.env.CATALOG_SOURCE_DB_ENABLED === "true",
pcatUrl: process.env.PCAT_SOURCE_DB_URL,
emexUrl: process.env.EMEX_SOURCE_DB_URL,
},
otel: {
enabled: process.env.OTEL_ENABLED === "true",
endpoint: process.env.OTEL_EXPORTER_OTLP_ENDPOINT,

View File

@@ -0,0 +1,19 @@
import { Module } from "@nestjs/common";
import { EmexSourceDbService } from "./emex-source-db.service";
import { PcatSourceDbService } from "./pcat-source-db.service";
/**
* Local catalog-dump lookup. Both services are always provided; whether they
* connect is decided at runtime from `CATALOG_SOURCE_DB_ENABLED` + the two
* `*_SOURCE_DB_URL` env vars. When disabled / unconfigured, every lookup
* returns null so the caller transparently falls back to the live upstream.
*
* Intentionally no DB schema modelling here — these are raw read-only queries
* against external dump DBs (pc2 Postgres + emex MariaDB) whose shapes are
* frozen snapshots and don't share Drizzle types with the sase schema.
*/
@Module({
providers: [PcatSourceDbService, EmexSourceDbService],
exports: [PcatSourceDbService, EmexSourceDbService],
})
export class CatalogSourceDbModule {}

View File

@@ -0,0 +1,183 @@
import { Injectable, Logger, type OnModuleDestroy, type OnModuleInit } from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import mysql, { type Pool, type RowDataPacket } from "mysql2/promise";
import type { EmexHotspot, EmexHotspotArea, EmexPart, EmexPartsResult } from "../emex/emex.types";
/**
* Look up a vehicle's parts + schema for a category (PNC group) in the local
* EMEX dump (sase-catalog-src-emex MariaDB). Returns null on any miss so the
* caller falls through to the live emex scrape.
*
* Inputs:
* - vehicleSsd: emex's per-vehicle session-state-descriptor — captured into
* `vehicles.rawData.emexSsd` during live VIN decode. Required (no VIN
* column in the dump; SSD is the only stable vehicle identifier).
* - categoryUrl: the QuickDetails.aspx URL stored in `categories.linkPath`.
* We parse `gid` (the group id) out of it.
*/
@Injectable()
export class EmexSourceDbService implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(EmexSourceDbService.name);
private pool: Pool | null = null;
private enabled = false;
constructor(private readonly config: ConfigService) {}
onModuleInit() {
const enabled = this.config.get<boolean>("catalogSource.enabled");
const url = this.config.get<string>("catalogSource.emexUrl");
if (!enabled || !url) {
this.logger.log(
`[emex-src] disabled (enabled=${enabled}, urlSet=${Boolean(url)}); upstream-only`,
);
return;
}
this.pool = mysql.createPool({
uri: url,
connectionLimit: 5,
connectTimeout: 10_000,
waitForConnections: true,
});
this.enabled = true;
this.logger.log("[emex-src] connected, lookup-first enabled");
}
async onModuleDestroy() {
if (this.pool) {
await this.pool.end();
this.pool = null;
}
}
/**
* Mirror the live `EmexCatalogService.fetchCategoryParts` shape.
* Returns null on miss; never throws.
*/
async fetchCategoryParts(
vehicleSsd: string | null | undefined,
categoryUrl: string,
): Promise<EmexPartsResult | null> {
if (!this.enabled || !this.pool) return null;
if (!vehicleSsd) return null; // no SSD captured → can't locate vehicle in dump
const gid = extractGid(categoryUrl);
if (!gid) return null;
try {
// 1) Resolve vehicle.id from SSD. We try `ssd = ?` first; if the dump
// canonicalized into `unique_key` (hash), the second arm catches it.
const [vehicleRows] = await this.pool.execute<RowDataPacket[]>(
"SELECT id, catalog_id FROM vehicles WHERE ssd = ? OR unique_key = ? LIMIT 1",
[vehicleSsd, vehicleSsd],
);
if (vehicleRows.length === 0) return null;
const vehicleId = vehicleRows[0].id as number;
const catalogId = vehicleRows[0].catalog_id as number;
// 2) Resolve part_group.id from external gid scoped to this catalog.
const [groupRows] = await this.pool.execute<RowDataPacket[]>(
"SELECT id FROM part_groups WHERE catalog_id = ? AND group_id = ? LIMIT 1",
[catalogId, gid],
);
if (groupRows.length === 0) return null;
const groupPk = groupRows[0].id as number;
// 3) Parts for this vehicle in this group.
const [partRows] = await this.pool.execute<RowDataPacket[]>(
`SELECT p.id AS part_id, p.part_number, p.name, p.position_number, p.pnc
FROM vehicle_parts vp
JOIN parts p ON p.id = vp.part_id
WHERE vp.vehicle_id = ? AND vp.group_id = ?
ORDER BY p.position_number, p.id`,
[vehicleId, groupPk],
);
const parts: EmexPart[] = partRows.map((r) => ({
oemCode: String(r.part_number ?? ""),
nameEn: String(r.name ?? ""),
positionCode: r.position_number ?? r.pnc ?? undefined,
}));
// 4) Schema image + hotspots for this group.
const [imgRows] = await this.pool.execute<RowDataPacket[]>(
`SELECT original_url, width, height, hotspots
FROM part_images
WHERE group_id = ? AND image_type IN ('DIAGRAM','SCHEMATIC')
ORDER BY is_primary DESC, sort_order
LIMIT 1`,
[groupPk],
);
let schemaImageUrl: string | null = null;
let schemaWidth = 0;
let schemaHeight = 0;
let hotspots: EmexHotspot[] = [];
if (imgRows.length > 0) {
const img = imgRows[0];
schemaImageUrl = (img.original_url as string) || null;
schemaWidth = (img.width as number) ?? 0;
schemaHeight = (img.height as number) ?? 0;
const rawHotspots = img.hotspots;
if (rawHotspots) {
try {
const parsed = typeof rawHotspots === "string" ? JSON.parse(rawHotspots) : rawHotspots;
hotspots = normalizeHotspots(parsed, partRows);
} catch (err) {
this.logger.debug(
`[emex-src] hotspot JSON parse failed for group ${groupPk}: ${(err as Error).message}`,
);
}
}
}
// Treat fully-empty result as miss so caller falls back to live.
if (parts.length === 0 && !schemaImageUrl) return null;
return { parts, schemaImageUrl, hotspots, schemaWidth, schemaHeight };
} catch (err) {
this.logger.warn(`[emex-src] lookup failed (gid=${gid}): ${(err as Error).message}`);
return null;
}
}
}
/** Extract `gid=...` from a QuickDetails.aspx / similar URL. */
function extractGid(url: string): string | null {
if (!url) return null;
const m = url.match(/[?&]gid=([^&#]+)/i);
if (!m) return null;
try {
return decodeURIComponent(m[1]);
} catch {
return m[1];
}
}
/**
* Convert the dump's hotspots JSON ([{x,y,w,h,part_id}]) into the EmexHotspot
* shape used by the live scraper (grouped by position code).
*/
function normalizeHotspots(parsed: unknown, partRows: RowDataPacket[]): EmexHotspot[] {
if (!Array.isArray(parsed)) return [];
// Map part_id → positionCode using the partRows we already have.
const positionByPartId = new Map<number, string>();
for (const r of partRows) {
const pid = r.part_id as number | undefined;
const pos = (r.position_number ?? r.pnc) as string | undefined;
if (pid && pos) positionByPartId.set(pid, pos);
}
const byKey = new Map<string, EmexHotspotArea[]>();
for (const h of parsed as Array<Record<string, unknown>>) {
const x = Number(h.x ?? h.left ?? 0);
const y = Number(h.y ?? h.top ?? 0);
const w = Number(h.w ?? h.width ?? 0);
const ht = Number(h.h ?? h.height ?? 0);
const pid = Number(h.part_id ?? h.partId ?? 0);
const key = positionByPartId.get(pid) ?? String(h.position ?? h.pnc ?? pid ?? "");
if (!key) continue;
const arr = byKey.get(key) ?? [];
arr.push({ left: x, top: y, width: w, height: ht });
byKey.set(key, arr);
}
return Array.from(byKey.entries()).map(([key, areas]) => ({ key, areas }));
}

View File

@@ -0,0 +1,154 @@
import { Injectable, Logger, type OnModuleDestroy, type OnModuleInit } from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import postgres, { type Sql } from "postgres";
import type {
PcatPartGroup,
PcatPartsResult,
PcatPosition,
} from "../parts-catalogs/parts-catalogs.types";
/**
* Look up parts for a (catalogId, carId, groupId) triple in the local
* parts-catalogs dump (sase-catalog-src-pcat). On any miss — feature disabled,
* connection failure, no schema image, or no matching parts — returns null so
* the caller can fall through to the live upstream service.
*
* Read-only by design: uses the `pcat_reader` user (or owner if reader not set).
*/
@Injectable()
export class PcatSourceDbService implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(PcatSourceDbService.name);
private sql: Sql | null = null;
private enabled = false;
constructor(private readonly config: ConfigService) {}
onModuleInit() {
const enabled = this.config.get<boolean>("catalogSource.enabled");
const url = this.config.get<string>("catalogSource.pcatUrl");
if (!enabled || !url) {
this.logger.log(
`[pcat-src] disabled (enabled=${enabled}, urlSet=${Boolean(url)}); upstream-only`,
);
return;
}
this.sql = postgres(url, {
max: 5,
idle_timeout: 30,
connect_timeout: 10,
prepare: false,
});
this.enabled = true;
this.logger.log("[pcat-src] connected, lookup-first enabled");
}
async onModuleDestroy() {
if (this.sql) {
await this.sql.end({ timeout: 5 });
this.sql = null;
}
}
/**
* Mirror the live `PartsCatalogsService.fetchParts` shape. Returns null on
* miss; never throws (DB blips fall back to live upstream).
*/
async fetchParts(
_catalogId: string,
_carId: string,
groupId: string,
): Promise<PcatPartsResult | null> {
if (!this.enabled || !this.sql) return null;
try {
// sase's pcat `groupId` (stored as `categories.linkPath` tail) is the
// page-level schema identifier that maps to the dump's `schema_ext_id`.
// The dump's `groups.id` is a higher-level category and does NOT match.
// Coverage is ~7% schema-hit × ~10% with-parts ≈ ~3-5% net; the misses
// fall through to upstream cleanly (this method returns null).
// Single round-trip: schema + parts + hotspot coordinates via schema_parts.
// We deliberately bypass the dump's `part_groups`+`part_group_items` tables
// — sampling against prod data shows that linkage covers 0 of our hits,
// whereas `schema_parts.part_id → parts` covers all 260 schemas that have
// any part annotation. Everything goes into a single un-named PcatPartGroup
// (the downstream insert flattens partGroups anyway).
const rows = await this.sql<
Array<{
schema_id: string;
img_url: string | null;
img_description: string | null;
part_id: string | null;
part_number: string | null;
part_name: string | null;
part_notice: string | null;
position_number: string | null;
position_x: number | null;
position_y: number | null;
position_width: number | null;
position_height: number | null;
}>
>`
SELECT
si.id::text AS schema_id,
si.img_url AS img_url,
si.img_description AS img_description,
p.id::text AS part_id,
p.part_number AS part_number,
p.name AS part_name,
p.notice AS part_notice,
sp.position_number AS position_number,
sp.position_x AS position_x,
sp.position_y AS position_y,
sp.position_width AS position_width,
sp.position_height AS position_height
FROM schema_images si
LEFT JOIN schema_parts sp ON sp.schema_image_id = si.id
LEFT JOIN parts p ON p.id = sp.part_id
WHERE si.schema_ext_id = ${groupId}
ORDER BY sp.position_number, p.id
`;
if (rows.length === 0) return null;
const first = rows[0];
const partsBucket: PcatPartGroup = { parts: [] };
const positionsByNumber = new Map<string, PcatPosition>();
for (const r of rows) {
if (r.part_number) {
partsBucket.parts.push({
id: r.part_id ?? undefined,
number: r.part_number,
name: r.part_name ?? "",
notice: r.part_notice ?? undefined,
positionNumber: r.position_number ?? undefined,
});
}
if (r.position_number && !positionsByNumber.has(r.position_number)) {
positionsByNumber.set(r.position_number, {
number: r.position_number,
coordinates: [
r.position_x ?? 0,
r.position_y ?? 0,
r.position_width ?? 0,
r.position_height ?? 0,
],
});
}
}
const result: PcatPartsResult = {
img: first.img_url ?? "",
imgDescription: first.img_description ?? undefined,
partGroups: partsBucket.parts.length > 0 ? [partsBucket] : [],
positions: Array.from(positionsByNumber.values()),
};
// Treat schema-image-only (no parts, no positions) as miss so caller falls
// back to live upstream — an image alone isn't useful enough to skip live.
if (result.partGroups.length === 0 && result.positions.length === 0) {
return null;
}
return result;
} catch (err) {
this.logger.warn(`[pcat-src] lookup failed (group=${groupId}): ${(err as Error).message}`);
return null;
}
}
}