feat(carcatonline): boş PL24 kataloglarının kategori ağacını gece carcatonline'dan doldur #273
@@ -81,6 +81,7 @@ function makeService(opts: {
|
||||
redis as never,
|
||||
pl24Service as never,
|
||||
{} as never,
|
||||
{ isEnabled: () => false, fetchAndStoreParts: vi.fn().mockResolvedValue(null) } as any,
|
||||
) as never as {
|
||||
healStaleBrowseRows: (brand: string, rows: Row[], where: unknown[]) => Promise<Row[] | null>;
|
||||
};
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { Module } from "@nestjs/common";
|
||||
import { CarcatonlineCatalogService } from "../integrations/carcatonline/carcatonline-catalog.service";
|
||||
import { PL24Module } from "../integrations/pl24/pl24.module";
|
||||
import { StorageModule } from "../storage/storage.module";
|
||||
import { SubscriptionsModule } from "../subscriptions/subscriptions.module";
|
||||
import { TranslationsModule } from "../translations/translations.module";
|
||||
import { CatalogController } from "./catalog.controller";
|
||||
import { CatalogService } from "./catalog.service";
|
||||
import { EmexCatalogController } from "./emex-catalog.controller";
|
||||
@@ -10,9 +12,9 @@ import { PcatCatalogController } from "./pcat-catalog.controller";
|
||||
import { PcatCatalogService } from "./pcat-catalog.service";
|
||||
|
||||
@Module({
|
||||
imports: [PL24Module, SubscriptionsModule, StorageModule],
|
||||
imports: [PL24Module, SubscriptionsModule, StorageModule, TranslationsModule],
|
||||
controllers: [CatalogController, EmexCatalogController, PcatCatalogController],
|
||||
providers: [CatalogService, EmexCatalogService, PcatCatalogService],
|
||||
providers: [CatalogService, EmexCatalogService, PcatCatalogService, CarcatonlineCatalogService],
|
||||
exports: [CatalogService, EmexCatalogService, PcatCatalogService],
|
||||
})
|
||||
export class CatalogModule {}
|
||||
|
||||
@@ -19,6 +19,7 @@ import {
|
||||
userBrands,
|
||||
userSubscriptions,
|
||||
} from "../database/schema/core";
|
||||
import { CarcatonlineCatalogService } from "../integrations/carcatonline/carcatonline-catalog.service";
|
||||
import { isPl24LeafNode, isStaleBrowseArchitecture } from "../integrations/pl24/pl24-tree";
|
||||
import { PL24Service } from "../integrations/pl24/pl24.service";
|
||||
import {
|
||||
@@ -40,6 +41,7 @@ export class CatalogService {
|
||||
private redis: RedisService,
|
||||
private pl24Service: PL24Service,
|
||||
private storage: StorageService,
|
||||
private carcatonline: CarcatonlineCatalogService,
|
||||
) {}
|
||||
|
||||
/**
|
||||
@@ -1021,7 +1023,9 @@ export class CatalogService {
|
||||
};
|
||||
}
|
||||
|
||||
if (linkPath && !this.isLeafPath(linkPath)) {
|
||||
// carcatonline trees are inserted in full by the night backfill; a node
|
||||
// without DB children is a leaf (parts below) — never send its link to PL24.
|
||||
if (linkPath && category.source !== "carcatonline" && !this.isLeafPath(linkPath)) {
|
||||
// Try to fetch subgroups
|
||||
const subGroups = await this.pl24Service.fetchSubGroupsByPath(
|
||||
linkPath,
|
||||
@@ -1208,6 +1212,18 @@ export class CatalogService {
|
||||
`Failed to fetch parts for category ${categoryId}: ${(err as Error).message}`,
|
||||
);
|
||||
}
|
||||
} else if ((needParts || needImage) && category.source === "carcatonline") {
|
||||
// Leaf seeded by the carcatonline backfill: fetch its parts + plate image on
|
||||
// demand (one paced upstream call, shared lockout/budget with the worker).
|
||||
const fetched = await this.carcatonline.fetchAndStoreParts({
|
||||
id: categoryId,
|
||||
catalogVehicleId,
|
||||
linkPath,
|
||||
});
|
||||
if (fetched) {
|
||||
if (needParts && fetched.parts.length > 0) dbParts = fetched.parts;
|
||||
if (needImage && fetched.pic) pics.push(fetched.pic);
|
||||
}
|
||||
}
|
||||
|
||||
// Parse hotspots
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { mapCarcatParts, parseCarcatLinkPath } from "./carcatonline-catalog.service";
|
||||
import type { CarcatPartsResponse } from "./carcatonline.client";
|
||||
|
||||
describe("parseCarcatLinkPath", () => {
|
||||
it("parses the carcat:<catalog>:<car>:<group> link path and rejects PL24 paths", () => {
|
||||
expect(parseCarcatLinkPath("carcat:pl_renault:pl_0bb9:pl_3f0b")).toEqual({
|
||||
catalogId: "pl_renault",
|
||||
carId: "pl_0bb9",
|
||||
groupId: "pl_3f0b",
|
||||
});
|
||||
expect(parseCarcatLinkPath("/p5renault/extern/details/bomDetails?catalog=XFE")).toBeNull();
|
||||
expect(parseCarcatLinkPath(null)).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("mapCarcatParts", () => {
|
||||
const payload: CarcatPartsResponse = {
|
||||
img: "static/images/renault/01063629.jpeg",
|
||||
partGroups: [
|
||||
{
|
||||
name: "Pedals",
|
||||
parts: [
|
||||
{
|
||||
id: "1",
|
||||
number: "03L 100 032 T",
|
||||
name: "COVER-PEDAL",
|
||||
positionNumber: "1",
|
||||
description: {
|
||||
qty: "\r\n\r\n 1",
|
||||
remark: "1.6ltr.",
|
||||
restriction: "+ FAMILY = KADJAR\n",
|
||||
},
|
||||
},
|
||||
{ id: "2", number: "253003RA0A", name: "SW ASSY-ASCD CANCEL", positionNumber: "10" },
|
||||
{ id: "3", number: "", name: "", positionNumber: "99" },
|
||||
],
|
||||
},
|
||||
],
|
||||
positions: [
|
||||
{ number: "1", coordinates: [342, 415, 15, 23] },
|
||||
{ number: "10", coordinates: [100, 50, 12, 12] },
|
||||
{ number: "bad", coordinates: [1] },
|
||||
],
|
||||
};
|
||||
|
||||
it("normalises OE codes, keeps originals, translates names and builds hotspots", () => {
|
||||
const tr = (raw: string) => (raw === "COVER-PEDAL" ? "PEDAL KAPAĞI" : raw);
|
||||
const out = mapCarcatParts(payload, { catalogVehicleId: "cv", categoryId: "cat" }, tr);
|
||||
expect(out.rows).toHaveLength(2); // the empty part is dropped
|
||||
expect(out.rows[0]).toMatchObject({
|
||||
catalogVehicleId: "cv",
|
||||
categoryId: "cat",
|
||||
oemCode: "03L100032T",
|
||||
name: "PEDAL KAPAĞI",
|
||||
nameOriginal: "COVER-PEDAL",
|
||||
quantity: 1,
|
||||
position: "1",
|
||||
hotspotIndex: 1,
|
||||
remark: "1.6ltr. | + FAMILY = KADJAR",
|
||||
source: "carcatonline",
|
||||
});
|
||||
expect(out.rows[1]).toMatchObject({
|
||||
oemCode: "253003RA0A",
|
||||
name: "SW ASSY-ASCD CANCEL",
|
||||
hotspotIndex: 10,
|
||||
});
|
||||
expect(out.imageUrl).toBe("https://api.carcatonline.com/static/images/renault/01063629.jpeg");
|
||||
expect(out.hotspots.items).toEqual([
|
||||
{ key: "1", areas: [{ left: 342, top: 415, width: 15, height: 23 }] },
|
||||
{ key: "10", areas: [{ left: 100, top: 50, width: 12, height: 12 }] },
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,188 @@
|
||||
import { Inject, Injectable, Logger } from "@nestjs/common";
|
||||
import { DATABASE, type Database } from "../../database/database.provider";
|
||||
import { parts, schemaPics } from "../../database/schema/core";
|
||||
import { RedisService } from "../../redis/redis.service";
|
||||
import { TranslationsService } from "../../translations/translations.service";
|
||||
import {
|
||||
type CarcatPartsResponse,
|
||||
CarcatonlineClient,
|
||||
CarcatonlineRateLimitError,
|
||||
carcatImageUrl,
|
||||
} from "./carcatonline.client";
|
||||
import {
|
||||
CarcatonlineThrottle,
|
||||
type RedisLike,
|
||||
carcatConfigFromEnv,
|
||||
redisTokenStore,
|
||||
} from "./carcatonline.pacing";
|
||||
|
||||
/** `categories.link_path` for carcatonline nodes: `carcat:<catalogId>:<carId>:<groupId>`. */
|
||||
export function parseCarcatLinkPath(linkPath: string | null | undefined): {
|
||||
catalogId: string;
|
||||
carId: string;
|
||||
groupId: string;
|
||||
} | null {
|
||||
if (!linkPath) return null;
|
||||
const m = linkPath.match(/^carcat:([^:]+):([^:]+):([^:]+)$/);
|
||||
return m ? { catalogId: m[1], carId: m[2], groupId: m[3] } : null;
|
||||
}
|
||||
|
||||
/** Map a parts2 payload to our `parts` rows and `schema_pics` hotspots. */
|
||||
export function mapCarcatParts(
|
||||
payload: CarcatPartsResponse,
|
||||
ctx: { catalogVehicleId: string; categoryId: string },
|
||||
translate: (raw: string) => string,
|
||||
) {
|
||||
const rows: (typeof parts.$inferInsert)[] = [];
|
||||
for (const group of payload.partGroups ?? []) {
|
||||
for (const p of group.parts ?? []) {
|
||||
if (!p.number && !p.name) continue;
|
||||
const clean = (s: string | undefined | null): string =>
|
||||
(s ?? "")
|
||||
.replace(/\r\n|\r|\n/g, " ")
|
||||
.replace(/\s+/g, " ")
|
||||
.trim();
|
||||
const rawName = clean(p.name) || clean(group.name) || "—";
|
||||
const remarkBits = [clean(p.description?.remark), clean(p.description?.restriction)].filter(
|
||||
Boolean,
|
||||
);
|
||||
const qtyText = clean(p.description?.qty);
|
||||
const qty = qtyText ? Number.parseInt(qtyText, 10) : Number.NaN;
|
||||
const hotspot = p.positionNumber ? Number.parseInt(p.positionNumber, 10) : Number.NaN;
|
||||
rows.push({
|
||||
catalogVehicleId: ctx.catalogVehicleId,
|
||||
vehicleId: null,
|
||||
categoryId: ctx.categoryId,
|
||||
oemCode: (p.number || "N/A").replace(/\s+/g, "").toUpperCase(),
|
||||
name: translate(rawName),
|
||||
nameOriginal: rawName,
|
||||
description: clean(p.description?.modelDescription) || clean(p.notice) || null,
|
||||
quantity: Number.isFinite(qty) && qty > 0 ? qty : null,
|
||||
position: p.positionNumber || null,
|
||||
hotspotIndex:
|
||||
Number.isFinite(hotspot) && hotspot > 0 && hotspot <= 2147483647 ? hotspot : null,
|
||||
unavailable: false,
|
||||
remark: remarkBits.length ? remarkBits.join(" | ").slice(0, 2000) : null,
|
||||
modelCodes: null,
|
||||
presel: false,
|
||||
price: null,
|
||||
currency: null,
|
||||
source: "carcatonline",
|
||||
});
|
||||
}
|
||||
}
|
||||
const hotspots = {
|
||||
width: null as number | null,
|
||||
height: null as number | null,
|
||||
items: (payload.positions ?? [])
|
||||
.filter((pos) => Array.isArray(pos.coordinates) && pos.coordinates.length >= 4)
|
||||
.map((pos) => ({
|
||||
key: pos.number,
|
||||
areas: [
|
||||
{
|
||||
left: pos.coordinates[0],
|
||||
top: pos.coordinates[1],
|
||||
width: pos.coordinates[2],
|
||||
height: pos.coordinates[3],
|
||||
},
|
||||
],
|
||||
})),
|
||||
};
|
||||
return { rows, imageUrl: carcatImageUrl(payload.img), hotspots };
|
||||
}
|
||||
|
||||
/**
|
||||
* API-side carcatonline access: on-demand parts for a category that the night
|
||||
* backfill seeded. Shares the Redis token / lockout / daily-budget / pacing with
|
||||
* the worker, so a user click never bursts the upstream API.
|
||||
*/
|
||||
@Injectable()
|
||||
export class CarcatonlineCatalogService {
|
||||
private readonly logger = new Logger(CarcatonlineCatalogService.name);
|
||||
private client: CarcatonlineClient | null = null;
|
||||
|
||||
constructor(
|
||||
@Inject(DATABASE) private readonly db: Database,
|
||||
private readonly redis: RedisService,
|
||||
private readonly translations: TranslationsService,
|
||||
) {}
|
||||
|
||||
isEnabled(): boolean {
|
||||
return process.env.CARCATONLINE_ENABLED === "true" && !!process.env.CARCATONLINE_EMAIL;
|
||||
}
|
||||
|
||||
private getClient(): CarcatonlineClient {
|
||||
if (!this.client) {
|
||||
const email = process.env.CARCATONLINE_EMAIL;
|
||||
const password = process.env.CARCATONLINE_PASSWORD;
|
||||
if (!email || !password) throw new Error("carcatonline credentials are not configured");
|
||||
this.client = new CarcatonlineClient(
|
||||
{
|
||||
email,
|
||||
password,
|
||||
logger: { log: (m) => this.logger.log(m), warn: (m) => this.logger.warn(m) },
|
||||
},
|
||||
redisTokenStore(this.redis.getClient() as unknown as RedisLike),
|
||||
);
|
||||
}
|
||||
return this.client;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch + persist parts and the schema image for one carcatonline leaf.
|
||||
* Returns null (and logs) on any upstream problem so the catalog page still renders.
|
||||
*/
|
||||
async fetchAndStoreParts(category: {
|
||||
id: string;
|
||||
catalogVehicleId: string | null;
|
||||
linkPath: string | null;
|
||||
}): Promise<{
|
||||
parts: (typeof parts.$inferSelect)[];
|
||||
pic: typeof schemaPics.$inferSelect | null;
|
||||
} | null> {
|
||||
if (!this.isEnabled() || !category.catalogVehicleId) return null;
|
||||
const ref = parseCarcatLinkPath(category.linkPath);
|
||||
if (!ref) return null;
|
||||
const redis = this.redis.getClient() as unknown as RedisLike;
|
||||
const throttle = new CarcatonlineThrottle(redis, carcatConfigFromEnv());
|
||||
try {
|
||||
await throttle.beforeCall();
|
||||
const payload = await this.getClient().parts(ref.catalogId, ref.carId, ref.groupId);
|
||||
const rawNames = [
|
||||
...new Set((payload.partGroups ?? []).flatMap((g) => g.parts.map((p) => p.name || g.name))),
|
||||
];
|
||||
const trMap = await this.translations.translateMany(rawNames.filter(Boolean));
|
||||
const mapped = mapCarcatParts(
|
||||
payload,
|
||||
{ catalogVehicleId: category.catalogVehicleId, categoryId: category.id },
|
||||
(raw) => trMap.get(raw) ?? raw,
|
||||
);
|
||||
const insertedParts = mapped.rows.length
|
||||
? await this.db.insert(parts).values(mapped.rows).onConflictDoNothing().returning()
|
||||
: [];
|
||||
let pic: typeof schemaPics.$inferSelect | null = null;
|
||||
if (mapped.imageUrl) {
|
||||
[pic] = await this.db
|
||||
.insert(schemaPics)
|
||||
.values({
|
||||
categoryId: category.id,
|
||||
imageUrl: mapped.imageUrl,
|
||||
hotspots: JSON.stringify(mapped.hotspots),
|
||||
source: "carcatonline",
|
||||
})
|
||||
.returning();
|
||||
}
|
||||
return { parts: insertedParts, pic };
|
||||
} catch (err) {
|
||||
if (err instanceof CarcatonlineRateLimitError) {
|
||||
await throttle.markLockout();
|
||||
this.logger.warn(`carcatonline rate-limited on ${ref.groupId} — lockout marked`);
|
||||
return null;
|
||||
}
|
||||
this.logger.warn(
|
||||
`carcatonline parts fetch failed for ${category.id}: ${(err as Error).message}`,
|
||||
);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
CarcatonlineAuthError,
|
||||
CarcatonlineClient,
|
||||
CarcatonlineHttpError,
|
||||
CarcatonlineRateLimitError,
|
||||
carcatImageUrl,
|
||||
loginCarcatonline,
|
||||
} from "./carcatonline.client";
|
||||
|
||||
const b64url = (s: string) => Buffer.from(s).toString("base64url");
|
||||
const jwt = (exp: number) =>
|
||||
`${b64url('{"alg":"HS256"}')}.${b64url(JSON.stringify({ sub: "e@x", exp }))}.sig`;
|
||||
|
||||
describe("loginCarcatonline", () => {
|
||||
it("posts the htmx login form, replays the session cookie and extracts the widget token", async () => {
|
||||
const exp = Math.floor(Date.now() / 1000) + 180 * 86400;
|
||||
const calls: { url: string; init?: RequestInit }[] = [];
|
||||
const fetchImpl = vi.fn(async (url: string | URL, init?: RequestInit) => {
|
||||
calls.push({ url: String(url), init });
|
||||
if (String(url).endsWith("/login")) {
|
||||
return new Response("<div>ok</div>", {
|
||||
status: 200,
|
||||
headers: { "set-cookie": 'session="abc"; HttpOnly; Path=/' },
|
||||
});
|
||||
}
|
||||
return new Response(
|
||||
`<script src="https://vinside.carcatonline.com/embed.js" data-token="${jwt(exp)}" data-language="en">`,
|
||||
{
|
||||
status: 200,
|
||||
},
|
||||
);
|
||||
});
|
||||
const t = await loginCarcatonline({
|
||||
email: "e@x",
|
||||
password: "pw",
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
});
|
||||
expect(t.token).toBe(jwt(exp));
|
||||
expect(t.expiresAt).toBe(exp * 1000);
|
||||
expect(String(calls[0].init?.body)).toBe("email=e%40x&password=pw");
|
||||
expect((calls[1].init?.headers as Record<string, string>).Cookie).toBe('session="abc"');
|
||||
});
|
||||
|
||||
it("fails clearly on wrong credentials (no session cookie)", async () => {
|
||||
const fetchImpl = vi.fn(async () => new Response("<form>", { status: 200 }));
|
||||
await expect(
|
||||
loginCarcatonline({
|
||||
email: "e",
|
||||
password: "bad",
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
}),
|
||||
).rejects.toBeInstanceOf(CarcatonlineAuthError);
|
||||
});
|
||||
});
|
||||
|
||||
describe("CarcatonlineClient.get", () => {
|
||||
const store = (token = jwt(Math.floor(Date.now() / 1000) + 30 * 86400)) => {
|
||||
let cur: { token: string; expiresAt: number } | null = {
|
||||
token,
|
||||
expiresAt: Date.now() + 30 * 86400 * 1000,
|
||||
};
|
||||
return {
|
||||
get: async () => cur,
|
||||
set: async (t: { token: string; expiresAt: number }) => {
|
||||
cur = t;
|
||||
},
|
||||
clear: async () => {
|
||||
cur = null;
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
it("sends the bearer token and surfaces 429 as a rate-limit error", async () => {
|
||||
const fetchImpl = vi.fn(async (_url: string | URL, init?: RequestInit) => {
|
||||
expect((init?.headers as Record<string, string>).Authorization).toMatch(/^Bearer /);
|
||||
return new Response('{"detail":"Customer Hourly VIN Limit Exceeded"}', { status: 429 });
|
||||
});
|
||||
const client = new CarcatonlineClient(
|
||||
{ email: "e", password: "p", fetchImpl: fetchImpl as unknown as typeof fetch },
|
||||
store(),
|
||||
);
|
||||
await expect(client.groups("pl_renault", "car")).rejects.toBeInstanceOf(
|
||||
CarcatonlineRateLimitError,
|
||||
);
|
||||
});
|
||||
|
||||
it("maps other HTTP errors with the API's detail", async () => {
|
||||
const fetchImpl = vi.fn(
|
||||
async () => new Response('{"detail":"No vehicle found"}', { status: 404 }),
|
||||
);
|
||||
const client = new CarcatonlineClient(
|
||||
{ email: "e", password: "p", fetchImpl: fetchImpl as unknown as typeof fetch },
|
||||
store(),
|
||||
);
|
||||
await expect(client.cars("pl_renault", "m", [])).rejects.toMatchObject({
|
||||
name: "CarcatonlineHttpError",
|
||||
status: 404,
|
||||
detail: "No vehicle found",
|
||||
} satisfies Partial<CarcatonlineHttpError>);
|
||||
});
|
||||
|
||||
it("builds the cascading parameter query", async () => {
|
||||
const seen: string[] = [];
|
||||
const fetchImpl = vi.fn(async (url: string | URL) => {
|
||||
seen.push(String(url));
|
||||
return new Response("[]", { status: 200 });
|
||||
});
|
||||
const client = new CarcatonlineClient(
|
||||
{ email: "e", password: "p", fetchImpl: fetchImpl as unknown as typeof fetch },
|
||||
store(),
|
||||
);
|
||||
await client.cars("pl_renault", "m1", ["a", "b"]);
|
||||
expect(seen[0]).toBe(
|
||||
"https://api.carcatonline.com/v1/catalogs/pl_renault/cars2/?modelId=m1¶meter=a¶meter=b",
|
||||
);
|
||||
});
|
||||
|
||||
it("resolves relative image paths against the API origin", () => {
|
||||
expect(carcatImageUrl("static/images/renault/01063629.jpeg")).toBe(
|
||||
"https://api.carcatonline.com/static/images/renault/01063629.jpeg",
|
||||
);
|
||||
expect(carcatImageUrl(null)).toBeNull();
|
||||
});
|
||||
});
|
||||
304
apps/api/src/integrations/carcatonline/carcatonline.client.ts
Normal file
304
apps/api/src/integrations/carcatonline/carcatonline.client.ts
Normal file
@@ -0,0 +1,304 @@
|
||||
/**
|
||||
* pro.carcatonline.com ("VINSide") client — a PartsLink24 mirror behind a
|
||||
* simple auth: `POST /login` (email+password) → session cookie → `GET /widget`
|
||||
* whose HTML embeds a 6-month JWT → `https://api.carcatonline.com/v1` with
|
||||
* `Authorization: Bearer <jwt>`. Verified live 2026-09-25; see
|
||||
* /home/s/ss/carcatonline-vinside-kesif-2026-09-25.md.
|
||||
*
|
||||
* Quotas (measured): VIN lookups ~30 per sliding window; browse endpoints lock
|
||||
* the whole API for ~36 min after a burst (~200 calls / 3 min) but sustain
|
||||
* 1 call / 2 s. The caller (worker) paces requests; this client only reports
|
||||
* 429 as `CarcatonlineRateLimitError` so the caller can back off.
|
||||
*/
|
||||
|
||||
export const CARCATONLINE_DEFAULTS = {
|
||||
portalUrl: "https://pro.carcatonline.com",
|
||||
apiUrl: "https://api.carcatonline.com/v1",
|
||||
} as const;
|
||||
|
||||
const USER_AGENT =
|
||||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36";
|
||||
|
||||
export class CarcatonlineRateLimitError extends Error {
|
||||
constructor(readonly path: string) {
|
||||
super(`carcatonline rate limit (429) on ${path}`);
|
||||
this.name = "CarcatonlineRateLimitError";
|
||||
}
|
||||
}
|
||||
|
||||
export class CarcatonlineAuthError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "CarcatonlineAuthError";
|
||||
}
|
||||
}
|
||||
|
||||
export class CarcatonlineHttpError extends Error {
|
||||
constructor(
|
||||
readonly status: number,
|
||||
readonly path: string,
|
||||
readonly detail: string,
|
||||
) {
|
||||
super(`carcatonline HTTP ${status} on ${path}: ${detail}`);
|
||||
this.name = "CarcatonlineHttpError";
|
||||
}
|
||||
}
|
||||
|
||||
export interface CarcatonlineToken {
|
||||
token: string;
|
||||
/** Epoch ms from the JWT `exp` claim. */
|
||||
expiresAt: number;
|
||||
}
|
||||
|
||||
export interface CarcatCatalog {
|
||||
id: string; // "pl_renault"
|
||||
name: string;
|
||||
img?: string;
|
||||
modelsCount?: number;
|
||||
}
|
||||
|
||||
export interface CarcatModel {
|
||||
id: string; // "pl_<hash>"
|
||||
name: string; // "KADJAR"
|
||||
img?: string;
|
||||
img_id?: string;
|
||||
description?: Record<string, string>;
|
||||
scope?: string;
|
||||
}
|
||||
|
||||
export interface CarcatParameterValue {
|
||||
idx: string;
|
||||
value: string;
|
||||
img?: string;
|
||||
img_id?: string;
|
||||
description?: Record<string, string>;
|
||||
}
|
||||
|
||||
export interface CarcatParameter {
|
||||
key: string; // "model_selections" | "engine" | "transmission" | …
|
||||
name: string;
|
||||
values: CarcatParameterValue[];
|
||||
sortOrder?: number;
|
||||
}
|
||||
|
||||
export interface CarcatCar {
|
||||
id: string; // carId "pl_<hash>"
|
||||
catalogId: string;
|
||||
brand?: string;
|
||||
modelId?: string;
|
||||
modelName?: string | null;
|
||||
name?: string;
|
||||
parameters?: { idx: string; key: string; name: string; value: string }[];
|
||||
}
|
||||
|
||||
export interface CarcatGroup {
|
||||
id: string;
|
||||
name: string;
|
||||
hasSubgroups: boolean;
|
||||
hasParts: boolean;
|
||||
img?: string;
|
||||
scope?: string;
|
||||
description?: {
|
||||
illustration?: string;
|
||||
remark?: string;
|
||||
modelDescription?: string;
|
||||
restriction?: string;
|
||||
};
|
||||
}
|
||||
|
||||
export interface CarcatPart {
|
||||
id?: string;
|
||||
number: string;
|
||||
name: string;
|
||||
notice?: string;
|
||||
positionNumber?: string;
|
||||
url?: string;
|
||||
description?: { qty?: string; remark?: string; modelDescription?: string; restriction?: string };
|
||||
}
|
||||
|
||||
export interface CarcatPartsResponse {
|
||||
img?: string;
|
||||
imgDescription?: string;
|
||||
partGroups: {
|
||||
name: string;
|
||||
number?: string;
|
||||
description?: string;
|
||||
positionNumber?: string;
|
||||
parts: CarcatPart[];
|
||||
}[];
|
||||
positions?: { number: string; coordinates: number[] }[];
|
||||
}
|
||||
|
||||
export interface CarcatonlineClientOptions {
|
||||
email: string;
|
||||
password: string;
|
||||
portalUrl?: string;
|
||||
apiUrl?: string;
|
||||
/** Called on every API request so the caller can meter the daily budget. */
|
||||
onRequest?: (path: string) => void;
|
||||
fetchImpl?: typeof fetch;
|
||||
logger?: { log: (m: string) => void; warn: (m: string) => void };
|
||||
}
|
||||
|
||||
export function decodeJwtExp(token: string): number {
|
||||
const payload = JSON.parse(Buffer.from(token.split(".")[1], "base64url").toString("utf8")) as {
|
||||
exp?: number;
|
||||
};
|
||||
if (!payload.exp) throw new CarcatonlineAuthError("widget token has no exp claim");
|
||||
return payload.exp * 1000;
|
||||
}
|
||||
|
||||
/**
|
||||
* Browser-free login: form POST → session cookie → /widget HTML → data-token.
|
||||
* Pure function so it can be unit-tested with a stubbed fetch.
|
||||
*/
|
||||
export async function loginCarcatonline(
|
||||
opts: Pick<CarcatonlineClientOptions, "email" | "password" | "portalUrl" | "fetchImpl">,
|
||||
): Promise<CarcatonlineToken> {
|
||||
const portal = opts.portalUrl ?? CARCATONLINE_DEFAULTS.portalUrl;
|
||||
const doFetch = opts.fetchImpl ?? fetch;
|
||||
const loginRes = await doFetch(`${portal}/login`, {
|
||||
method: "POST",
|
||||
redirect: "manual",
|
||||
headers: {
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
"HX-Request": "true",
|
||||
"User-Agent": USER_AGENT,
|
||||
},
|
||||
body: new URLSearchParams({ email: opts.email, password: opts.password }).toString(),
|
||||
});
|
||||
const setCookies: string[] =
|
||||
typeof (loginRes.headers as { getSetCookie?: () => string[] }).getSetCookie === "function"
|
||||
? (loginRes.headers as unknown as { getSetCookie: () => string[] }).getSetCookie()
|
||||
: [];
|
||||
const session = setCookies.map((c) => c.split(";")[0]).find((kv) => kv.startsWith("session="));
|
||||
if (!loginRes.ok || !session) {
|
||||
throw new CarcatonlineAuthError(
|
||||
`login failed (HTTP ${loginRes.status}, session cookie ${session ? "present" : "missing"}) — wrong credentials?`,
|
||||
);
|
||||
}
|
||||
const widgetRes = await doFetch(`${portal}/widget`, {
|
||||
headers: { Cookie: session, "HX-Request": "true", "User-Agent": USER_AGENT },
|
||||
});
|
||||
const html = await widgetRes.text();
|
||||
const m = html.match(/data-token="([^"]+)"/);
|
||||
if (!widgetRes.ok || !m) {
|
||||
throw new CarcatonlineAuthError(
|
||||
`widget page did not expose a token (HTTP ${widgetRes.status})`,
|
||||
);
|
||||
}
|
||||
return { token: m[1], expiresAt: decodeJwtExp(m[1]) };
|
||||
}
|
||||
|
||||
export class CarcatonlineClient {
|
||||
private readonly apiUrl: string;
|
||||
private readonly doFetch: typeof fetch;
|
||||
|
||||
constructor(
|
||||
private readonly opts: CarcatonlineClientOptions,
|
||||
private readonly tokenStore: {
|
||||
get(): Promise<CarcatonlineToken | null>;
|
||||
set(t: CarcatonlineToken): Promise<void>;
|
||||
clear(): Promise<void>;
|
||||
},
|
||||
) {
|
||||
this.apiUrl = opts.apiUrl ?? CARCATONLINE_DEFAULTS.apiUrl;
|
||||
this.doFetch = opts.fetchImpl ?? fetch;
|
||||
}
|
||||
|
||||
/** Cached token unless it expires within 7 days; otherwise a fresh login. */
|
||||
async getToken(force = false): Promise<CarcatonlineToken> {
|
||||
if (!force) {
|
||||
const cached = await this.tokenStore.get();
|
||||
if (cached && cached.expiresAt - Date.now() > 7 * 24 * 3600 * 1000) return cached;
|
||||
}
|
||||
const fresh = await loginCarcatonline(this.opts);
|
||||
await this.tokenStore.set(fresh);
|
||||
this.opts.logger?.log(
|
||||
`[carcatonline] logged in, token valid until ${new Date(fresh.expiresAt).toISOString()}`,
|
||||
);
|
||||
return fresh;
|
||||
}
|
||||
|
||||
async get<T>(path: string): Promise<T> {
|
||||
let token = await this.getToken();
|
||||
for (let attempt = 0; attempt < 2; attempt += 1) {
|
||||
this.opts.onRequest?.(path);
|
||||
const res = await this.doFetch(`${this.apiUrl}${path}`, {
|
||||
headers: { Authorization: `Bearer ${token.token}`, "User-Agent": USER_AGENT },
|
||||
});
|
||||
if (res.status === 429) throw new CarcatonlineRateLimitError(path);
|
||||
if (res.status === 401 && attempt === 0) {
|
||||
this.opts.logger?.warn("[carcatonline] 401 — refreshing token");
|
||||
await this.tokenStore.clear();
|
||||
token = await this.getToken(true);
|
||||
continue;
|
||||
}
|
||||
const text = await res.text();
|
||||
if (!res.ok) {
|
||||
let detail = text.slice(0, 200);
|
||||
try {
|
||||
detail = (JSON.parse(text) as { detail?: string }).detail ?? detail;
|
||||
} catch {
|
||||
/* keep raw */
|
||||
}
|
||||
throw new CarcatonlineHttpError(res.status, path, detail);
|
||||
}
|
||||
return JSON.parse(text) as T;
|
||||
}
|
||||
throw new CarcatonlineAuthError("token refresh did not resolve the 401");
|
||||
}
|
||||
|
||||
catalogs(): Promise<CarcatCatalog[]> {
|
||||
return this.get("/catalogs/");
|
||||
}
|
||||
|
||||
models(catalogId: string): Promise<CarcatModel[]> {
|
||||
return this.get(`/catalogs/${encodeURIComponent(catalogId)}/models/`);
|
||||
}
|
||||
|
||||
carsParameters(
|
||||
catalogId: string,
|
||||
modelId: string,
|
||||
selected: string[],
|
||||
): Promise<CarcatParameter[]> {
|
||||
return this.get(
|
||||
`/catalogs/${encodeURIComponent(catalogId)}/cars-parameters/?${paramsQuery(modelId, selected)}`,
|
||||
);
|
||||
}
|
||||
|
||||
cars(catalogId: string, modelId: string, selected: string[]): Promise<CarcatCar[]> {
|
||||
return this.get(
|
||||
`/catalogs/${encodeURIComponent(catalogId)}/cars2/?${paramsQuery(modelId, selected)}`,
|
||||
);
|
||||
}
|
||||
|
||||
groups(catalogId: string, carId: string, groupId?: string): Promise<CarcatGroup[]> {
|
||||
const q = new URLSearchParams({ carId });
|
||||
if (groupId) q.set("groupId", groupId);
|
||||
return this.get(`/catalogs/${encodeURIComponent(catalogId)}/groups2?${q.toString()}`);
|
||||
}
|
||||
|
||||
parts(catalogId: string, carId: string, groupId: string): Promise<CarcatPartsResponse> {
|
||||
const q = new URLSearchParams({ carId, groupId });
|
||||
return this.get(`/catalogs/${encodeURIComponent(catalogId)}/parts2?${q.toString()}`);
|
||||
}
|
||||
}
|
||||
|
||||
function paramsQuery(modelId: string, selected: string[]): string {
|
||||
const q = new URLSearchParams();
|
||||
q.set("modelId", modelId);
|
||||
for (const s of selected) q.append("parameter", s);
|
||||
return q.toString();
|
||||
}
|
||||
|
||||
/** Absolute URL for an `img` value returned by the API ("static/images/…"). */
|
||||
export function carcatImageUrl(
|
||||
img: string | undefined | null,
|
||||
apiUrl = CARCATONLINE_DEFAULTS.apiUrl,
|
||||
): string | null {
|
||||
if (!img) return null;
|
||||
if (/^https?:\/\//.test(img)) return img;
|
||||
const origin = new URL(apiUrl).origin;
|
||||
return `${origin}/${img.replace(/^\/+/, "")}`;
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
catalogsForBrand,
|
||||
isNonVehicleModel,
|
||||
matchCarcatModel,
|
||||
normalizeLabel,
|
||||
relaxLabel,
|
||||
} from "./carcatonline.matcher";
|
||||
|
||||
const models = (names: string[]) => names.map((name, i) => ({ id: `pl_${i}`, name }));
|
||||
|
||||
describe("carcatonline model matcher", () => {
|
||||
it("matches identical PartsLink24 labels exactly (Hyundai/Kia/Volvo style)", () => {
|
||||
const list = models([
|
||||
"ACCENT 00",
|
||||
"ACCENT 06: -DEC.2006",
|
||||
"ACCENT 06: JAN.2007-",
|
||||
"XC90 (16-)",
|
||||
]);
|
||||
expect(matchCarcatModel("ACCENT 06: JAN.2007-", list)?.model.name).toBe("ACCENT 06: JAN.2007-");
|
||||
expect(matchCarcatModel("XC90 (16-)", list)?.tier).toBe("exact");
|
||||
});
|
||||
|
||||
it("folds diacritics and PL24 backslash escapes", () => {
|
||||
const list = models(["ARKANA RUSYA", "LOGAN\\-SANDERO 3 / TALIANT", "KADJAR CHINE"]);
|
||||
expect(matchCarcatModel("LOGAN-SANDERO 3 / TALIANT", list)?.model.name).toBe(
|
||||
"LOGAN\\-SANDERO 3 / TALIANT",
|
||||
);
|
||||
expect(normalizeLabel("KADJAR ÇİN")).toBe("KADJAR CIN");
|
||||
});
|
||||
|
||||
it("strips region and year decorations for the relaxed tier (Nissan/Ford style)", () => {
|
||||
const list = models(["ALTIMA", "370Z", "B-MAX", "C-MAX"]);
|
||||
expect(matchCarcatModel("ALTIMA (EL)", list)).toMatchObject({
|
||||
tier: "relaxed",
|
||||
model: { name: "ALTIMA" },
|
||||
});
|
||||
expect(matchCarcatModel("B-MAX (2012- 2017)", list)?.model.name).toBe("B-MAX");
|
||||
expect(relaxLabel("A04 ASTRA-H 2004 - 2014")).toBe("A04 ASTRA H");
|
||||
});
|
||||
|
||||
it("resolves a bare body code against a label that contains it (Mercedes style)", () => {
|
||||
const list = models([
|
||||
"Mercedes-Benz W100 600",
|
||||
"Mercedes-Benz R107 / C107 SL / SLC",
|
||||
"Mercedes-Benz W124",
|
||||
]);
|
||||
expect(matchCarcatModel("C107", list)).toMatchObject({
|
||||
tier: "code",
|
||||
model: { name: "Mercedes-Benz R107 / C107 SL / SLC" },
|
||||
});
|
||||
expect(matchCarcatModel("W124", list)?.tier).toBe("code");
|
||||
});
|
||||
|
||||
it("returns null when ambiguous or absent", () => {
|
||||
const list = models(["Mercedes-Benz W205 C-Class", "Mercedes-Benz W205 AMG"]);
|
||||
expect(matchCarcatModel("W205", list)).toBeNull();
|
||||
expect(matchCarcatModel("Porsche 996", models(["Porsche 991", "Porsche 992"]))).toBeNull();
|
||||
expect(matchCarcatModel("", list)).toBeNull();
|
||||
});
|
||||
|
||||
it("knows PL24 pseudo-models that are not vehicles", () => {
|
||||
expect(isNonVehicleModel("Kimyasal maddeler")).toBe(true);
|
||||
expect(isNonVehicleModel("Özel kataloglar")).toBe(true);
|
||||
expect(isNonVehicleModel("Golf Sportsvan")).toBe(false);
|
||||
});
|
||||
|
||||
it("maps brands to ordered catalog candidates", () => {
|
||||
expect(catalogsForBrand("Mercedes-Benz")).toEqual([
|
||||
"pl_mercedes",
|
||||
"pl_mercedes_vans",
|
||||
"pl_mercedes_truck",
|
||||
]);
|
||||
expect(catalogsForBrand("Opel")[0]).toBe("pl_opel");
|
||||
expect(catalogsForBrand("Unknown Brand")).toEqual([]);
|
||||
});
|
||||
});
|
||||
133
apps/api/src/integrations/carcatonline/carcatonline.matcher.ts
Normal file
133
apps/api/src/integrations/carcatonline/carcatonline.matcher.ts
Normal file
@@ -0,0 +1,133 @@
|
||||
/**
|
||||
* Map our PL24 `catalog_vehicles` (brand + model label) onto carcatonline
|
||||
* catalogs/models. carcatonline's `pl_*` catalogs mirror PartsLink24, so most
|
||||
* model labels are identical (measured 2026-09-25 on prod's empty catalogs:
|
||||
* Hyundai 226/226, Kia 145/145, Volvo 59/59, Mitsubishi 56/56, Toyota 39/39,
|
||||
* Alfa 23/23 exact; Renault 45/65, VW 38/58, Porsche 26/64 exact). The rest
|
||||
* differ by region suffixes ("ALTIMA (EL)" vs "ALTIMA"), year ranges
|
||||
* ("B-MAX (2012- 2017)" vs "B-MAX"), or body-code-only labels (Mercedes "C107"
|
||||
* vs "Mercedes-Benz R107 / C107 …"), handled by the relaxed / code tiers.
|
||||
*/
|
||||
import type { CarcatModel } from "./carcatonline.client";
|
||||
|
||||
/** Ordered catalog candidates per brand; the first one with a model match wins. */
|
||||
export const BRAND_CATALOGS: Record<string, string[]> = {
|
||||
abarth: ["pl_abarth"],
|
||||
"alfa romeo": ["pl_alfa_romeo"],
|
||||
alpine: ["pl_alpine"],
|
||||
audi: ["pl_audi"],
|
||||
bentley: ["pl_bentley"],
|
||||
bmw: ["pl_bmw"],
|
||||
citroen: ["pl_citroen"],
|
||||
citroën: ["pl_citroen"],
|
||||
cupra: ["pl_cupra", "pl_seat"],
|
||||
dacia: ["pl_dacia", "pl_renault"],
|
||||
ds: ["pl_ds"],
|
||||
fiat: ["pl_fiat", "pl_fiat_professional"],
|
||||
ford: ["pl_ford", "pl_ford_pro"],
|
||||
hyundai: ["pl_hyundai"],
|
||||
infiniti: ["pl_infiniti", "pl_nissan"],
|
||||
iveco: ["pl_iveco"],
|
||||
jaguar: ["pl_jaguar"],
|
||||
jeep: ["pl_jeep"],
|
||||
kia: ["pl_kia"],
|
||||
lancia: ["pl_lancia"],
|
||||
"land rover": ["pl_land_rover"],
|
||||
lexus: ["pl_lexus", "pl_toyota"],
|
||||
man: ["pl_man"],
|
||||
"mercedes-benz": ["pl_mercedes", "pl_mercedes_vans", "pl_mercedes_truck"],
|
||||
mercedes: ["pl_mercedes", "pl_mercedes_vans", "pl_mercedes_truck"],
|
||||
mini: ["pl_mini"],
|
||||
mitsubishi: ["pl_mitsubishi"],
|
||||
nissan: ["pl_nissan"],
|
||||
opel: ["pl_opel", "pl_psa_opel", "pl_vauxhall"],
|
||||
peugeot: ["pl_peugeot"],
|
||||
polestar: ["pl_polestar"],
|
||||
porsche: ["pl_porsche", "pl_porsche_classic"],
|
||||
renault: ["pl_renault"],
|
||||
seat: ["pl_seat", "pl_cupra"],
|
||||
skoda: ["pl_skoda"],
|
||||
smart: ["pl_smart"],
|
||||
suzuki: ["pl_suzuki"],
|
||||
toyota: ["pl_toyota"],
|
||||
vauxhall: ["pl_vauxhall", "pl_psa_vauxhall"],
|
||||
volkswagen: ["pl_volkswagen", "pl_volkswagen_commercial", "pl_volkswagen_classic"],
|
||||
volvo: ["pl_volvo"],
|
||||
};
|
||||
|
||||
export function catalogsForBrand(brandName: string | null | undefined): string[] {
|
||||
const key = (brandName ?? "").trim().toLowerCase();
|
||||
return BRAND_CATALOGS[key] ?? [];
|
||||
}
|
||||
|
||||
/** PL24 pseudo-models that are not vehicles (VW group "Kimyasal maddeler", "Özel kataloglar" …). */
|
||||
const NON_VEHICLE_LABELS = new Set<string>([
|
||||
"KIMYASAL MADDELER",
|
||||
"OZEL KATALOGLAR",
|
||||
"DIGER URUNLER",
|
||||
"URETICI SAYFALARI",
|
||||
"KISALTMALAR",
|
||||
"AOS",
|
||||
"ELEKTRIKL BAGLANTI",
|
||||
"KAMP ARACI",
|
||||
"KAMYONET",
|
||||
"KOMB ISI GUC SANTR",
|
||||
]);
|
||||
|
||||
export function isNonVehicleModel(label: string): boolean {
|
||||
return NON_VEHICLE_LABELS.has(normalizeLabel(label));
|
||||
}
|
||||
|
||||
/** Uppercase, diacritic-folded, punctuation collapsed, PL24 backslash escapes removed. */
|
||||
export function normalizeLabel(s: string | null | undefined): string {
|
||||
return (s ?? "")
|
||||
.toUpperCase()
|
||||
.normalize("NFD")
|
||||
.replace(/\p{M}/gu, "")
|
||||
.replace(/\\/g, "")
|
||||
.replace(/[^A-Z0-9]+/g, " ")
|
||||
.trim();
|
||||
}
|
||||
|
||||
/** Drop region suffixes "(EL)/(ER)", year ranges "(2012- 2017)", "2004 - 2014", and 4-digit years. */
|
||||
export function relaxLabel(s: string | null | undefined): string {
|
||||
const stripped = (s ?? "")
|
||||
.replace(/\((EL|ER|EUR|JPN|USA|GCC|RUS)\)/gi, " ")
|
||||
.replace(/\((19|20)\d\d\s*-\s*(19|20)?\d{0,2}\)/g, " ")
|
||||
.replace(/\b(19|20)\d\d\s*-\s*(19|20)?\d{0,2}\b/g, " ")
|
||||
.replace(/\b(19|20)\d\d\b/g, " ");
|
||||
return normalizeLabel(stripped);
|
||||
}
|
||||
|
||||
export interface ModelMatch {
|
||||
model: CarcatModel;
|
||||
tier: "exact" | "relaxed" | "code";
|
||||
}
|
||||
|
||||
/**
|
||||
* Pick the carcatonline model for one of our PL24 model labels.
|
||||
* - exact: normalized labels identical
|
||||
* - relaxed: identical after stripping region/year decorations on both sides
|
||||
* - code: our label is a short body/chassis code (≤6 chars, has a digit) that
|
||||
* appears as a whole token in exactly ONE carcatonline label (Mercedes C107)
|
||||
* Returns null when ambiguous or absent — the caller records "unmatched".
|
||||
*/
|
||||
export function matchCarcatModel(ourLabel: string, models: CarcatModel[]): ModelMatch | null {
|
||||
const target = normalizeLabel(ourLabel);
|
||||
if (!target) return null;
|
||||
const exact = models.filter((m) => normalizeLabel(m.name) === target);
|
||||
if (exact.length === 1) return { model: exact[0], tier: "exact" };
|
||||
if (exact.length > 1) return { model: exact[0], tier: "exact" }; // identical labels → same PL24 model, first wins
|
||||
|
||||
const relaxedTarget = relaxLabel(ourLabel);
|
||||
if (relaxedTarget) {
|
||||
const relaxed = models.filter((m) => relaxLabel(m.name) === relaxedTarget);
|
||||
if (relaxed.length === 1) return { model: relaxed[0], tier: "relaxed" };
|
||||
}
|
||||
|
||||
if (/^[A-Z]{1,2}\d{2,4}[A-Z]?$/.test(target) && target.length <= 6) {
|
||||
const code = models.filter((m) => normalizeLabel(m.name).split(" ").includes(target));
|
||||
if (code.length === 1) return { model: code[0], tier: "code" };
|
||||
}
|
||||
return null;
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
CARCAT_KEYS,
|
||||
CarcatonlineBudgetError,
|
||||
CarcatonlineLockedError,
|
||||
CarcatonlineThrottle,
|
||||
type RedisLike,
|
||||
carcatConfigFromEnv,
|
||||
dailyCallsKey,
|
||||
isWithinWindow,
|
||||
msUntilWindowOpens,
|
||||
redisTokenStore,
|
||||
} from "./carcatonline.pacing";
|
||||
|
||||
function fakeRedis(): RedisLike & { store: Map<string, string>; ttl: Map<string, number> } {
|
||||
const store = new Map<string, string>();
|
||||
const ttl = new Map<string, number>();
|
||||
const set = async (key: string, value: string, mode: "EX" | "PX", n: number, cond?: "NX") => {
|
||||
if (cond === "NX" && store.has(key)) return null;
|
||||
store.set(key, value);
|
||||
ttl.set(key, mode === "EX" ? n : n / 1000);
|
||||
return "OK";
|
||||
};
|
||||
return {
|
||||
store,
|
||||
ttl,
|
||||
set: set as RedisLike["set"],
|
||||
async get(k) {
|
||||
return store.get(k) ?? null;
|
||||
},
|
||||
async del(k) {
|
||||
store.delete(k);
|
||||
},
|
||||
async incr(k) {
|
||||
const n = Number(store.get(k) ?? 0) + 1;
|
||||
store.set(k, String(n));
|
||||
return n;
|
||||
},
|
||||
async expire(k, s) {
|
||||
ttl.set(k, s);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const at = (iso: string) => new Date(iso);
|
||||
|
||||
describe("night window (Europe/Istanbul, UTC+3)", () => {
|
||||
it("wraps past midnight: 20:00–07:00", () => {
|
||||
expect(isWithinWindow(at("2026-09-26T17:00:00Z"), 20, 7)).toBe(true); // 20:00 Istanbul
|
||||
expect(isWithinWindow(at("2026-09-26T23:30:00Z"), 20, 7)).toBe(true); // 02:30
|
||||
expect(isWithinWindow(at("2026-09-27T03:59:00Z"), 20, 7)).toBe(true); // 06:59
|
||||
expect(isWithinWindow(at("2026-09-27T04:00:00Z"), 20, 7)).toBe(false); // 07:00
|
||||
expect(isWithinWindow(at("2026-09-26T12:00:00Z"), 20, 7)).toBe(false); // 15:00
|
||||
});
|
||||
|
||||
it("computes the wait until the window opens", () => {
|
||||
expect(msUntilWindowOpens(at("2026-09-26T12:00:00Z"), 20, 7)).toBe(5 * 3600 * 1000); // 15:00 → 20:00
|
||||
expect(msUntilWindowOpens(at("2026-09-26T23:00:00Z"), 20, 7)).toBe(0);
|
||||
});
|
||||
|
||||
it("keys the daily budget by Istanbul day", () => {
|
||||
expect(dailyCallsKey(at("2026-09-26T22:30:00Z"))).toBe(`${CARCAT_KEYS.callsPrefix}2026-09-27`);
|
||||
});
|
||||
|
||||
it("reads config from env with defaults", () => {
|
||||
const cfg = carcatConfigFromEnv({
|
||||
CARCATONLINE_DAILY_CALL_CAP: "100",
|
||||
CARCATONLINE_WINDOW_START: "x",
|
||||
});
|
||||
expect(cfg).toMatchObject({
|
||||
dailyCallCap: 100,
|
||||
windowStartHour: 20,
|
||||
windowEndHour: 7,
|
||||
minIntervalMs: 7000,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("CarcatonlineThrottle", () => {
|
||||
const cfg = {
|
||||
minIntervalMs: 2000,
|
||||
dailyCallCap: 3,
|
||||
windowStartHour: 20,
|
||||
windowEndHour: 7,
|
||||
lockoutSeconds: 2400,
|
||||
};
|
||||
|
||||
it("counts calls per day and enforces the cap", async () => {
|
||||
const redis = fakeRedis();
|
||||
const sleep = vi.fn().mockResolvedValue(undefined);
|
||||
let t = Date.parse("2026-09-26T20:00:00Z");
|
||||
const throttle = new CarcatonlineThrottle(redis, cfg, () => new Date(t), sleep);
|
||||
for (let i = 0; i < 3; i += 1) {
|
||||
await throttle.beforeCall();
|
||||
redis.store.delete(CARCAT_KEYS.lastCall); // simulate the PX slot expiring
|
||||
t += 3000;
|
||||
}
|
||||
expect(await throttle.callsToday()).toBe(3);
|
||||
await expect(throttle.beforeCall()).rejects.toBeInstanceOf(CarcatonlineBudgetError);
|
||||
});
|
||||
|
||||
it("waits for the shared min-interval slot instead of bursting", async () => {
|
||||
const redis = fakeRedis();
|
||||
const sleep = vi.fn(async () => {
|
||||
redis.store.delete(CARCAT_KEYS.lastCall);
|
||||
});
|
||||
const throttle = new CarcatonlineThrottle(redis, cfg, () => new Date(), sleep);
|
||||
await throttle.beforeCall();
|
||||
await throttle.beforeCall(); // slot taken → sleeps once, then acquires
|
||||
expect(sleep).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("refuses to call while locked out and records lockouts with the configured length", async () => {
|
||||
const redis = fakeRedis();
|
||||
const now = new Date("2026-09-26T21:00:00Z");
|
||||
const throttle = new CarcatonlineThrottle(redis, cfg, () => now);
|
||||
await throttle.markLockout();
|
||||
expect(await throttle.lockoutRemainingMs()).toBe(2400 * 1000);
|
||||
await expect(throttle.beforeCall()).rejects.toBeInstanceOf(CarcatonlineLockedError);
|
||||
});
|
||||
});
|
||||
|
||||
describe("redisTokenStore", () => {
|
||||
it("stores the token with a TTL derived from its expiry", async () => {
|
||||
const redis = fakeRedis();
|
||||
const store = redisTokenStore(redis);
|
||||
await store.set({ token: "jwt", expiresAt: Date.now() + 100 * 3600 * 1000 });
|
||||
expect((await store.get())?.token).toBe("jwt");
|
||||
expect(redis.ttl.get(CARCAT_KEYS.token)).toBeGreaterThan(99 * 3600);
|
||||
await store.clear();
|
||||
expect(await store.get()).toBeNull();
|
||||
});
|
||||
});
|
||||
184
apps/api/src/integrations/carcatonline/carcatonline.pacing.ts
Normal file
184
apps/api/src/integrations/carcatonline/carcatonline.pacing.ts
Normal file
@@ -0,0 +1,184 @@
|
||||
/**
|
||||
* Shared throttling primitives for carcatonline (used by the worker backfill and
|
||||
* the API's on-demand parts fetch). All state lives in Redis so the API and the
|
||||
* worker share one budget:
|
||||
* - min interval between calls (measured 2026-09-26: ~110 calls at 1/2.2 s → 429 + ~40 min lockout;
|
||||
* the quota looks like ~100 calls per window, so the default is 1 call / 7 s ≈ 8.5/min)
|
||||
* - lockout flag set on any 429
|
||||
* - per-Istanbul-day call counter with a hard cap
|
||||
* - the night window (default 20:00–07:00 Europe/Istanbul) for bulk work
|
||||
*/
|
||||
import type { CarcatonlineToken } from "./carcatonline.client";
|
||||
|
||||
export interface RedisLike {
|
||||
get(key: string): Promise<string | null>;
|
||||
set(key: string, value: string, mode: "EX", ttlSeconds: number): Promise<unknown>;
|
||||
del(key: string): Promise<unknown>;
|
||||
incr(key: string): Promise<number>;
|
||||
expire(key: string, seconds: number): Promise<unknown>;
|
||||
/** SET key value PX ms NX — returns "OK" or null. */
|
||||
set(key: string, value: string, mode: "PX", ttlMs: number, cond: "NX"): Promise<unknown>;
|
||||
}
|
||||
|
||||
export const CARCAT_KEYS = {
|
||||
token: "carcatonline:token",
|
||||
lockout: "carcatonline:lockout",
|
||||
lastCall: "carcatonline:last-call",
|
||||
callsPrefix: "carcatonline:calls:",
|
||||
modelsPrefix: "carcatonline:models:",
|
||||
} as const;
|
||||
|
||||
export interface CarcatConfig {
|
||||
minIntervalMs: number;
|
||||
dailyCallCap: number;
|
||||
windowStartHour: number;
|
||||
windowEndHour: number;
|
||||
lockoutSeconds: number;
|
||||
}
|
||||
|
||||
export function carcatConfigFromEnv(env: NodeJS.ProcessEnv = process.env): CarcatConfig {
|
||||
const num = (v: string | undefined, d: number): number => {
|
||||
const n = Number(v);
|
||||
return Number.isFinite(n) && n >= 0 ? n : d;
|
||||
};
|
||||
return {
|
||||
minIntervalMs: num(env.CARCATONLINE_MIN_INTERVAL_MS, 7000),
|
||||
dailyCallCap: num(env.CARCATONLINE_DAILY_CALL_CAP, 15000),
|
||||
windowStartHour: num(env.CARCATONLINE_WINDOW_START, 20),
|
||||
windowEndHour: num(env.CARCATONLINE_WINDOW_END, 7),
|
||||
lockoutSeconds: num(env.CARCATONLINE_LOCKOUT_SECONDS, 40 * 60),
|
||||
};
|
||||
}
|
||||
|
||||
const ISTANBUL_OFFSET_MS = 3 * 60 * 60 * 1000; // fixed UTC+3 (no DST since 2016)
|
||||
|
||||
/** Hour (0-23) and "YYYY-MM-DD" in Europe/Istanbul. */
|
||||
export function istanbulParts(date: Date): { hour: number; minute: number; day: string } {
|
||||
const shifted = new Date(date.getTime() + ISTANBUL_OFFSET_MS);
|
||||
return {
|
||||
hour: shifted.getUTCHours(),
|
||||
minute: shifted.getUTCMinutes(),
|
||||
day: shifted.toISOString().slice(0, 10),
|
||||
};
|
||||
}
|
||||
|
||||
/** True when `date` falls inside the bulk window. Wraps past midnight (20 → 7). */
|
||||
export function isWithinWindow(date: Date, startHour: number, endHour: number): boolean {
|
||||
const { hour } = istanbulParts(date);
|
||||
if (startHour === endHour) return true;
|
||||
if (startHour < endHour) return hour >= startHour && hour < endHour;
|
||||
return hour >= startHour || hour < endHour;
|
||||
}
|
||||
|
||||
/** Milliseconds until the next window opens (0 when already open). */
|
||||
export function msUntilWindowOpens(date: Date, startHour: number, endHour: number): number {
|
||||
if (isWithinWindow(date, startHour, endHour)) return 0;
|
||||
const shifted = new Date(date.getTime() + ISTANBUL_OFFSET_MS);
|
||||
const next = new Date(shifted);
|
||||
next.setUTCHours(startHour, 0, 0, 0);
|
||||
if (next.getTime() <= shifted.getTime()) next.setUTCDate(next.getUTCDate() + 1);
|
||||
return next.getTime() - shifted.getTime();
|
||||
}
|
||||
|
||||
export function dailyCallsKey(date: Date): string {
|
||||
return `${CARCAT_KEYS.callsPrefix}${istanbulParts(date).day}`;
|
||||
}
|
||||
|
||||
export class CarcatonlineLockedError extends Error {
|
||||
constructor(readonly retryAfterMs: number) {
|
||||
super(`carcatonline is locked out for ${Math.round(retryAfterMs / 1000)}s`);
|
||||
this.name = "CarcatonlineLockedError";
|
||||
}
|
||||
}
|
||||
|
||||
export class CarcatonlineBudgetError extends Error {
|
||||
constructor(
|
||||
readonly used: number,
|
||||
readonly cap: number,
|
||||
) {
|
||||
super(`carcatonline daily call cap reached (${used}/${cap})`);
|
||||
this.name = "CarcatonlineBudgetError";
|
||||
}
|
||||
}
|
||||
|
||||
export class CarcatonlineWindowClosedError extends Error {
|
||||
constructor(readonly retryAfterMs: number) {
|
||||
super(`carcatonline bulk window closed; reopens in ${Math.round(retryAfterMs / 60000)} min`);
|
||||
this.name = "CarcatonlineWindowClosedError";
|
||||
}
|
||||
}
|
||||
|
||||
/** Redis-backed token store shared by API and worker. */
|
||||
export function redisTokenStore(redis: RedisLike) {
|
||||
return {
|
||||
async get(): Promise<CarcatonlineToken | null> {
|
||||
const raw = await redis.get(CARCAT_KEYS.token);
|
||||
if (!raw) return null;
|
||||
try {
|
||||
const t = JSON.parse(raw) as CarcatonlineToken;
|
||||
return t.token && t.expiresAt ? t : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
},
|
||||
async set(t: CarcatonlineToken): Promise<void> {
|
||||
const ttl = Math.max(60, Math.floor((t.expiresAt - Date.now()) / 1000));
|
||||
await redis.set(CARCAT_KEYS.token, JSON.stringify(t), "EX", ttl);
|
||||
},
|
||||
async clear(): Promise<void> {
|
||||
await redis.del(CARCAT_KEYS.token);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Coordinates every outbound carcatonline call. `beforeCall()` must be awaited
|
||||
* before each request: it throws when locked out or over the daily cap, and
|
||||
* otherwise sleeps until the shared min-interval slot is free and counts the call.
|
||||
*/
|
||||
export class CarcatonlineThrottle {
|
||||
constructor(
|
||||
private readonly redis: RedisLike,
|
||||
private readonly cfg: CarcatConfig,
|
||||
private readonly now: () => Date = () => new Date(),
|
||||
private readonly sleep: (ms: number) => Promise<void> = (ms) =>
|
||||
new Promise((r) => setTimeout(r, ms)),
|
||||
) {}
|
||||
|
||||
async lockoutRemainingMs(): Promise<number> {
|
||||
const until = Number((await this.redis.get(CARCAT_KEYS.lockout)) ?? 0);
|
||||
return until > this.now().getTime() ? until - this.now().getTime() : 0;
|
||||
}
|
||||
|
||||
async markLockout(): Promise<void> {
|
||||
const until = this.now().getTime() + this.cfg.lockoutSeconds * 1000;
|
||||
await this.redis.set(CARCAT_KEYS.lockout, String(until), "EX", this.cfg.lockoutSeconds + 60);
|
||||
}
|
||||
|
||||
async callsToday(): Promise<number> {
|
||||
return Number((await this.redis.get(dailyCallsKey(this.now()))) ?? 0);
|
||||
}
|
||||
|
||||
async beforeCall(): Promise<void> {
|
||||
const locked = await this.lockoutRemainingMs();
|
||||
if (locked > 0) throw new CarcatonlineLockedError(locked);
|
||||
const used = await this.callsToday();
|
||||
if (used >= this.cfg.dailyCallCap)
|
||||
throw new CarcatonlineBudgetError(used, this.cfg.dailyCallCap);
|
||||
// Shared min-interval slot: SET NX PX; spin (bounded) until acquired.
|
||||
for (let i = 0; i < 50; i += 1) {
|
||||
const ok = await this.redis.set(
|
||||
CARCAT_KEYS.lastCall,
|
||||
"1",
|
||||
"PX",
|
||||
this.cfg.minIntervalMs,
|
||||
"NX",
|
||||
);
|
||||
if (ok === "OK") break;
|
||||
await this.sleep(Math.max(100, Math.floor(this.cfg.minIntervalMs / 4)));
|
||||
}
|
||||
const key = dailyCallsKey(this.now());
|
||||
const n = await this.redis.incr(key);
|
||||
if (n === 1) await this.redis.expire(key, 3 * 24 * 3600);
|
||||
}
|
||||
}
|
||||
167
apps/api/src/integrations/carcatonline/carcatonline.tree.spec.ts
Normal file
167
apps/api/src/integrations/carcatonline/carcatonline.tree.spec.ts
Normal file
@@ -0,0 +1,167 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { CarcatCar, CarcatGroup, CarcatParameter } from "./carcatonline.client";
|
||||
import {
|
||||
type CrawledNode,
|
||||
crawlGroupTree,
|
||||
disambiguateNames,
|
||||
resolveRepresentativeCar,
|
||||
} from "./carcatonline.tree";
|
||||
|
||||
const hooks = { beforeCall: vi.fn().mockResolvedValue(undefined) };
|
||||
|
||||
describe("resolveRepresentativeCar", () => {
|
||||
it("cascades model_selections → engine → transmission and prefers diesel/manual values", async () => {
|
||||
const steps: CarcatParameter[][] = [
|
||||
[{ key: "model_selections", name: "Model", values: [{ idx: "sel-suv", value: "SUV" }] }],
|
||||
[
|
||||
{ key: "model_selections", name: "Model", values: [{ idx: "sel-suv", value: "SUV" }] },
|
||||
{
|
||||
key: "engine",
|
||||
name: "Engine",
|
||||
values: [
|
||||
{ idx: "e12", value: "1.2 TCE PETROL ENGINE" },
|
||||
{ idx: "e15", value: "1.5 DCI DIESEL ENGINE" },
|
||||
],
|
||||
},
|
||||
],
|
||||
[
|
||||
{ key: "model_selections", name: "Model", values: [{ idx: "sel-suv", value: "SUV" }] },
|
||||
{ key: "engine", name: "Engine", values: [{ idx: "e15", value: "1.5 DCI DIESEL ENGINE" }] },
|
||||
{
|
||||
key: "transmission",
|
||||
name: "Transmission",
|
||||
values: [
|
||||
{ idx: "dc4", value: "6-SPEED DUAL CLUTCH GEARBOX: DC4" },
|
||||
{ idx: "tl4", value: "6-SPEED MANUAL GEARBOX" },
|
||||
],
|
||||
},
|
||||
],
|
||||
];
|
||||
let call = 0;
|
||||
const client = {
|
||||
carsParameters: vi.fn(async () => steps[Math.min(call++, steps.length - 1)]),
|
||||
cars: vi.fn(async (_c: string, _m: string, selected: string[]) => {
|
||||
if (selected.length < 3)
|
||||
throw Object.assign(new Error("Not all required parameters are selected"), {
|
||||
status: 404,
|
||||
});
|
||||
return [{ id: "car-1", catalogId: "pl_renault" } as CarcatCar];
|
||||
}),
|
||||
groups: vi.fn(),
|
||||
};
|
||||
const car = await resolveRepresentativeCar(client, hooks, "pl_renault", "model-1");
|
||||
expect(car?.carId).toBe("car-1");
|
||||
expect(car?.parameters.map((p) => p.idx)).toEqual(["sel-suv", "e15", "tl4"]);
|
||||
expect(hooks.beforeCall).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns null when parameters run out without a car", async () => {
|
||||
const client = {
|
||||
carsParameters: vi.fn(async () => [] as CarcatParameter[]),
|
||||
cars: vi.fn(async () => [] as CarcatCar[]),
|
||||
groups: vi.fn(),
|
||||
};
|
||||
expect(await resolveRepresentativeCar(client, hooks, "pl_x", "m")).toBeNull();
|
||||
});
|
||||
|
||||
it("propagates non-404 errors (rate limit)", async () => {
|
||||
const client = {
|
||||
carsParameters: vi.fn(
|
||||
async () =>
|
||||
[{ key: "k", name: "K", values: [{ idx: "a", value: "A" }] }] as CarcatParameter[],
|
||||
),
|
||||
cars: vi.fn(async () => {
|
||||
throw Object.assign(new Error("429"), { status: 429 });
|
||||
}),
|
||||
groups: vi.fn(),
|
||||
};
|
||||
await expect(resolveRepresentativeCar(client, hooks, "pl_x", "m")).rejects.toThrow("429");
|
||||
});
|
||||
});
|
||||
|
||||
describe("crawlGroupTree", () => {
|
||||
it("walks every node that has subgroups, once, breadth-first with depth", async () => {
|
||||
const tree: Record<string, CarcatGroup[]> = {
|
||||
root: [
|
||||
{ id: "g1", name: "Engine", hasSubgroups: true, hasParts: false },
|
||||
{
|
||||
id: "g2",
|
||||
name: "Brakes",
|
||||
hasSubgroups: false,
|
||||
hasParts: true,
|
||||
description: { illustration: "47-010" },
|
||||
},
|
||||
],
|
||||
g1: [
|
||||
{ id: "g1a", name: "Oil", hasSubgroups: false, hasParts: true },
|
||||
{ id: "g1b", name: "Cooling", hasSubgroups: true, hasParts: false },
|
||||
],
|
||||
g1b: [{ id: "g1b1", name: "Radiator", hasSubgroups: false, hasParts: true }],
|
||||
};
|
||||
const client = {
|
||||
carsParameters: vi.fn(),
|
||||
cars: vi.fn(),
|
||||
groups: vi.fn(
|
||||
async (_c: string, _car: string, groupId?: string) => tree[groupId ?? "root"] ?? [],
|
||||
),
|
||||
};
|
||||
const { nodes, calls } = await crawlGroupTree(client, hooks, "pl_renault", "car-1");
|
||||
expect(calls).toBe(3);
|
||||
expect(nodes.map((n) => [n.id, n.parentId, n.depth])).toEqual([
|
||||
["g1", null, 1],
|
||||
["g2", null, 1],
|
||||
["g1a", "g1", 2],
|
||||
["g1b", "g1", 2],
|
||||
["g1b1", "g1b", 3],
|
||||
]);
|
||||
expect(nodes.find((n) => n.id === "g2")?.illustration).toBe("47-010");
|
||||
});
|
||||
|
||||
it("aborts when the window closes mid-crawl", async () => {
|
||||
const client = {
|
||||
carsParameters: vi.fn(),
|
||||
cars: vi.fn(),
|
||||
groups: vi.fn(
|
||||
async () => [{ id: "x", name: "X", hasSubgroups: true, hasParts: false }] as CarcatGroup[],
|
||||
),
|
||||
};
|
||||
let calls = 0;
|
||||
const abortHooks = {
|
||||
beforeCall: vi.fn().mockResolvedValue(undefined),
|
||||
checkAbort: () => {
|
||||
if (calls++ >= 1) throw new Error("window closed");
|
||||
},
|
||||
};
|
||||
await expect(crawlGroupTree(client, abortHooks, "pl_x", "car")).rejects.toThrow(
|
||||
"window closed",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("disambiguateNames", () => {
|
||||
const node = (id: string, parentId: string | null, name: string, depth: number): CrawledNode => ({
|
||||
id,
|
||||
parentId,
|
||||
name,
|
||||
hasSubgroups: false,
|
||||
hasParts: true,
|
||||
img: null,
|
||||
illustration: null,
|
||||
depth,
|
||||
});
|
||||
|
||||
it("suffixes duplicate names with the parent name, then an ordinal", () => {
|
||||
const nodes = [
|
||||
node("a", null, "Engine", 1),
|
||||
node("b", null, "Body", 1),
|
||||
node("a1", "a", "01059926", 2),
|
||||
node("b1", "b", "01059926", 2),
|
||||
node("b2", "b", "01059926", 2),
|
||||
];
|
||||
const names = disambiguateNames(nodes);
|
||||
expect(names.get("a1")).toBe("01059926 (Engine)");
|
||||
expect(names.get("b1")).toBe("01059926 (Body)");
|
||||
expect(names.get("b2")).toBe("01059926 (Body) #2");
|
||||
expect(names.get("a")).toBe("Engine");
|
||||
});
|
||||
});
|
||||
181
apps/api/src/integrations/carcatonline/carcatonline.tree.ts
Normal file
181
apps/api/src/integrations/carcatonline/carcatonline.tree.ts
Normal file
@@ -0,0 +1,181 @@
|
||||
/**
|
||||
* Resolve a carcatonline model to one representative car (cascading parameter
|
||||
* selection) and crawl its full group tree. Pure orchestration over an injected
|
||||
* client; the caller supplies `beforeCall` (throttle) and `checkAbort` (window).
|
||||
*/
|
||||
import type { CarcatCar, CarcatGroup, CarcatParameter } from "./carcatonline.client";
|
||||
|
||||
export interface TreeClient {
|
||||
carsParameters(
|
||||
catalogId: string,
|
||||
modelId: string,
|
||||
selected: string[],
|
||||
): Promise<CarcatParameter[]>;
|
||||
cars(catalogId: string, modelId: string, selected: string[]): Promise<CarcatCar[]>;
|
||||
groups(catalogId: string, carId: string, groupId?: string): Promise<CarcatGroup[]>;
|
||||
}
|
||||
|
||||
export interface CrawlHooks {
|
||||
beforeCall(): Promise<void>;
|
||||
/** Throw to abort (e.g. bulk window closed). */
|
||||
checkAbort?(): void;
|
||||
}
|
||||
|
||||
export interface SelectedParameter {
|
||||
key: string;
|
||||
name: string;
|
||||
idx: string;
|
||||
value: string;
|
||||
}
|
||||
|
||||
export interface ResolvedCar {
|
||||
carId: string;
|
||||
parameters: SelectedParameter[];
|
||||
car: CarcatCar;
|
||||
}
|
||||
|
||||
/** Preferred values when a parameter offers several (first match wins, else first value). */
|
||||
const PREFERRED_TOKENS = [
|
||||
"1.5",
|
||||
"1.6",
|
||||
"DIESEL",
|
||||
"DIZEL",
|
||||
"MANUAL",
|
||||
"MEKANIK",
|
||||
"LHD",
|
||||
"EUROPE",
|
||||
"EU",
|
||||
];
|
||||
|
||||
function pickValue(param: CarcatParameter): CarcatParameter["values"][number] {
|
||||
for (const tok of PREFERRED_TOKENS) {
|
||||
const hit = param.values.find((v) => v.value.toUpperCase().includes(tok));
|
||||
if (hit) return hit;
|
||||
}
|
||||
return param.values[0];
|
||||
}
|
||||
|
||||
/**
|
||||
* Walk `cars-parameters` picking one value per required parameter until
|
||||
* `cars2` returns a car. At most `maxSteps` API calls per step pair.
|
||||
*/
|
||||
export async function resolveRepresentativeCar(
|
||||
client: TreeClient,
|
||||
hooks: CrawlHooks,
|
||||
catalogId: string,
|
||||
modelId: string,
|
||||
maxSteps = 8,
|
||||
): Promise<ResolvedCar | null> {
|
||||
const selected: SelectedParameter[] = [];
|
||||
for (let step = 0; step < maxSteps; step += 1) {
|
||||
hooks.checkAbort?.();
|
||||
await hooks.beforeCall();
|
||||
const params = await client.carsParameters(
|
||||
catalogId,
|
||||
modelId,
|
||||
selected.map((s) => s.idx),
|
||||
);
|
||||
const chosen = new Set(selected.map((s) => s.idx));
|
||||
const pending = params.filter(
|
||||
(p) => p.values?.length > 0 && !p.values.some((v) => chosen.has(v.idx)),
|
||||
);
|
||||
if (pending.length > 0) {
|
||||
const p = pending[0];
|
||||
const v = pickValue(p);
|
||||
selected.push({ key: p.key, name: p.name, idx: v.idx, value: v.value });
|
||||
}
|
||||
hooks.checkAbort?.();
|
||||
await hooks.beforeCall();
|
||||
try {
|
||||
const cars = await client.cars(
|
||||
catalogId,
|
||||
modelId,
|
||||
selected.map((s) => s.idx),
|
||||
);
|
||||
if (Array.isArray(cars) && cars.length > 0) {
|
||||
return { carId: cars[0].id, parameters: selected, car: cars[0] };
|
||||
}
|
||||
} catch (err) {
|
||||
// 404 "Not all required parameters are selected" → keep cascading;
|
||||
// anything else (429/401/5xx) propagates.
|
||||
const status = (err as { status?: number }).status;
|
||||
if (status !== 404) throw err;
|
||||
}
|
||||
if (pending.length === 0) return null; // nothing left to select and still no car
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export interface CrawledNode {
|
||||
id: string; // carcatonline group id
|
||||
parentId: string | null; // carcatonline parent group id
|
||||
name: string;
|
||||
hasSubgroups: boolean;
|
||||
hasParts: boolean;
|
||||
img: string | null;
|
||||
illustration: string | null;
|
||||
depth: number;
|
||||
}
|
||||
|
||||
/** Breadth-first crawl of the whole group tree (one call per node that has subgroups). */
|
||||
export async function crawlGroupTree(
|
||||
client: TreeClient,
|
||||
hooks: CrawlHooks,
|
||||
catalogId: string,
|
||||
carId: string,
|
||||
maxNodes = 5000,
|
||||
): Promise<{ nodes: CrawledNode[]; calls: number }> {
|
||||
const nodes: CrawledNode[] = [];
|
||||
let calls = 0;
|
||||
const queue: { groupId: string | null; depth: number }[] = [{ groupId: null, depth: 1 }];
|
||||
const seen = new Set<string>();
|
||||
while (queue.length > 0 && nodes.length < maxNodes) {
|
||||
const { groupId, depth } = queue.shift() as { groupId: string | null; depth: number };
|
||||
hooks.checkAbort?.();
|
||||
await hooks.beforeCall();
|
||||
const groups = await client.groups(catalogId, carId, groupId ?? undefined);
|
||||
calls += 1;
|
||||
for (const g of groups) {
|
||||
if (!g?.id || seen.has(g.id)) continue;
|
||||
seen.add(g.id);
|
||||
nodes.push({
|
||||
id: g.id,
|
||||
parentId: groupId,
|
||||
name: (g.name || "").trim() || g.id,
|
||||
hasSubgroups: !!g.hasSubgroups,
|
||||
hasParts: !!g.hasParts,
|
||||
img: g.img || null,
|
||||
illustration: g.description?.illustration || null,
|
||||
depth,
|
||||
});
|
||||
if (g.hasSubgroups) queue.push({ groupId: g.id, depth: depth + 1 });
|
||||
}
|
||||
}
|
||||
return { nodes, calls };
|
||||
}
|
||||
|
||||
/**
|
||||
* The `categories` unique index is (vehicle, catalog_vehicle, name, source), so
|
||||
* two nodes with the same display name under one catalog vehicle would collide
|
||||
* and orphan a branch. Disambiguate duplicates with their parent's name, then
|
||||
* with an ordinal.
|
||||
*/
|
||||
export function disambiguateNames(nodes: CrawledNode[]): Map<string, string> {
|
||||
const byId = new Map(nodes.map((n) => [n.id, n]));
|
||||
const counts = new Map<string, number>();
|
||||
for (const n of nodes) counts.set(n.name, (counts.get(n.name) ?? 0) + 1);
|
||||
const used = new Set<string>();
|
||||
const out = new Map<string, string>();
|
||||
for (const n of nodes) {
|
||||
let name = n.name;
|
||||
if ((counts.get(n.name) ?? 0) > 1 && n.parentId) {
|
||||
const parent = byId.get(n.parentId);
|
||||
if (parent) name = `${n.name} (${parent.name})`;
|
||||
}
|
||||
let candidate = name;
|
||||
for (let i = 2; used.has(candidate); i += 1) candidate = `${name} #${i}`;
|
||||
used.add(candidate);
|
||||
out.set(n.id, candidate);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
@@ -37,5 +37,6 @@ export const QUEUE_NAMES = {
|
||||
PART_PRICE_REFRESH: "part-price-refresh",
|
||||
VINPIN_DECODE: "vinpin-decode",
|
||||
RPARTSTORE_DECODE: "rpartstore-decode",
|
||||
CARCATONLINE_BACKFILL: "carcatonline-backfill",
|
||||
CANONICAL_BACKFILL: "canonical-backfill",
|
||||
} as const;
|
||||
|
||||
@@ -7,6 +7,11 @@ import {
|
||||
CANONICAL_BACKFILL_QUEUE,
|
||||
CanonicalBackfillQueueProvider,
|
||||
} from "./queues/canonical-backfill.queue";
|
||||
import {
|
||||
CARCATONLINE_BACKFILL_QUEUE,
|
||||
CARCAT_JOB,
|
||||
CarcatonlineBackfillQueueProvider,
|
||||
} from "./queues/carcatonline-backfill.queue";
|
||||
import {
|
||||
CATALOG_PREFETCH_FAST_QUEUE,
|
||||
CATALOG_PREFETCH_QUEUE,
|
||||
@@ -45,6 +50,7 @@ import { VINPIN_DECODE_QUEUE, VinpinDecodeQueueProvider } from "./queues/vinpin-
|
||||
VinpinDecodeQueueProvider,
|
||||
RpartstoreDecodeQueueProvider,
|
||||
CanonicalBackfillQueueProvider,
|
||||
CarcatonlineBackfillQueueProvider,
|
||||
PrefetchWorkerService,
|
||||
],
|
||||
exports: [
|
||||
@@ -59,6 +65,7 @@ import { VINPIN_DECODE_QUEUE, VinpinDecodeQueueProvider } from "./queues/vinpin-
|
||||
VINPIN_DECODE_QUEUE,
|
||||
RPARTSTORE_DECODE_QUEUE,
|
||||
CANONICAL_BACKFILL_QUEUE,
|
||||
CARCATONLINE_BACKFILL_QUEUE,
|
||||
],
|
||||
})
|
||||
export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
@@ -70,6 +77,7 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
@Inject(EXPERT_REWARDS_QUEUE) private expertRewardsQueue: Queue,
|
||||
@Inject(PART_PRICE_REFRESH_QUEUE) private partPriceRefreshQueue: Queue,
|
||||
@Inject(CANONICAL_BACKFILL_QUEUE) private canonicalBackfillQueue: Queue,
|
||||
@Inject(CARCATONLINE_BACKFILL_QUEUE) private carcatonlineBackfillQueue: Queue,
|
||||
) {}
|
||||
|
||||
async onModuleInit() {
|
||||
@@ -204,6 +212,29 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
console.log(
|
||||
"[jobs] Registered canonical-backfill cron: 0 5 * * * (Europe/Istanbul, incremental)",
|
||||
);
|
||||
|
||||
// carcatonline category backfill: every 30 min the scan job tops the queue
|
||||
// up with empty PL24 catalog vehicles. The scan and the vehicle jobs both
|
||||
// refuse to work outside the night window (CARCATONLINE_WINDOW_START/_END,
|
||||
// default 20:00–07:00 Europe/Istanbul), so the frequent pattern is harmless
|
||||
// by day. Registered only when the source is enabled; removed otherwise so a
|
||||
// disabled env leaves no orphan scheduler behind.
|
||||
if (process.env.CARCATONLINE_ENABLED === "true") {
|
||||
await this.carcatonlineBackfillQueue.upsertJobScheduler(
|
||||
"carcatonline-backfill-scan",
|
||||
{ pattern: "*/30 * * * *", tz: "Europe/Istanbul" },
|
||||
{
|
||||
name: CARCAT_JOB.scan,
|
||||
data: {},
|
||||
opts: { removeOnComplete: { count: 48 }, removeOnFail: { count: 48 } },
|
||||
},
|
||||
);
|
||||
console.log("[jobs] Registered carcatonline-backfill scan: */30 * * * * (night window only)");
|
||||
} else {
|
||||
await this.carcatonlineBackfillQueue
|
||||
.removeJobScheduler("carcatonline-backfill-scan")
|
||||
.catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
@@ -213,6 +244,7 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
|
||||
this.catalogPrefetchQueue.close(),
|
||||
this.lifecycleEmailQueue.close(),
|
||||
this.expertRewardsQueue.close(),
|
||||
this.carcatonlineBackfillQueue.close(),
|
||||
this.partPriceRefreshQueue.close(),
|
||||
this.canonicalBackfillQueue.close(),
|
||||
]);
|
||||
|
||||
@@ -0,0 +1,269 @@
|
||||
import { DelayedError } from "bullmq";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { CarcatGroup } from "../../integrations/carcatonline/carcatonline.client";
|
||||
import { CarcatonlineRateLimitError } from "../../integrations/carcatonline/carcatonline.client";
|
||||
import { CARCAT_KEYS } from "../../integrations/carcatonline/carcatonline.pacing";
|
||||
import { CARCAT_JOB } from "../queues/carcatonline-backfill.queue";
|
||||
import { processCarcatonlineBackfill } from "./carcatonline-backfill.processor";
|
||||
|
||||
const NIGHT = new Date("2026-09-26T19:30:00Z"); // 22:30 Istanbul
|
||||
const DAY = new Date("2026-09-26T09:00:00Z"); // 12:00 Istanbul
|
||||
const CFG = {
|
||||
minIntervalMs: 0,
|
||||
dailyCallCap: 1000,
|
||||
windowStartHour: 20,
|
||||
windowEndHour: 7,
|
||||
lockoutSeconds: 2400,
|
||||
};
|
||||
|
||||
function fakeRedis() {
|
||||
const store = new Map<string, string>();
|
||||
return {
|
||||
store,
|
||||
async get(k: string) {
|
||||
return store.get(k) ?? null;
|
||||
},
|
||||
set: (async (k: string, v: string, mode: string, n: number, cond?: string) => {
|
||||
if (cond === "NX" && store.has(k)) return null;
|
||||
// A zero-length PX slot expires immediately (the tests run with minIntervalMs: 0).
|
||||
if (mode === "PX" && n <= 0) return "OK";
|
||||
store.set(k, v);
|
||||
return "OK";
|
||||
}) as any,
|
||||
async del(k: string) {
|
||||
store.delete(k);
|
||||
},
|
||||
async incr(k: string) {
|
||||
const n = Number(store.get(k) ?? 0) + 1;
|
||||
store.set(k, String(n));
|
||||
return n;
|
||||
},
|
||||
async expire() {},
|
||||
async keys(pattern: string) {
|
||||
const prefix = pattern.replace(/\*$/, "");
|
||||
return [...store.keys()].filter((k) => k.startsWith(prefix));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** Drizzle-ish mock: select chains resolve to queued row sets, updates/inserts are recorded. */
|
||||
function fakeDb(selectResults: unknown[][]) {
|
||||
const updates: Record<string, unknown>[] = [];
|
||||
const inserts: Record<string, unknown>[][] = [];
|
||||
const queue = [...selectResults];
|
||||
const chain = () => {
|
||||
const rows = queue.shift() ?? [];
|
||||
const c: any = {
|
||||
from: () => c,
|
||||
where: () => c,
|
||||
orderBy: () => c,
|
||||
limit: async () => rows,
|
||||
};
|
||||
return c;
|
||||
};
|
||||
let idSeq = 0;
|
||||
const db: any = {
|
||||
select: vi.fn(() => chain()),
|
||||
update: vi.fn(() => ({
|
||||
set: (payload: Record<string, unknown>) => {
|
||||
updates.push(payload);
|
||||
return { where: async () => undefined };
|
||||
},
|
||||
})),
|
||||
insert: vi.fn(() => ({
|
||||
values: (rows: Record<string, unknown>[]) => {
|
||||
inserts.push(rows);
|
||||
return {
|
||||
onConflictDoNothing: () => ({
|
||||
returning: async () =>
|
||||
rows.map((r) => ({ id: `db-${++idSeq}`, externalId: r.externalId })),
|
||||
}),
|
||||
};
|
||||
},
|
||||
})),
|
||||
};
|
||||
return { db, updates, inserts };
|
||||
}
|
||||
|
||||
const cv = {
|
||||
id: "cv-1",
|
||||
source: "pl24",
|
||||
brandName: "Renault",
|
||||
model: "KADJAR",
|
||||
architecture: "P5_MODERN",
|
||||
metadata: null,
|
||||
};
|
||||
|
||||
function fakeClient(tree: Record<string, CarcatGroup[]>) {
|
||||
return {
|
||||
models: vi.fn(async () => [
|
||||
{ id: "m-kadjar", name: "KADJAR" },
|
||||
{ id: "m-clio", name: "CLIO 4 / LUTECIA 4" },
|
||||
]),
|
||||
carsParameters: vi.fn(async (_c: string, _m: string, sel: string[]) =>
|
||||
sel.length === 0
|
||||
? [{ key: "model_selections", name: "Model", values: [{ idx: "suv", value: "SUV" }] }]
|
||||
: [],
|
||||
),
|
||||
cars: vi.fn(async (_c: string, _m: string, sel: string[]) =>
|
||||
sel.length ? [{ id: "car-1", catalogId: "pl_renault" }] : [],
|
||||
),
|
||||
groups: vi.fn(async (_c: string, _car: string, gid?: string) => tree[gid ?? "root"] ?? []),
|
||||
};
|
||||
}
|
||||
|
||||
const job = (name: string, data: Record<string, unknown> = {}) =>
|
||||
({ id: "j1", name, data, moveToDelayed: vi.fn().mockResolvedValue(undefined) }) as any;
|
||||
|
||||
describe("processCarcatonlineBackfill", () => {
|
||||
beforeEach(() => {
|
||||
process.env.CARCATONLINE_ENABLED = "true";
|
||||
});
|
||||
afterEach(() => {
|
||||
process.env.CARCATONLINE_ENABLED = "false";
|
||||
});
|
||||
|
||||
const baseDeps = (db: any, redis: any, client: any, now: Date) => ({
|
||||
db,
|
||||
redis,
|
||||
queue: {
|
||||
add: vi.fn(),
|
||||
getWaitingCount: async () => 0,
|
||||
getDelayedCount: async () => 0,
|
||||
getActiveCount: async () => 0,
|
||||
},
|
||||
enqueueTranslation: vi.fn().mockResolvedValue(undefined),
|
||||
client: () => client,
|
||||
config: CFG,
|
||||
now: () => now,
|
||||
logger: { log: () => undefined, warn: () => undefined },
|
||||
});
|
||||
|
||||
it("is a no-op when disabled", async () => {
|
||||
process.env.CARCATONLINE_ENABLED = "false";
|
||||
const { db } = fakeDb([]);
|
||||
const r = await processCarcatonlineBackfill(
|
||||
job(CARCAT_JOB.scan),
|
||||
undefined,
|
||||
baseDeps(db, fakeRedis(), {}, NIGHT),
|
||||
);
|
||||
expect(r).toEqual({ skipped: true });
|
||||
});
|
||||
|
||||
it("scan: skips outside the night window, enqueues day-scoped vehicle jobs inside it", async () => {
|
||||
const { db } = fakeDb([[{ id: "cv-1" }, { id: "cv-2" }]]);
|
||||
const deps = baseDeps(db, fakeRedis(), {}, DAY);
|
||||
expect(await processCarcatonlineBackfill(job(CARCAT_JOB.scan), undefined, deps)).toEqual({
|
||||
skipped: "outside-window",
|
||||
});
|
||||
const night = baseDeps(db, fakeRedis(), {}, NIGHT);
|
||||
const r = await processCarcatonlineBackfill(job(CARCAT_JOB.scan), undefined, night);
|
||||
expect(r).toEqual({ enqueued: 2 });
|
||||
expect(night.queue.add).toHaveBeenCalledWith(
|
||||
CARCAT_JOB.vehicle,
|
||||
{ catalogVehicleId: "cv-1" },
|
||||
{ jobId: "carcat-cv-1-2026-09-26" },
|
||||
);
|
||||
});
|
||||
|
||||
it("vehicle: outside the window it re-schedules itself to the window start", async () => {
|
||||
const { db } = fakeDb([]);
|
||||
const j = job(CARCAT_JOB.vehicle, { catalogVehicleId: "cv-1" });
|
||||
await expect(
|
||||
processCarcatonlineBackfill(j, "tok", baseDeps(db, fakeRedis(), {}, DAY)),
|
||||
).rejects.toBeInstanceOf(DelayedError);
|
||||
const [ts] = j.moveToDelayed.mock.calls[0];
|
||||
expect(ts).toBe(Date.parse("2026-09-26T17:00:00Z") + 60_000); // 20:00 Istanbul + 1 min
|
||||
});
|
||||
|
||||
it("vehicle: fills the tree, marks the catalog, invalidates the tree cache and queues translations", async () => {
|
||||
const tree: Record<string, CarcatGroup[]> = {
|
||||
root: [
|
||||
{ id: "g1", name: "Engine And Peripherals", hasSubgroups: true, hasParts: false },
|
||||
{ id: "g2", name: "Routine Vehicle Maintenance", hasSubgroups: false, hasParts: true },
|
||||
],
|
||||
g1: [
|
||||
{
|
||||
id: "g1a",
|
||||
name: "01063629",
|
||||
hasSubgroups: false,
|
||||
hasParts: true,
|
||||
description: { illustration: "01063629" },
|
||||
},
|
||||
],
|
||||
};
|
||||
const { db, updates, inserts } = fakeDb([[cv], []]);
|
||||
const redis = fakeRedis();
|
||||
redis.store.set("cat:catalog:tree:cv-1", "stale");
|
||||
const client = fakeClient(tree);
|
||||
const deps = baseDeps(db, redis, client, NIGHT);
|
||||
const r = await processCarcatonlineBackfill(
|
||||
job(CARCAT_JOB.vehicle, { catalogVehicleId: "cv-1" }),
|
||||
"tok",
|
||||
deps,
|
||||
);
|
||||
expect(r).toMatchObject({ status: "done", nodes: 3 });
|
||||
// level 1 then level 2, parent resolved to the DB id of g1
|
||||
expect(inserts).toHaveLength(2);
|
||||
expect(inserts[0].map((x) => [x.name, x.parentId, x.linkPath])).toEqual([
|
||||
["Engine And Peripherals", null, "carcat:pl_renault:car-1:g1"],
|
||||
["Routine Vehicle Maintenance", null, "carcat:pl_renault:car-1:g2"],
|
||||
]);
|
||||
expect(inserts[1][0]).toMatchObject({
|
||||
name: "01063629",
|
||||
parentId: "db-1",
|
||||
source: "carcatonline",
|
||||
hasParts: true,
|
||||
linkWid: "01063629",
|
||||
});
|
||||
expect(updates.some((u) => u.categoriesFetched === true)).toBe(true);
|
||||
expect(redis.store.has("cat:catalog:tree:cv-1")).toBe(false);
|
||||
expect(deps.enqueueTranslation).toHaveBeenCalledWith([
|
||||
"Engine And Peripherals",
|
||||
"Routine Vehicle Maintenance",
|
||||
"01063629",
|
||||
]);
|
||||
// models list cached for 24h
|
||||
expect(redis.store.has(`${CARCAT_KEYS.modelsPrefix}pl_renault`)).toBe(true);
|
||||
});
|
||||
|
||||
it("vehicle: records unmatched models without calling the tree endpoints", async () => {
|
||||
const { db, updates } = fakeDb([[{ ...cv, model: "XJH AİLESİ" }], []]);
|
||||
const client = fakeClient({});
|
||||
const r = await processCarcatonlineBackfill(
|
||||
job(CARCAT_JOB.vehicle, { catalogVehicleId: "cv-1" }),
|
||||
"tok",
|
||||
baseDeps(db, fakeRedis(), client, NIGHT),
|
||||
);
|
||||
expect(r).toMatchObject({ status: "unmatched", reason: "model" });
|
||||
expect(client.groups).not.toHaveBeenCalled();
|
||||
expect(updates.at(-1)).toHaveProperty("metadata");
|
||||
});
|
||||
|
||||
it("vehicle: a 429 marks the shared lockout and re-schedules past it", async () => {
|
||||
const { db } = fakeDb([[cv], []]);
|
||||
const redis = fakeRedis();
|
||||
const client = fakeClient({});
|
||||
client.models = vi.fn(async () => {
|
||||
throw new CarcatonlineRateLimitError("/catalogs/pl_renault/models/");
|
||||
});
|
||||
const j = job(CARCAT_JOB.vehicle, { catalogVehicleId: "cv-1" });
|
||||
await expect(
|
||||
processCarcatonlineBackfill(j, "tok", baseDeps(db, redis, client, NIGHT)),
|
||||
).rejects.toBeInstanceOf(DelayedError);
|
||||
expect(Number(redis.store.get(CARCAT_KEYS.lockout))).toBe(NIGHT.getTime() + 2400 * 1000);
|
||||
expect(j.moveToDelayed.mock.calls[0][0]).toBeGreaterThan(NIGHT.getTime() + 2400 * 1000);
|
||||
});
|
||||
|
||||
it("vehicle: skips catalogs that already have categories", async () => {
|
||||
const { db } = fakeDb([[cv], [{ one: 1 }]]);
|
||||
const client = fakeClient({});
|
||||
const r = await processCarcatonlineBackfill(
|
||||
job(CARCAT_JOB.vehicle, { catalogVehicleId: "cv-1" }),
|
||||
"tok",
|
||||
baseDeps(db, fakeRedis(), client, NIGHT),
|
||||
);
|
||||
expect(r).toEqual({ skipped: "already-has-categories" });
|
||||
expect(client.models).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
388
apps/api/src/jobs/processors/carcatonline-backfill.processor.ts
Normal file
388
apps/api/src/jobs/processors/carcatonline-backfill.processor.ts
Normal file
@@ -0,0 +1,388 @@
|
||||
import { DelayedError, Job, type Queue } from "bullmq";
|
||||
import { and, eq, inArray, notExists, sql } from "drizzle-orm";
|
||||
import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
|
||||
import { catalogVehicles, categories } from "../../database/schema/core";
|
||||
import {
|
||||
type CarcatModel,
|
||||
CarcatonlineClient,
|
||||
CarcatonlineRateLimitError,
|
||||
} from "../../integrations/carcatonline/carcatonline.client";
|
||||
import {
|
||||
catalogsForBrand,
|
||||
isNonVehicleModel,
|
||||
matchCarcatModel,
|
||||
} from "../../integrations/carcatonline/carcatonline.matcher";
|
||||
import {
|
||||
CARCAT_KEYS,
|
||||
type CarcatConfig,
|
||||
CarcatonlineBudgetError,
|
||||
CarcatonlineLockedError,
|
||||
CarcatonlineThrottle,
|
||||
CarcatonlineWindowClosedError,
|
||||
type RedisLike,
|
||||
carcatConfigFromEnv,
|
||||
isWithinWindow,
|
||||
msUntilWindowOpens,
|
||||
redisTokenStore,
|
||||
} from "../../integrations/carcatonline/carcatonline.pacing";
|
||||
import {
|
||||
type CrawledNode,
|
||||
type TreeClient,
|
||||
crawlGroupTree,
|
||||
disambiguateNames,
|
||||
resolveRepresentativeCar,
|
||||
} from "../../integrations/carcatonline/carcatonline.tree";
|
||||
import { CARCAT_JOB, type CarcatonlineVehicleJobData } from "../queues/carcatonline-backfill.queue";
|
||||
|
||||
type Database = PostgresJsDatabase<Record<string, unknown>>;
|
||||
|
||||
/** Only architectures whose catalog page is served from the DB tree (see CatalogService.getCategoryTree). */
|
||||
export const CARCAT_BACKFILL_ARCHITECTURES = ["P5_MODERN"];
|
||||
|
||||
export interface CarcatonlineProcessorDeps {
|
||||
db: Database;
|
||||
redis: RedisLike & { keys(pattern: string): Promise<string[]> };
|
||||
queue: Pick<Queue, "add" | "getWaitingCount" | "getDelayedCount" | "getActiveCount">;
|
||||
/** Enqueue names for the LLM translation pipeline (Turkish). */
|
||||
enqueueTranslation: (terms: string[]) => Promise<void>;
|
||||
/** Lazily built so missing credentials fail the job, not worker boot. */
|
||||
client: () => TreeClient & { models(catalogId: string): Promise<CarcatModel[]> };
|
||||
config?: CarcatConfig;
|
||||
scanBatch?: number;
|
||||
now?: () => Date;
|
||||
logger?: { log: (m: string) => void; warn: (m: string) => void };
|
||||
}
|
||||
|
||||
export interface CarcatMetadata {
|
||||
status: "matched" | "unmatched" | "no_car" | "done" | "failed";
|
||||
catalogId?: string;
|
||||
modelId?: string;
|
||||
modelName?: string;
|
||||
matchTier?: string;
|
||||
carId?: string;
|
||||
parameters?: { key: string; value: string; idx: string }[];
|
||||
nodes?: number;
|
||||
calls?: number;
|
||||
error?: string;
|
||||
at: string;
|
||||
}
|
||||
|
||||
/** Don't retry unmatched / failed vehicles more often than this. */
|
||||
const RETRY_AFTER_DAYS = 7;
|
||||
|
||||
/** Worker-side client; pacing/lockout/budget are applied by the crawl hooks, not here. */
|
||||
export function buildCarcatClient(redis: RedisLike): CarcatonlineClient {
|
||||
const email = process.env.CARCATONLINE_EMAIL;
|
||||
const password = process.env.CARCATONLINE_PASSWORD;
|
||||
if (!email || !password) {
|
||||
throw new Error("CARCATONLINE_EMAIL / CARCATONLINE_PASSWORD are not configured");
|
||||
}
|
||||
return new CarcatonlineClient(
|
||||
{ email, password, logger: { log: (m) => console.log(m), warn: (m) => console.warn(m) } },
|
||||
redisTokenStore(redis),
|
||||
);
|
||||
}
|
||||
|
||||
export async function processCarcatonlineBackfill(
|
||||
job: Job,
|
||||
token: string | undefined,
|
||||
deps: CarcatonlineProcessorDeps,
|
||||
): Promise<Record<string, unknown>> {
|
||||
const log = deps.logger ?? { log: (m) => console.log(m), warn: (m) => console.warn(m) };
|
||||
if (process.env.CARCATONLINE_ENABLED !== "true") {
|
||||
log.log(`[carcatonline] disabled (CARCATONLINE_ENABLED!=true) — job ${job.id} no-op`);
|
||||
return { skipped: true };
|
||||
}
|
||||
const cfg = deps.config ?? carcatConfigFromEnv();
|
||||
const now = deps.now ?? (() => new Date());
|
||||
if (job.name === CARCAT_JOB.scan) return scan(job, deps, cfg, now, log);
|
||||
return fillVehicle(job as Job<CarcatonlineVehicleJobData>, token, deps, cfg, now, log);
|
||||
}
|
||||
|
||||
/** Periodic scan: inside the window, top the queue up with empty PL24 catalogs. */
|
||||
async function scan(
|
||||
job: Job,
|
||||
deps: CarcatonlineProcessorDeps,
|
||||
cfg: CarcatConfig,
|
||||
now: () => Date,
|
||||
log: NonNullable<CarcatonlineProcessorDeps["logger"]>,
|
||||
): Promise<Record<string, unknown>> {
|
||||
if (!isWithinWindow(now(), cfg.windowStartHour, cfg.windowEndHour)) {
|
||||
return { skipped: "outside-window" };
|
||||
}
|
||||
const pending =
|
||||
(await deps.queue.getWaitingCount()) +
|
||||
(await deps.queue.getDelayedCount()) +
|
||||
(await deps.queue.getActiveCount());
|
||||
const batch = deps.scanBatch ?? 20;
|
||||
if (pending >= batch) return { skipped: "queue-busy", pending };
|
||||
|
||||
const cutoff = new Date(now().getTime() - RETRY_AFTER_DAYS * 24 * 3600 * 1000).toISOString();
|
||||
const rows = await deps.db
|
||||
.select({ id: catalogVehicles.id })
|
||||
.from(catalogVehicles)
|
||||
.where(
|
||||
and(
|
||||
eq(catalogVehicles.source, "pl24"),
|
||||
inArray(catalogVehicles.architecture, CARCAT_BACKFILL_ARCHITECTURES),
|
||||
notExists(
|
||||
deps.db
|
||||
.select({ one: sql`1` })
|
||||
.from(categories)
|
||||
.where(eq(categories.catalogVehicleId, catalogVehicles.id)),
|
||||
),
|
||||
// never tried, or tried more than RETRY_AFTER_DAYS ago
|
||||
sql`(${catalogVehicles.metadata}->'carcatonline'->>'at') IS NULL OR (${catalogVehicles.metadata}->'carcatonline'->>'at') < ${cutoff}`,
|
||||
),
|
||||
)
|
||||
.orderBy(
|
||||
sql`${catalogVehicles.metadata}->'carcatonline'->>'at' NULLS FIRST, ${catalogVehicles.createdAt}`,
|
||||
)
|
||||
.limit(batch - pending);
|
||||
|
||||
for (const r of rows) {
|
||||
await deps.queue.add(
|
||||
CARCAT_JOB.vehicle,
|
||||
{ catalogVehicleId: r.id } satisfies CarcatonlineVehicleJobData,
|
||||
{ jobId: `carcat-${r.id}-${now().toISOString().slice(0, 10)}` },
|
||||
);
|
||||
}
|
||||
log.log(`[carcatonline] scan enqueued ${rows.length} vehicle(s) (pending before: ${pending})`);
|
||||
return { enqueued: rows.length };
|
||||
}
|
||||
|
||||
async function fillVehicle(
|
||||
job: Job<CarcatonlineVehicleJobData>,
|
||||
token: string | undefined,
|
||||
deps: CarcatonlineProcessorDeps,
|
||||
cfg: CarcatConfig,
|
||||
now: () => Date,
|
||||
log: NonNullable<CarcatonlineProcessorDeps["logger"]>,
|
||||
): Promise<Record<string, unknown>> {
|
||||
const { catalogVehicleId, force } = job.data;
|
||||
const throttle = new CarcatonlineThrottle(deps.redis, cfg, now);
|
||||
|
||||
const reschedule = async (ms: number, reason: string): Promise<never> => {
|
||||
log.warn(
|
||||
`[carcatonline] ${catalogVehicleId}: ${reason} → retry in ${Math.round(ms / 60000)} min`,
|
||||
);
|
||||
await job.moveToDelayed(now().getTime() + ms, token);
|
||||
throw new DelayedError();
|
||||
};
|
||||
|
||||
if (!force && !isWithinWindow(now(), cfg.windowStartHour, cfg.windowEndHour)) {
|
||||
await reschedule(
|
||||
msUntilWindowOpens(now(), cfg.windowStartHour, cfg.windowEndHour) + 60_000,
|
||||
"outside window",
|
||||
);
|
||||
}
|
||||
const locked = await throttle.lockoutRemainingMs();
|
||||
if (locked > 0) await reschedule(locked + 30_000, "locked out");
|
||||
|
||||
const [cv] = await deps.db
|
||||
.select()
|
||||
.from(catalogVehicles)
|
||||
.where(eq(catalogVehicles.id, catalogVehicleId))
|
||||
.limit(1);
|
||||
if (!cv) return { skipped: "missing" };
|
||||
const [existing] = await deps.db
|
||||
.select({ one: sql<number>`1` })
|
||||
.from(categories)
|
||||
.where(eq(categories.catalogVehicleId, catalogVehicleId))
|
||||
.limit(1);
|
||||
if (existing) return { skipped: "already-has-categories" };
|
||||
|
||||
const client = deps.client();
|
||||
const hooks = {
|
||||
beforeCall: () => throttle.beforeCall(),
|
||||
checkAbort: () => {
|
||||
if (!force && !isWithinWindow(now(), cfg.windowStartHour, cfg.windowEndHour)) {
|
||||
throw new CarcatonlineWindowClosedError(
|
||||
msUntilWindowOpens(now(), cfg.windowStartHour, cfg.windowEndHour),
|
||||
);
|
||||
}
|
||||
},
|
||||
};
|
||||
const setMeta = async (meta: CarcatMetadata): Promise<void> => {
|
||||
await deps.db
|
||||
.update(catalogVehicles)
|
||||
.set({
|
||||
metadata: sql`coalesce(${catalogVehicles.metadata}, '{}'::jsonb) || ${JSON.stringify({ carcatonline: meta })}::jsonb`,
|
||||
updatedAt: now(),
|
||||
})
|
||||
.where(eq(catalogVehicles.id, catalogVehicleId));
|
||||
};
|
||||
|
||||
try {
|
||||
// 1. brand → catalog candidates → model
|
||||
if (isNonVehicleModel(cv.model)) {
|
||||
await setMeta({
|
||||
status: "unmatched",
|
||||
error: "non-vehicle pseudo-model",
|
||||
at: now().toISOString(),
|
||||
});
|
||||
return { status: "unmatched", reason: "non-vehicle" };
|
||||
}
|
||||
const catalogIds = catalogsForBrand(cv.brandName);
|
||||
if (catalogIds.length === 0) {
|
||||
await setMeta({
|
||||
status: "unmatched",
|
||||
error: `no carcatonline catalog for brand ${cv.brandName}`,
|
||||
at: now().toISOString(),
|
||||
});
|
||||
return { status: "unmatched", reason: "brand" };
|
||||
}
|
||||
let match: { catalogId: string; modelId: string; modelName: string; tier: string } | null =
|
||||
null;
|
||||
for (const catalogId of catalogIds) {
|
||||
const models = await cachedModels(deps, client, hooks, catalogId);
|
||||
const m = matchCarcatModel(cv.model, models);
|
||||
if (m) {
|
||||
match = { catalogId, modelId: m.model.id, modelName: m.model.name, tier: m.tier };
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!match) {
|
||||
await setMeta({
|
||||
status: "unmatched",
|
||||
error: `model "${cv.model}" not in ${catalogIds.join(",")}`,
|
||||
at: now().toISOString(),
|
||||
});
|
||||
log.warn(
|
||||
`[carcatonline] ${cv.brandName} "${cv.model}" → unmatched (${catalogIds.join(",")})`,
|
||||
);
|
||||
return { status: "unmatched", reason: "model" };
|
||||
}
|
||||
|
||||
// 2. representative car
|
||||
const car = await resolveRepresentativeCar(client, hooks, match.catalogId, match.modelId);
|
||||
if (!car) {
|
||||
await setMeta({ status: "no_car", ...match, at: now().toISOString() });
|
||||
return { status: "no_car" };
|
||||
}
|
||||
|
||||
// 3. full group tree
|
||||
const { nodes, calls } = await crawlGroupTree(client, hooks, match.catalogId, car.carId);
|
||||
if (nodes.length === 0) {
|
||||
await setMeta({
|
||||
status: "failed",
|
||||
...match,
|
||||
carId: car.carId,
|
||||
error: "empty tree",
|
||||
at: now().toISOString(),
|
||||
});
|
||||
return { status: "empty" };
|
||||
}
|
||||
|
||||
// 4. persist (level by level so parent ids resolve), invalidate tree cache, queue translations
|
||||
const inserted = await insertTree(deps.db, catalogVehicleId, match.catalogId, car.carId, nodes);
|
||||
await deps.db
|
||||
.update(catalogVehicles)
|
||||
.set({ categoriesFetched: true, updatedAt: now() })
|
||||
.where(eq(catalogVehicles.id, catalogVehicleId));
|
||||
await setMeta({
|
||||
status: "done",
|
||||
...match,
|
||||
carId: car.carId,
|
||||
parameters: car.parameters.map((p) => ({ key: p.key, value: p.value, idx: p.idx })),
|
||||
nodes: inserted,
|
||||
calls: calls + 2,
|
||||
at: now().toISOString(),
|
||||
});
|
||||
for (const key of await deps.redis.keys(`cat:catalog:tree:${catalogVehicleId}*`))
|
||||
await deps.redis.del(key);
|
||||
await deps.enqueueTranslation([...new Set(nodes.map((n) => n.name))]);
|
||||
|
||||
log.log(
|
||||
`[carcatonline] ${cv.brandName} "${cv.model}" → ${match.catalogId}/${match.modelName} (${match.tier}) car=${car.carId} nodes=${inserted} calls=${calls}`,
|
||||
);
|
||||
return { status: "done", nodes: inserted, calls };
|
||||
} catch (err) {
|
||||
if (err instanceof DelayedError) throw err;
|
||||
if (err instanceof CarcatonlineRateLimitError) {
|
||||
await throttle.markLockout();
|
||||
await reschedule(cfg.lockoutSeconds * 1000 + 30_000, "429 rate limit (lockout marked)");
|
||||
}
|
||||
if (err instanceof CarcatonlineLockedError)
|
||||
await reschedule(err.retryAfterMs + 30_000, "locked out");
|
||||
if (err instanceof CarcatonlineBudgetError) {
|
||||
await reschedule(
|
||||
msUntilWindowOpens(now(), cfg.windowStartHour, cfg.windowEndHour) + 3600_000,
|
||||
err.message,
|
||||
);
|
||||
}
|
||||
if (err instanceof CarcatonlineWindowClosedError)
|
||||
await reschedule(err.retryAfterMs + 60_000, "window closed mid-crawl");
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
await setMeta({ status: "failed", error: message.slice(0, 300), at: now().toISOString() });
|
||||
log.warn(`[carcatonline] ${catalogVehicleId} failed: ${message}`);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
async function cachedModels(
|
||||
deps: CarcatonlineProcessorDeps,
|
||||
client: { models(catalogId: string): Promise<CarcatModel[]> },
|
||||
hooks: { beforeCall(): Promise<void> },
|
||||
catalogId: string,
|
||||
): Promise<CarcatModel[]> {
|
||||
const key = `${CARCAT_KEYS.modelsPrefix}${catalogId}`;
|
||||
const raw = await deps.redis.get(key);
|
||||
if (raw) {
|
||||
try {
|
||||
return JSON.parse(raw) as CarcatModel[];
|
||||
} catch {
|
||||
/* refetch */
|
||||
}
|
||||
}
|
||||
await hooks.beforeCall();
|
||||
const models = await client.models(catalogId);
|
||||
await deps.redis.set(
|
||||
key,
|
||||
JSON.stringify(models.map((m) => ({ id: m.id, name: m.name }))),
|
||||
"EX",
|
||||
24 * 3600,
|
||||
);
|
||||
return models;
|
||||
}
|
||||
|
||||
/** Insert the crawled tree; returns the number of rows written. */
|
||||
export async function insertTree(
|
||||
db: Database,
|
||||
catalogVehicleId: string,
|
||||
catalogId: string,
|
||||
carId: string,
|
||||
nodes: CrawledNode[],
|
||||
): Promise<number> {
|
||||
const names = disambiguateNames(nodes);
|
||||
const dbIdByGroup = new Map<string, string>();
|
||||
let written = 0;
|
||||
const maxDepth = Math.max(...nodes.map((n) => n.depth));
|
||||
for (let depth = 1; depth <= maxDepth; depth += 1) {
|
||||
const level = nodes.filter(
|
||||
(n) => n.depth === depth && (n.parentId === null || dbIdByGroup.has(n.parentId)),
|
||||
);
|
||||
if (level.length === 0) continue;
|
||||
const rows = level.map((n) => ({
|
||||
catalogVehicleId,
|
||||
vehicleId: null as string | null,
|
||||
name: names.get(n.id) ?? n.name,
|
||||
nameOriginal: n.name,
|
||||
parentId: n.parentId ? (dbIdByGroup.get(n.parentId) ?? null) : null,
|
||||
externalId: n.id,
|
||||
linkPath: `carcat:${catalogId}:${carId}:${n.id}`,
|
||||
linkWid: n.illustration ?? null,
|
||||
source: "carcatonline",
|
||||
hasSubgroups: n.hasSubgroups,
|
||||
hasParts: n.hasParts,
|
||||
}));
|
||||
const inserted = await db
|
||||
.insert(categories)
|
||||
.values(rows)
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: categories.id, externalId: categories.externalId });
|
||||
for (const r of inserted) if (r.externalId) dbIdByGroup.set(r.externalId, r.id);
|
||||
written += inserted.length;
|
||||
}
|
||||
return written;
|
||||
}
|
||||
@@ -116,14 +116,14 @@ export async function processTranslation(
|
||||
await db.execute(drizzleSql`
|
||||
UPDATE categories
|
||||
SET name = ${tr}
|
||||
WHERE source IN ('emex', 'parts-catalogs')
|
||||
WHERE source IN ('emex', 'parts-catalogs', 'carcatonline')
|
||||
AND name_original = ${orig}
|
||||
AND name = name_original
|
||||
`);
|
||||
await db.execute(drizzleSql`
|
||||
UPDATE parts
|
||||
SET name = ${tr}
|
||||
WHERE source IN ('emex', 'parts-catalogs')
|
||||
WHERE source IN ('emex', 'parts-catalogs', 'carcatonline')
|
||||
AND name_original = ${orig}
|
||||
AND name = name_original
|
||||
`);
|
||||
|
||||
43
apps/api/src/jobs/queues/carcatonline-backfill.queue.ts
Normal file
43
apps/api/src/jobs/queues/carcatonline-backfill.queue.ts
Normal file
@@ -0,0 +1,43 @@
|
||||
import { Provider } from "@nestjs/common";
|
||||
import { type JobsOptions, Queue } from "bullmq";
|
||||
import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config";
|
||||
|
||||
export const CARCATONLINE_BACKFILL_QUEUE = "CARCATONLINE_BACKFILL_QUEUE";
|
||||
|
||||
/** Job names on the carcatonline backfill queue. */
|
||||
export const CARCAT_JOB = {
|
||||
/** Periodic: pick empty PL24 catalog vehicles and enqueue `vehicle` jobs. */
|
||||
scan: "carcatonline-scan",
|
||||
/** Fill one catalog vehicle's category tree from carcatonline. */
|
||||
vehicle: "carcatonline-vehicle",
|
||||
} as const;
|
||||
|
||||
export interface CarcatonlineVehicleJobData {
|
||||
catalogVehicleId: string;
|
||||
/** Skip the night-window check (manual runs). Lockout and daily cap still apply. */
|
||||
force?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* `attempts: 1`: the processor itself re-schedules (moveToDelayed) on lockout,
|
||||
* budget or window-closed, and records terminal outcomes in
|
||||
* `catalog_vehicles.metadata.carcatonline`, so BullMQ retries would only burn
|
||||
* the shared call budget.
|
||||
*/
|
||||
export const CARCATONLINE_BACKFILL_JOB_OPTIONS: JobsOptions = {
|
||||
attempts: 1,
|
||||
removeOnComplete: { count: 200 },
|
||||
removeOnFail: { count: 200 },
|
||||
};
|
||||
|
||||
export const CarcatonlineBackfillQueueProvider: Provider = {
|
||||
provide: CARCATONLINE_BACKFILL_QUEUE,
|
||||
useFactory: () => {
|
||||
const telemetry = getBullTelemetry();
|
||||
return new Queue(QUEUE_NAMES.CARCATONLINE_BACKFILL, {
|
||||
connection: getBullConnection(),
|
||||
...(telemetry ? { telemetry } : {}),
|
||||
defaultJobOptions: CARCATONLINE_BACKFILL_JOB_OPTIONS,
|
||||
});
|
||||
},
|
||||
};
|
||||
@@ -2,7 +2,7 @@ import { Sentry } from "./instrument-worker"; // MUST be first — initializes S
|
||||
import "./telemetry/worker-tracing"; // MUST be early — instruments modules before they load
|
||||
|
||||
import "dotenv/config";
|
||||
import { Worker } from "bullmq";
|
||||
import { Queue, Worker } from "bullmq";
|
||||
import { drizzle } from "drizzle-orm/postgres-js";
|
||||
import Redis from "ioredis";
|
||||
import OpenAI from "openai";
|
||||
@@ -10,6 +10,10 @@ import postgres from "postgres";
|
||||
import { getVinpinDaemon } from "./integrations/vinpin/vinpin-daemon.service";
|
||||
import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "./jobs/bull.config";
|
||||
import { processCanonicalBackfill } from "./jobs/processors/canonical-backfill.processor";
|
||||
import {
|
||||
buildCarcatClient,
|
||||
processCarcatonlineBackfill,
|
||||
} from "./jobs/processors/carcatonline-backfill.processor";
|
||||
import { processEmexScrape } from "./jobs/processors/emex-scrape.processor";
|
||||
import { processExpertRewards } from "./jobs/processors/expert-rewards.processor";
|
||||
import { processLifecycleEmails } from "./jobs/processors/lifecycle-email.processor";
|
||||
@@ -297,6 +301,71 @@ rpartstoreDecodeWorker.on("failed", (job, err) => {
|
||||
|
||||
workers.push(rpartstoreDecodeWorker);
|
||||
|
||||
// carcatonline category backfill (night window only, 1 upstream call / ~2 s,
|
||||
// shared Redis lockout + daily budget with the API's on-demand parts fetch).
|
||||
// Concurrency 1: the crawl is sequential by design; the processor re-schedules
|
||||
// itself on lockout / budget / window-closed. No-op unless CARCATONLINE_ENABLED.
|
||||
const carcatRedis = new Redis({
|
||||
host: process.env.REDIS_HOST || "localhost",
|
||||
port: Number(process.env.REDIS_PORT) || 6379,
|
||||
password: process.env.REDIS_PASSWORD || undefined,
|
||||
maxRetriesPerRequest: null,
|
||||
lazyConnect: true,
|
||||
});
|
||||
const carcatQueue = new Queue(QUEUE_NAMES.CARCATONLINE_BACKFILL, { connection });
|
||||
const carcatTranslationQueue = new Queue(QUEUE_NAMES.TRANSLATION, { connection });
|
||||
let carcatClient: ReturnType<typeof buildCarcatClient> | null = null;
|
||||
const carcatonlineBackfillWorker = new Worker(
|
||||
QUEUE_NAMES.CARCATONLINE_BACKFILL,
|
||||
async (job, token) => {
|
||||
return processCarcatonlineBackfill(job, token, {
|
||||
db,
|
||||
redis: carcatRedis,
|
||||
queue: carcatQueue,
|
||||
client: () => {
|
||||
carcatClient ??= buildCarcatClient(carcatRedis);
|
||||
return carcatClient;
|
||||
},
|
||||
// Same dedupe + jobId scheme as TranslationsService.enqueueTranslation so
|
||||
// API-side and worker-side enqueues collapse onto the same jobs.
|
||||
enqueueTranslation: async (terms) => {
|
||||
const fresh: string[] = [];
|
||||
for (const t of terms) {
|
||||
if ((await carcatRedis.set(`tr:queued:${t}`, "1", "EX", 300, "NX")) === "OK")
|
||||
fresh.push(t);
|
||||
}
|
||||
for (let i = 0; i < fresh.length; i += 50) {
|
||||
const chunk = fresh.slice(i, i + 50);
|
||||
await carcatTranslationQueue.add(
|
||||
"translate",
|
||||
{ terms: chunk },
|
||||
{ jobId: `tr-${Buffer.from(chunk.sort().join("|")).toString("base64").slice(0, 32)}` },
|
||||
);
|
||||
}
|
||||
},
|
||||
scanBatch: Number(process.env.CARCATONLINE_SCAN_BATCH) || 20,
|
||||
});
|
||||
},
|
||||
{
|
||||
connection,
|
||||
concurrency: 1,
|
||||
...(telemetry ? { telemetry } : {}),
|
||||
},
|
||||
);
|
||||
|
||||
carcatonlineBackfillWorker.on("completed", (job, result) => {
|
||||
console.log(`[worker] carcatonline-backfill job ${job.id} completed → ${JSON.stringify(result)}`);
|
||||
});
|
||||
|
||||
carcatonlineBackfillWorker.on("failed", (job, err) => {
|
||||
console.error(`[worker] carcatonline-backfill job ${job?.id} failed: ${err.message}`);
|
||||
Sentry.captureException(err, {
|
||||
tags: { queue: QUEUE_NAMES.CARCATONLINE_BACKFILL, jobId: job?.id },
|
||||
});
|
||||
});
|
||||
|
||||
workers.push(carcatonlineBackfillWorker);
|
||||
|
||||
// Vinpin warm-session daemon: holds the single Vinpin seat warm (browser + login
|
||||
// + Fiat ePER / Renault Rpartstore / Dialogys windows open) during business hours
|
||||
// (08:00–21:00 Europe/Istanbul), keepalive-nudged every ~75s, so decodes run on
|
||||
@@ -408,8 +477,10 @@ async function shutdown(signal: string) {
|
||||
await vinpinDaemon.stop();
|
||||
console.log("[worker] Vinpin warm daemon stopped");
|
||||
|
||||
// 2c. Release the RPartStore token/counter Redis client.
|
||||
// 2c. Release the RPartStore / carcatonline Redis clients + producer queues.
|
||||
rpartstoreRedis.disconnect();
|
||||
carcatRedis.disconnect();
|
||||
await Promise.all([carcatQueue.close(), carcatTranslationQueue.close()]);
|
||||
|
||||
// 3. Close database connection
|
||||
await sql.end();
|
||||
|
||||
@@ -78,6 +78,17 @@ services:
|
||||
- RPARTSTORE_DAILY_CAP=${RPARTSTORE_DAILY_CAP:-10}
|
||||
- RPARTSTORE_BROKER_URL=${RPARTSTORE_BROKER_URL:-wss://1po-bff.renault-edh.com/ws}
|
||||
- RPARTSTORE_APP_VERSION=${RPARTSTORE_APP_VERSION:-1.34.0.6}
|
||||
# carcatonline (PartsLink24 mirror) — night-window (20:00-07:00 Istanbul) category backfill
|
||||
# for empty PL24 catalogs + on-demand parts; 1 call/7 s (~100 calls/window quota), 429 → 40 min lockout, daily cap.
|
||||
- CARCATONLINE_ENABLED=${CARCATONLINE_ENABLED:-false}
|
||||
- CARCATONLINE_EMAIL=${CARCATONLINE_EMAIL:-}
|
||||
- CARCATONLINE_PASSWORD=${CARCATONLINE_PASSWORD:-}
|
||||
- CARCATONLINE_DAILY_CALL_CAP=${CARCATONLINE_DAILY_CALL_CAP:-15000}
|
||||
- CARCATONLINE_WINDOW_START=${CARCATONLINE_WINDOW_START:-20}
|
||||
- CARCATONLINE_WINDOW_END=${CARCATONLINE_WINDOW_END:-7}
|
||||
- CARCATONLINE_MIN_INTERVAL_MS=${CARCATONLINE_MIN_INTERVAL_MS:-7000}
|
||||
- CARCATONLINE_LOCKOUT_SECONDS=${CARCATONLINE_LOCKOUT_SECONDS:-2400}
|
||||
- CARCATONLINE_SCAN_BATCH=${CARCATONLINE_SCAN_BATCH:-20}
|
||||
- POSTAL_API_URL=${POSTAL_API_URL:-}
|
||||
- POSTAL_API_KEY=${POSTAL_API_KEY:-}
|
||||
- POSTAL_FROM_ADDRESS=${POSTAL_FROM_ADDRESS:-noreply@sase.tr}
|
||||
@@ -243,6 +254,17 @@ services:
|
||||
- RPARTSTORE_DAILY_CAP=${RPARTSTORE_DAILY_CAP:-10}
|
||||
- RPARTSTORE_BROKER_URL=${RPARTSTORE_BROKER_URL:-wss://1po-bff.renault-edh.com/ws}
|
||||
- RPARTSTORE_APP_VERSION=${RPARTSTORE_APP_VERSION:-1.34.0.6}
|
||||
# carcatonline (PartsLink24 mirror) — night-window (20:00-07:00 Istanbul) category backfill
|
||||
# for empty PL24 catalogs + on-demand parts; 1 call/7 s (~100 calls/window quota), 429 → 40 min lockout, daily cap.
|
||||
- CARCATONLINE_ENABLED=${CARCATONLINE_ENABLED:-false}
|
||||
- CARCATONLINE_EMAIL=${CARCATONLINE_EMAIL:-}
|
||||
- CARCATONLINE_PASSWORD=${CARCATONLINE_PASSWORD:-}
|
||||
- CARCATONLINE_DAILY_CALL_CAP=${CARCATONLINE_DAILY_CALL_CAP:-15000}
|
||||
- CARCATONLINE_WINDOW_START=${CARCATONLINE_WINDOW_START:-20}
|
||||
- CARCATONLINE_WINDOW_END=${CARCATONLINE_WINDOW_END:-7}
|
||||
- CARCATONLINE_MIN_INTERVAL_MS=${CARCATONLINE_MIN_INTERVAL_MS:-7000}
|
||||
- CARCATONLINE_LOCKOUT_SECONDS=${CARCATONLINE_LOCKOUT_SECONDS:-2400}
|
||||
- CARCATONLINE_SCAN_BATCH=${CARCATONLINE_SCAN_BATCH:-20}
|
||||
# Novu lifecycle e-mail automation — the worker fires trial-ending + win-back
|
||||
- NOVU_API_URL=${NOVU_API_URL:-https://api.bildirim.semih.ai}
|
||||
- NOVU_API_KEY=${NOVU_API_KEY:-}
|
||||
|
||||
@@ -96,6 +96,22 @@ export const envSchema = z.object({
|
||||
z.string().default("1.34.0.6"),
|
||||
),
|
||||
|
||||
// carcatonline (pro.carcatonline.com, PartsLink24 mirror) — night-window
|
||||
// category backfill for empty PL24 catalog vehicles + on-demand parts for the
|
||||
// seeded leaves. Shared Redis pacing/lockout/daily budget across api+worker.
|
||||
CARCATONLINE_ENABLED: z
|
||||
.string()
|
||||
.transform((v) => v === "true")
|
||||
.default("false"),
|
||||
CARCATONLINE_EMAIL: z.string().optional(),
|
||||
CARCATONLINE_PASSWORD: z.string().optional(),
|
||||
CARCATONLINE_DAILY_CALL_CAP: z.coerce.number().int().min(0).default(15000),
|
||||
CARCATONLINE_WINDOW_START: z.coerce.number().int().min(0).max(23).default(20),
|
||||
CARCATONLINE_WINDOW_END: z.coerce.number().int().min(0).max(23).default(7),
|
||||
CARCATONLINE_MIN_INTERVAL_MS: z.coerce.number().int().min(500).default(7000),
|
||||
CARCATONLINE_LOCKOUT_SECONDS: z.coerce.number().int().min(60).default(2400),
|
||||
CARCATONLINE_SCAN_BATCH: z.coerce.number().int().min(1).default(20),
|
||||
|
||||
// Parts-Catalogs (Playwright JWT capture + DataImpulse proxy)
|
||||
PCAT_USE_PROXY: z.string().default("true"),
|
||||
PCAT_PROXY_HOST: z.string().default("gw.dataimpulse.com"),
|
||||
|
||||
@@ -5,7 +5,7 @@ export interface Category {
|
||||
nameOriginal: string | null;
|
||||
parentId: string | null;
|
||||
externalId: string | null;
|
||||
source: "pl24" | "emex" | "parts-catalogs";
|
||||
source: "pl24" | "emex" | "parts-catalogs" | "carcatonline";
|
||||
createdAt: Date;
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@ export interface SchemaPic {
|
||||
categoryId: string;
|
||||
imageUrl: string;
|
||||
hotspots: Hotspot[];
|
||||
source: "pl24" | "emex" | "parts-catalogs";
|
||||
source: "pl24" | "emex" | "parts-catalogs" | "carcatonline";
|
||||
createdAt: Date;
|
||||
}
|
||||
|
||||
|
||||
@@ -13,7 +13,7 @@ export interface Part {
|
||||
createdAt: Date;
|
||||
}
|
||||
|
||||
export type PartSource = "pl24" | "emex";
|
||||
export type PartSource = "pl24" | "emex" | "parts-catalogs" | "carcatonline";
|
||||
|
||||
export interface PartSearchResult {
|
||||
part: Part;
|
||||
|
||||
Reference in New Issue
Block a user