feat: add parts catalogs integration, catalog prefetch worker, and vehicle select modal
Integrate external parts catalogs API with auth service, add BullMQ-based catalog prefetch worker for background data caching, expand vehicles service with shared vehicle support, and add vehicle select modal to frontend. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Sase Dev
@@ -69,6 +69,7 @@
|
||||
"playwright": "^1.50.0",
|
||||
"tsx": "^4.19.0",
|
||||
"typescript": "^5.7.0",
|
||||
"undici": "^7.22.0",
|
||||
"vitest": "^3.0.0"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,9 +3,10 @@ import { CategoriesController } from "./categories.controller";
|
||||
import { CategoriesService } from "./categories.service";
|
||||
import { PL24Module } from "../integrations/pl24/pl24.module";
|
||||
import { EmexModule } from "../integrations/emex/emex.module";
|
||||
import { PartsCatalogsModule } from "../integrations/parts-catalogs/parts-catalogs.module";
|
||||
|
||||
@Module({
|
||||
imports: [PL24Module, EmexModule],
|
||||
imports: [PL24Module, EmexModule, PartsCatalogsModule],
|
||||
controllers: [CategoriesController],
|
||||
providers: [CategoriesService],
|
||||
exports: [CategoriesService],
|
||||
|
||||
@@ -18,7 +18,11 @@ function createService(db: any) {
|
||||
const storage = {
|
||||
upload: vi.fn().mockResolvedValue("https://storage.sase.tr/sase-schemas/test.png"),
|
||||
};
|
||||
const service = new CategoriesService(db as any, redis as any, pl24Service as any, emexService as any, storage as any);
|
||||
const partsCatalogsService = {
|
||||
fetchGroups: vi.fn().mockResolvedValue([]),
|
||||
fetchParts: vi.fn().mockResolvedValue(null),
|
||||
};
|
||||
const service = new CategoriesService(db as any, redis as any, pl24Service as any, emexService as any, partsCatalogsService as any, storage as any);
|
||||
return { service, db, redis, pl24Service };
|
||||
}
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ import { categories, vehicles, schemaPics, parts } from "../database/schema/core
|
||||
import { RedisService } from "../redis/redis.service";
|
||||
import { PL24Service } from "../integrations/pl24/pl24.service";
|
||||
import { EmexService } from "../integrations/emex/emex.service";
|
||||
import { PartsCatalogsService } from "../integrations/parts-catalogs/parts-catalogs.service";
|
||||
import { StorageService } from "../storage/storage.service";
|
||||
|
||||
@Injectable()
|
||||
@@ -16,6 +17,7 @@ export class CategoriesService {
|
||||
private redis: RedisService,
|
||||
private pl24Service: PL24Service,
|
||||
private emexService: EmexService,
|
||||
private partsCatalogsService: PartsCatalogsService,
|
||||
private storage: StorageService,
|
||||
) {}
|
||||
|
||||
@@ -80,6 +82,46 @@ export class CategoriesService {
|
||||
}
|
||||
}
|
||||
|
||||
// If still no categories, try PartsCatalogs
|
||||
if (dbCategories.length === 0 && vehicle.rawData) {
|
||||
const rawData = vehicle.rawData as any;
|
||||
if (rawData?.source === "parts-catalogs" && rawData.catalogId && rawData.carId) {
|
||||
try {
|
||||
const carParams = this.buildPcatCarParams(rawData.parameters);
|
||||
const groups = await this.partsCatalogsService.fetchGroups(
|
||||
rawData.catalogId,
|
||||
rawData.carId,
|
||||
undefined,
|
||||
carParams,
|
||||
);
|
||||
|
||||
if (groups.length > 0) {
|
||||
const insertData = groups.map((g) => ({
|
||||
vehicleId,
|
||||
name: g.name,
|
||||
nameOriginal: g.name,
|
||||
parentId: null as string | null,
|
||||
externalId: g.id,
|
||||
linkPath: `pcat:${rawData.catalogId}:${rawData.carId}:${g.id}`,
|
||||
linkWid: null as string | null,
|
||||
source: "parts-catalogs" as const,
|
||||
}));
|
||||
|
||||
dbCategories = await this.db.insert(categories).values(insertData).onConflictDoNothing().returning();
|
||||
if (dbCategories.length < insertData.length) {
|
||||
dbCategories = await this.db
|
||||
.select()
|
||||
.from(categories)
|
||||
.where(eq(categories.vehicleId, vehicleId));
|
||||
}
|
||||
this.logger.log(`Stored ${dbCategories.length} PartsCatalogs categories for ${vehicle.vin}`);
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.warn(`PartsCatalogs category fetch failed for ${vehicleId}: ${(err as Error).message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If still no categories, try EMEX fallback
|
||||
if (dbCategories.length === 0 && vehicle.vin) {
|
||||
this.logger.log(`No PL24 categories for ${vehicle.vin}, trying EMEX fallback`);
|
||||
@@ -219,6 +261,51 @@ export class CategoriesService {
|
||||
const catalogInfo = rawData?.catalogInfo;
|
||||
const linkPath = category.linkPath;
|
||||
|
||||
// PartsCatalogs subgroups (on-demand drill-down)
|
||||
if (category.source === "parts-catalogs" && rawData?.source === "parts-catalogs" && category.externalId) {
|
||||
try {
|
||||
const carParams = this.buildPcatCarParams(rawData.parameters);
|
||||
const subGroups = await this.partsCatalogsService.fetchGroups(
|
||||
rawData.catalogId,
|
||||
rawData.carId,
|
||||
category.externalId,
|
||||
carParams,
|
||||
);
|
||||
|
||||
if (subGroups.length > 0) {
|
||||
const insertData = subGroups.map((g) => ({
|
||||
vehicleId: category.vehicleId,
|
||||
name: g.name,
|
||||
nameOriginal: g.name,
|
||||
parentId: categoryId,
|
||||
externalId: g.id,
|
||||
linkPath: `pcat:${rawData.catalogId}:${rawData.carId}:${g.id}`,
|
||||
linkWid: null as string | null,
|
||||
source: "parts-catalogs" as const,
|
||||
}));
|
||||
|
||||
children = await this.db
|
||||
.insert(categories)
|
||||
.values(insertData)
|
||||
.onConflictDoNothing()
|
||||
.returning();
|
||||
|
||||
if (children.length < insertData.length) {
|
||||
children = await this.db
|
||||
.select()
|
||||
.from(categories)
|
||||
.where(eq(categories.parentId, categoryId));
|
||||
}
|
||||
|
||||
await this.redis.del(`cat:tree:${category.vehicleId}`);
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.error(`PartsCatalogs subgroup fetch failed for ${categoryId} (externalId=${category.externalId}): ${(err as Error).message}`);
|
||||
}
|
||||
|
||||
return children.length > 0 ? this.enrichWithSchemaImages(children) : children;
|
||||
}
|
||||
|
||||
if (!catalogInfo?.serviceName || !linkPath) {
|
||||
return [];
|
||||
}
|
||||
@@ -290,11 +377,16 @@ export class CategoriesService {
|
||||
if (!category) throw new NotFoundException("Kategori bulunamadı");
|
||||
|
||||
// Check if this category has children
|
||||
const children = await this.db
|
||||
let children = await this.db
|
||||
.select()
|
||||
.from(categories)
|
||||
.where(eq(categories.parentId, categoryId));
|
||||
|
||||
// For parts-catalogs categories with no DB children, fetch from API first
|
||||
if (children.length === 0 && category.source === "parts-catalogs" && category.linkPath?.startsWith("pcat:")) {
|
||||
children = await this.getChildren(categoryId);
|
||||
}
|
||||
|
||||
if (children.length > 0) {
|
||||
return {
|
||||
id: category.id,
|
||||
@@ -333,7 +425,117 @@ export class CategoriesService {
|
||||
.where(eq(vehicles.id, category.vehicleId))
|
||||
.limit(1);
|
||||
|
||||
if (vehicle && category.source === "emex") {
|
||||
if (vehicle && category.source === "parts-catalogs" && category.linkPath?.startsWith("pcat:")) {
|
||||
// PartsCatalogs: fetch parts + schema image via API
|
||||
try {
|
||||
// Format: pcat:{catalogId}:{carId}:{groupId} — split only on first 3 colons
|
||||
const withoutPrefix = category.linkPath.slice("pcat:".length);
|
||||
const firstColon = withoutPrefix.indexOf(":");
|
||||
const secondColon = withoutPrefix.indexOf(":", firstColon + 1);
|
||||
const catalogId = withoutPrefix.slice(0, firstColon);
|
||||
const carId = withoutPrefix.slice(firstColon + 1, secondColon);
|
||||
const groupId = withoutPrefix.slice(secondColon + 1);
|
||||
const vehicleRawData = vehicle.rawData as any;
|
||||
const carParams = this.buildPcatCarParams(vehicleRawData?.parameters);
|
||||
|
||||
const partsResult = await this.partsCatalogsService.fetchParts(
|
||||
catalogId,
|
||||
carId,
|
||||
groupId,
|
||||
carParams,
|
||||
);
|
||||
|
||||
if (partsResult) {
|
||||
// Flatten part groups into parts
|
||||
if (needParts) {
|
||||
const allParts: Array<typeof parts.$inferInsert> = [];
|
||||
|
||||
for (const pg of partsResult.partGroups) {
|
||||
for (const p of pg.parts) {
|
||||
if (!p.number) continue;
|
||||
const posNum = p.positionNumber || pg.positionNumber || null;
|
||||
allParts.push({
|
||||
vehicleId: vehicle.id,
|
||||
categoryId,
|
||||
oemCode: p.number,
|
||||
name: p.name || "Unknown",
|
||||
nameOriginal: p.name || null,
|
||||
description: p.notice || null,
|
||||
quantity: null,
|
||||
position: posNum,
|
||||
hotspotIndex: posNum ? parseInt(posNum, 10) || null : null,
|
||||
unavailable: false,
|
||||
remark: null as string | null,
|
||||
modelCodes: null as string | null,
|
||||
presel: false,
|
||||
source: "parts-catalogs" as const,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (allParts.length > 0) {
|
||||
dbParts = await this.db.insert(parts).values(allParts).returning();
|
||||
this.logger.log(`Stored ${dbParts.length} PartsCatalogs parts for category ${categoryId}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Download schema image and store hotspots
|
||||
if (needImage && partsResult.img) {
|
||||
try {
|
||||
const imgUrl = partsResult.img.startsWith("//") ? `https:${partsResult.img}` : partsResult.img;
|
||||
const imgResp = await fetch(imgUrl, {
|
||||
signal: AbortSignal.timeout(15000),
|
||||
});
|
||||
if (imgResp.ok) {
|
||||
const buf = Buffer.from(await imgResp.arrayBuffer());
|
||||
const ext = partsResult.img.includes(".gif") ? "gif" : "png";
|
||||
const key = `schemas/pcat-${categoryId}.${ext}`;
|
||||
const minioUrl = await this.storage.upload(key, buf, `image/${ext}`);
|
||||
|
||||
const dims = this.getImageDimensions(buf, ext);
|
||||
|
||||
// Convert positions to hotspot format
|
||||
const hotspotItems = partsResult.positions.map((pos) => ({
|
||||
key: pos.number,
|
||||
label: pos.number,
|
||||
areas: [
|
||||
{
|
||||
left: pos.coordinates[0],
|
||||
top: pos.coordinates[1],
|
||||
width: pos.coordinates[2],
|
||||
height: pos.coordinates[3],
|
||||
},
|
||||
],
|
||||
}));
|
||||
|
||||
const [inserted] = await this.db
|
||||
.insert(schemaPics)
|
||||
.values({
|
||||
categoryId,
|
||||
imageUrl: minioUrl,
|
||||
hotspots: JSON.stringify({
|
||||
width: dims.width,
|
||||
height: dims.height,
|
||||
items: hotspotItems,
|
||||
}),
|
||||
source: "parts-catalogs",
|
||||
})
|
||||
.returning();
|
||||
|
||||
pics.push(inserted);
|
||||
this.logger.log(
|
||||
`Stored PartsCatalogs schema image for category ${categoryId}: ${minioUrl}`,
|
||||
);
|
||||
}
|
||||
} catch (imgErr) {
|
||||
this.logger.warn(`Failed to download PC schema image: ${(imgErr as Error).message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.error(`Failed to fetch PartsCatalogs parts for category ${categoryId}: ${(err as Error).message}`);
|
||||
}
|
||||
} 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);
|
||||
@@ -585,9 +787,12 @@ export class CategoriesService {
|
||||
const dbChildCount = childCountMap.get(c.id) || 0;
|
||||
// EMEX: leaf only if linkPath exists and no DB children
|
||||
// PL24: leaf if BOM/servicepart-items linkPath, or no linkPath and no DB children
|
||||
// PartsCatalogs: leaf if linkPath starts with "pcat:" and no DB children
|
||||
const isLeaf = c.source === "emex"
|
||||
? (!!c.linkPath && dbChildCount === 0)
|
||||
: (c.linkPath?.includes("/bom/") || c.linkPath?.includes("/bomdetails") || c.linkPath?.includes("/partinfo/") || c.linkPath?.includes("/servicepart/vin_items") || (!c.linkPath && dbChildCount === 0));
|
||||
: c.source === "parts-catalogs"
|
||||
? (!!c.linkPath?.startsWith("pcat:") && dbChildCount === 0)
|
||||
: (c.linkPath?.includes("/bom/") || c.linkPath?.includes("/bomdetails") || c.linkPath?.includes("/partinfo/") || c.linkPath?.includes("/servicepart/vin_items") || (!c.linkPath && dbChildCount === 0));
|
||||
return {
|
||||
...c,
|
||||
schemaImageUrl: picMap.get(c.id) || null,
|
||||
@@ -596,6 +801,21 @@ export class CategoriesService {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Build query params from PartsCatalogs car parameters.
|
||||
* Parameters are [{key, idx, value}] — API expects {key: idx} as query params.
|
||||
*/
|
||||
private buildPcatCarParams(parameters?: Array<{ key: string; idx: string; value: string }>): Record<string, string> {
|
||||
if (!parameters || !Array.isArray(parameters)) return {};
|
||||
const params: Record<string, string> = {};
|
||||
for (const p of parameters) {
|
||||
if (p.key && p.idx) {
|
||||
params[p.key] = p.idx;
|
||||
}
|
||||
}
|
||||
return params;
|
||||
}
|
||||
|
||||
private getImageDimensions(buf: Buffer, ext: string): { width: number; height: number } {
|
||||
try {
|
||||
if (ext === 'gif' && buf.length >= 10) {
|
||||
|
||||
@@ -7,17 +7,6 @@ export interface TransformedResponse<T> {
|
||||
meta?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
function stripSource(data: unknown): unknown {
|
||||
if (Array.isArray(data)) return data.map(stripSource);
|
||||
if (data && typeof data === "object" && !(data instanceof Date)) {
|
||||
const { source, ...rest } = data as Record<string, unknown>;
|
||||
return Object.fromEntries(
|
||||
Object.entries(rest).map(([k, v]) => [k, stripSource(v)]),
|
||||
);
|
||||
}
|
||||
return data;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class TransformInterceptor<T> implements NestInterceptor<T, TransformedResponse<T>> {
|
||||
intercept(
|
||||
@@ -28,21 +17,21 @@ export class TransformInterceptor<T> implements NestInterceptor<T, TransformedRe
|
||||
map((data) => {
|
||||
// If already wrapped, pass through
|
||||
if (data && typeof data === "object" && "success" in data) {
|
||||
return stripSource(data) as TransformedResponse<T>;
|
||||
return data as TransformedResponse<T>;
|
||||
}
|
||||
|
||||
// Handle pagination response
|
||||
if (data && typeof data === "object" && "items" in data && "meta" in data) {
|
||||
return {
|
||||
success: true,
|
||||
data: stripSource(data.items),
|
||||
data: data.items,
|
||||
meta: data.meta,
|
||||
} as TransformedResponse<T>;
|
||||
}
|
||||
|
||||
return {
|
||||
success: true,
|
||||
data: stripSource(data),
|
||||
data,
|
||||
} as TransformedResponse<T>;
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -231,14 +231,11 @@ export const oemCodeCopies = pgTable(
|
||||
],
|
||||
);
|
||||
|
||||
// ─── Vehicles ───────────────────────────────────────
|
||||
// ─── Vehicles (shared config — one record per VIN) ──
|
||||
export const vehicles = pgTable(
|
||||
"vehicles",
|
||||
{
|
||||
id: uuid("id").primaryKey().defaultRandom(),
|
||||
userId: uuid("user_id")
|
||||
.notNull()
|
||||
.references(() => users.id, { onDelete: "cascade" }),
|
||||
vin: varchar("vin", { length: 17 }).notNull(),
|
||||
brandId: uuid("brand_id").references(() => brands.id),
|
||||
brandName: varchar("brand_name", { length: 100 }),
|
||||
@@ -254,9 +251,27 @@ export const vehicles = pgTable(
|
||||
updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
(table) => [
|
||||
index("vehicles_user_id_idx").on(table.userId),
|
||||
index("vehicles_vin_idx").on(table.vin),
|
||||
uniqueIndex("vehicles_user_vin_idx").on(table.userId, table.vin),
|
||||
uniqueIndex("vehicles_vin_unique_idx").on(table.vin),
|
||||
],
|
||||
);
|
||||
|
||||
// ─── User Vehicles (junction — user ↔ shared vehicle) ─
|
||||
export const userVehicles = pgTable(
|
||||
"user_vehicles",
|
||||
{
|
||||
id: uuid("id").primaryKey().defaultRandom(),
|
||||
userId: uuid("user_id")
|
||||
.notNull()
|
||||
.references(() => users.id, { onDelete: "cascade" }),
|
||||
vehicleId: uuid("vehicle_id")
|
||||
.notNull()
|
||||
.references(() => vehicles.id, { onDelete: "cascade" }),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
lastAccessedAt: timestamp("last_accessed_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
(table) => [
|
||||
uniqueIndex("user_vehicles_user_vehicle_idx").on(table.userId, table.vehicleId),
|
||||
index("user_vehicles_user_id_idx").on(table.userId),
|
||||
],
|
||||
);
|
||||
|
||||
@@ -271,7 +286,7 @@ export const categories = pgTable(
|
||||
name: varchar("name", { length: 500 }).notNull(),
|
||||
nameOriginal: varchar("name_original", { length: 500 }),
|
||||
parentId: uuid("parent_id"),
|
||||
externalId: varchar("external_id", { length: 100 }),
|
||||
externalId: varchar("external_id", { length: 500 }),
|
||||
linkPath: text("link_path"),
|
||||
linkWid: varchar("link_wid", { length: 100 }),
|
||||
unavailable: boolean("unavailable").default(false).notNull(),
|
||||
@@ -285,24 +300,6 @@ export const categories = pgTable(
|
||||
],
|
||||
);
|
||||
|
||||
// ─── Vehicle Categories (junction) ──────────────────
|
||||
export const vehicleCategories = pgTable(
|
||||
"vehicle_categories",
|
||||
{
|
||||
id: uuid("id").primaryKey().defaultRandom(),
|
||||
vehicleId: uuid("vehicle_id")
|
||||
.notNull()
|
||||
.references(() => vehicles.id, { onDelete: "cascade" }),
|
||||
categoryId: uuid("category_id")
|
||||
.notNull()
|
||||
.references(() => categories.id, { onDelete: "cascade" }),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
(table) => [
|
||||
uniqueIndex("vehicle_categories_unique_idx").on(table.vehicleId, table.categoryId),
|
||||
],
|
||||
);
|
||||
|
||||
// ─── Parts ──────────────────────────────────────────
|
||||
export const parts = pgTable(
|
||||
"parts",
|
||||
|
||||
82
apps/api/src/database/schema/parts-catalogs.ts
Normal file
82
apps/api/src/database/schema/parts-catalogs.ts
Normal file
@@ -0,0 +1,82 @@
|
||||
import {
|
||||
pgTable,
|
||||
uuid,
|
||||
varchar,
|
||||
text,
|
||||
boolean,
|
||||
serial,
|
||||
timestamp,
|
||||
jsonb,
|
||||
index,
|
||||
uniqueIndex,
|
||||
} from "drizzle-orm/pg-core";
|
||||
|
||||
// ─── PCAT Vehicles — VIN decode results ────────────
|
||||
export const pcatVehicles = pgTable(
|
||||
"pcat_vehicles",
|
||||
{
|
||||
id: text("id").primaryKey(), // parts-catalogs car ID
|
||||
catalogId: text("catalog_id").notNull(),
|
||||
vin: varchar("vin", { length: 17 }),
|
||||
name: text("name").notNull(),
|
||||
description: text("description"),
|
||||
parameters: jsonb("parameters"), // [{key, idx, value}]
|
||||
rawData: jsonb("raw_data"),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
(table) => [
|
||||
index("pcat_vehicles_vin_idx").on(table.vin),
|
||||
index("pcat_vehicles_catalog_id_idx").on(table.catalogId),
|
||||
],
|
||||
);
|
||||
|
||||
// ─── PCAT Part Groups — Hierarchical categories ────
|
||||
export const pcatPartGroups = pgTable(
|
||||
"pcat_part_groups",
|
||||
{
|
||||
id: text("id").primaryKey(), // parts-catalogs group ID
|
||||
catalogId: text("catalog_id").notNull(),
|
||||
parentId: text("parent_id"),
|
||||
name: text("name").notNull(),
|
||||
imgUrl: text("img_url"),
|
||||
hasSubgroups: boolean("has_subgroups").default(false).notNull(),
|
||||
hasParts: boolean("has_parts").default(false).notNull(),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
(table) => [
|
||||
index("pcat_part_groups_catalog_id_idx").on(table.catalogId),
|
||||
index("pcat_part_groups_parent_id_idx").on(table.parentId),
|
||||
],
|
||||
);
|
||||
|
||||
// ─── PCAT Parts — OEM parts ────────────────────────
|
||||
export const pcatParts = pgTable(
|
||||
"pcat_parts",
|
||||
{
|
||||
id: text("id").primaryKey(), // OEM number as PK
|
||||
name: text("name").notNull(),
|
||||
notice: text("notice"),
|
||||
description: text("description"),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
(table) => [
|
||||
index("pcat_parts_name_idx").on(table.name),
|
||||
],
|
||||
);
|
||||
|
||||
// ─── PCAT Schema Pics — Schema images + hotspots ───
|
||||
export const pcatSchemaPics = pgTable(
|
||||
"pcat_schema_pics",
|
||||
{
|
||||
id: serial("id").primaryKey(),
|
||||
groupId: text("group_id").notNull(),
|
||||
carId: text("car_id").notNull(),
|
||||
imgUrl: text("img_url").notNull(),
|
||||
imgDescription: text("img_description"),
|
||||
hotspots: jsonb("hotspots"), // positions array from API
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
|
||||
},
|
||||
(table) => [
|
||||
index("pcat_schema_pics_group_car_idx").on(table.groupId, table.carId),
|
||||
],
|
||||
);
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
payments,
|
||||
queryLogs,
|
||||
vehicles,
|
||||
userVehicles,
|
||||
categories,
|
||||
parts,
|
||||
schemaPics,
|
||||
@@ -23,7 +24,7 @@ export const usersRelations = relations(users, ({ many }) => ({
|
||||
userBrands: many(userBrands),
|
||||
payments: many(payments),
|
||||
queryLogs: many(queryLogs),
|
||||
vehicles: many(vehicles),
|
||||
userVehicles: many(userVehicles),
|
||||
referralsGiven: many(referrals, { relationName: "referrer" }),
|
||||
referralsReceived: many(referrals, { relationName: "referred" }),
|
||||
}));
|
||||
@@ -74,12 +75,17 @@ export const queryLogsRelations = relations(queryLogs, ({ one }) => ({
|
||||
}));
|
||||
|
||||
export const vehiclesRelations = relations(vehicles, ({ one, many }) => ({
|
||||
user: one(users, { fields: [vehicles.userId], references: [users.id] }),
|
||||
brand: one(brands, { fields: [vehicles.brandId], references: [brands.id] }),
|
||||
userVehicles: many(userVehicles),
|
||||
categories: many(categories),
|
||||
parts: many(parts),
|
||||
}));
|
||||
|
||||
export const userVehiclesRelations = relations(userVehicles, ({ one }) => ({
|
||||
user: one(users, { fields: [userVehicles.userId], references: [users.id] }),
|
||||
vehicle: one(vehicles, { fields: [userVehicles.vehicleId], references: [vehicles.id] }),
|
||||
}));
|
||||
|
||||
export const categoriesRelations = relations(categories, ({ one, many }) => ({
|
||||
vehicle: one(vehicles, { fields: [categories.vehicleId], references: [vehicles.id] }),
|
||||
parent: one(categories, {
|
||||
|
||||
@@ -28,6 +28,7 @@ import {
|
||||
} from './emex.types';
|
||||
import { mapEmexResponse, createEmptyDecodedVehicle } from './emex.mapper';
|
||||
import { EmexBrowserService } from './emex.browser';
|
||||
import { RedisService } from '../../redis/redis.service';
|
||||
|
||||
// Type definition for the imported scraper module
|
||||
interface EmexScraperModule {
|
||||
@@ -66,6 +67,7 @@ export class EmexService {
|
||||
constructor(
|
||||
private configService: ConfigService,
|
||||
private browserService: EmexBrowserService,
|
||||
private redis: RedisService,
|
||||
) {
|
||||
// __dirname is apps/api/src/integrations/emex/ or dist/integrations/emex/
|
||||
// Scraper lives at <monorepo-root>/scripts/emex-vin-scraper.js
|
||||
@@ -82,6 +84,15 @@ export class EmexService {
|
||||
this.logger.log(`EMEX Service initialized with scraper path: ${this.scraperPath}`);
|
||||
}
|
||||
|
||||
/** Mark EMEX as actively used (5min TTL) to defer prefetch worker */
|
||||
private async touchActivity(): Promise<void> {
|
||||
try {
|
||||
await this.redis.set("prefetch:activity:emex", String(Date.now()), 300);
|
||||
} catch {
|
||||
// Non-critical — don't break the request
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Lazily initialize the scraper module
|
||||
*/
|
||||
@@ -184,6 +195,7 @@ export class EmexService {
|
||||
|
||||
const supported = this.isSupported(cleanVin);
|
||||
this.logger.log(`Decoding VIN: ${cleanVin} (catalog supported: ${supported})`);
|
||||
await this.touchActivity();
|
||||
|
||||
let release: (() => Promise<void>) | null = null;
|
||||
|
||||
@@ -375,6 +387,7 @@ export class EmexService {
|
||||
}
|
||||
|
||||
this.logger.log(`Fetching parts from category URL: ${categoryUrl}`);
|
||||
await this.touchActivity();
|
||||
|
||||
let release: (() => Promise<void>) | null = null;
|
||||
|
||||
|
||||
@@ -0,0 +1,685 @@
|
||||
/**
|
||||
* Parts-Catalogs Auth Service — JWT warm pool via Playwright + DataImpulse proxy
|
||||
*
|
||||
* JWT is captured by navigating to partner sites and intercepting
|
||||
* the Authorization header from requests to parts-catalogs.com.
|
||||
* JWT is IP-bound (~10 min TTL), so the same proxy port must be used for both
|
||||
* browser capture and subsequent API calls.
|
||||
*
|
||||
* Warm pool behavior:
|
||||
* 09:00-19:00 Istanbul → proactive: maintain >= 1 slot, auto-refresh before expiry
|
||||
* 19:00-09:00 → on-demand only: capture only when needed
|
||||
*
|
||||
* Each slot manages its own refresh timer (no polling loop).
|
||||
* Dynamic scaling: 1 JWT per 6 req/min, capped at 5 slots.
|
||||
*/
|
||||
|
||||
import {
|
||||
Injectable,
|
||||
Logger,
|
||||
OnModuleInit,
|
||||
OnModuleDestroy,
|
||||
} from "@nestjs/common";
|
||||
import { ConfigService } from "@nestjs/config";
|
||||
import type { Browser, BrowserContext } from "playwright";
|
||||
import type { PcatJwtToken, JwtSlot, PcatSession } from "./parts-catalogs.types";
|
||||
|
||||
const REFRESH_BUFFER = 90; // Refresh 90s before expiry
|
||||
const CAPTURE_POLL_INTERVAL = 500; // ms
|
||||
const CAPTURE_POLL_MAX = 40; // 40 × 500ms = 20s max wait
|
||||
const PAGE_TIMEOUT = 30_000; // 30s navigation timeout
|
||||
const CONTEXT_CLOSE_TIMEOUT = 5_000;
|
||||
const SITE_COOLDOWN = 10 * 60 * 1000; // 10 min per site
|
||||
const MAX_POOL_SIZE = 5;
|
||||
const RPM_WINDOW = 60_000; // 1-minute rolling window
|
||||
const RPM_PER_SLOT = 6; // 1 JWT per 6 req/min
|
||||
|
||||
/**
|
||||
* Sites that embed the parts-catalogs.com widget.
|
||||
* Widget loads JS → calls /api/start → then calls /v1/catalogs/ with JWT.
|
||||
* Each site uses a different proxy port (IP) to avoid rate limiting.
|
||||
*/
|
||||
const JWT_SITES = [
|
||||
"https://www.e-acca.com/cats/#/catalogs",
|
||||
"https://www.alkatalog.com/cats/#/catalogs",
|
||||
"https://auto-komplekt.ru/goodvin#/catalogs",
|
||||
"https://www.autotrade.md/cats/#/catalogs",
|
||||
"https://www.e-trak.ru/cats/#/catalogs",
|
||||
"https://www.autopolyus.ru/cats/#/catalogs",
|
||||
"https://knkauto.ru/goodvin#/catalogs",
|
||||
"https://www.autodo.kz/#/catalogs",
|
||||
"https://avtoman124.ru/goodvin#/catalogs",
|
||||
"https://flynestauto.com/auto-parts-oem-catalog",
|
||||
"http://en.demo.tradesoft.hk.com/cats/#/catalogs",
|
||||
];
|
||||
|
||||
// DataImpulse proxy defaults (port-based IP rotation)
|
||||
const DI_HOST = "gw.dataimpulse.com";
|
||||
const DI_PORT_MIN = 10000;
|
||||
const DI_PORT_MAX = 10999;
|
||||
const DI_DEFAULT_USER = "1726bbe361918676d44e";
|
||||
const DI_DEFAULT_PASS = "f11c7b6128cc86c6";
|
||||
|
||||
/** Simple counting semaphore (same pattern as EmexBrowserService) */
|
||||
class Semaphore {
|
||||
private current = 0;
|
||||
private queue: Array<() => void> = [];
|
||||
|
||||
constructor(private readonly max: number) {}
|
||||
|
||||
acquire(): Promise<void> {
|
||||
if (this.current < this.max) {
|
||||
this.current++;
|
||||
return Promise.resolve();
|
||||
}
|
||||
return new Promise<void>((resolve) => {
|
||||
this.queue.push(resolve);
|
||||
});
|
||||
}
|
||||
|
||||
release(): void {
|
||||
const next = this.queue.shift();
|
||||
if (next) {
|
||||
next();
|
||||
} else {
|
||||
this.current--;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly logger = new Logger(PartsCatalogsAuthService.name);
|
||||
|
||||
private browser: Browser | null = null;
|
||||
private launching: Promise<void> | null = null;
|
||||
private readonly semaphore = new Semaphore(1); // Max 1 concurrent JWT capture
|
||||
|
||||
// Pool state
|
||||
private pool: JwtSlot[] = [];
|
||||
private siteLastUsedAt = new Map<string, number>();
|
||||
private siteIndex = 0; // round-robin across JWT_SITES
|
||||
private requestRoundRobin = 0; // round-robin across pool slots
|
||||
|
||||
// Business hours scheduling
|
||||
private businessHoursTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
// RPM tracking
|
||||
private requestTimestamps: number[] = [];
|
||||
|
||||
// Config
|
||||
private readonly useProxy: boolean;
|
||||
private readonly proxyHost: string;
|
||||
private readonly proxyUser: string;
|
||||
private readonly proxyPass: string;
|
||||
|
||||
constructor(private configService: ConfigService) {
|
||||
this.useProxy =
|
||||
this.configService.get<string>("PCAT_USE_PROXY", "true") === "true";
|
||||
this.proxyHost = this.configService.get<string>("PCAT_PROXY_HOST", DI_HOST);
|
||||
this.proxyUser = this.configService.get<string>(
|
||||
"PCAT_PROXY_USER",
|
||||
DI_DEFAULT_USER,
|
||||
);
|
||||
this.proxyPass = this.configService.get<string>(
|
||||
"PCAT_PROXY_PASS",
|
||||
DI_DEFAULT_PASS,
|
||||
);
|
||||
}
|
||||
|
||||
async onModuleInit(): Promise<void> {
|
||||
try {
|
||||
await this.launchBrowser();
|
||||
this.logger.log("Browser launched for JWT capture");
|
||||
} catch (err) {
|
||||
this.logger.error(
|
||||
`Failed to launch browser on init: ${(err as Error).message}`,
|
||||
);
|
||||
}
|
||||
|
||||
// Start business hours scheduling
|
||||
if (this.isBusinessHours()) {
|
||||
this.captureToPool().catch((err) => {
|
||||
this.logger.error(`Initial pool capture failed: ${(err as Error).message}`);
|
||||
});
|
||||
}
|
||||
this.scheduleBusinessHours();
|
||||
}
|
||||
|
||||
async onModuleDestroy(): Promise<void> {
|
||||
this.clearAllRefreshTimers();
|
||||
if (this.businessHoursTimer) {
|
||||
clearTimeout(this.businessHoursTimer);
|
||||
this.businessHoursTimer = null;
|
||||
}
|
||||
await this.closeBrowser();
|
||||
this.logger.log("Browser closed on module destroy");
|
||||
}
|
||||
|
||||
// ─── Public API ───────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Acquire a matched JWT + proxy session from the pool.
|
||||
* Returns a valid slot (round-robin) or captures on-demand if pool is empty.
|
||||
*/
|
||||
async acquireSession(): Promise<PcatSession> {
|
||||
this.trackRequest();
|
||||
|
||||
// Try to find a valid slot in the pool
|
||||
const slot = this.pickValidSlot();
|
||||
if (slot) {
|
||||
// Check if we should scale up in the background
|
||||
this.maybeScaleUp();
|
||||
return this.slotToSession(slot);
|
||||
}
|
||||
|
||||
// No valid slot — capture on-demand
|
||||
this.logger.log("JWT pool empty — capturing on-demand...");
|
||||
const newSlot = await this.captureToPool();
|
||||
return this.slotToSession(newSlot);
|
||||
}
|
||||
|
||||
/**
|
||||
* Invalidate a session after 401/403, remove the slot, and capture a replacement.
|
||||
*/
|
||||
async invalidateSession(session: PcatSession): Promise<void> {
|
||||
const idx = this.pool.indexOf(session._slot);
|
||||
if (idx !== -1) {
|
||||
this.clearSlotTimer(this.pool[idx]);
|
||||
this.pool.splice(idx, 1);
|
||||
this.logger.log(
|
||||
`JWT pool: slot invalidated (port ${session._slot.proxyPort}), ${this.pool.length} remaining`,
|
||||
);
|
||||
}
|
||||
|
||||
// Capture replacement in background (don't block the caller's retry)
|
||||
if (this.isBusinessHours() || this.pool.length === 0) {
|
||||
this.captureToPool().catch((err) => {
|
||||
this.logger.error(`Replacement capture failed: ${(err as Error).message}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Pool management ──────────────────────────────────────
|
||||
|
||||
private pickValidSlot(): JwtSlot | null {
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
// Remove expired slots
|
||||
this.pool = this.pool.filter((s) => {
|
||||
if (s.jwt.exp - now < 30) {
|
||||
this.clearSlotTimer(s);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
});
|
||||
|
||||
if (this.pool.length === 0) return null;
|
||||
|
||||
// Round-robin across valid slots
|
||||
this.requestRoundRobin = this.requestRoundRobin % this.pool.length;
|
||||
const slot = this.pool[this.requestRoundRobin];
|
||||
this.requestRoundRobin++;
|
||||
return slot;
|
||||
}
|
||||
|
||||
private slotToSession(slot: JwtSlot): PcatSession {
|
||||
const proxyUrl = this.useProxy
|
||||
? `http://${this.proxyUser}:${this.proxyPass}@${this.proxyHost}:${slot.proxyPort}`
|
||||
: null;
|
||||
const proxyConfig = this.useProxy
|
||||
? {
|
||||
server: `http://${this.proxyHost}:${slot.proxyPort}`,
|
||||
username: this.proxyUser,
|
||||
password: this.proxyPass,
|
||||
}
|
||||
: null;
|
||||
return {
|
||||
authorization: slot.jwt.raw,
|
||||
proxyUrl,
|
||||
proxyConfig,
|
||||
_slot: slot,
|
||||
};
|
||||
}
|
||||
|
||||
// ─── Core capture + pool add ──────────────────────────────
|
||||
|
||||
private async captureToPool(): Promise<JwtSlot> {
|
||||
await this.semaphore.acquire();
|
||||
try {
|
||||
await this.ensureBrowser();
|
||||
|
||||
const maxRetries = 4;
|
||||
for (let attempt = 0; attempt < maxRetries; attempt++) {
|
||||
const port = this.allocatePort();
|
||||
const siteUrl = this.getAvailableSite();
|
||||
|
||||
const jwt = await this.attemptCapture(siteUrl, port);
|
||||
if (jwt) {
|
||||
const slot: JwtSlot = {
|
||||
jwt,
|
||||
proxyPort: port,
|
||||
siteUsed: siteUrl,
|
||||
capturedAt: Date.now(),
|
||||
refreshTimer: null,
|
||||
};
|
||||
|
||||
// Mark site as used
|
||||
this.siteLastUsedAt.set(siteUrl, Date.now());
|
||||
|
||||
// Add to pool
|
||||
this.pool.push(slot);
|
||||
|
||||
// Schedule refresh if in business hours
|
||||
if (this.isBusinessHours()) {
|
||||
this.scheduleSlotRefresh(slot);
|
||||
}
|
||||
|
||||
const ttl = jwt.exp - Math.floor(Date.now() / 1000);
|
||||
const refreshIn = this.isBusinessHours()
|
||||
? Math.max(ttl - REFRESH_BUFFER, 30)
|
||||
: null;
|
||||
this.logger.log(
|
||||
`JWT pool: slot captured, TTL: ${ttl}s${refreshIn ? `, refresh in ${refreshIn}s` : ""}, pool size: ${this.pool.length}`,
|
||||
);
|
||||
return slot;
|
||||
}
|
||||
this.logger.warn(
|
||||
`JWT capture attempt ${attempt + 1}/${maxRetries} failed (${new URL(siteUrl).hostname}), trying next site...`,
|
||||
);
|
||||
}
|
||||
|
||||
throw new Error("Failed to capture JWT after all retries");
|
||||
} finally {
|
||||
this.semaphore.release();
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Timer-based refresh per slot ─────────────────────────
|
||||
|
||||
private scheduleSlotRefresh(slot: JwtSlot): void {
|
||||
this.clearSlotTimer(slot);
|
||||
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
const delayMs = Math.max((slot.jwt.exp - now - REFRESH_BUFFER) * 1000, 30_000);
|
||||
|
||||
slot.refreshTimer = setTimeout(async () => {
|
||||
// Don't refresh outside business hours
|
||||
if (!this.isBusinessHours()) {
|
||||
this.logger.debug("JWT pool: refresh timer fired outside business hours, skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const newSlot = await this.captureToPool();
|
||||
// Remove old slot
|
||||
const idx = this.pool.indexOf(slot);
|
||||
if (idx !== -1) {
|
||||
this.pool.splice(idx, 1);
|
||||
}
|
||||
this.logger.log(
|
||||
`JWT pool: slot refreshed, TTL: ${newSlot.jwt.exp - Math.floor(Date.now() / 1000)}s`,
|
||||
);
|
||||
} catch (err) {
|
||||
this.logger.error(`JWT pool: refresh failed: ${(err as Error).message}`);
|
||||
// Schedule a retry in 30s if still in business hours
|
||||
if (this.isBusinessHours()) {
|
||||
slot.refreshTimer = setTimeout(() => {
|
||||
this.scheduleSlotRefresh(slot);
|
||||
}, 30_000);
|
||||
}
|
||||
}
|
||||
}, delayMs);
|
||||
}
|
||||
|
||||
private clearSlotTimer(slot: JwtSlot): void {
|
||||
if (slot.refreshTimer) {
|
||||
clearTimeout(slot.refreshTimer);
|
||||
slot.refreshTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private clearAllRefreshTimers(): void {
|
||||
for (const slot of this.pool) {
|
||||
this.clearSlotTimer(slot);
|
||||
}
|
||||
this.logger.log("JWT pool: all refresh timers cleared");
|
||||
}
|
||||
|
||||
// ─── Business hours scheduling ────────────────────────────
|
||||
|
||||
/**
|
||||
* Get current Istanbul hour and minute using Intl.DateTimeFormat.
|
||||
* This works correctly regardless of the server's local timezone.
|
||||
*/
|
||||
private getIstanbulTime(): { hour: number; minute: number } {
|
||||
const parts = new Intl.DateTimeFormat("en-US", {
|
||||
timeZone: "Europe/Istanbul",
|
||||
hour: "numeric",
|
||||
minute: "numeric",
|
||||
hour12: false,
|
||||
}).formatToParts(new Date());
|
||||
|
||||
const hour = parseInt(parts.find((p) => p.type === "hour")!.value, 10);
|
||||
const minute = parseInt(parts.find((p) => p.type === "minute")!.value, 10);
|
||||
return { hour, minute };
|
||||
}
|
||||
|
||||
private isBusinessHours(): boolean {
|
||||
const { hour } = this.getIstanbulTime();
|
||||
return hour >= 9 && hour < 19;
|
||||
}
|
||||
|
||||
private scheduleBusinessHours(): void {
|
||||
if (this.businessHoursTimer) {
|
||||
clearTimeout(this.businessHoursTimer);
|
||||
this.businessHoursTimer = null;
|
||||
}
|
||||
|
||||
const { hour, minute } = this.getIstanbulTime();
|
||||
|
||||
let delayMs: number;
|
||||
let nextEvent: string;
|
||||
|
||||
if (this.isBusinessHours()) {
|
||||
// Schedule 19:00 stop
|
||||
const minsUntil19 = (19 - hour - 1) * 60 + (60 - minute);
|
||||
delayMs = minsUntil19 * 60 * 1000;
|
||||
nextEvent = "stop (19:00)";
|
||||
|
||||
this.businessHoursTimer = setTimeout(() => {
|
||||
this.logger.log("JWT pool: business hours ended, timers cleared");
|
||||
this.clearAllRefreshTimers();
|
||||
this.scheduleBusinessHours(); // schedule next 09:00 start
|
||||
}, delayMs);
|
||||
} else {
|
||||
// Schedule 09:00 start
|
||||
let minsUntil9: number;
|
||||
if (hour >= 19) {
|
||||
// Same day evening → next day 09:00
|
||||
minsUntil9 = (24 - hour + 9 - 1) * 60 + (60 - minute);
|
||||
} else {
|
||||
// Before 09:00
|
||||
minsUntil9 = (9 - hour - 1) * 60 + (60 - minute);
|
||||
}
|
||||
delayMs = minsUntil9 * 60 * 1000;
|
||||
nextEvent = "start (09:00)";
|
||||
|
||||
this.businessHoursTimer = setTimeout(() => {
|
||||
this.logger.log("JWT pool: business hours started, capturing initial slot");
|
||||
this.captureToPool().catch((err) => {
|
||||
this.logger.error(`Business hours initial capture failed: ${(err as Error).message}`);
|
||||
});
|
||||
this.scheduleBusinessHours(); // schedule 19:00 stop
|
||||
}, delayMs);
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`JWT pool: next ${nextEvent} in ${Math.round(delayMs / 60_000)}min (Istanbul: ${hour}:${String(minute).padStart(2, "0")})`,
|
||||
);
|
||||
}
|
||||
|
||||
// ─── RPM tracking & dynamic scaling ───────────────────────
|
||||
|
||||
private trackRequest(): void {
|
||||
const now = Date.now();
|
||||
this.requestTimestamps.push(now);
|
||||
// Prune old timestamps outside the window
|
||||
const cutoff = now - RPM_WINDOW;
|
||||
while (this.requestTimestamps.length > 0 && this.requestTimestamps[0] < cutoff) {
|
||||
this.requestTimestamps.shift();
|
||||
}
|
||||
}
|
||||
|
||||
private getRPM(): number {
|
||||
const now = Date.now();
|
||||
const cutoff = now - RPM_WINDOW;
|
||||
return this.requestTimestamps.filter((t) => t >= cutoff).length;
|
||||
}
|
||||
|
||||
private getDesiredPoolSize(): number {
|
||||
const rpm = this.getRPM();
|
||||
return Math.min(Math.max(1, Math.ceil(rpm / RPM_PER_SLOT)), MAX_POOL_SIZE);
|
||||
}
|
||||
|
||||
private maybeScaleUp(): void {
|
||||
if (!this.isBusinessHours()) return;
|
||||
|
||||
const desired = this.getDesiredPoolSize();
|
||||
if (this.pool.length < desired) {
|
||||
this.logger.log(
|
||||
`JWT pool: scaling up, ${this.pool.length}/${desired} slots (RPM: ${this.getRPM()})`,
|
||||
);
|
||||
this.captureToPool().catch((err) => {
|
||||
this.logger.warn(`JWT pool: scale-up capture failed: ${(err as Error).message}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Site cooldown & port allocation ──────────────────────
|
||||
|
||||
private getAvailableSite(): string {
|
||||
const now = Date.now();
|
||||
|
||||
// Try round-robin, preferring sites not on cooldown
|
||||
for (let i = 0; i < JWT_SITES.length; i++) {
|
||||
const idx = (this.siteIndex + i) % JWT_SITES.length;
|
||||
const site = JWT_SITES[idx];
|
||||
const lastUsed = this.siteLastUsedAt.get(site) || 0;
|
||||
|
||||
if (now - lastUsed >= SITE_COOLDOWN) {
|
||||
this.siteIndex = (idx + 1) % JWT_SITES.length;
|
||||
return site;
|
||||
}
|
||||
}
|
||||
|
||||
// All on cooldown — pick the one with oldest usage
|
||||
let oldestIdx = 0;
|
||||
let oldestTime = Infinity;
|
||||
for (let i = 0; i < JWT_SITES.length; i++) {
|
||||
const lastUsed = this.siteLastUsedAt.get(JWT_SITES[i]) || 0;
|
||||
if (lastUsed < oldestTime) {
|
||||
oldestTime = lastUsed;
|
||||
oldestIdx = i;
|
||||
}
|
||||
}
|
||||
this.siteIndex = (oldestIdx + 1) % JWT_SITES.length;
|
||||
return JWT_SITES[oldestIdx];
|
||||
}
|
||||
|
||||
private allocatePort(): number {
|
||||
return DI_PORT_MIN + Math.floor(Math.random() * (DI_PORT_MAX - DI_PORT_MIN + 1));
|
||||
}
|
||||
|
||||
// ─── JWT capture via Playwright ───────────────────────────
|
||||
|
||||
private async attemptCapture(
|
||||
siteUrl: string,
|
||||
port: number,
|
||||
): Promise<PcatJwtToken | null> {
|
||||
let context: BrowserContext | null = null;
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
// Build context options with proxy
|
||||
const contextOptions: Record<string, unknown> = {};
|
||||
if (this.useProxy) {
|
||||
contextOptions.proxy = {
|
||||
server: `http://${this.proxyHost}:${port}`,
|
||||
username: this.proxyUser,
|
||||
password: this.proxyPass,
|
||||
};
|
||||
}
|
||||
|
||||
context = await this.browser!.newContext(contextOptions);
|
||||
const page = await context.newPage();
|
||||
|
||||
// Intercept requests to parts-catalogs.com
|
||||
let capturedJwt: string | null = null;
|
||||
|
||||
page.on("request", (request) => {
|
||||
if (capturedJwt) return;
|
||||
const url = request.url();
|
||||
if (
|
||||
url.includes("parts-catalogs.com") ||
|
||||
url.includes("api.parts-catalogs.com")
|
||||
) {
|
||||
const auth = request.headers()["authorization"];
|
||||
if (auth) {
|
||||
capturedJwt = auth;
|
||||
this.logger.debug("JWT intercepted from request");
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Block heavy resources to save proxy bandwidth
|
||||
await page.route("**/*", (route) => {
|
||||
const url = route.request().url();
|
||||
const type = route.request().resourceType();
|
||||
|
||||
// Block images, fonts, media
|
||||
if (["image", "font", "media"].includes(type)) {
|
||||
return route.abort();
|
||||
}
|
||||
|
||||
// Block known trackers/analytics
|
||||
if (
|
||||
url.includes("google-analytics.com") ||
|
||||
url.includes("googletagmanager.com") ||
|
||||
url.includes("mc.yandex.ru") ||
|
||||
url.includes("facebook.net") ||
|
||||
url.includes("doubleclick.net") ||
|
||||
url.includes("hotjar.com")
|
||||
) {
|
||||
return route.abort();
|
||||
}
|
||||
|
||||
return route.continue();
|
||||
});
|
||||
|
||||
// Navigate — networkidle waits for widget JS to load + make API calls
|
||||
try {
|
||||
await page.goto(siteUrl, {
|
||||
timeout: PAGE_TIMEOUT,
|
||||
waitUntil: "networkidle",
|
||||
});
|
||||
} catch (navErr) {
|
||||
// Navigation may timeout but JWT could still be captured
|
||||
this.logger.debug(
|
||||
`Navigation ended: ${(navErr as Error).message?.slice(0, 80)}`,
|
||||
);
|
||||
}
|
||||
|
||||
// Poll for JWT
|
||||
for (let i = 0; i < CAPTURE_POLL_MAX; i++) {
|
||||
if (capturedJwt) break;
|
||||
await new Promise((r) => setTimeout(r, CAPTURE_POLL_INTERVAL));
|
||||
}
|
||||
|
||||
const elapsed = Date.now() - startTime;
|
||||
|
||||
if (capturedJwt) {
|
||||
const token = this.parseJwt(capturedJwt);
|
||||
this.logger.log(
|
||||
`JWT captured in ${elapsed}ms from ${new URL(siteUrl).hostname}`,
|
||||
);
|
||||
return token;
|
||||
}
|
||||
|
||||
this.logger.debug(`No JWT after ${elapsed}ms from ${siteUrl}`);
|
||||
return null;
|
||||
} catch (err) {
|
||||
this.logger.warn(`JWT capture error: ${(err as Error).message}`);
|
||||
return null;
|
||||
} finally {
|
||||
if (context) {
|
||||
try {
|
||||
await Promise.race([
|
||||
context.close(),
|
||||
new Promise((r) => setTimeout(r, CONTEXT_CLOSE_TIMEOUT)),
|
||||
]);
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private parseJwt(rawToken: string): PcatJwtToken {
|
||||
const parts = rawToken.split(".");
|
||||
if (parts.length !== 3) {
|
||||
throw new Error("Invalid JWT format");
|
||||
}
|
||||
|
||||
// Decode payload with proper base64url padding
|
||||
let payloadB64 = parts[1];
|
||||
const padding = 4 - (payloadB64.length % 4);
|
||||
if (padding !== 4) {
|
||||
payloadB64 += "=".repeat(padding);
|
||||
}
|
||||
|
||||
const payload = JSON.parse(
|
||||
Buffer.from(payloadB64, "base64url").toString("utf-8"),
|
||||
);
|
||||
|
||||
return {
|
||||
raw: rawToken,
|
||||
exp: payload.exp || 0,
|
||||
host: payload.host || "",
|
||||
apiKey: payload.apiKey || "",
|
||||
apiPath: payload.apiPath || "",
|
||||
ip: payload.ip || "",
|
||||
hash: payload.h || "",
|
||||
};
|
||||
}
|
||||
|
||||
// ─── Browser lifecycle ───────────────────────────────────
|
||||
|
||||
private async launchBrowser(): Promise<void> {
|
||||
if (this.launching) {
|
||||
return this.launching;
|
||||
}
|
||||
this.launching = this._doLaunch();
|
||||
try {
|
||||
await this.launching;
|
||||
} finally {
|
||||
this.launching = null;
|
||||
}
|
||||
}
|
||||
|
||||
private async _doLaunch(): Promise<void> {
|
||||
const { chromium } = await import("playwright");
|
||||
|
||||
this.browser = await chromium.launch({
|
||||
headless: true,
|
||||
args: [
|
||||
"--no-sandbox",
|
||||
"--disable-setuid-sandbox",
|
||||
"--disable-dev-shm-usage",
|
||||
"--disable-accelerated-2d-canvas",
|
||||
"--disable-gpu",
|
||||
],
|
||||
});
|
||||
|
||||
this.browser.on("disconnected", () => {
|
||||
this.logger.warn("Browser disconnected — will relaunch on next request");
|
||||
this.browser = null;
|
||||
});
|
||||
}
|
||||
|
||||
private async closeBrowser(): Promise<void> {
|
||||
if (this.browser) {
|
||||
try {
|
||||
await this.browser.close();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
this.browser = null;
|
||||
}
|
||||
}
|
||||
|
||||
private async ensureBrowser(): Promise<void> {
|
||||
if (this.browser?.isConnected()) return;
|
||||
this.logger.log("Browser not connected — relaunching");
|
||||
await this.launchBrowser();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
import { Module } from "@nestjs/common";
|
||||
import { PartsCatalogsAuthService } from "./parts-catalogs-auth.service";
|
||||
import { PartsCatalogsService } from "./parts-catalogs.service";
|
||||
|
||||
@Module({
|
||||
providers: [PartsCatalogsAuthService, PartsCatalogsService],
|
||||
exports: [PartsCatalogsService],
|
||||
})
|
||||
export class PartsCatalogsModule {}
|
||||
@@ -0,0 +1,240 @@
|
||||
/**
|
||||
* Parts-Catalogs API Service — HTTP client for parts-catalogs.com
|
||||
*
|
||||
* All requests go through the same DataImpulse proxy as the JWT capture
|
||||
* to ensure the JWT's IP-bound constraint is satisfied.
|
||||
*/
|
||||
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { PartsCatalogsAuthService } from "./parts-catalogs-auth.service";
|
||||
import { RedisService } from "../../redis/redis.service";
|
||||
import type {
|
||||
PcatVinResult,
|
||||
PcatCar,
|
||||
PcatGroup,
|
||||
PcatPartsResult,
|
||||
PcatSession,
|
||||
} from "./parts-catalogs.types";
|
||||
|
||||
const API_BASE = "https://api.parts-catalogs.com/v1";
|
||||
const REQUEST_TIMEOUT = 30_000;
|
||||
|
||||
@Injectable()
|
||||
export class PartsCatalogsService {
|
||||
private readonly logger = new Logger(PartsCatalogsService.name);
|
||||
|
||||
constructor(
|
||||
private authService: PartsCatalogsAuthService,
|
||||
private redis: RedisService,
|
||||
) {}
|
||||
|
||||
/** Mark parts-catalogs as actively used (5min TTL) to defer prefetch worker */
|
||||
private async touchActivity(): Promise<void> {
|
||||
try {
|
||||
await this.redis.set("prefetch:activity:parts-catalogs", String(Date.now()), 300);
|
||||
} catch {
|
||||
// Non-critical
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* VIN decode — returns one or more car matches.
|
||||
*/
|
||||
async decodeVin(vin: string): Promise<PcatVinResult | null> {
|
||||
try {
|
||||
const data = await this.fetchWithAuth("/car/info", { q: vin });
|
||||
|
||||
if (!data || typeof data !== "object") {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Response can be a single car or array of cars depending on VIN
|
||||
const rawCars = Array.isArray(data) ? data : [data];
|
||||
|
||||
const cars: PcatCar[] = [];
|
||||
for (const item of rawCars) {
|
||||
if (!item) continue;
|
||||
|
||||
// Each car result may have nested catalog info
|
||||
const catalogId = item.catalogId || item.catalog?.id || "";
|
||||
const carId = item.id || item.carId || "";
|
||||
|
||||
if (!carId) continue;
|
||||
|
||||
cars.push({
|
||||
id: String(carId),
|
||||
name: item.name || item.title || "",
|
||||
description: item.description || item.modelName || undefined,
|
||||
parameters: Array.isArray(item.parameters) ? item.parameters : undefined,
|
||||
catalogId: String(catalogId),
|
||||
});
|
||||
}
|
||||
|
||||
if (cars.length === 0) return null;
|
||||
|
||||
return { cars };
|
||||
} catch (err) {
|
||||
this.logger.warn(`VIN decode failed for ${vin}: ${(err as Error).message}`);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get category groups for a car.
|
||||
* Pass groupId for subgroups.
|
||||
*/
|
||||
async fetchGroups(
|
||||
catalogId: string,
|
||||
carId: string,
|
||||
groupId?: string,
|
||||
carParams?: Record<string, string>,
|
||||
): Promise<PcatGroup[]> {
|
||||
await this.touchActivity();
|
||||
const params: Record<string, string> = { carId };
|
||||
if (groupId) params.groupId = groupId;
|
||||
if (carParams) Object.assign(params, carParams);
|
||||
|
||||
const data = await this.fetchWithAuth(
|
||||
`/catalogs/${catalogId}/groups2/`,
|
||||
params,
|
||||
);
|
||||
|
||||
if (!Array.isArray(data)) return [];
|
||||
|
||||
return data.map((g: any) => ({
|
||||
id: String(g.id),
|
||||
parentId: g.parentId ? String(g.parentId) : undefined,
|
||||
name: g.name || "",
|
||||
img: g.img || undefined,
|
||||
hasSubgroups: !!g.hasSubgroups,
|
||||
hasParts: !!g.hasParts,
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Get parts + schema image + hotspot positions for a group.
|
||||
*/
|
||||
async fetchParts(
|
||||
catalogId: string,
|
||||
carId: string,
|
||||
groupId: string,
|
||||
carParams?: Record<string, string>,
|
||||
): Promise<PcatPartsResult | null> {
|
||||
await this.touchActivity();
|
||||
const params: Record<string, string> = { carId, groupId };
|
||||
if (carParams) Object.assign(params, carParams);
|
||||
|
||||
const data = await this.fetchWithAuth(
|
||||
`/catalogs/${catalogId}/parts2`,
|
||||
params,
|
||||
);
|
||||
|
||||
if (!data || typeof data !== "object") return null;
|
||||
|
||||
return {
|
||||
img: data.img || "",
|
||||
imgDescription: data.imgDescription || undefined,
|
||||
partGroups: Array.isArray(data.partGroups)
|
||||
? data.partGroups.map((pg: any) => ({
|
||||
name: pg.name || undefined,
|
||||
number: pg.number || undefined,
|
||||
positionNumber: pg.positionNumber || undefined,
|
||||
parts: Array.isArray(pg.parts)
|
||||
? pg.parts.map((p: any) => ({
|
||||
id: p.id ? String(p.id) : undefined,
|
||||
number: p.number || "",
|
||||
name: p.name || "",
|
||||
nameId: p.nameId || undefined,
|
||||
notice: p.notice || undefined,
|
||||
positionNumber: p.positionNumber || undefined,
|
||||
}))
|
||||
: [],
|
||||
}))
|
||||
: [],
|
||||
positions: Array.isArray(data.positions)
|
||||
? data.positions.map((pos: any) => ({
|
||||
number: String(pos.number),
|
||||
coordinates: pos.coordinates || [0, 0, 0, 0],
|
||||
}))
|
||||
: [],
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a VIN is potentially supported.
|
||||
* Parts-catalogs.com covers most brands, so this is broadly true.
|
||||
*/
|
||||
isSupported(_vin: string): boolean {
|
||||
return true;
|
||||
}
|
||||
|
||||
// ─── Private ─────────────────────────────────────────────
|
||||
|
||||
private async fetchWithAuth(
|
||||
endpoint: string,
|
||||
params?: Record<string, string>,
|
||||
): Promise<any> {
|
||||
const maxRetries = 2;
|
||||
|
||||
let session: PcatSession | null = null;
|
||||
|
||||
for (let attempt = 0; attempt <= maxRetries; attempt++) {
|
||||
session = await this.authService.acquireSession();
|
||||
|
||||
const url = new URL(`${API_BASE}${endpoint}`);
|
||||
if (params) {
|
||||
for (const [key, value] of Object.entries(params)) {
|
||||
url.searchParams.set(key, value);
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const fetchOptions: RequestInit & { dispatcher?: any } = {
|
||||
method: "GET",
|
||||
headers: {
|
||||
Authorization: session.authorization,
|
||||
Accept: "application/json",
|
||||
"User-Agent":
|
||||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||
},
|
||||
signal: AbortSignal.timeout(REQUEST_TIMEOUT),
|
||||
};
|
||||
|
||||
// Use undici ProxyAgent if proxy is configured
|
||||
if (session.proxyUrl) {
|
||||
const { ProxyAgent } = await import("undici");
|
||||
fetchOptions.dispatcher = new ProxyAgent(session.proxyUrl);
|
||||
}
|
||||
|
||||
const response = await fetch(url.toString(), fetchOptions);
|
||||
|
||||
if (response.ok) {
|
||||
return await response.json();
|
||||
}
|
||||
|
||||
if (response.status === 401 || response.status === 403) {
|
||||
this.logger.warn(
|
||||
`Auth error (${response.status}) on ${endpoint}, attempt ${attempt + 1}/${maxRetries + 1}`,
|
||||
);
|
||||
if (attempt < maxRetries) {
|
||||
await this.authService.invalidateSession(session);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
const text = await response.text().catch(() => "");
|
||||
throw new Error(
|
||||
`HTTP ${response.status} from ${endpoint}: ${text.slice(0, 200)}`,
|
||||
);
|
||||
} catch (err) {
|
||||
if ((err as Error).name === "TimeoutError") {
|
||||
this.logger.warn(`Timeout on ${endpoint}, attempt ${attempt + 1}`);
|
||||
if (attempt < maxRetries) continue;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error(`Max retries exceeded for ${endpoint}`);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
export interface PcatJwtToken {
|
||||
raw: string;
|
||||
exp: number;
|
||||
host: string;
|
||||
apiKey: string;
|
||||
apiPath: string;
|
||||
ip: string;
|
||||
hash: string;
|
||||
}
|
||||
|
||||
export interface JwtSlot {
|
||||
jwt: PcatJwtToken;
|
||||
proxyPort: number;
|
||||
siteUsed: string;
|
||||
capturedAt: number;
|
||||
refreshTimer: ReturnType<typeof setTimeout> | null;
|
||||
}
|
||||
|
||||
export interface PcatSession {
|
||||
authorization: string;
|
||||
proxyUrl: string | null;
|
||||
proxyConfig: { server: string; username: string; password: string } | null;
|
||||
_slot: JwtSlot;
|
||||
}
|
||||
|
||||
export interface PcatCarParameter {
|
||||
key: string;
|
||||
idx: string;
|
||||
value: string;
|
||||
}
|
||||
|
||||
export interface PcatCar {
|
||||
id: string;
|
||||
name: string;
|
||||
description?: string;
|
||||
parameters?: PcatCarParameter[];
|
||||
catalogId: string;
|
||||
}
|
||||
|
||||
export interface PcatVinResult {
|
||||
cars: PcatCar[];
|
||||
}
|
||||
|
||||
export interface PcatGroup {
|
||||
id: string;
|
||||
parentId?: string;
|
||||
name: string;
|
||||
img?: string;
|
||||
hasSubgroups: boolean;
|
||||
hasParts: boolean;
|
||||
}
|
||||
|
||||
export interface PcatPart {
|
||||
id?: string;
|
||||
number: string; // OEM code
|
||||
name: string;
|
||||
nameId?: number;
|
||||
notice?: string;
|
||||
positionNumber?: string;
|
||||
}
|
||||
|
||||
export interface PcatPartGroup {
|
||||
name?: string;
|
||||
number?: string;
|
||||
positionNumber?: string;
|
||||
parts: PcatPart[];
|
||||
}
|
||||
|
||||
export interface PcatPosition {
|
||||
number: string;
|
||||
coordinates: [number, number, number, number]; // x, y, w, h
|
||||
}
|
||||
|
||||
export interface PcatPartsResult {
|
||||
img: string;
|
||||
imgDescription?: string;
|
||||
partGroups: PcatPartGroup[];
|
||||
positions: PcatPosition[];
|
||||
}
|
||||
@@ -261,6 +261,7 @@ export class PL24Service {
|
||||
return this.fordLegacyService.fetchPartsByPath(linkPath, serviceName);
|
||||
}
|
||||
|
||||
await this.touchActivity();
|
||||
const pathHash = createHash("sha256").update(linkPath).digest("hex").substring(0, 16);
|
||||
const cacheKey = `${PL24_DEFAULTS.CACHE_PREFIX}parts:path:${pathHash}`;
|
||||
const cached = await this.redis.getJson<PL24PartsResponse>(cacheKey);
|
||||
@@ -384,6 +385,7 @@ export class PL24Service {
|
||||
}
|
||||
|
||||
this.logger.log(`Fetching sub-groups by path: ${linkPath}`);
|
||||
await this.touchActivity();
|
||||
|
||||
try {
|
||||
await this.authService.authorizeService(serviceName);
|
||||
@@ -411,6 +413,7 @@ export class PL24Service {
|
||||
serviceName: string,
|
||||
mainGroupsPath: string,
|
||||
): Promise<PL24DecodedCategory[]> {
|
||||
await this.touchActivity();
|
||||
try {
|
||||
await this.authService.authorizeService(serviceName);
|
||||
const headers = await this.authService.buildAuthHeaders(serviceName);
|
||||
@@ -435,6 +438,7 @@ export class PL24Service {
|
||||
hotspots: PL24Hotspot[];
|
||||
} | null> {
|
||||
if (!imageUrl) return null;
|
||||
await this.touchActivity();
|
||||
|
||||
// Extract image ID for dedup
|
||||
const imageId = this.extractImageIdFromUrl(imageUrl);
|
||||
@@ -572,6 +576,15 @@ export class PL24Service {
|
||||
return SERVICE_TO_BRAND[serviceName] || null;
|
||||
}
|
||||
|
||||
/** Mark PL24 as actively used (5min TTL) to defer prefetch worker */
|
||||
private async touchActivity(): Promise<void> {
|
||||
try {
|
||||
await this.redis.set("prefetch:activity:pl24", String(Date.now()), 300);
|
||||
} catch {
|
||||
// Non-critical — don't break the request
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== PRIVATE: Request helpers ====================
|
||||
|
||||
/**
|
||||
|
||||
@@ -24,4 +24,5 @@ export const QUEUE_NAMES = {
|
||||
EMEX_SCRAPE: "emex-scrape",
|
||||
SUBSCRIPTION_EXPIRY: "subscription-expiry",
|
||||
QUERY_CLEANUP: "query-cleanup",
|
||||
CATALOG_PREFETCH: "catalog-prefetch",
|
||||
} as const;
|
||||
|
||||
@@ -6,19 +6,29 @@ import {
|
||||
SUBSCRIPTION_EXPIRY_QUEUE,
|
||||
} from "./queues/subscription-expiry.queue";
|
||||
import { QueryCleanupQueueProvider, QUERY_CLEANUP_QUEUE } from "./queues/query-cleanup.queue";
|
||||
import {
|
||||
CatalogPrefetchQueueProvider,
|
||||
CATALOG_PREFETCH_QUEUE,
|
||||
} from "./queues/catalog-prefetch.queue";
|
||||
import { PrefetchWorkerService } from "./prefetch-worker.service";
|
||||
import { CategoriesModule } from "../categories/categories.module";
|
||||
|
||||
@Module({
|
||||
imports: [CategoriesModule],
|
||||
providers: [
|
||||
EmexScrapeQueueProvider,
|
||||
SubscriptionExpiryQueueProvider,
|
||||
QueryCleanupQueueProvider,
|
||||
CatalogPrefetchQueueProvider,
|
||||
PrefetchWorkerService,
|
||||
],
|
||||
exports: [EMEX_SCRAPE_QUEUE, SUBSCRIPTION_EXPIRY_QUEUE, QUERY_CLEANUP_QUEUE],
|
||||
exports: [EMEX_SCRAPE_QUEUE, SUBSCRIPTION_EXPIRY_QUEUE, QUERY_CLEANUP_QUEUE, CATALOG_PREFETCH_QUEUE],
|
||||
})
|
||||
export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
constructor(
|
||||
@Inject(SUBSCRIPTION_EXPIRY_QUEUE) private subscriptionExpiryQueue: Queue,
|
||||
@Inject(QUERY_CLEANUP_QUEUE) private queryCleanupQueue: Queue,
|
||||
@Inject(CATALOG_PREFETCH_QUEUE) private catalogPrefetchQueue: Queue,
|
||||
) {}
|
||||
|
||||
async onModuleInit() {
|
||||
@@ -59,6 +69,7 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
await Promise.all([
|
||||
this.subscriptionExpiryQueue.close(),
|
||||
this.queryCleanupQueue.close(),
|
||||
this.catalogPrefetchQueue.close(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
144
apps/api/src/jobs/prefetch-utils.ts
Normal file
144
apps/api/src/jobs/prefetch-utils.ts
Normal file
@@ -0,0 +1,144 @@
|
||||
import { RedisService } from "../redis/redis.service";
|
||||
|
||||
/**
|
||||
* Custom error that tells BullMQ to retry after a delay.
|
||||
* The worker catches this and re-queues the job with the specified delay.
|
||||
*/
|
||||
export class RateLimitError extends Error {
|
||||
constructor(public readonly retryAfterMs: number) {
|
||||
super(`Rate limited — retry after ${retryAfterMs}ms`);
|
||||
this.name = "RateLimitError";
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a user is actively using the source.
|
||||
* Throws RateLimitError (1min retry) if cooldown key exists.
|
||||
*/
|
||||
export async function checkCooldown(
|
||||
redis: RedisService,
|
||||
source: string,
|
||||
): Promise<void> {
|
||||
const key = `prefetch:activity:${source}`;
|
||||
const exists = await redis.exists(key);
|
||||
if (exists) {
|
||||
const retryMs = source === "parts-catalogs" ? 120_000 : 60_000;
|
||||
throw new RateLimitError(retryMs);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check PL24 business hours (09:00–18:00 Europe/Istanbul).
|
||||
* Throws RateLimitError with delay until next 09:00 if outside window.
|
||||
*/
|
||||
export function checkTimeWindow(source: string): void {
|
||||
if (source !== "pl24" && source !== "parts-catalogs") return;
|
||||
|
||||
const hourStr = new Intl.DateTimeFormat("en-US", {
|
||||
timeZone: "Europe/Istanbul",
|
||||
hour: "numeric",
|
||||
hour12: false,
|
||||
}).format(new Date());
|
||||
const h = parseInt(hourStr, 10);
|
||||
|
||||
const endHour = source === "parts-catalogs" ? 19 : 18;
|
||||
if (h < 9 || h >= endHour) {
|
||||
throw new RateLimitError(msUntilNext9AM());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Milliseconds until the next 09:00 Europe/Istanbul.
|
||||
*/
|
||||
export function msUntilNext9AM(): number {
|
||||
const now = new Date();
|
||||
|
||||
// Get current Istanbul time components
|
||||
const istParts = new Intl.DateTimeFormat("en-US", {
|
||||
timeZone: "Europe/Istanbul",
|
||||
year: "numeric",
|
||||
month: "2-digit",
|
||||
day: "2-digit",
|
||||
hour: "2-digit",
|
||||
minute: "2-digit",
|
||||
second: "2-digit",
|
||||
hour12: false,
|
||||
}).formatToParts(now);
|
||||
|
||||
const get = (type: string) =>
|
||||
parseInt(istParts.find((p) => p.type === type)?.value || "0", 10);
|
||||
|
||||
const hour = get("hour");
|
||||
const minute = get("minute");
|
||||
const second = get("second");
|
||||
|
||||
// If before 09:00 → wait until 09:00 today
|
||||
// If 18:00+ → wait until 09:00 tomorrow
|
||||
let hoursToWait: number;
|
||||
if (hour < 9) {
|
||||
hoursToWait = 9 - hour;
|
||||
} else {
|
||||
hoursToWait = 24 - hour + 9;
|
||||
}
|
||||
|
||||
const ms =
|
||||
hoursToWait * 3600_000 -
|
||||
minute * 60_000 -
|
||||
second * 1000;
|
||||
|
||||
// At least 1 minute, at most 15 hours
|
||||
return Math.max(60_000, Math.min(ms, 15 * 3600_000));
|
||||
}
|
||||
|
||||
/** Redis hash key for prefetch progress */
|
||||
export function progressKey(vehicleId: string): string {
|
||||
return `prefetch:progress:${vehicleId}`;
|
||||
}
|
||||
|
||||
export interface PrefetchProgress {
|
||||
status: "running" | "completed" | "error";
|
||||
total: number;
|
||||
completed: number;
|
||||
errors: number;
|
||||
startedAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export async function initProgress(
|
||||
redis: RedisService,
|
||||
vehicleId: string,
|
||||
): Promise<void> {
|
||||
const now = new Date().toISOString();
|
||||
await redis.setJson(progressKey(vehicleId), {
|
||||
status: "running",
|
||||
total: 0,
|
||||
completed: 0,
|
||||
errors: 0,
|
||||
startedAt: now,
|
||||
updatedAt: now,
|
||||
} satisfies PrefetchProgress, 86400);
|
||||
}
|
||||
|
||||
export async function updateProgress(
|
||||
redis: RedisService,
|
||||
vehicleId: string,
|
||||
update: Partial<PrefetchProgress>,
|
||||
): Promise<void> {
|
||||
const key = progressKey(vehicleId);
|
||||
const current = await redis.getJson<PrefetchProgress>(key);
|
||||
if (!current) return;
|
||||
|
||||
const updated: PrefetchProgress = {
|
||||
...current,
|
||||
...update,
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
await redis.setJson(key, updated, 86400);
|
||||
}
|
||||
|
||||
export async function getProgress(
|
||||
redis: RedisService,
|
||||
vehicleId: string,
|
||||
): Promise<PrefetchProgress | null> {
|
||||
return redis.getJson<PrefetchProgress>(progressKey(vehicleId));
|
||||
}
|
||||
395
apps/api/src/jobs/prefetch-worker.service.ts
Normal file
395
apps/api/src/jobs/prefetch-worker.service.ts
Normal file
@@ -0,0 +1,395 @@
|
||||
import {
|
||||
Inject,
|
||||
Injectable,
|
||||
Logger,
|
||||
OnModuleDestroy,
|
||||
OnModuleInit,
|
||||
} from "@nestjs/common";
|
||||
import { Job, Queue, Worker } from "bullmq";
|
||||
import { eq, and, isNull } from "drizzle-orm";
|
||||
import { CATALOG_PREFETCH_QUEUE } from "./queues/catalog-prefetch.queue";
|
||||
import { getBullConnection, QUEUE_NAMES } from "./bull.config";
|
||||
import {
|
||||
PrefetchInitJobData,
|
||||
PrefetchCategoryJobData,
|
||||
} from "./prefetch.types";
|
||||
import {
|
||||
RateLimitError,
|
||||
checkCooldown,
|
||||
checkTimeWindow,
|
||||
initProgress,
|
||||
updateProgress,
|
||||
} from "./prefetch-utils";
|
||||
import { CategoriesService } from "../categories/categories.service";
|
||||
import { RedisService } from "../redis/redis.service";
|
||||
import { DATABASE, Database } from "../database/database.provider";
|
||||
import { categories, parts, vehicles } from "../database/schema/core";
|
||||
|
||||
const MAX_DEPTH = 5;
|
||||
|
||||
@Injectable()
|
||||
export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly logger = new Logger(PrefetchWorkerService.name);
|
||||
private worker: Worker | null = null;
|
||||
private pcatJobIndex = 0;
|
||||
|
||||
constructor(
|
||||
@Inject(CATALOG_PREFETCH_QUEUE) private queue: Queue,
|
||||
private categoriesService: CategoriesService,
|
||||
private redis: RedisService,
|
||||
@Inject(DATABASE) private db: Database,
|
||||
) {}
|
||||
|
||||
onModuleInit() {
|
||||
this.worker = new Worker(
|
||||
QUEUE_NAMES.CATALOG_PREFETCH,
|
||||
(job) => this.process(job),
|
||||
{
|
||||
connection: getBullConnection(),
|
||||
concurrency: 1,
|
||||
limiter: { max: 5, duration: 60_000 },
|
||||
},
|
||||
);
|
||||
|
||||
this.worker.on("failed", (job, err) => {
|
||||
if (err instanceof RateLimitError) {
|
||||
this.logger.debug(
|
||||
`[prefetch] Job ${job?.name} rate-limited, will retry in ${err.retryAfterMs}ms`,
|
||||
);
|
||||
} else {
|
||||
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=1, 5 jobs/min)");
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
if (this.worker) {
|
||||
await this.worker.close();
|
||||
this.worker = null;
|
||||
}
|
||||
}
|
||||
|
||||
private async process(job: Job): Promise<void> {
|
||||
const data = job.data as { source?: string };
|
||||
if (data.source === "parts-catalogs") {
|
||||
await new Promise((r) => setTimeout(r, 15_000));
|
||||
}
|
||||
|
||||
if (job.name === "prefetch-init") {
|
||||
return this.processInit(job as Job<PrefetchInitJobData>);
|
||||
}
|
||||
if (job.name === "prefetch-children") {
|
||||
return this.processChildren(job as Job<PrefetchCategoryJobData>);
|
||||
}
|
||||
if (job.name === "prefetch-parts") {
|
||||
return this.processParts(job as Job<PrefetchCategoryJobData>);
|
||||
}
|
||||
this.logger.warn(`[prefetch] Unknown job name: ${job.name}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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}`);
|
||||
|
||||
this.pcatJobIndex = 0;
|
||||
|
||||
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
|
||||
const topCategories = await this.db
|
||||
.select()
|
||||
.from(categories)
|
||||
.where(
|
||||
and(
|
||||
eq(categories.vehicleId, vehicleId),
|
||||
isNull(categories.parentId),
|
||||
),
|
||||
);
|
||||
|
||||
if (topCategories.length === 0) {
|
||||
this.logger.log(`[prefetch] No categories for vehicle=${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)) {
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Helpers ====================
|
||||
|
||||
private async queueCategoryJob(
|
||||
cat: { id: string; linkPath: string | null; source: string; unavailable: boolean },
|
||||
vehicleId: string,
|
||||
source: string,
|
||||
depth: number,
|
||||
): Promise<void> {
|
||||
if (cat.unavailable) return;
|
||||
|
||||
if (this.isLeafLinkPath(cat.linkPath, cat.source)) {
|
||||
// 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): boolean {
|
||||
if (!linkPath) return false;
|
||||
if (source === "emex") return true; // EMEX leaves always have a URL
|
||||
if (source === "parts-catalogs") return linkPath.startsWith("pcat:"); // pcat: prefix = leaf
|
||||
// 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> = {
|
||||
jobId: `prefetch:${data.vehicleId}:${data.categoryId}:${data.action}`,
|
||||
};
|
||||
|
||||
if (data.source === "parts-catalogs") {
|
||||
opts.delay = ++this.pcatJobIndex * 20_000;
|
||||
}
|
||||
|
||||
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}`);
|
||||
// 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,
|
||||
});
|
||||
}
|
||||
}
|
||||
16
apps/api/src/jobs/prefetch.types.ts
Normal file
16
apps/api/src/jobs/prefetch.types.ts
Normal file
@@ -0,0 +1,16 @@
|
||||
export type PrefetchSource = "pl24" | "emex" | "parts-catalogs";
|
||||
|
||||
/** Initial job: walks the category tree and queues sub-jobs */
|
||||
export interface PrefetchInitJobData {
|
||||
vehicleId: string;
|
||||
source: PrefetchSource;
|
||||
}
|
||||
|
||||
/** Per-category job: fetches children OR parts */
|
||||
export interface PrefetchCategoryJobData {
|
||||
vehicleId: string;
|
||||
categoryId: string;
|
||||
source: PrefetchSource;
|
||||
action: "children" | "parts";
|
||||
depth: number;
|
||||
}
|
||||
25
apps/api/src/jobs/queues/catalog-prefetch.queue.ts
Normal file
25
apps/api/src/jobs/queues/catalog-prefetch.queue.ts
Normal file
@@ -0,0 +1,25 @@
|
||||
import { Provider } from "@nestjs/common";
|
||||
import { Queue } from "bullmq";
|
||||
import { getBullConnection, getBullTelemetry, QUEUE_NAMES } from "../bull.config";
|
||||
|
||||
export const CATALOG_PREFETCH_QUEUE = "CATALOG_PREFETCH_QUEUE";
|
||||
|
||||
export const CatalogPrefetchQueueProvider: Provider = {
|
||||
provide: CATALOG_PREFETCH_QUEUE,
|
||||
useFactory: () => {
|
||||
const telemetry = getBullTelemetry();
|
||||
return new Queue(QUEUE_NAMES.CATALOG_PREFETCH, {
|
||||
connection: getBullConnection(),
|
||||
...(telemetry ? { telemetry } : {}),
|
||||
defaultJobOptions: {
|
||||
attempts: 3,
|
||||
backoff: {
|
||||
type: "exponential",
|
||||
delay: 30000,
|
||||
},
|
||||
removeOnComplete: { count: 1000 },
|
||||
removeOnFail: { count: 5000 },
|
||||
},
|
||||
});
|
||||
},
|
||||
};
|
||||
@@ -24,8 +24,9 @@ export class VehiclesController {
|
||||
async decode(
|
||||
@CurrentUser("id") userId: string,
|
||||
@Body("vin", VinValidationPipe) vin: string,
|
||||
@Body("pcatCarId") pcatCarId?: string,
|
||||
) {
|
||||
return this.vehiclesService.decodeVin(vin, userId);
|
||||
return this.vehiclesService.decodeVin(vin, userId, pcatCarId);
|
||||
}
|
||||
|
||||
@Get("history")
|
||||
@@ -63,6 +64,15 @@ export class VehiclesController {
|
||||
return { sent: true };
|
||||
}
|
||||
|
||||
@Get(":vehicleId/prefetch-status")
|
||||
async prefetchStatus(
|
||||
@Param("vehicleId") vehicleId: string,
|
||||
@CurrentUser("id") userId: string,
|
||||
) {
|
||||
await this.vehiclesService.getById(vehicleId, userId);
|
||||
return this.vehiclesService.getPrefetchStatus(vehicleId);
|
||||
}
|
||||
|
||||
@Get(":vehicleId/categories/:categoryId")
|
||||
async getCategoryParts(
|
||||
@Param("vehicleId") vehicleId: string,
|
||||
|
||||
@@ -5,11 +5,13 @@ import { CorgiModule } from "../integrations/corgi/corgi.module";
|
||||
import { PL24Module } from "../integrations/pl24/pl24.module";
|
||||
import { VinApiModule } from "../integrations/vin-api/vin-api.module";
|
||||
import { EmexModule } from "../integrations/emex/emex.module";
|
||||
import { PartsCatalogsModule } from "../integrations/parts-catalogs/parts-catalogs.module";
|
||||
import { BrandsModule } from "../brands/brands.module";
|
||||
import { CategoriesModule } from "../categories/categories.module";
|
||||
import { JobsModule } from "../jobs/jobs.module";
|
||||
|
||||
@Module({
|
||||
imports: [CorgiModule, PL24Module, VinApiModule, EmexModule, BrandsModule, CategoriesModule],
|
||||
imports: [CorgiModule, PL24Module, VinApiModule, EmexModule, PartsCatalogsModule, BrandsModule, CategoriesModule, JobsModule],
|
||||
controllers: [VehiclesController],
|
||||
providers: [VehiclesService],
|
||||
exports: [VehiclesService],
|
||||
|
||||
@@ -57,16 +57,27 @@ function createService(dbOrOverrides: any = {}) {
|
||||
del: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
|
||||
const partsCatalogsService = {
|
||||
decodeVin: vi.fn().mockResolvedValue(null),
|
||||
isSupported: vi.fn().mockReturnValue(true),
|
||||
};
|
||||
|
||||
const prefetchQueue = {
|
||||
add: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
|
||||
const service = new VehiclesService(
|
||||
db as any,
|
||||
prefetchQueue as any,
|
||||
corgiService as any,
|
||||
pl24Service as any,
|
||||
vinApiService as any,
|
||||
emexService as any,
|
||||
partsCatalogsService as any,
|
||||
redisService as any,
|
||||
);
|
||||
|
||||
return { service, db, corgiService, pl24Service, vinApiService, emexService, redisService };
|
||||
return { service, db, corgiService, pl24Service, vinApiService, emexService, partsCatalogsService, redisService };
|
||||
}
|
||||
|
||||
describe("VehiclesService", () => {
|
||||
|
||||
@@ -6,13 +6,27 @@ import {
|
||||
Logger,
|
||||
NotFoundException,
|
||||
} from "@nestjs/common";
|
||||
import { Queue } from "bullmq";
|
||||
import { eq, and, desc, or } from "drizzle-orm";
|
||||
import { DATABASE, Database } from "../database/database.provider";
|
||||
import { vehicles, queryLogs, brands, userBrands, userSubscriptions, plans } from "../database/schema/core";
|
||||
import { CATALOG_PREFETCH_QUEUE } from "../jobs/queues/catalog-prefetch.queue";
|
||||
import type { PrefetchSource } from "../jobs/prefetch.types";
|
||||
import {
|
||||
vehicles,
|
||||
userVehicles,
|
||||
queryLogs,
|
||||
brands,
|
||||
userBrands,
|
||||
userSubscriptions,
|
||||
plans,
|
||||
parts,
|
||||
} from "../database/schema/core";
|
||||
import { CorgiService } from "../integrations/corgi/corgi.service";
|
||||
import { PL24Service } from "../integrations/pl24/pl24.service";
|
||||
import { VinApiService } from "../integrations/vin-api/vin-api.service";
|
||||
import { EmexService } from "../integrations/emex/emex.service";
|
||||
import { PartsCatalogsService } from "../integrations/parts-catalogs/parts-catalogs.service";
|
||||
import type { PcatCar } from "../integrations/parts-catalogs/parts-catalogs.types";
|
||||
import { RedisService } from "../redis/redis.service";
|
||||
import { isValidVin } from "@sase/shared";
|
||||
|
||||
@@ -27,6 +41,8 @@ interface VinResolveResult {
|
||||
source: string;
|
||||
corgiKnown: boolean;
|
||||
corgiResult: any;
|
||||
/** When source is "parts-catalogs" and multiple cars found, these are the candidates */
|
||||
pcatCandidates?: PcatCar[];
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
@@ -35,57 +51,68 @@ export class VehiclesService {
|
||||
|
||||
constructor(
|
||||
@Inject(DATABASE) private db: Database,
|
||||
@Inject(CATALOG_PREFETCH_QUEUE) private prefetchQueue: Queue,
|
||||
private corgiService: CorgiService,
|
||||
private pl24Service: PL24Service,
|
||||
private vinApiService: VinApiService,
|
||||
private emexService: EmexService,
|
||||
private partsCatalogsService: PartsCatalogsService,
|
||||
private redis: RedisService,
|
||||
) {}
|
||||
|
||||
async decodeVin(vin: string, userId: string) {
|
||||
async decodeVin(vin: string, userId: string, pcatCarId?: string) {
|
||||
const startTime = Date.now();
|
||||
|
||||
if (!isValidVin(vin)) {
|
||||
throw new BadRequestException("Geçersiz şase numarası");
|
||||
}
|
||||
|
||||
// 1. Cache check
|
||||
const cached = await this.db
|
||||
// 1. Check for shared vehicle config by VIN (no userId filter)
|
||||
const [existing] = await this.db
|
||||
.select()
|
||||
.from(vehicles)
|
||||
.where(and(eq(vehicles.userId, userId), eq(vehicles.vin, vin)))
|
||||
.where(eq(vehicles.vin, vin))
|
||||
.limit(1);
|
||||
|
||||
if (cached.length > 0) {
|
||||
const vehicle = cached[0];
|
||||
const age = Date.now() - new Date(vehicle.updatedAt).getTime();
|
||||
if (existing && !pcatCarId) {
|
||||
const age = Date.now() - new Date(existing.updatedAt).getTime();
|
||||
if (age < 24 * 60 * 60 * 1000) {
|
||||
// Fresh cache (<24h)
|
||||
await this.logQuery(userId, vin, vehicle.brandId, "cache", true, Date.now() - startTime);
|
||||
return vehicle;
|
||||
// Fresh cache (<24h) — check brand access, link user, return
|
||||
if (existing.brandId) {
|
||||
await this.checkBrandAccess(userId, existing.brandId);
|
||||
}
|
||||
await this.ensureUserVehicleLink(userId, existing.id);
|
||||
await this.logQuery(userId, vin, existing.brandId, "cache", true, Date.now() - startTime);
|
||||
return existing;
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Resolve VIN via cached decode chain (Corgi → PL24 → EMEX)
|
||||
const resolved = await this.resolveVin(vin);
|
||||
// 2. Resolve VIN via cached decode chain (Corgi → PartsCatalogs → PL24 → EMEX)
|
||||
const resolved = await this.resolveVin(vin, pcatCarId);
|
||||
|
||||
if (!resolved) {
|
||||
await this.logQuery(userId, vin, null, "corgi", false, Date.now() - startTime, "Unknown VIN/brand");
|
||||
throw new BadRequestException("Şase numarası tanınamadı. Marka desteklenmiyor.");
|
||||
}
|
||||
|
||||
// 2b. If resolveVin returned multiple candidates, return them for frontend selection
|
||||
if (resolved.pcatCandidates && resolved.pcatCandidates.length > 1) {
|
||||
await this.logQuery(userId, vin, null, "parts-catalogs", true, Date.now() - startTime);
|
||||
return { candidates: resolved.pcatCandidates, vin, source: "parts-catalogs" };
|
||||
}
|
||||
|
||||
// 3. Brand access check
|
||||
let brandId: string | null = null;
|
||||
let brandName = resolved.brandName;
|
||||
const brandName = resolved.brandName;
|
||||
if (brandName) {
|
||||
const brand = await this.db
|
||||
const [brand] = await this.db
|
||||
.select()
|
||||
.from(brands)
|
||||
.where(eq(brands.name, brandName))
|
||||
.limit(1);
|
||||
|
||||
if (brand.length > 0) {
|
||||
brandId = brand[0].id;
|
||||
if (brand) {
|
||||
brandId = brand.id;
|
||||
await this.checkBrandAccess(userId, brandId);
|
||||
}
|
||||
}
|
||||
@@ -98,9 +125,8 @@ export class VehiclesService {
|
||||
source = vinApiData ? "vin-api" : "corgi";
|
||||
}
|
||||
|
||||
// 5. Save to DB
|
||||
// 5. Upsert shared vehicle config (ON CONFLICT vin → UPDATE)
|
||||
const vehicleData = {
|
||||
userId,
|
||||
vin,
|
||||
brandId,
|
||||
brandName,
|
||||
@@ -115,17 +141,21 @@ export class VehiclesService {
|
||||
updatedAt: new Date(),
|
||||
};
|
||||
|
||||
let savedVehicle;
|
||||
if (cached.length > 0) {
|
||||
const [updated] = await this.db
|
||||
.update(vehicles)
|
||||
.set(vehicleData)
|
||||
.where(eq(vehicles.id, cached[0].id))
|
||||
.returning();
|
||||
savedVehicle = updated;
|
||||
} else {
|
||||
const [inserted] = await this.db.insert(vehicles).values(vehicleData).returning();
|
||||
savedVehicle = inserted;
|
||||
const [savedVehicle] = await this.db
|
||||
.insert(vehicles)
|
||||
.values(vehicleData)
|
||||
.onConflictDoUpdate({
|
||||
target: vehicles.vin,
|
||||
set: vehicleData,
|
||||
})
|
||||
.returning();
|
||||
|
||||
// 6. Link user to this shared vehicle
|
||||
await this.ensureUserVehicleLink(userId, savedVehicle.id);
|
||||
|
||||
// 7. Schedule background catalog prefetch
|
||||
if (source === "pl24" || source === "emex" || source === "parts-catalogs") {
|
||||
await this.schedulePrefetch(savedVehicle.id, source as PrefetchSource);
|
||||
}
|
||||
|
||||
await this.logQuery(userId, vin, brandId, source, true, Date.now() - startTime);
|
||||
@@ -158,9 +188,16 @@ export class VehiclesService {
|
||||
|
||||
/**
|
||||
* Shared VIN decode chain with 5-minute Redis cache.
|
||||
* Corgi (offline) → PL24 → EMEX fallback.
|
||||
* Corgi (offline) → PartsCatalogs → PL24 fallback → EMEX fallback.
|
||||
*
|
||||
* @param pcatCarId If provided, skip resolve chain and use this specific PC car
|
||||
*/
|
||||
private async resolveVin(vin: string): Promise<VinResolveResult | null> {
|
||||
private async resolveVin(vin: string, pcatCarId?: string): Promise<VinResolveResult | null> {
|
||||
// If user selected a specific PC car from candidates, resolve directly
|
||||
if (pcatCarId) {
|
||||
return this.resolvePcatCarById(vin, pcatCarId);
|
||||
}
|
||||
|
||||
const cacheKey = `vin:resolve:${vin}`;
|
||||
const cached = await this.redis.getJson<VinResolveResult>(cacheKey);
|
||||
if (cached) {
|
||||
@@ -173,7 +210,44 @@ export class VehiclesService {
|
||||
const corgiKnown = !!(corgiResult && corgiResult.isKnown);
|
||||
let brandName = corgiKnown ? corgiResult.brandName : null;
|
||||
|
||||
// 2. PL24 decode
|
||||
// 2. PartsCatalogs (first external source)
|
||||
let pcatCandidates: PcatCar[] | null = null;
|
||||
try {
|
||||
const pcatResult = await this.partsCatalogsService.decodeVin(vin);
|
||||
if (pcatResult?.cars?.length === 1) {
|
||||
// Single car → use directly
|
||||
const car = pcatResult.cars[0];
|
||||
if (!brandName) brandName = this.extractBrandFromPcatCar(car) || null;
|
||||
const result: VinResolveResult = {
|
||||
brandName,
|
||||
model: car.name || null,
|
||||
year: this.extractYearFromPcatCar(car) || corgiResult?.modelYear || null,
|
||||
engine: this.extractParamFromPcatCar(car, "engine") || null,
|
||||
transmission: this.extractParamFromPcatCar(car, "transmission") || null,
|
||||
bodyType: this.extractParamFromPcatCar(car, "body") || null,
|
||||
rawData: {
|
||||
source: "parts-catalogs",
|
||||
catalogId: car.catalogId,
|
||||
carId: car.id,
|
||||
parameters: car.parameters || [],
|
||||
pcatCar: car,
|
||||
},
|
||||
source: "parts-catalogs",
|
||||
corgiKnown,
|
||||
corgiResult: corgiResult || null,
|
||||
};
|
||||
await this.redis.setJson(cacheKey, result, 300);
|
||||
return result;
|
||||
}
|
||||
if (pcatResult?.cars && pcatResult.cars.length > 1) {
|
||||
pcatCandidates = pcatResult.cars;
|
||||
this.logger.log(`PartsCatalogs returned ${pcatCandidates.length} candidates for ${vin}`);
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.warn(`PartsCatalogs decode failed for ${vin}: ${(err as Error).message}`);
|
||||
}
|
||||
|
||||
// 3. PL24 (if PC had multiple results, or PC failed entirely)
|
||||
let pl24Vehicle: any = null;
|
||||
if (this.pl24Service.isSupported(vin)) {
|
||||
try {
|
||||
@@ -186,7 +260,42 @@ export class VehiclesService {
|
||||
}
|
||||
}
|
||||
|
||||
// 3. EMEX fallback
|
||||
// If PL24 succeeded, use PL24 regardless of PC candidates
|
||||
if (pl24Vehicle) {
|
||||
const result: VinResolveResult = {
|
||||
brandName: brandName || corgiResult?.brandName || null,
|
||||
model: pl24Vehicle.model || null,
|
||||
year: pl24Vehicle.year || corgiResult?.modelYear || null,
|
||||
engine: pl24Vehicle.engineType || pl24Vehicle.engineCode || null,
|
||||
transmission: pl24Vehicle.transmission || null,
|
||||
bodyType: pl24Vehicle.bodyType || null,
|
||||
rawData: pl24Vehicle,
|
||||
source: "pl24",
|
||||
corgiKnown,
|
||||
corgiResult: corgiResult || null,
|
||||
};
|
||||
await this.redis.setJson(cacheKey, result, 300);
|
||||
return result;
|
||||
}
|
||||
|
||||
// 3b. PL24 failed + PC had multiple candidates → return candidates for user selection
|
||||
if (pcatCandidates && pcatCandidates.length > 1) {
|
||||
return {
|
||||
brandName,
|
||||
model: null,
|
||||
year: null,
|
||||
engine: null,
|
||||
transmission: null,
|
||||
bodyType: null,
|
||||
rawData: null,
|
||||
source: "parts-catalogs",
|
||||
corgiKnown,
|
||||
corgiResult: corgiResult || null,
|
||||
pcatCandidates,
|
||||
};
|
||||
}
|
||||
|
||||
// 4. EMEX fallback (slowest, browser-based)
|
||||
let emexVehicle: import("../integrations/emex/emex.types").DecodedVehicle | null = null;
|
||||
if (!pl24Vehicle) {
|
||||
try {
|
||||
@@ -201,25 +310,20 @@ export class VehiclesService {
|
||||
}
|
||||
|
||||
// Nothing recognized this VIN
|
||||
if (!pl24Vehicle && !emexVehicle && !corgiKnown) {
|
||||
if (!emexVehicle && !corgiKnown) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const source = pl24Vehicle ? "pl24" : emexVehicle ? "emex" : "corgi";
|
||||
const resolvedSource = emexVehicle ? "emex" : "corgi";
|
||||
const result: VinResolveResult = {
|
||||
brandName: brandName || corgiResult?.brandName || emexVehicle?.brand || null,
|
||||
model: pl24Vehicle?.model || emexVehicle?.model || null,
|
||||
year: pl24Vehicle?.year || emexVehicle?.year || corgiResult?.modelYear || null,
|
||||
engine:
|
||||
pl24Vehicle?.engineType ||
|
||||
pl24Vehicle?.engineCode ||
|
||||
emexVehicle?.engineCode ||
|
||||
emexVehicle?.engineType ||
|
||||
null,
|
||||
transmission: pl24Vehicle?.transmission || emexVehicle?.transmission || null,
|
||||
bodyType: pl24Vehicle?.bodyType || emexVehicle?.bodyType || null,
|
||||
rawData: pl24Vehicle || emexVehicle?.raw || null,
|
||||
source,
|
||||
model: emexVehicle?.model || null,
|
||||
year: emexVehicle?.year || corgiResult?.modelYear || null,
|
||||
engine: emexVehicle?.engineCode || emexVehicle?.engineType || null,
|
||||
transmission: emexVehicle?.transmission || null,
|
||||
bodyType: emexVehicle?.bodyType || null,
|
||||
rawData: emexVehicle?.raw || null,
|
||||
source: resolvedSource,
|
||||
corgiKnown,
|
||||
corgiResult: corgiResult || null,
|
||||
};
|
||||
@@ -228,38 +332,176 @@ export class VehiclesService {
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a specific PartsCatalogs car by ID (after user selects from candidates).
|
||||
*/
|
||||
private async resolvePcatCarById(vin: string, pcatCarId: string): Promise<VinResolveResult | null> {
|
||||
const corgiResult = this.corgiService.decodeVin(vin);
|
||||
const corgiKnown = !!(corgiResult && corgiResult.isKnown);
|
||||
let brandName = corgiKnown ? corgiResult.brandName : null;
|
||||
|
||||
// Re-decode VIN to get fresh car list, then find the selected car
|
||||
const pcatResult = await this.partsCatalogsService.decodeVin(vin);
|
||||
const car = pcatResult?.cars?.find((c) => c.id === pcatCarId);
|
||||
|
||||
if (!car) {
|
||||
this.logger.warn(`PartsCatalogs car ${pcatCarId} not found for ${vin}`);
|
||||
return null;
|
||||
}
|
||||
|
||||
if (!brandName) brandName = this.extractBrandFromPcatCar(car) || null;
|
||||
|
||||
return {
|
||||
brandName,
|
||||
model: car.name || null,
|
||||
year: this.extractYearFromPcatCar(car) || corgiResult?.modelYear || null,
|
||||
engine: this.extractParamFromPcatCar(car, "engine") || null,
|
||||
transmission: this.extractParamFromPcatCar(car, "transmission") || null,
|
||||
bodyType: this.extractParamFromPcatCar(car, "body") || null,
|
||||
rawData: {
|
||||
source: "parts-catalogs",
|
||||
catalogId: car.catalogId,
|
||||
carId: car.id,
|
||||
parameters: car.parameters || [],
|
||||
pcatCar: car,
|
||||
},
|
||||
source: "parts-catalogs",
|
||||
corgiKnown,
|
||||
corgiResult: corgiResult || null,
|
||||
};
|
||||
}
|
||||
|
||||
// ─── PartsCatalogs helpers ─────────────────────────────
|
||||
|
||||
/** Map common parts-catalogs catalog IDs to brand display names */
|
||||
private static readonly PCAT_CATALOG_BRAND_MAP: Record<string, string> = {
|
||||
vw: "Volkswagen", volkswagen: "Volkswagen",
|
||||
bmw: "BMW", mercedes: "Mercedes-Benz", "mercedes-benz": "Mercedes-Benz",
|
||||
audi: "Audi", porsche: "Porsche", skoda: "Skoda",
|
||||
seat: "Seat", ford: "Ford", opel: "Opel",
|
||||
renault: "Renault", peugeot: "Peugeot", citroen: "Citroen",
|
||||
fiat: "Fiat", toyota: "Toyota", honda: "Honda",
|
||||
hyundai: "Hyundai", kia: "Kia", nissan: "Nissan",
|
||||
mazda: "Mazda", subaru: "Subaru", volvo: "Volvo",
|
||||
jaguar: "Jaguar", "land-rover": "Land Rover", landrover: "Land Rover",
|
||||
mini: "Mini", dacia: "Dacia", suzuki: "Suzuki",
|
||||
mitsubishi: "Mitsubishi", chevrolet: "Chevrolet",
|
||||
};
|
||||
|
||||
private extractBrandFromPcatCar(car: PcatCar): string | null {
|
||||
if (!car.catalogId) return null;
|
||||
const key = car.catalogId.toLowerCase();
|
||||
return VehiclesService.PCAT_CATALOG_BRAND_MAP[key] || null;
|
||||
}
|
||||
|
||||
private extractYearFromPcatCar(car: PcatCar): number | null {
|
||||
if (!car.parameters) return null;
|
||||
const yearParam = car.parameters.find(
|
||||
(p) => p.key.toLowerCase().includes("year") || p.key.toLowerCase().includes("model_year"),
|
||||
);
|
||||
if (yearParam?.value) {
|
||||
const num = parseInt(yearParam.value, 10);
|
||||
if (num > 1900 && num < 2100) return num;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private extractParamFromPcatCar(car: PcatCar, keyword: string): string | null {
|
||||
if (!car.parameters) return null;
|
||||
const param = car.parameters.find((p) => p.key.toLowerCase().includes(keyword));
|
||||
return param?.value || null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get user's vehicle history via junction table.
|
||||
* Ordered by lastAccessedAt (most recent first).
|
||||
*/
|
||||
async getHistory(userId: string, page = 1, limit = 20) {
|
||||
const offset = (page - 1) * limit;
|
||||
return this.db
|
||||
.select()
|
||||
.from(vehicles)
|
||||
.where(eq(vehicles.userId, userId))
|
||||
.orderBy(desc(vehicles.updatedAt))
|
||||
.select({
|
||||
id: vehicles.id,
|
||||
vin: vehicles.vin,
|
||||
brandId: vehicles.brandId,
|
||||
brandName: vehicles.brandName,
|
||||
model: vehicles.model,
|
||||
year: vehicles.year,
|
||||
engine: vehicles.engine,
|
||||
transmission: vehicles.transmission,
|
||||
bodyType: vehicles.bodyType,
|
||||
market: vehicles.market,
|
||||
rawData: vehicles.rawData,
|
||||
source: vehicles.source,
|
||||
createdAt: vehicles.createdAt,
|
||||
updatedAt: vehicles.updatedAt,
|
||||
lastAccessedAt: userVehicles.lastAccessedAt,
|
||||
})
|
||||
.from(userVehicles)
|
||||
.innerJoin(vehicles, eq(userVehicles.vehicleId, vehicles.id))
|
||||
.where(eq(userVehicles.userId, userId))
|
||||
.orderBy(desc(userVehicles.lastAccessedAt))
|
||||
.limit(limit)
|
||||
.offset(offset);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get vehicle by ID — verify user has access via junction table.
|
||||
*/
|
||||
async getById(id: string, userId: string) {
|
||||
const result = await this.db
|
||||
.select()
|
||||
const [result] = await this.db
|
||||
.select({
|
||||
id: vehicles.id,
|
||||
vin: vehicles.vin,
|
||||
brandId: vehicles.brandId,
|
||||
brandName: vehicles.brandName,
|
||||
model: vehicles.model,
|
||||
year: vehicles.year,
|
||||
engine: vehicles.engine,
|
||||
transmission: vehicles.transmission,
|
||||
bodyType: vehicles.bodyType,
|
||||
market: vehicles.market,
|
||||
rawData: vehicles.rawData,
|
||||
source: vehicles.source,
|
||||
createdAt: vehicles.createdAt,
|
||||
updatedAt: vehicles.updatedAt,
|
||||
})
|
||||
.from(vehicles)
|
||||
.where(and(eq(vehicles.id, id), eq(vehicles.userId, userId)))
|
||||
.innerJoin(userVehicles, eq(userVehicles.vehicleId, vehicles.id))
|
||||
.where(and(eq(vehicles.id, id), eq(userVehicles.userId, userId)))
|
||||
.limit(1);
|
||||
|
||||
if (result.length === 0) throw new NotFoundException("Araç bulunamadı");
|
||||
return result[0];
|
||||
if (!result) throw new NotFoundException("Araç bulunamadı");
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete user's link to a vehicle (junction record only).
|
||||
* The shared vehicle config and its categories/parts remain intact.
|
||||
*/
|
||||
async deleteVehicle(id: string, userId: string) {
|
||||
const result = await this.db
|
||||
.delete(vehicles)
|
||||
.where(and(eq(vehicles.id, id), eq(vehicles.userId, userId)))
|
||||
.delete(userVehicles)
|
||||
.where(and(eq(userVehicles.vehicleId, id), eq(userVehicles.userId, userId)))
|
||||
.returning();
|
||||
|
||||
if (result.length === 0) throw new NotFoundException("Araç bulunamadı");
|
||||
return { deleted: true };
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure a user ↔ vehicle link exists in the junction table.
|
||||
* Uses ON CONFLICT to update lastAccessedAt if already linked.
|
||||
*/
|
||||
private async ensureUserVehicleLink(userId: string, vehicleId: string) {
|
||||
await this.db
|
||||
.insert(userVehicles)
|
||||
.values({ userId, vehicleId, lastAccessedAt: new Date() })
|
||||
.onConflictDoUpdate({
|
||||
target: [userVehicles.userId, userVehicles.vehicleId],
|
||||
set: { lastAccessedAt: new Date() },
|
||||
});
|
||||
}
|
||||
|
||||
private async checkBrandAccess(userId: string, brandId: string) {
|
||||
const [sub] = await this.db
|
||||
.select({
|
||||
@@ -295,6 +537,50 @@ export class VehiclesService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule background catalog prefetch for a vehicle.
|
||||
* Skips if vehicle already has parts in DB (= already prefetched).
|
||||
* Uses a short-lived Redis key to prevent duplicate scheduling within the delay window.
|
||||
*/
|
||||
private async schedulePrefetch(vehicleId: string, source: PrefetchSource) {
|
||||
// Skip if already queued (5min TTL covers the initial delay window)
|
||||
const key = `prefetch:scheduled:${vehicleId}`;
|
||||
const queued = await this.redis.exists(key);
|
||||
if (queued) return;
|
||||
|
||||
// Skip if vehicle already has parts in DB (= previously prefetched)
|
||||
const [partCheck] = await this.db
|
||||
.select({ id: parts.id })
|
||||
.from(parts)
|
||||
.where(eq(parts.vehicleId, vehicleId))
|
||||
.limit(1);
|
||||
if (partCheck) return;
|
||||
|
||||
try {
|
||||
await this.prefetchQueue.add(
|
||||
"prefetch-init",
|
||||
{ vehicleId, source },
|
||||
{
|
||||
jobId: `prefetch:${vehicleId}`,
|
||||
delay: 5 * 60 * 1000, // 5 min initial delay
|
||||
},
|
||||
);
|
||||
|
||||
await this.redis.set(key, "1", 600); // 10min TTL — just to prevent double-scheduling
|
||||
this.logger.log(`[prefetch] Scheduled for vehicle=${vehicleId}, source=${source}`);
|
||||
} catch (err) {
|
||||
this.logger.warn(`[prefetch] Failed to schedule: ${(err as Error).message}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get prefetch progress for a vehicle (Redis hash).
|
||||
*/
|
||||
async getPrefetchStatus(vehicleId: string) {
|
||||
const { getProgress } = await import("../jobs/prefetch-utils");
|
||||
return getProgress(this.redis, vehicleId);
|
||||
}
|
||||
|
||||
private async logQuery(
|
||||
userId: string,
|
||||
vin: string,
|
||||
|
||||
@@ -14,6 +14,7 @@ interface Category {
|
||||
schemaImageUrl?: string | null;
|
||||
parentId?: string | null;
|
||||
unavailable?: boolean;
|
||||
source?: string;
|
||||
}
|
||||
|
||||
interface CategoryGridProps {
|
||||
@@ -46,7 +47,8 @@ export function CategoryGrid({ categories, vehicleId }: CategoryGridProps) {
|
||||
(c) =>
|
||||
c.children !== undefined &&
|
||||
c.children.length === 0 &&
|
||||
!c.schemaImageUrl,
|
||||
!c.schemaImageUrl &&
|
||||
c.source !== "parts-catalogs",
|
||||
);
|
||||
|
||||
if (leafsWithoutImage.length === 0) {
|
||||
|
||||
@@ -14,6 +14,7 @@ interface Category {
|
||||
schemaImageUrl?: string | null;
|
||||
parentId?: string | null;
|
||||
unavailable?: boolean;
|
||||
source?: string;
|
||||
}
|
||||
|
||||
export function CategoryTree({ categories, vehicleId }: { categories: Category[]; vehicleId: string }) {
|
||||
@@ -68,7 +69,7 @@ function CategoryNode({ category, vehicleId, level, parentPrefetching }: {
|
||||
useEffect(() => {
|
||||
if (!expanded || prefetchedRef.current) return;
|
||||
const leafs = children.filter(
|
||||
(c) => c.children !== undefined && c.children.length === 0 && !c.schemaImageUrl,
|
||||
(c) => c.children !== undefined && c.children.length === 0 && !c.schemaImageUrl && c.source !== "parts-catalogs",
|
||||
);
|
||||
if (leafs.length === 0) return;
|
||||
prefetchedRef.current = true;
|
||||
|
||||
154
apps/web/src/components/vehicles/vehicle-select-modal.tsx
Normal file
154
apps/web/src/components/vehicles/vehicle-select-modal.tsx
Normal file
@@ -0,0 +1,154 @@
|
||||
import { useState } from "react";
|
||||
import {
|
||||
Dialog,
|
||||
DialogContent,
|
||||
DialogDescription,
|
||||
DialogHeader,
|
||||
DialogTitle,
|
||||
Button,
|
||||
Badge,
|
||||
Separator,
|
||||
} from "@sase/ui";
|
||||
import { Car, Loader2, ChevronRight } from "lucide-react";
|
||||
|
||||
interface PcatCandidate {
|
||||
id: string;
|
||||
name: string;
|
||||
description?: string;
|
||||
parameters?: Array<{ key: string; idx: string; value: string }>;
|
||||
catalogId: string;
|
||||
}
|
||||
|
||||
interface VehicleSelectModalProps {
|
||||
open: boolean;
|
||||
onClose: () => void;
|
||||
candidates: PcatCandidate[];
|
||||
vin: string;
|
||||
onSelect: (carId: string) => void;
|
||||
loading?: boolean;
|
||||
}
|
||||
|
||||
export function VehicleSelectModal({
|
||||
open,
|
||||
onClose,
|
||||
candidates,
|
||||
vin,
|
||||
onSelect,
|
||||
loading,
|
||||
}: VehicleSelectModalProps) {
|
||||
const [selectedId, setSelectedId] = useState<string | null>(null);
|
||||
|
||||
function getParamValue(
|
||||
params: PcatCandidate["parameters"],
|
||||
keyword: string,
|
||||
): string | null {
|
||||
if (!params) return null;
|
||||
const p = params.find((param) =>
|
||||
param.key.toLowerCase().includes(keyword),
|
||||
);
|
||||
return p?.value || null;
|
||||
}
|
||||
|
||||
return (
|
||||
<Dialog open={open} onOpenChange={(v) => !v && onClose()}>
|
||||
<DialogContent className="max-h-[85vh] overflow-y-auto sm:max-w-lg">
|
||||
<DialogHeader>
|
||||
<DialogTitle className="font-[family-name:var(--font-display)]">
|
||||
Araç Seçimi
|
||||
</DialogTitle>
|
||||
<DialogDescription>
|
||||
<span className="font-mono text-xs">{vin}</span> için birden fazla
|
||||
araç bulundu. Lütfen aracınızı seçin.
|
||||
</DialogDescription>
|
||||
</DialogHeader>
|
||||
|
||||
<Separator className="my-2" />
|
||||
|
||||
<div className="space-y-2">
|
||||
{candidates.map((car) => {
|
||||
const year = getParamValue(car.parameters, "year");
|
||||
const engine = getParamValue(car.parameters, "engine");
|
||||
const body = getParamValue(car.parameters, "body");
|
||||
const isSelected = selectedId === car.id;
|
||||
|
||||
return (
|
||||
<button
|
||||
key={car.id}
|
||||
type="button"
|
||||
onClick={() => setSelectedId(car.id)}
|
||||
className={`group flex w-full items-center gap-4 rounded-xl border p-4 text-left transition-colors ${
|
||||
isSelected
|
||||
? "border-primary bg-primary/5"
|
||||
: "border-border bg-background hover:bg-accent"
|
||||
}`}
|
||||
>
|
||||
<div
|
||||
className={`flex size-10 shrink-0 items-center justify-center rounded-xl ${
|
||||
isSelected
|
||||
? "bg-primary/10 text-primary"
|
||||
: "bg-muted text-muted-foreground"
|
||||
}`}
|
||||
>
|
||||
<Car className="size-5" />
|
||||
</div>
|
||||
<div className="min-w-0 flex-1">
|
||||
<p className="truncate text-sm font-medium">{car.name}</p>
|
||||
{car.description && (
|
||||
<p className="truncate text-xs text-muted-foreground">
|
||||
{car.description}
|
||||
</p>
|
||||
)}
|
||||
<div className="mt-1.5 flex flex-wrap gap-1.5">
|
||||
{year && (
|
||||
<Badge variant="secondary" className="text-[10px]">
|
||||
{year}
|
||||
</Badge>
|
||||
)}
|
||||
{engine && (
|
||||
<Badge variant="secondary" className="text-[10px]">
|
||||
{engine}
|
||||
</Badge>
|
||||
)}
|
||||
{body && (
|
||||
<Badge variant="secondary" className="text-[10px]">
|
||||
{body}
|
||||
</Badge>
|
||||
)}
|
||||
<Badge variant="outline" className="text-[10px]">
|
||||
{car.catalogId}
|
||||
</Badge>
|
||||
</div>
|
||||
</div>
|
||||
<ChevronRight
|
||||
className={`size-4 shrink-0 transition-colors ${
|
||||
isSelected
|
||||
? "text-primary"
|
||||
: "text-muted-foreground/50 group-hover:text-muted-foreground"
|
||||
}`}
|
||||
/>
|
||||
</button>
|
||||
);
|
||||
})}
|
||||
</div>
|
||||
|
||||
<Separator className="my-2" />
|
||||
|
||||
<div className="flex justify-end gap-2">
|
||||
<Button variant="outline" onClick={onClose} className="rounded-xl">
|
||||
Vazgeç
|
||||
</Button>
|
||||
<Button
|
||||
disabled={!selectedId || loading}
|
||||
onClick={() => selectedId && onSelect(selectedId)}
|
||||
className="rounded-xl"
|
||||
>
|
||||
{loading ? (
|
||||
<Loader2 className="mr-2 size-4 animate-spin" />
|
||||
) : null}
|
||||
Seç ve Devam Et
|
||||
</Button>
|
||||
</div>
|
||||
</DialogContent>
|
||||
</Dialog>
|
||||
);
|
||||
}
|
||||
@@ -14,6 +14,7 @@ import { api, ApiError } from "@/lib/api-client";
|
||||
import { startAction } from "@/lib/faro";
|
||||
import { capture } from "@/lib/posthog";
|
||||
import { toast } from "@/lib/toast";
|
||||
import { VehicleSelectModal } from "@/components/vehicles/vehicle-select-modal";
|
||||
|
||||
// ─── HELPERS ──────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -51,6 +52,11 @@ function SearchPage() {
|
||||
const [reportSending, setReportSending] = useState(false);
|
||||
const [reportSent, setReportSent] = useState(false);
|
||||
|
||||
// Vehicle candidate selection (PartsCatalogs multi-result)
|
||||
const [candidates, setCandidates] = useState<any[] | null>(null);
|
||||
const [candidateVin, setCandidateVin] = useState("");
|
||||
const [selectLoading, setSelectLoading] = useState(false);
|
||||
|
||||
// Live preview state
|
||||
const [preview, setPreview] = useState<{
|
||||
brandName: string;
|
||||
@@ -142,6 +148,19 @@ function SearchPage() {
|
||||
setLoading(true);
|
||||
try {
|
||||
const data = await api.post<any>("/vehicles/decode", { vin: cleanVin });
|
||||
|
||||
// Handle multiple vehicle candidates (PartsCatalogs)
|
||||
if (data.candidates && Array.isArray(data.candidates)) {
|
||||
setCandidates(data.candidates);
|
||||
setCandidateVin(cleanVin);
|
||||
capture("vin_decode_candidates", {
|
||||
vin: cleanVin,
|
||||
count: data.candidates.length,
|
||||
});
|
||||
setLoading(false);
|
||||
return;
|
||||
}
|
||||
|
||||
capture("vin_decode_success", { vin: cleanVin, vehicle_id: data.id });
|
||||
navigate({
|
||||
to: "/dashboard/vehicles/$id",
|
||||
@@ -190,6 +209,36 @@ function SearchPage() {
|
||||
}
|
||||
}
|
||||
|
||||
async function handleCandidateSelect(pcatCarId: string) {
|
||||
setSelectLoading(true);
|
||||
try {
|
||||
const data = await api.post<any>("/vehicles/decode", {
|
||||
vin: candidateVin,
|
||||
pcatCarId,
|
||||
});
|
||||
capture("vin_decode_candidate_selected", {
|
||||
vin: candidateVin,
|
||||
pcatCarId,
|
||||
vehicle_id: data.id,
|
||||
});
|
||||
setCandidates(null);
|
||||
navigate({
|
||||
to: "/dashboard/vehicles/$id",
|
||||
params: { id: data.id },
|
||||
});
|
||||
} catch (err) {
|
||||
const message =
|
||||
err instanceof ApiError
|
||||
? err.message
|
||||
: "Bir hata oluştu. Lütfen tekrar deneyin.";
|
||||
setError(message);
|
||||
setCandidates(null);
|
||||
toast.error("Araç seçimi başarısız");
|
||||
} finally {
|
||||
setSelectLoading(false);
|
||||
}
|
||||
}
|
||||
|
||||
function fillExampleVin() {
|
||||
setVin("WVWZZZ1JZ3W597935");
|
||||
inputRef.current?.focus();
|
||||
@@ -424,6 +473,18 @@ function SearchPage() {
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* ─── Vehicle Selection Modal (PartsCatalogs multi-result) ──── */}
|
||||
{candidates && (
|
||||
<VehicleSelectModal
|
||||
open={!!candidates}
|
||||
onClose={() => setCandidates(null)}
|
||||
candidates={candidates}
|
||||
vin={candidateVin}
|
||||
onSelect={handleCandidateSelect}
|
||||
loading={selectLoading}
|
||||
/>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user