diff --git a/apps/api/drizzle/0037_rpartstore_decodes.sql b/apps/api/drizzle/0037_rpartstore_decodes.sql new file mode 100644 index 0000000..42d2a71 --- /dev/null +++ b/apps/api/drizzle/0037_rpartstore_decodes.sql @@ -0,0 +1,34 @@ +-- RPartStore (rpartstore.renault.com, Renault Group dealer portal) VIN-decode +-- fallback cache (feature-flagged: RPARTSTORE_ENABLED). One row per VIN. The +-- rpartstore-decode BullMQ job asks the RPartStore BFF (STOMP over WebSocket) +-- for Renault/Dacia VINs the normal chain (pcat/PL24/emex) can't identify, then +-- matches the decoded model to an EXISTING PL24 catalog_vehicle. RPartStore is +-- ONLY a decode oracle here — parts are served from PL24's existing catalog. +-- status: 'pending' | 'decoded' | 'not_found' | 'capped' | 'failed'. +CREATE TABLE IF NOT EXISTS "rpartstore_decodes" ( + "vin" text PRIMARY KEY NOT NULL, + "status" text DEFAULT 'pending' NOT NULL, + "brand_name" text, + "model" text, + "model_code" text, + "family_code" text, + "model_year" text, + "engine" text, + "gearbox" text, + "energy_type" text, + "manufacturing_date" text, + "catalog_vehicle_id" uuid, + "raw" jsonb, + "attempts" integer DEFAULT 0 NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone, + "decoded_at" timestamp with time zone +); +--> statement-breakpoint +DO $$ BEGIN + ALTER TABLE "rpartstore_decodes" ADD CONSTRAINT "rpartstore_decodes_catalog_vehicle_id_catalog_vehicles_id_fk" FOREIGN KEY ("catalog_vehicle_id") REFERENCES "public"."catalog_vehicles"("id") ON DELETE set null ON UPDATE no action; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "rpartstore_decodes_status_idx" ON "rpartstore_decodes" USING btree ("status"); diff --git a/apps/api/drizzle/meta/_journal.json b/apps/api/drizzle/meta/_journal.json index 5d5d111..9b53e4a 100644 --- a/apps/api/drizzle/meta/_journal.json +++ b/apps/api/drizzle/meta/_journal.json @@ -260,6 +260,13 @@ "when": 1786521893000, "tag": "0036_pl24_mitsubishi_partinfo_cleanup", "breakpoints": true + }, + { + "idx": 37, + "version": "7", + "when": 1790332800000, + "tag": "0037_rpartstore_decodes", + "breakpoints": true } ] -} \ No newline at end of file +} diff --git a/apps/api/src/database/schema/core.ts b/apps/api/src/database/schema/core.ts index 8e3121b..0c6409d 100644 --- a/apps/api/src/database/schema/core.ts +++ b/apps/api/src/database/schema/core.ts @@ -409,6 +409,36 @@ export const vinpinDecodes = pgTable("vinpin_decodes", { decodedAt: timestamp("decoded_at", { withTimezone: true }), }); +// ─── RPartStore decodes (Renault/Dacia decode-oracle cache — one row per VIN) ── +// Feature-flagged (RPARTSTORE_ENABLED). Populated by the rpartstore-decode BullMQ +// job which asks the RPartStore BFF (rpartstore.renault.com, STOMP/WebSocket) for +// Renault/Dacia VINs the normal chain (pcat/PL24/emex) can't identify. On success +// the decoded model is matched to an EXISTING PL24 catalog_vehicle and parts are +// served from there — RPartStore is ONLY a decode oracle. Hard daily cap on +// searches (RPARTSTORE_DAILY_CAP); over-cap VINs are 'capped' and retried after 24 h. +export const rpartstoreDecodes = pgTable("rpartstore_decodes", { + vin: text("vin").primaryKey(), + // 'pending' | 'decoded' | 'not_found' | 'capped' | 'failed' + status: text("status").notNull().default("pending"), + brandName: text("brand_name"), + model: text("model"), + modelCode: text("model_code"), + familyCode: text("family_code"), + modelYear: text("model_year"), + engine: text("engine"), + gearbox: text("gearbox"), + energyType: text("energy_type"), + manufacturingDate: text("manufacturing_date"), + catalogVehicleId: uuid("catalog_vehicle_id").references(() => catalogVehicles.id, { + onDelete: "set null", + }), + raw: jsonb("raw"), + attempts: integer("attempts").notNull().default(0), + createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(), + updatedAt: timestamp("updated_at", { withTimezone: true }), + decodedAt: timestamp("decoded_at", { withTimezone: true }), +}); + // ─── Vehicles (shared config — one record per VIN) ── export const vehicles = pgTable( "vehicles", diff --git a/apps/api/src/integrations/rpartstore/rpartstore.auth.spec.ts b/apps/api/src/integrations/rpartstore/rpartstore.auth.spec.ts new file mode 100644 index 0000000..9e75777 --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.auth.spec.ts @@ -0,0 +1,148 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { RpartstoreAuthError, decodeJwtExpiry, loginRpartstore } from "./rpartstore.auth"; + +const b64url = (s: string): string => Buffer.from(s).toString("base64url"); +const makeJwt = (claims: Record): string => + `${b64url('{"alg":"RS256"}')}.${b64url(JSON.stringify(claims))}.sig`; + +function jsonResponse( + body: unknown, + init: { status?: number; headers?: Record } = {}, +) { + return new Response(JSON.stringify(body), { + status: init.status ?? 200, + headers: { "content-type": "application/json", ...(init.headers ?? {}) }, + }); +} + +describe("loginRpartstore (Okta IDX, browser-free)", () => { + afterEach(() => vi.unstubAllGlobals()); + + it("walks authorize → introspect → identify → answer → redirect → token and returns the access token", async () => { + const exp = Math.floor(Date.now() / 1000) + 3600; + const jwt = makeJwt({ sub: "G123326", exp }); + const calls: { url: string; body?: string; cookie?: string }[] = []; + let redirectState = ""; + + const fetchImpl = vi.fn(async (url: string | URL, init?: RequestInit) => { + const u = String(url); + const headers = (init?.headers ?? {}) as Record; + calls.push({ + url: u, + body: init?.body ? String(init.body) : undefined, + cookie: headers.Cookie, + }); + if (u.includes("/v1/authorize")) { + redirectState = new URL(u).searchParams.get("state") ?? ""; + return new Response("", { + status: 200, + headers: { "set-cookie": "JSESSIONID=js1; Path=/; HttpOnly" }, + }); + } + if (u.endsWith("/idp/idx/introspect")) return jsonResponse({ stateHandle: "sh-1" }); + if (u.endsWith("/idp/idx/identify")) return jsonResponse({ stateHandle: "sh-2" }); + if (u.endsWith("/idp/idx/challenge/answer")) { + return jsonResponse({ + success: { href: "https://sso.renault.com/login/token/redirect?stateToken=st" }, + }); + } + if (u.includes("/login/token/redirect")) { + return new Response(null, { + status: 302, + headers: { + location: `https://rpartstore.renault.com/idp-redirect?code=CODE1&state=${redirectState}`, + }, + }); + } + if (u.endsWith("/v1/token")) { + return jsonResponse({ + token_type: "Bearer", + expires_in: 3600, + access_token: jwt, + scope: "openid", + }); + } + throw new Error(`unexpected url ${u}`); + }); + + const token = await loginRpartstore({ + username: "G123326", + password: "pw", + fetchImpl: fetchImpl as unknown as typeof fetch, + }); + + expect(token.accessToken).toBe(jwt); + expect(token.subject).toBe("G123326"); + expect(token.expiresAt).toBe(exp * 1000); + + // stateToken unescaped (\x2D → "-") and carried into introspect + expect(calls[1].body).toBe(JSON.stringify({ stateToken: "02.id.abc-def" })); + // identify uses the introspect handle, answer the identify handle + expect(JSON.parse(calls[2].body!)).toEqual({ identifier: "G123326", stateHandle: "sh-1" }); + expect(JSON.parse(calls[3].body!)).toEqual({ + credentials: { passcode: "pw" }, + stateHandle: "sh-2", + }); + // cookies from the authorize page are replayed on later hops + expect(calls[3].cookie).toContain("JSESSIONID=js1"); + // PKCE: the token exchange sends the code and a verifier, never the password + const tokenBody = new URLSearchParams(calls.at(-1)!.body); + expect(tokenBody.get("grant_type")).toBe("authorization_code"); + expect(tokenBody.get("code")).toBe("CODE1"); + expect(tokenBody.get("code_verifier")).toBeTruthy(); + expect(calls.at(-1)!.body).not.toContain("pw"); + }); + + it("fails with a challenge error when Okta returns no success redirect (bad password / MFA)", async () => { + const fetchImpl = vi.fn(async (url: string | URL) => { + const u = String(url); + if (u.includes("/v1/authorize")) return new Response("stateToken = 'x'", { status: 200 }); + if (u.endsWith("/introspect")) return jsonResponse({ stateHandle: "sh" }); + if (u.endsWith("/identify")) return jsonResponse({ stateHandle: "sh" }); + if (u.endsWith("/challenge/answer")) { + return jsonResponse({ messages: { value: [{ message: "Authentication failed" }] } }); + } + throw new Error(`unexpected url ${u}`); + }); + await expect( + loginRpartstore({ + username: "u", + password: "bad", + fetchImpl: fetchImpl as unknown as typeof fetch, + }), + ).rejects.toMatchObject({ name: "RpartstoreAuthError", step: "challenge" }); + }); + + it("rejects an OAuth state mismatch on the callback", async () => { + const fetchImpl = vi.fn(async (url: string | URL) => { + const u = String(url); + if (u.includes("/v1/authorize")) return new Response("stateToken = 'x'", { status: 200 }); + if (u.endsWith("/introspect")) return jsonResponse({ stateHandle: "sh" }); + if (u.endsWith("/identify")) return jsonResponse({ stateHandle: "sh" }); + if (u.endsWith("/challenge/answer")) + return jsonResponse({ success: { href: "https://sso.renault.com/r" } }); + if (u === "https://sso.renault.com/r") { + return new Response(null, { + status: 302, + headers: { location: "https://rpartstore.renault.com/idp-redirect?code=C&state=forged" }, + }); + } + throw new Error(`unexpected url ${u}`); + }); + await expect( + loginRpartstore({ + username: "u", + password: "p", + fetchImpl: fetchImpl as unknown as typeof fetch, + }), + ).rejects.toBeInstanceOf(RpartstoreAuthError); + }); + + it("decodes exp/sub from the access token", () => { + expect(decodeJwtExpiry(makeJwt({ sub: "G1", exp: 1790334740 }))).toEqual({ + exp: 1790334740, + sub: "G1", + }); + expect(() => decodeJwtExpiry(makeJwt({ sub: "G1" }))).toThrow(RpartstoreAuthError); + }); +}); diff --git a/apps/api/src/integrations/rpartstore/rpartstore.auth.ts b/apps/api/src/integrations/rpartstore/rpartstore.auth.ts new file mode 100644 index 0000000..dc9c5a9 --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.auth.ts @@ -0,0 +1,238 @@ +/** + * Browser-free Okta (OIE / IDX) login for rpartstore.renault.com. + * + * Flow (verified live 2026-09-25, ~1.8 s, no MFA / captcha): + * 1. GET {issuer}/v1/authorize?...PKCE... → hosted page HTML containing `stateToken` + * 2. POST /idp/idx/introspect {stateToken} → stateHandle + * 3. POST /idp/idx/identify {identifier, stateHandle} + * 4. POST /idp/idx/challenge/answer {credentials:{passcode}, stateHandle} → success.href + * 5. GET success.href (follow redirects manually) → …/idp-redirect?code=…&state=… + * 6. POST {issuer}/v1/token (authorization_code + code_verifier) → access_token (1 h, no refresh_token) + */ +import { createHash, randomBytes } from "node:crypto"; + +export interface RpartstoreAuthConfig { + username: string; + password: string; + issuer?: string; + clientId?: string; + redirectUri?: string; + scope?: string; + /** Injectable for tests. */ + fetchImpl?: typeof fetch; +} + +export interface RpartstoreToken { + accessToken: string; + /** Epoch ms. */ + expiresAt: number; + subject: string; +} + +export class RpartstoreAuthError extends Error { + constructor( + message: string, + readonly step: string, + readonly detail?: unknown, + ) { + super(message); + this.name = "RpartstoreAuthError"; + } +} + +const DEFAULTS = { + issuer: "https://sso.renault.com/oauth2/aus133y6mks4ptDss417", + clientId: "irn-72795_ope_pkce_4hcafvxlbcil", + redirectUri: "https://rpartstore.renault.com/idp-redirect", + scope: "openid alliance_profile apis.default", +}; + +const USER_AGENT = + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"; +const ION = "application/ion+json; okta-version=1.0.0"; + +const b64url = (buf: Buffer): string => buf.toString("base64url"); + +/** Tiny cookie jar: Okta needs `idx`/`JSESSIONID` cookies across the IDX steps. */ +class CookieJar { + private readonly jar = new Map(); + + absorb(res: Response): void { + const setCookies: string[] = + typeof (res.headers as { getSetCookie?: () => string[] }).getSetCookie === "function" + ? (res.headers as unknown as { getSetCookie: () => string[] }).getSetCookie() + : []; + for (const sc of setCookies) { + const kv = sc.split(";")[0]; + const i = kv.indexOf("="); + if (i > 0) this.jar.set(kv.slice(0, i).trim(), kv.slice(i + 1).trim()); + } + } + + header(): string { + return Array.from(this.jar.entries()) + .map(([k, v]) => `${k}=${v}`) + .join("; "); + } +} + +export function decodeJwtExpiry(accessToken: string): { exp: number; sub: string } { + const payload = JSON.parse( + Buffer.from(accessToken.split(".")[1], "base64url").toString("utf8"), + ) as { + exp?: number; + sub?: string; + }; + if (!payload.exp) throw new RpartstoreAuthError("access token has no exp claim", "token"); + return { exp: payload.exp, sub: payload.sub ?? "" }; +} + +export async function loginRpartstore(cfg: RpartstoreAuthConfig): Promise { + const issuer = cfg.issuer ?? DEFAULTS.issuer; + const clientId = cfg.clientId ?? DEFAULTS.clientId; + const redirectUri = cfg.redirectUri ?? DEFAULTS.redirectUri; + const scope = cfg.scope ?? DEFAULTS.scope; + const doFetch = cfg.fetchImpl ?? fetch; + const jar = new CookieJar(); + + const request = async (url: string, init: RequestInit = {}): Promise => { + const res = await doFetch(url, { + redirect: "manual", + ...init, + headers: { + "User-Agent": USER_AGENT, + Cookie: jar.header(), + ...(init.headers as Record | undefined), + }, + }); + jar.absorb(res); + return res; + }; + + const verifier = b64url(randomBytes(48)); + const challenge = b64url(createHash("sha256").update(verifier).digest()); + const state = randomBytes(16).toString("hex"); + const nonce = randomBytes(16).toString("hex"); + + // 1. hosted authorize page → stateToken + const authorize = new URL(`${issuer}/v1/authorize`); + for (const [k, v] of Object.entries({ + client_id: clientId, + redirect_uri: redirectUri, + response_type: "code", + scope, + state, + nonce, + code_challenge: challenge, + code_challenge_method: "S256", + response_mode: "query", + })) { + authorize.searchParams.set(k, v); + } + const authorizeRes = await request(authorize.toString()); + const html = await authorizeRes.text(); + const stateTokenMatch = + html.match(/stateToken\s*=\s*'([^']+)'/) ?? html.match(/"stateToken":"([^"]+)"/); + if (authorizeRes.status !== 200 || !stateTokenMatch) { + throw new RpartstoreAuthError( + `authorize page did not expose a stateToken (HTTP ${authorizeRes.status})`, + "authorize", + ); + } + const stateToken = stateTokenMatch[1].replace(/\\x([0-9A-Fa-f]{2})/g, (_, hex: string) => + String.fromCharCode(Number.parseInt(hex, 16)), + ); + + const idx = async ( + path: string, + body: Record, + step: string, + ): Promise> => { + const res = await request(`https://sso.renault.com/idp/idx/${path}`, { + method: "POST", + headers: { "Content-Type": ION, Accept: ION, Origin: "https://sso.renault.com" }, + body: JSON.stringify(body), + }); + const json = (await res.json().catch(() => ({}))) as Record; + if (!res.ok) { + throw new RpartstoreAuthError( + `Okta ${step} failed (HTTP ${res.status})`, + step, + json.messages ?? json, + ); + } + return json; + }; + + // 2-4. IDX remediation + const introspected = await idx("introspect", { stateToken }, "introspect"); + const identified = await idx( + "identify", + { identifier: cfg.username, stateHandle: introspected.stateHandle }, + "identify", + ); + const answered = await idx( + "challenge/answer", + { + credentials: { passcode: cfg.password }, + stateHandle: identified.stateHandle ?? introspected.stateHandle, + }, + "challenge", + ); + const successHref: string | undefined = answered.success?.href; + if (!successHref) { + throw new RpartstoreAuthError( + "Okta did not return a success redirect (wrong password, locked account or MFA now required)", + "challenge", + answered.messages, + ); + } + + // 5. follow the redirect chain until the app callback carries ?code= + let next = successHref; + let code: string | undefined; + for (let hop = 0; hop < 6 && !code; hop += 1) { + const res = await request(next); + const location = res.headers.get("location"); + if (!location) break; + const target = new URL(location, next); + const gotCode = target.searchParams.get("code"); + if (gotCode) { + if (target.searchParams.get("state") !== state) + throw new RpartstoreAuthError("OAuth state mismatch", "redirect"); + code = gotCode; + } + next = target.toString(); + } + if (!code) + throw new RpartstoreAuthError("redirect chain ended without an authorization code", "redirect"); + + // 6. PKCE token exchange + const tokenRes = await request(`${issuer}/v1/token`, { + method: "POST", + headers: { + "Content-Type": "application/x-www-form-urlencoded", + Origin: "https://rpartstore.renault.com", + }, + body: new URLSearchParams({ + grant_type: "authorization_code", + redirect_uri: redirectUri, + code, + code_verifier: verifier, + client_id: clientId, + }).toString(), + }); + const token = (await tokenRes.json().catch(() => ({}))) as { + access_token?: string; + error?: string; + error_description?: string; + }; + if (!tokenRes.ok || !token.access_token) { + throw new RpartstoreAuthError( + `token exchange failed (HTTP ${tokenRes.status}): ${token.error_description ?? token.error ?? ""}`, + "token", + ); + } + const { exp, sub } = decodeJwtExpiry(token.access_token); + return { accessToken: token.access_token, expiresAt: exp * 1000, subject: sub }; +} diff --git a/apps/api/src/integrations/rpartstore/rpartstore.client.ts b/apps/api/src/integrations/rpartstore/rpartstore.client.ts new file mode 100644 index 0000000..20a6269 --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.client.ts @@ -0,0 +1,157 @@ +import { type RpartstoreToken, loginRpartstore } from "./rpartstore.auth"; +import { canonicalRpartstoreBrand } from "./rpartstore.routing"; +import { + RpartstoreBffError, + RpartstoreRateLimitError, + RpartstoreSession, +} from "./rpartstore.session"; +import type { + RpartstoreDecoded, + RpartstoreVehicle, + SearchVehicleResponsePayload, +} from "./rpartstore.types"; + +export const RPARTSTORE_DEFAULTS = { + brokerUrl: "wss://1po-bff.renault-edh.com/ws", + appVersion: "1.34.0.6", + webLanguage: "tr", + country: "TR", +} as const; + +/** Where the (1 h, non-refreshable) Okta access token is cached between decodes. */ +export interface TokenStore { + get(): Promise; + set(token: RpartstoreToken, ttlSeconds: number): Promise; + clear(): Promise; +} + +export interface RpartstoreClientOptions { + username: string; + password: string; + tokenStore: TokenStore; + brokerUrl?: string; + appVersion?: string; + webLanguage?: string; + country?: string; + /** Injectable for tests. */ + login?: typeof loginRpartstore; + sessionFactory?: (opts: ConstructorParameters[0]) => RpartstoreSession; + logger?: { log: (msg: string) => void; warn: (msg: string) => void }; +} + +/** Refresh the token this many ms before its `exp`. */ +const TOKEN_SKEW_MS = 5 * 60 * 1000; + +export class RpartstoreClient { + private readonly login: typeof loginRpartstore; + private readonly sessionFactory: NonNullable; + + constructor(private readonly opts: RpartstoreClientOptions) { + this.login = opts.login ?? loginRpartstore; + this.sessionFactory = opts.sessionFactory ?? ((o) => new RpartstoreSession(o)); + } + + /** Cached token when still valid (with skew), otherwise a fresh Okta login. */ + async getToken(force = false): Promise { + if (!force) { + const cached = await this.opts.tokenStore.get(); + if (cached && cached.expiresAt - Date.now() > TOKEN_SKEW_MS) return cached; + } + const token = await this.login({ username: this.opts.username, password: this.opts.password }); + const ttl = Math.max(60, Math.floor((token.expiresAt - Date.now() - TOKEN_SKEW_MS) / 1000)); + await this.opts.tokenStore.set(token, ttl); + this.opts.logger?.log(`[rpartstore] logged in as ${token.subject}, token ttl ${ttl}s`); + return token; + } + + /** + * One VIN search on a fresh STOMP session (volume is capped at a handful per + * day, so a persistent socket buys nothing). Resolves to the decoded vehicle, + * `null` when RPartStore answers NOT_FOUND, and throws `RpartstoreRateLimitError` + * / other errors for the caller to handle. A CONNECT failure is retried once + * with a forced re-login (expired or revoked token). + */ + async searchVin(vin: string): Promise { + let token = await this.getToken(); + const session = await this.openSession(token.accessToken).catch(async (err: Error) => { + this.opts.logger?.warn(`[rpartstore] CONNECT failed (${err.message}); re-login and retry`); + await this.opts.tokenStore.clear(); + token = await this.getToken(true); + return this.openSession(token.accessToken); + }); + try { + const country = this.opts.country ?? RPARTSTORE_DEFAULTS.country; + const lang = this.opts.webLanguage ?? RPARTSTORE_DEFAULTS.webLanguage; + const msg = await session.request( + "/vehicles/search/vin-or-vrn", + { + requestId: crypto.randomUUID(), + value: vin, + queryType: "VIN", + userContext: { + webLanguage: lang, + documentLanguage: lang, + documentCountryLanguage: country, + documentFallbackLanguage: lang, + documentFallbackCountryLanguage: country, + userCountry: country, + r1Country: country, + }, + countryCode: country, + includeEstimate: false, + }, + { expect: ["1PO/CATALOG/SEARCH_VEHICLE_RESPONSE"], timeoutMs: 15_000 }, + ); + const vehicle = msg.payload?.vehicles?.[0]; + if (!vehicle) return null; + return normalizeVehicle(vehicle); + } catch (err) { + if (err instanceof RpartstoreBffError) { + const errorType = (err.payload as { errorType?: string } | undefined)?.errorType; + if (err.type === "1PO/CATALOG/SEARCH_VEHICLE_ERROR" && errorType === "NOT_FOUND") + return null; + if (err.type === "1PO/CATALOG/SEARCH_VEHICLE_NOT_COVERED_IN_COUNTRY") return null; + } + if (err instanceof RpartstoreRateLimitError) throw err; + throw err; + } finally { + session.close(); + } + } + + private async openSession(accessToken: string): Promise { + const session = this.sessionFactory({ + brokerUrl: this.opts.brokerUrl ?? RPARTSTORE_DEFAULTS.brokerUrl, + accessToken, + profile: this.opts.username, + appVersion: this.opts.appVersion ?? RPARTSTORE_DEFAULTS.appVersion, + webLanguage: this.opts.webLanguage ?? RPARTSTORE_DEFAULTS.webLanguage, + logger: this.opts.logger + ? { debug: () => undefined, warn: (m) => this.opts.logger?.warn(`[rpartstore] ${m}`) } + : undefined, + }); + await session.connect(); + return session; + } +} + +export function normalizeVehicle(v: RpartstoreVehicle): RpartstoreDecoded { + const dh = v.dataHubVehicle ?? {}; + const brandName = canonicalRpartstoreBrand(v.vehicleBrand) ?? v.vehicleBrand; + const clean = (s: string | undefined): string | null => { + const t = (s ?? "").trim(); + return t.length > 0 ? t : null; + }; + return { + brandName, + model: clean(v.model), + modelCode: clean(dh.modelTypeCode), + familyCode: clean(dh.familyCode), + modelYear: clean(dh.modelYear), + engine: clean(dh.engine), + gearbox: clean(dh.gearbox), + energyType: clean(dh.energyType), + manufacturingDate: clean(v.manufacturingDate), + raw: v, + }; +} diff --git a/apps/api/src/integrations/rpartstore/rpartstore.matcher.spec.ts b/apps/api/src/integrations/rpartstore/rpartstore.matcher.spec.ts new file mode 100644 index 0000000..e07731d --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.matcher.spec.ts @@ -0,0 +1,140 @@ +import { describe, expect, it } from "vitest"; +import { + type RpartstoreCatalogCandidate, + pickRpartstoreCatalogMatch, + rpartstoreModelTokens, +} from "./rpartstore.matcher"; + +// Real PL24 `catalog_vehicles.model` values for Renault (prod, 2026-09-25). +const RENAULT = [ + "ALASKAN", + "ARKANA EUROPE / XM3", + "ARKANA RUSYA", + "AUSTRAL/ESPACE VI/RAFALE", + "CAPTUR / QM3", + "CAPTUR II EUROPE/SYMBIOZ", + "CAPTUR II ÇİN", + "CAPTUR/KAPTUR", + "CLIO 4 / LUTECIA 4", + "CLIO 5/LUTECIA 5", + "DUSTER 1", + "DUSTER 2", + "DUSTER III / BIGSTER", + "EXPRESS", + "FLUENCE / FLUENCE Z.E.", + "KADJAR", + "KADJAR ÇİN", + "KANGOO 1", + "KANGOO 2 / KANGOO Z.E.", + "KANGOO 3", + "LATITUDE / SAFRANE 2", + "LOGAN\\-SANDERO 1/TONDAR 1", + "LOGAN\\-SANDERO 3 / TALIANT", + "MASTER 3", + "MASTER 4 VAN", + "MEGANE 1", + "MEGANE 2 / SCENIC 2", + "MEGANE 3 / SCENIC 3", + "MEGANE 4", + "MEGANE 4 SEDAN", + "RENAULT 5 EXPRESS / RAPID", + "RENAULT 9 / 11", + "TRAFIC 2", + "TRAFIC 3", + "X62 CHINE", +]; + +const DACIA = [ + "DOKKER", + "DUSTER 1", + "DUSTER 2", + "DUSTER 3/BIGSTER", + "LOGAN\\-SANDERO 1/TONDAR 1", + "LOGAN\\-SANDERO 2/SYMBOL 2", + "LOGAN\\-SANDERO 3 / TALIANT", +]; + +const cands = (models: string[]): RpartstoreCatalogCandidate[] => + models.map((model) => ({ id: model, model })); + +describe("rpartstoreModelTokens", () => { + it("drops the model code in parentheses and converts roman generations", () => { + expect(rpartstoreModelTokens("Clio IV / Lutecia IV (B98)")).toEqual([ + "CLIO", + "4", + "LUTECIA", + "4", + ]); + expect(rpartstoreModelTokens("Duster III (SUV)")).toEqual(["DUSTER", "3"]); + expect(rpartstoreModelTokens("Megane I Classic (L64)")).toEqual(["MEGANE", "1", "CLASSIC"]); + }); + + it("keeps single-digit generation tokens and folds diacritics", () => { + expect(rpartstoreModelTokens("MEGANE 4 SEDAN")).toEqual(["MEGANE", "4", "SEDAN"]); + expect(rpartstoreModelTokens("KADJAR ÇİN")).toEqual(["KADJAR", "CIN"]); + expect(rpartstoreModelTokens("LOGAN\\-SANDERO 3 / TALIANT")).toEqual([ + "LOGAN", + "SANDERO", + "3", + "TALIANT", + ]); + }); +}); + +describe("pickRpartstoreCatalogMatch (prod benchmark models → PL24 Renault catalog)", () => { + const expectPick = (model: string, expected: string | null, pool = RENAULT) => + expect(pickRpartstoreCatalogMatch(model, cands(pool))).toBe(expected); + + it("matches roman-numeral generations to digit generations", () => { + expectPick("Clio IV / Lutecia IV (B98)", "CLIO 4 / LUTECIA 4"); + expectPick("Trafic III (J82)", "TRAFIC 3"); + expectPick("Master III (F62)", "MASTER 3"); + expectPick("Kangoo II (K61)", "KANGOO 2 / KANGOO Z.E."); + expectPick("Logan III (LJF)", "LOGAN\\-SANDERO 3 / TALIANT"); + expectPick("Duster III (SUV)", "DUSTER III / BIGSTER"); + }); + + it("prefers the body-qualified catalog when the decode carries the qualifier, and the plain one otherwise", () => { + expectPick("Megane IV Sedan (LFF)", "MEGANE 4 SEDAN"); + expectPick("Megane IV (BFB)", "MEGANE 4"); + expectPick("Megane I Classic (L64)", "MEGANE 1"); + }); + + it("never picks a wrong-market catalog when a mainstream one exists", () => { + expectPick("Kadjar (HFE)", "KADJAR"); + expectPick("Captur II (HJB)", "CAPTUR II EUROPE/SYMBIOZ"); + }); + + it("does not let a newer generation steal a decode without a generation", () => { + expectPick("Captur (J87)", "CAPTUR / QM3"); + expectPick("Express (KJK)", "EXPRESS"); + }); + + it("matches multi-name catalogs and pre-2000 models", () => { + expectPick("Latitude / Safrane II (L43)", "LATITUDE / SAFRANE 2"); + expectPick("Fluence (L38)", "FLUENCE / FLUENCE Z.E."); + expectPick("Renault 9 / 11 (L42)", "RENAULT 9 / 11"); + }); + + it("matches Dacia models within the Dacia catalog", () => { + expectPick("Duster I (H79)", "DUSTER 1", DACIA); + expectPick("Duster II (HJD)", "DUSTER 2", DACIA); + expectPick("Dokker (K67)", "DOKKER", DACIA); + expectPick("Logan I (L90)", "LOGAN\\-SANDERO 1/TONDAR 1", DACIA); + }); + + it("returns null when nothing qualifies or the model is missing", () => { + expectPick("Arkana (LJL)", "ARKANA EUROPE / XM3"); + expectPick("Twingo Z.E.", null); + expect(pickRpartstoreCatalogMatch(undefined, cands(RENAULT))).toBeNull(); + expect(pickRpartstoreCatalogMatch("", cands(RENAULT))).toBeNull(); + }); + + it("breaks exact ties by the fuller catalog", () => { + const pool: RpartstoreCatalogCandidate[] = [ + { id: "a", model: "KADJAR", categoryCount: 12 }, + { id: "b", model: "KADJAR", categoryCount: 44 }, + ]; + expect(pickRpartstoreCatalogMatch("Kadjar (HFE)", pool)).toBe("b"); + }); +}); diff --git a/apps/api/src/integrations/rpartstore/rpartstore.matcher.ts b/apps/api/src/integrations/rpartstore/rpartstore.matcher.ts new file mode 100644 index 0000000..c73fafe --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.matcher.ts @@ -0,0 +1,133 @@ +import { MARKET_QUALIFIERS } from "../vinpin/vinpin.matcher"; + +/** + * Map an RPartStore model label ("Megane IV Sedan (LFF)", "Clio IV / Lutecia IV + * (B98)", "Duster III (SUV)") onto an EXISTING PL24 `catalog_vehicles` row of the + * same brand ("MEGANE 4 SEDAN", "CLIO 4 / LUTECIA 4", "DUSTER III / BIGSTER"). + * + * Differences from the Vinpin matcher that made a dedicated picker necessary: + * - RPartStore writes generations as roman numerals, PL24 mostly as digits + * ("IV" vs "4") — both sides are normalised to digits. + * - Generation digits are single characters; the Vinpin tokenizer drops tokens + * shorter than 2 chars, which would make "MEGANE 4" ≡ "MEGANE". + * - Body qualifiers (SEDAN, CLASSIC, …) are optional on the catalog side: + * "Megane I Classic" must still match "MEGANE 1". + * - PL24 Renault/Dacia rows carry no year, so there is no year scoring. + */ + +export interface RpartstoreCatalogCandidate { + id: string; + model: string | null; + /** Tiebreak only: fuller catalog wins. */ + categoryCount?: number | null; +} + +const ROMAN: Record = { + I: "1", + II: "2", + III: "3", + IV: "4", + V: "5", + VI: "6", + VII: "7", + VIII: "8", +}; + +/** Body / trim words that PL24 may omit; they only add a small bonus when both sides have them. */ +const BODY_QUALIFIERS = new Set([ + "SEDAN", + "CLASSIC", + "HATCHBACK", + "HB", + "ESTATE", + "GRANDTOUR", + "SPORTTOURER", + "SW", + "BREAK", + "VAN", + "COMBI", + "KOMBI", + "CABRIO", + "COUPE", + "SASI", + "CHASSIS", + "PICKUP", + "PHASE", + "PH", +]); + +/** Tokenize a model label: fold diacritics, drop parenthetical codes/years, split on + * non-alphanumerics, convert roman generation numerals to digits. Keeps 1-char + * numeric tokens (generations) and drops 1-char alpha noise ("Z.E." → "ZE" kept + * as-is is not needed for matching). */ +export function rpartstoreModelTokens(s: string | null | undefined): string[] { + if (!s) return []; + return s + .toUpperCase() + .normalize("NFD") + .replace(/\p{M}/gu, "") + .replace(/\([^)]*\)/g, " ") + .replace(/[^A-Z0-9]+/g, " ") + .split(" ") + .map((t) => t.trim()) + .filter((t) => t.length > 0) + .map((t) => ROMAN[t] ?? t) + .filter((t) => t.length >= 2 || /^\d$/.test(t)); +} + +const isNumeric = (t: string): boolean => /^\d+$/.test(t); + +/** + * Returns the id of the best candidate or null. A candidate qualifies when every + * "core" decoded token (everything except body qualifiers) appears among its + * tokens. Score, strongest first: market-qualifier penalty (wrong-market catalogs + * only win when alone), exact-match bonus, body-qualifier overlap bonus, a hard + * penalty for candidates that add a generation number the decode lacks + * ("Captur" must not pick "CAPTUR II"), a mild penalty per extra token, then the + * fuller catalog as tiebreak. + */ +export function pickRpartstoreCatalogMatch( + model: string | null | undefined, + candidates: RpartstoreCatalogCandidate[], +): string | null { + const decoded = rpartstoreModelTokens(model); + if (decoded.length === 0) return null; + const core = decoded.filter((t) => !BODY_QUALIFIERS.has(t)); + const required = core.length > 0 ? core : decoded; + const decodedSet = new Set(decoded); + const decodedHasGeneration = decoded.some(isNumeric); + + let best: { id: string; score: number; categoryCount: number } | null = null; + for (const c of candidates) { + const tokens = rpartstoreModelTokens(c.model); + if (tokens.length === 0) continue; + const tokenSet = new Set(tokens); + if (!required.every((t) => tokenSet.has(t))) continue; + + const extras = tokens.filter((t) => !decodedSet.has(t)); + const marketExtras = extras.filter((t) => MARKET_QUALIFIERS.has(t)).length; + const generationExtras = decodedHasGeneration ? 0 : extras.filter(isNumeric).length; + const qualifierOverlap = decoded.filter( + (t) => BODY_QUALIFIERS.has(t) && tokenSet.has(t), + ).length; + const exact = extras.length === 0 && tokens.length === decoded.length; + + const score = + 1000 - + marketExtras * 5000 - + generationExtras * 150 - + extras.length * 20 + + qualifierOverlap * 100 + + (exact ? 300 : 0); + const categoryCount = c.categoryCount ?? 0; + + if ( + !best || + score > best.score || + (score === best.score && categoryCount > best.categoryCount) + ) { + best = { id: c.id, score, categoryCount }; + } + } + return best?.id ?? null; +} diff --git a/apps/api/src/integrations/rpartstore/rpartstore.routing.ts b/apps/api/src/integrations/rpartstore/rpartstore.routing.ts new file mode 100644 index 0000000..e863cce --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.routing.ts @@ -0,0 +1,32 @@ +import { getBrandFromWmi } from "@sase/shared"; + +/** WMIs routed to RPartStore even when the shared WMI table is silent. VF1/VF2 = + * Renault (France), VF6 = Renault (Trucks/Sofasa), UU1 = Dacia (Romania). VF7 is + * Citroën and is deliberately NOT here. */ +const RPARTSTORE_WMIS = new Set(["VF1", "VF2", "VF6", "UU1"]); + +const RPARTSTORE_BRANDS = new Set(["renault", "dacia"]); + +/** Canonical brand names as they appear in `catalog_vehicles.brand_name`. */ +export function canonicalRpartstoreBrand( + brand: string | null | undefined, +): "Renault" | "Dacia" | null { + const b = (brand ?? "").trim().toLowerCase(); + if (b === "renault") return "Renault"; + if (b === "dacia") return "Dacia"; + return null; +} + +/** + * Should this VIN be offered to the RPartStore decode fallback? Renault/Dacia + * only: decided by the WMI first (shared table, then the explicit allowlist), + * with the identified browse brand as a last resort (Dacia-badged cars built + * under a Renault WMI still say "Dacia" in the identification). + */ +export function isRpartstoreVin(vin: string, browseBrand?: string | null): boolean { + const wmi = (vin || "").toUpperCase().slice(0, 3); + const wmiBrand = getBrandFromWmi(wmi); + if (wmiBrand && RPARTSTORE_BRANDS.has(wmiBrand.toLowerCase())) return true; + if (RPARTSTORE_WMIS.has(wmi)) return true; + return !!browseBrand && RPARTSTORE_BRANDS.has(browseBrand.trim().toLowerCase()); +} diff --git a/apps/api/src/integrations/rpartstore/rpartstore.session.ts b/apps/api/src/integrations/rpartstore/rpartstore.session.ts new file mode 100644 index 0000000..fc2c183 --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.session.ts @@ -0,0 +1,256 @@ +/** + * One STOMP-over-WebSocket session against the RPartStore BFF. + * + * Request/response is correlated by the `trace-id` header we send and the `traceId` + * header the server echoes. One request can yield several MESSAGE frames + * (e.g. a VIN search → SEARCH_VEHICLE_RESPONSE, COMMAND_PROCESSING_RESPONSE, + * EXPLODED_TREE_RESPONSE), so a request resolves on the first frame whose `type` + * matches one of the expected terminal types, and errors on a frame from + * `/user/queue/error` or a `*_RATE_LIMIT_EXCEEDED` type. + * + * Uses Node 22's built-in WebSocket (no Origin / subprotocol required by the server). + */ +import { randomUUID } from "node:crypto"; +import { decodeFrame, encodeFrame } from "./rpartstore.stomp"; + +export interface BffMessage { + type: string; + payload: T; +} + +export interface SessionOptions { + brokerUrl: string; + accessToken: string; + profile: string; + appVersion: string; + webLanguage: string; + connectTimeoutMs?: number; + /** STOMP heart-beat interval (ms) negotiated with the server. */ + heartbeatMs?: number; + onClose?: (reason: string) => void; + logger?: { debug: (msg: string) => void; warn: (msg: string) => void }; +} + +export interface RequestOptions { + /** Message `type`s that complete the request. */ + expect: string[]; + timeoutMs?: number; +} + +export class RpartstoreRateLimitError extends Error { + constructor( + readonly retryAfterSeconds: number, + readonly limitType: string, + ) { + super(`RPartStore rate limit (${limitType}), retry after ${retryAfterSeconds}s`); + this.name = "RpartstoreRateLimitError"; + } +} + +export class RpartstoreBffError extends Error { + constructor( + readonly type: string, + readonly payload: unknown, + ) { + super(`RPartStore BFF error ${type}`); + this.name = "RpartstoreBffError"; + } +} + +interface Pending { + expect: Set; + resolve: (msg: BffMessage) => void; + reject: (err: Error) => void; + timer: NodeJS.Timeout; +} + +export class RpartstoreSession { + private ws: WebSocket | null = null; + private readonly pending = new Map(); + private heartbeatTimer: NodeJS.Timeout | null = null; + private closed = false; + + constructor(private readonly opts: SessionOptions) {} + + get isOpen(): boolean { + return !this.closed && this.ws !== null && this.ws.readyState === WebSocket.OPEN; + } + + /** Opens the socket, sends CONNECT and subscribes to the user queues. Resolves on CONNECTED. */ + connect(): Promise { + const { + brokerUrl, + accessToken, + profile, + appVersion, + webLanguage, + connectTimeoutMs = 10_000, + heartbeatMs = 60_000, + } = this.opts; + return new Promise((resolve, reject) => { + let settled = false; + const fail = (err: Error): void => { + if (!settled) { + settled = true; + reject(err); + } + this.teardown(err.message); + }; + const timer = setTimeout(() => fail(new Error("STOMP CONNECT timeout")), connectTimeoutMs); + let ws: WebSocket; + try { + ws = new WebSocket(brokerUrl, ["v12.stomp"]); + } catch (err) { + clearTimeout(timer); + reject(err instanceof Error ? err : new Error(String(err))); + return; + } + this.ws = ws; + ws.onopen = () => { + ws.send( + encodeFrame("CONNECT", { + "trace-id": randomUUID(), + "x-auth-token": accessToken, + "selected-profile": profile, + "app-version": appVersion, + "web-language": webLanguage, + "accept-version": "1.2,1.1,1.0", + "heart-beat": `${heartbeatMs},${heartbeatMs}`, + }), + ); + }; + ws.onmessage = (event: MessageEvent) => { + const frame = decodeFrame(typeof event.data === "string" ? event.data : String(event.data)); + if (!frame) return; // heartbeat + if (frame.command === "CONNECTED") { + clearTimeout(timer); + ws.send(encodeFrame("SUBSCRIBE", { id: "sub-0", destination: "/user/queue/main" })); + ws.send(encodeFrame("SUBSCRIBE", { id: "sub-1", destination: "/user/queue/error" })); + this.startHeartbeat(heartbeatMs); + settled = true; + resolve(); + return; + } + if (frame.command === "ERROR") { + fail( + new Error(`STOMP ERROR: ${frame.headers.message ?? ""} ${frame.body.slice(0, 200)}`), + ); + return; + } + if (frame.command === "MESSAGE") this.onMessage(frame.headers, frame.body); + }; + ws.onerror = () => fail(new Error("WebSocket error")); + ws.onclose = (event: { code: number; reason: string }) => { + clearTimeout(timer); + const reason = `WebSocket closed (${event.code}${event.reason ? ` ${event.reason}` : ""})`; + if (!settled) fail(new Error(reason)); + else this.teardown(reason); + }; + }); + } + + /** Publishes `{payload}` to `/app/` and resolves on the first expected message type. */ + request( + path: string, + payload: unknown, + options: RequestOptions, + ): Promise> { + const ws = this.ws; + if (!this.isOpen || !ws) return Promise.reject(new Error("STOMP session is not connected")); + const traceId = randomUUID(); + const body = JSON.stringify({ payload }); + const { expect, timeoutMs = 15_000 } = options; + return new Promise>((resolve, reject) => { + const timer = setTimeout(() => { + this.pending.delete(traceId); + reject(new Error(`RPartStore request ${path} timed out after ${timeoutMs}ms`)); + }, timeoutMs); + this.pending.set(traceId, { + expect: new Set(expect), + resolve: (msg) => resolve(msg as BffMessage), + reject, + timer, + }); + ws.send(encodeFrame("SEND", { destination: `/app${path}`, "trace-id": traceId }, body)); + this.opts.logger?.debug(`→ ${path} trace=${traceId}`); + }); + } + + close(): void { + this.teardown("closed by client"); + } + + private onMessage(headers: Record, body: string): void { + const traceId = headers.traceId; + const pending = traceId ? this.pending.get(traceId) : undefined; + let msg: BffMessage; + try { + msg = JSON.parse(body) as BffMessage; + } catch { + this.opts.logger?.warn( + `unparseable BFF message on ${headers.destination ?? "?"} trace=${traceId ?? "?"}`, + ); + return; + } + if (!pending) return; // unsolicited push (search-history refresh etc.) + const isError = headers.destination === "/user/queue/error"; + if (/RATE_LIMIT_EXCEEDED$/.test(msg.type)) { + const p = msg.payload as { retryAfterSeconds?: number; limitType?: string }; + this.settle(traceId, pending, (pend) => + pend.reject( + new RpartstoreRateLimitError(p.retryAfterSeconds ?? 10, p.limitType ?? "SHORT_TERM"), + ), + ); + return; + } + if (isError) { + this.settle(traceId, pending, (pend) => + pend.reject(new RpartstoreBffError(msg.type, msg.payload)), + ); + return; + } + if (pending.expect.has(msg.type)) { + this.settle(traceId, pending, (pend) => pend.resolve(msg)); + } + // other frames on the same trace (COMMAND_PROCESSING_RESPONSE, EXPLODED_TREE_RESPONSE…) are ignored here + } + + private settle(traceId: string, pending: Pending, fn: (p: Pending) => void): void { + clearTimeout(pending.timer); + this.pending.delete(traceId); + fn(pending); + } + + private startHeartbeat(intervalMs: number): void { + this.stopHeartbeat(); + this.heartbeatTimer = setInterval(() => { + if (this.isOpen) this.ws?.send("\n"); + }, intervalMs); + } + + private stopHeartbeat(): void { + if (this.heartbeatTimer) clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + + private teardown(reason: string): void { + if (this.closed) return; + this.closed = true; + this.stopHeartbeat(); + for (const [traceId, p] of this.pending) { + clearTimeout(p.timer); + p.reject(new Error(`STOMP session ended: ${reason}`)); + this.pending.delete(traceId); + } + const ws = this.ws; + this.ws = null; + if (ws && ws.readyState !== WebSocket.CLOSED) { + try { + ws.close(); + } catch { + /* ignore */ + } + } + this.opts.onClose?.(reason); + } +} diff --git a/apps/api/src/integrations/rpartstore/rpartstore.stomp.spec.ts b/apps/api/src/integrations/rpartstore/rpartstore.stomp.spec.ts new file mode 100644 index 0000000..7e63192 --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.stomp.spec.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "vitest"; +import { decodeFrame, encodeFrame } from "./rpartstore.stomp"; + +describe("rpartstore STOMP codec", () => { + it("encodes a SEND frame exactly like the RPartStore web app", () => { + const body = '{"payload":{"requestId":"r","value":"VF1RFE00653633190","queryType":"VIN"}}'; + const frame = encodeFrame( + "SEND", + { destination: "/app/vehicles/search/vin-or-vrn", "trace-id": "t-1" }, + body, + ); + expect(frame).toBe( + `SEND\ndestination:/app/vehicles/search/vin-or-vrn\ntrace-id:t-1\ncontent-length:${Buffer.byteLength(body)}\n\n${body}\u0000`, + ); + }); + + it("does not escape CONNECT header values (STOMP 1.2 rule) but escapes others", () => { + expect(encodeFrame("CONNECT", { "x-auth-token": "a:b" })).toBe( + "CONNECT\nx-auth-token:a:b\n\n\u0000", + ); + expect(encodeFrame("SEND", { destination: "/app/x", note: "a:b\nc" })).toContain( + "note:a\\cb\\nc", + ); + }); + + it("decodes a live MESSAGE frame with headers and JSON body", () => { + const raw = + "MESSAGE\ntraceId:f68cc229\ncontent-type:text/plain;charset=UTF-8\ndestination:/user/queue/main\nsubscription:sub-0\nmessage-id:975afec7-1\ncontent-length:64\n\n" + + '{"type":"1PO/COMMON/COMMAND_PROCESSING_RESPONSE","payload":true}\u0000'; + const frame = decodeFrame(raw); + expect(frame?.command).toBe("MESSAGE"); + expect(frame?.headers.traceId).toBe("f68cc229"); + expect(frame?.headers["content-type"]).toBe("text/plain;charset=UTF-8"); + expect(JSON.parse(frame!.body)).toEqual({ + type: "1PO/COMMON/COMMAND_PROCESSING_RESPONSE", + payload: true, + }); + }); + + it("treats heartbeat newlines as no frame", () => { + expect(decodeFrame("\n")).toBeNull(); + expect(decodeFrame("")).toBeNull(); + }); + + it("keeps the first value of a repeated header", () => { + const frame = decodeFrame("MESSAGE\nk:first\nk:second\n\nbody\u0000"); + expect(frame?.headers.k).toBe("first"); + expect(frame?.body).toBe("body"); + }); +}); diff --git a/apps/api/src/integrations/rpartstore/rpartstore.stomp.ts b/apps/api/src/integrations/rpartstore/rpartstore.stomp.ts new file mode 100644 index 0000000..74c340c --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.stomp.ts @@ -0,0 +1,69 @@ +/** + * Minimal STOMP 1.2 codec used by the RPartStore BFF client. + * The BFF speaks STOMP over a plain WebSocket (`wss://1po-bff.renault-edh.com/ws`): + * CONNECT → CONNECTED, SUBSCRIBE /user/queue/main + /user/queue/error, + * SEND /app/ with a `trace-id` header, and 1..N MESSAGE frames back + * carrying the same `traceId` header and a `{type, payload}` JSON body. + */ + +export interface StompFrame { + command: string; + headers: Record; + body: string; +} + +const NULL = "\u0000"; + +/** STOMP 1.2 header value escaping (RFC: `\\`, `\n` → `\\n`, `:` → `\\c`, `\r` → `\\r`). */ +function escapeHeader(value: string): string { + return value + .replace(/\\/g, "\\\\") + .replace(/\r/g, "\\r") + .replace(/\n/g, "\\n") + .replace(/:/g, "\\c"); +} + +function unescapeHeader(value: string): string { + return value + .replace(/\\n/g, "\n") + .replace(/\\r/g, "\r") + .replace(/\\c/g, ":") + .replace(/\\\\/g, "\\"); +} + +export function encodeFrame( + command: string, + headers: Record, + body = "", +): string { + const lines = [command]; + for (const [k, v] of Object.entries(headers)) { + lines.push(`${k}:${command === "CONNECT" ? String(v) : escapeHeader(String(v))}`); + } + if (body && headers["content-length"] === undefined) { + lines.push(`content-length:${Buffer.byteLength(body, "utf8")}`); + } + return `${lines.join("\n")}\n\n${body}${NULL}`; +} + +/** Heartbeat frames are a bare newline; they decode to `null`. */ +export function decodeFrame(raw: string): StompFrame | null { + if (raw === "\n" || raw === "\r\n" || raw === "") return null; + const headerEnd = raw.indexOf("\n\n"); + if (headerEnd < 0) return null; + const headerBlock = raw.slice(0, headerEnd).split("\n"); + const command = headerBlock[0].trim(); + const headers: Record = {}; + for (const line of headerBlock.slice(1)) { + const i = line.indexOf(":"); + if (i < 0) continue; + const key = line.slice(0, i); + // STOMP: the first occurrence of a repeated header wins. + if (headers[key] === undefined) + headers[key] = + command === "CONNECTED" ? line.slice(i + 1) : unescapeHeader(line.slice(i + 1)); + } + let body = raw.slice(headerEnd + 2); + if (body.endsWith(NULL)) body = body.slice(0, -1); + return { command, headers, body }; +} diff --git a/apps/api/src/integrations/rpartstore/rpartstore.types.ts b/apps/api/src/integrations/rpartstore/rpartstore.types.ts new file mode 100644 index 0000000..f2fabb7 --- /dev/null +++ b/apps/api/src/integrations/rpartstore/rpartstore.types.ts @@ -0,0 +1,54 @@ +/** + * Shapes of the RPartStore BFF `SEARCH_VEHICLE_RESPONSE` payload (DATAHUB + * catalog source, TR market). Captured live 2026-09-25; see + * /home/s/ss/rpartstore-dogrudan-vin-decode-2026-09-25.md for the protocol notes. + */ + +export interface RpartstoreDataHubVehicle { + name?: string; // "RENAULT Kadjar (HFE)" + modelType?: string; // "SUV" + modelTypeCode?: string; // "HFE" + familyCode?: string; // "XFE" + bodyType?: string; // "HFE" + engine?: string; // "1.5 DCI DİZEL MOTOR [K9K]" + gearbox?: string; // "6 VİTESLİ KAVRAMA VİTES KUTUSU:DC4 [DC4]" + energyType?: string; // "MOTORIN" + powerKw?: string; // "066 KW POWER" | "" + modelYear?: string; // "2015" + vehicleAge?: number; + wheelbaseLength?: string; + roofHeight?: string; +} + +export interface RpartstoreVehicle { + catalogSource: string; // "DATAHUB" + vin: string; + vehicleKey: string; + model: string | undefined; // "Kadjar (HFE)" — undefined on some pre-2000 cars + vehicleBrand: string; // "RENAULT" | "DACIA" + country: string; // "TR" + manufacturingDate?: string; // "2015-07-28" + imageUrl?: string; + dataHubVehicle?: RpartstoreDataHubVehicle; + vehicleIdentifiedBy?: string; +} + +export interface SearchVehicleResponsePayload { + vehicles: RpartstoreVehicle[]; + requestId: string; + searchedCountry: string; +} + +/** Normalised decode result stored in `rpartstore_decodes`. */ +export interface RpartstoreDecoded { + brandName: string; // "Renault" | "Dacia" (canonical casing, matches catalog_vehicles.brand_name) + model: string | null; // "Kadjar (HFE)" + modelCode: string | null; // "HFE" + familyCode: string | null; // "XFE" + modelYear: string | null; // "2015" + engine: string | null; + gearbox: string | null; + energyType: string | null; + manufacturingDate: string | null; + raw: RpartstoreVehicle; +} diff --git a/apps/api/src/jobs/bull.config.ts b/apps/api/src/jobs/bull.config.ts index 3bb90f9..fed0fbf 100644 --- a/apps/api/src/jobs/bull.config.ts +++ b/apps/api/src/jobs/bull.config.ts @@ -36,5 +36,6 @@ export const QUEUE_NAMES = { EXPERT_REWARDS: "expert-rewards", PART_PRICE_REFRESH: "part-price-refresh", VINPIN_DECODE: "vinpin-decode", + RPARTSTORE_DECODE: "rpartstore-decode", CANONICAL_BACKFILL: "canonical-backfill", } as const; diff --git a/apps/api/src/jobs/jobs.module.ts b/apps/api/src/jobs/jobs.module.ts index d29b313..a3efb5c 100644 --- a/apps/api/src/jobs/jobs.module.ts +++ b/apps/api/src/jobs/jobs.module.ts @@ -21,6 +21,10 @@ import { PartPriceRefreshQueueProvider, } from "./queues/part-price-refresh.queue"; import { QUERY_CLEANUP_QUEUE, QueryCleanupQueueProvider } from "./queues/query-cleanup.queue"; +import { + RPARTSTORE_DECODE_QUEUE, + RpartstoreDecodeQueueProvider, +} from "./queues/rpartstore-decode.queue"; import { SUBSCRIPTION_EXPIRY_QUEUE, SubscriptionExpiryQueueProvider, @@ -39,6 +43,7 @@ import { VINPIN_DECODE_QUEUE, VinpinDecodeQueueProvider } from "./queues/vinpin- ExpertRewardsQueueProvider, PartPriceRefreshQueueProvider, VinpinDecodeQueueProvider, + RpartstoreDecodeQueueProvider, CanonicalBackfillQueueProvider, PrefetchWorkerService, ], @@ -52,6 +57,7 @@ import { VINPIN_DECODE_QUEUE, VinpinDecodeQueueProvider } from "./queues/vinpin- EXPERT_REWARDS_QUEUE, PART_PRICE_REFRESH_QUEUE, VINPIN_DECODE_QUEUE, + RPARTSTORE_DECODE_QUEUE, CANONICAL_BACKFILL_QUEUE, ], }) diff --git a/apps/api/src/jobs/processors/rpartstore-decode.processor.spec.ts b/apps/api/src/jobs/processors/rpartstore-decode.processor.spec.ts new file mode 100644 index 0000000..a47cd32 --- /dev/null +++ b/apps/api/src/jobs/processors/rpartstore-decode.processor.spec.ts @@ -0,0 +1,243 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { RpartstoreRateLimitError } from "../../integrations/rpartstore/rpartstore.session"; +import type { RpartstoreDecoded } from "../../integrations/rpartstore/rpartstore.types"; +import { + type RedisLike, + istanbulDayKey, + processRpartstoreDecode, + redisTokenStore, + rpartstoreDailyKey, +} from "./rpartstore-decode.processor"; + +/** In-memory ioredis stand-in covering the surface the processor uses. */ +function fakeRedis(): RedisLike & { store: Map; ttls: Map } { + const store = new Map(); + const ttls = new Map(); + return { + store, + ttls, + async get(k) { + return store.get(k) ?? null; + }, + async set(k, v, _mode, ttl) { + store.set(k, v); + ttls.set(k, ttl); + }, + 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) { + ttls.set(k, s); + }, + }; +} + +/** Chainable drizzle mock: records every `.set()` payload and answers selects with `rows`. */ +function fakeDb(rows: unknown[] = []) { + const sets: Record[] = []; + const update = vi.fn(() => ({ + set: vi.fn((payload: Record) => { + sets.push(payload); + return { where: vi.fn().mockResolvedValue(undefined) }; + }), + })); + const select = vi.fn(() => ({ + from: vi.fn(() => ({ where: vi.fn().mockResolvedValue(rows) })), + })); + return { db: { update, select } as any, sets }; +} + +const decodedKadjar: RpartstoreDecoded = { + brandName: "Renault", + model: "Kadjar (HFE)", + modelCode: "HFE", + familyCode: "XFE", + modelYear: "2015", + engine: "1.5 DCI DİZEL MOTOR [K9K]", + gearbox: "DC4", + energyType: "MOTORIN", + manufacturingDate: "2015-07-28", + raw: { + catalogSource: "DATAHUB", + vin: "VF1RFE00653633190", + vehicleKey: "VF1RFE00653633190", + model: "Kadjar (HFE)", + vehicleBrand: "RENAULT", + country: "TR", + }, +}; + +const job = (vin: string) => + ({ id: `j-${vin}`, data: { vin }, attemptsMade: 0, opts: { attempts: 1 } }) as any; +const NOW = new Date("2026-09-25T20:30:00+03:00"); + +describe("processRpartstoreDecode", () => { + beforeEach(() => { + process.env.RPARTSTORE_ENABLED = "true"; + vi.spyOn(console, "log").mockImplementation(() => undefined); + vi.spyOn(console, "warn").mockImplementation(() => undefined); + vi.spyOn(console, "error").mockImplementation(() => undefined); + }); + afterEach(() => { + process.env.RPARTSTORE_ENABLED = "false"; + vi.restoreAllMocks(); + }); + + it("is a no-op when RPARTSTORE_ENABLED is not true", async () => { + process.env.RPARTSTORE_ENABLED = "false"; + const { db, sets } = fakeDb(); + const searchVin = vi.fn(); + const r = await processRpartstoreDecode(job("VF1X"), { + db, + redis: fakeRedis(), + decoder: () => ({ searchVin }), + dailyCap: 10, + }); + expect(r.skipped).toBe(true); + expect(searchVin).not.toHaveBeenCalled(); + expect(sets).toEqual([]); + }); + + it("decodes, matches the PL24 catalog vehicle and records the decoded row", async () => { + const { db, sets } = fakeDb([ + { id: "cv-kadjar", model: "KADJAR", categoryCount: 44 }, + { id: "cv-kadjar-cn", model: "KADJAR ÇİN", categoryCount: 22 }, + ]); + const redis = fakeRedis(); + const searchVin = vi.fn().mockResolvedValue(decodedKadjar); + const r = await processRpartstoreDecode(job("VF1RFE00653633190"), { + db, + redis, + decoder: () => ({ searchVin }), + dailyCap: 10, + now: () => NOW, + }); + expect(r).toEqual({ status: "decoded", catalogVehicleId: "cv-kadjar" }); + expect(searchVin).toHaveBeenCalledWith("VF1RFE00653633190"); + // one cap unit reserved on the Istanbul day, with an expiry + expect(redis.store.get(rpartstoreDailyKey(NOW))).toBe("1"); + expect(redis.ttls.get(rpartstoreDailyKey(NOW))).toBeGreaterThan(0); + const final = sets.at(-1)!; + expect(final).toMatchObject({ + status: "decoded", + brandName: "Renault", + model: "Kadjar (HFE)", + modelCode: "HFE", + modelYear: "2015", + catalogVehicleId: "cv-kadjar", + }); + }); + + it("records not_found when RPartStore has no vehicle for the VIN", async () => { + const { db, sets } = fakeDb(); + const r = await processRpartstoreDecode(job("VF1NOPE"), { + db, + redis: fakeRedis(), + decoder: () => ({ searchVin: vi.fn().mockResolvedValue(null) }), + dailyCap: 10, + }); + expect(r.status).toBe("not_found"); + expect(sets.at(-1)).toMatchObject({ status: "not_found" }); + }); + + it("stops sending once the daily cap is spent and marks the row capped", async () => { + const redis = fakeRedis(); + const key = rpartstoreDailyKey(NOW); + redis.store.set(key, "10"); // ten searches already sent today + const { db, sets } = fakeDb(); + const searchVin = vi.fn().mockResolvedValue(decodedKadjar); + const r = await processRpartstoreDecode(job("VF1CAP"), { + db, + redis, + decoder: () => ({ searchVin }), + dailyCap: 10, + now: () => NOW, + }); + expect(r.status).toBe("capped"); + expect(searchVin).not.toHaveBeenCalled(); + expect(sets.at(-1)).toMatchObject({ status: "capped" }); + }); + + it("allows exactly `dailyCap` searches per day", async () => { + const redis = fakeRedis(); + const searchVin = vi.fn().mockResolvedValue(null); + const statuses: string[] = []; + for (let i = 0; i < 12; i += 1) { + const { db } = fakeDb(); + const r = await processRpartstoreDecode(job(`VF1${i}`), { + db, + redis, + decoder: () => ({ searchVin }), + dailyCap: 10, + now: () => NOW, + }); + statuses.push(r.status); + } + expect(searchVin).toHaveBeenCalledTimes(10); + expect(statuses.filter((s) => s === "capped")).toHaveLength(2); + }); + + it("waits retryAfterSeconds and retries once on a short-term rate limit without a second reservation", async () => { + const redis = fakeRedis(); + const { db } = fakeDb(); + const searchVin = vi + .fn() + .mockRejectedValueOnce(new RpartstoreRateLimitError(10, "SHORT_TERM")) + .mockResolvedValueOnce(null); + const sleep = vi.fn().mockResolvedValue(undefined); + const r = await processRpartstoreDecode(job("VF1RL"), { + db, + redis, + decoder: () => ({ searchVin }), + dailyCap: 10, + now: () => NOW, + sleep, + }); + expect(r.status).toBe("not_found"); + expect(searchVin).toHaveBeenCalledTimes(2); + expect(sleep).toHaveBeenCalledWith(10_500); + expect(redis.store.get(rpartstoreDailyKey(NOW))).toBe("1"); + }); + + it("marks the row failed and rethrows on an infrastructure error (single attempt)", async () => { + const { db, sets } = fakeDb(); + await expect( + processRpartstoreDecode(job("VF1ERR"), { + db, + redis: fakeRedis(), + decoder: () => { + throw new Error("RPARTSTORE_USER / RPARTSTORE_PASS are not configured"); + }, + dailyCap: 10, + }), + ).rejects.toThrow(/not configured/); + expect(sets.at(-1)).toMatchObject({ status: "failed" }); + }); +}); + +describe("istanbulDayKey", () => { + it("counts the cap against the Istanbul calendar day, not UTC", () => { + // 23:30 UTC on the 25th is already the 26th in Istanbul (UTC+3). + expect(istanbulDayKey(new Date("2026-09-25T23:30:00Z"))).toBe("2026-09-26"); + expect(istanbulDayKey(new Date("2026-09-25T20:59:00Z"))).toBe("2026-09-25"); + }); +}); + +describe("redisTokenStore", () => { + it("round-trips a token with its ttl and ignores garbage", async () => { + const redis = fakeRedis(); + const store = redisTokenStore(redis); + await store.set({ accessToken: "t", expiresAt: 1, subject: "s" }, 120); + expect(await store.get()).toEqual({ accessToken: "t", expiresAt: 1, subject: "s" }); + expect(redis.ttls.get("rpartstore:token")).toBe(120); + redis.store.set("rpartstore:token", "{not json"); + expect(await store.get()).toBeNull(); + await store.clear(); + expect(redis.store.has("rpartstore:token")).toBe(false); + }); +}); diff --git a/apps/api/src/jobs/processors/rpartstore-decode.processor.ts b/apps/api/src/jobs/processors/rpartstore-decode.processor.ts new file mode 100644 index 0000000..7dcb287 --- /dev/null +++ b/apps/api/src/jobs/processors/rpartstore-decode.processor.ts @@ -0,0 +1,252 @@ +import { Job } from "bullmq"; +import { and, eq, sql } from "drizzle-orm"; +import { PostgresJsDatabase } from "drizzle-orm/postgres-js"; +import { catalogVehicles, rpartstoreDecodes } from "../../database/schema/core"; +import type { RpartstoreToken } from "../../integrations/rpartstore/rpartstore.auth"; +import { RpartstoreClient, type TokenStore } from "../../integrations/rpartstore/rpartstore.client"; +import { + type RpartstoreCatalogCandidate, + pickRpartstoreCatalogMatch, +} from "../../integrations/rpartstore/rpartstore.matcher"; +import { RpartstoreRateLimitError } from "../../integrations/rpartstore/rpartstore.session"; +import type { RpartstoreDecoded } from "../../integrations/rpartstore/rpartstore.types"; + +type Database = PostgresJsDatabase>; + +export interface RpartstoreDecodeJobData { + vin: string; +} + +/** Minimal ioredis surface the processor needs (also satisfied by RedisService.getClient()). */ +export interface RedisLike { + get(key: string): Promise; + set(key: string, value: string, mode: "EX", ttlSeconds: number): Promise; + del(key: string): Promise; + incr(key: string): Promise; + expire(key: string, seconds: number): Promise; +} + +export interface RpartstoreDecoder { + searchVin(vin: string): Promise; +} + +export interface RpartstoreProcessorDeps { + db: Database; + redis: RedisLike; + /** Built lazily so a missing credential only fails the job, not worker boot. */ + decoder: () => RpartstoreDecoder; + /** Hard cap on VIN searches sent to RPartStore per Istanbul calendar day. */ + dailyCap: number; + now?: () => Date; + sleep?: (ms: number) => Promise; +} + +export const RPARTSTORE_TOKEN_KEY = "rpartstore:token"; +export const RPARTSTORE_DAILY_KEY_PREFIX = "rpartstore:daily:"; +/** Counter keys live two days so a late-night job never sees a vanished key. */ +const DAILY_KEY_TTL_SECONDS = 2 * 24 * 60 * 60; + +/** "YYYY-MM-DD" in Europe/Istanbul — the day the cap is counted against. */ +export function istanbulDayKey(date: Date): string { + const parts = new Intl.DateTimeFormat("en-CA", { + timeZone: "Europe/Istanbul", + year: "numeric", + month: "2-digit", + day: "2-digit", + }).formatToParts(date); + const get = (t: string): string => parts.find((p) => p.type === t)?.value ?? ""; + return `${get("year")}-${get("month")}-${get("day")}`; +} + +export const rpartstoreDailyKey = (date: Date): string => + `${RPARTSTORE_DAILY_KEY_PREFIX}${istanbulDayKey(date)}`; + +/** Redis-backed cache for the 1 h Okta access token (shared by worker restarts). */ +export function redisTokenStore(redis: RedisLike): TokenStore { + return { + async get() { + const raw = await redis.get(RPARTSTORE_TOKEN_KEY); + if (!raw) return null; + try { + const t = JSON.parse(raw) as RpartstoreToken; + return t.accessToken && t.expiresAt ? t : null; + } catch { + return null; + } + }, + async set(token, ttlSeconds) { + await redis.set(RPARTSTORE_TOKEN_KEY, JSON.stringify(token), "EX", ttlSeconds); + }, + async clear() { + await redis.del(RPARTSTORE_TOKEN_KEY); + }, + }; +} + +export function buildRpartstoreClient(redis: RedisLike): RpartstoreClient { + const username = process.env.RPARTSTORE_USER; + const password = process.env.RPARTSTORE_PASS; + if (!username || !password) { + throw new Error("RPARTSTORE_USER / RPARTSTORE_PASS are not configured"); + } + return new RpartstoreClient({ + username, + password, + tokenStore: redisTokenStore(redis), + brokerUrl: process.env.RPARTSTORE_BROKER_URL || undefined, + appVersion: process.env.RPARTSTORE_APP_VERSION || undefined, + logger: { log: (m) => console.log(m), warn: (m) => console.warn(m) }, + }); +} + +export function rpartstoreDailyCapFromEnv(): number { + const n = Number(process.env.RPARTSTORE_DAILY_CAP); + return Number.isFinite(n) && n >= 0 ? Math.floor(n) : 10; +} + +/** + * RPartStore decode processor (worker, concurrency 1, ≥6 s between jobs via the + * queue limiter — RPartStore allows 2 VIN searches per 10 s). + * + * Reserves one unit of the daily cap BEFORE sending anything: `INCR` on the + * Istanbul-day key; when the reservation lands above the cap the row is marked + * `capped` and nothing is sent, so at most `dailyCap` searches reach RPartStore + * per day even under concurrent enqueues. A rate-limited search waits + * `retryAfterSeconds` and is retried once without a second reservation. + * + * Outcomes recorded in `rpartstore_decodes`: decoded (with an optional PL24 + * catalog_vehicle match), not_found (definitive), capped (retry tomorrow), + * failed (infra/auth error — user-retriable after 24 h). + */ +export async function processRpartstoreDecode( + job: Job, + deps: RpartstoreProcessorDeps, +): Promise<{ status: string; catalogVehicleId: string | null; skipped?: boolean }> { + const { vin } = job.data; + const { db, redis } = deps; + const now = deps.now ?? (() => new Date()); + const sleep = deps.sleep ?? ((ms: number) => new Promise((r) => setTimeout(r, ms))); + + if (process.env.RPARTSTORE_ENABLED !== "true") { + console.log(`[rpartstore-decode] disabled (RPARTSTORE_ENABLED!=true) — job ${job.id} no-op`); + return { status: "pending", catalogVehicleId: null, skipped: true }; + } + + await db + .update(rpartstoreDecodes) + .set({ attempts: sql`${rpartstoreDecodes.attempts} + 1`, updatedAt: now() }) + .where(eq(rpartstoreDecodes.vin, vin)); + + // Daily cap reservation. + const dayKey = rpartstoreDailyKey(now()); + const reserved = await redis.incr(dayKey); + if (reserved === 1) await redis.expire(dayKey, DAILY_KEY_TTL_SECONDS); + if (reserved > deps.dailyCap) { + await db + .update(rpartstoreDecodes) + .set({ status: "capped", updatedAt: now() }) + .where(eq(rpartstoreDecodes.vin, vin)); + console.warn( + `[rpartstore-decode] ${vin} → capped (${reserved - 1}/${deps.dailyCap} searches used on ${dayKey})`, + ); + return { status: "capped", catalogVehicleId: null }; + } + + console.log( + `[rpartstore-decode] job ${job.id} decoding ${vin} (${reserved}/${deps.dailyCap} today)`, + ); + + try { + const decoder = deps.decoder(); + let decoded: RpartstoreDecoded | null; + try { + decoded = await decoder.searchVin(vin); + } catch (err) { + if (!(err instanceof RpartstoreRateLimitError)) throw err; + const waitMs = Math.min(Math.max(err.retryAfterSeconds, 1), 30) * 1000 + 500; + console.warn( + `[rpartstore-decode] ${vin} rate-limited (${err.limitType}); retrying in ${waitMs}ms`, + ); + await sleep(waitMs); + decoded = await decoder.searchVin(vin); + } + + if (!decoded) { + await db + .update(rpartstoreDecodes) + .set({ status: "not_found", updatedAt: now() }) + .where(eq(rpartstoreDecodes.vin, vin)); + console.log(`[rpartstore-decode] ${vin} → not_found`); + return { status: "not_found", catalogVehicleId: null }; + } + + const catalogVehicleId = await matchCatalogVehicle(db, decoded); + + await db + .update(rpartstoreDecodes) + .set({ + status: "decoded", + brandName: decoded.brandName, + model: decoded.model, + modelCode: decoded.modelCode, + familyCode: decoded.familyCode, + modelYear: decoded.modelYear, + engine: decoded.engine, + gearbox: decoded.gearbox, + energyType: decoded.energyType, + manufacturingDate: decoded.manufacturingDate, + catalogVehicleId, + raw: decoded.raw, + decodedAt: now(), + updatedAt: now(), + }) + .where(eq(rpartstoreDecodes.vin, vin)); + + console.log( + `[rpartstore-decode] ${vin} → decoded ${decoded.brandName} "${decoded.model ?? "?"}" ${decoded.modelYear ?? ""} catalog_vehicle=${catalogVehicleId ?? "null"}`, + ); + return { status: "decoded", catalogVehicleId }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + console.error(`[rpartstore-decode] job ${job.id} failed for ${vin}: ${message}`); + const isLastAttempt = job.attemptsMade + 1 >= (job.opts.attempts ?? 1); + if (isLastAttempt) { + await db + .update(rpartstoreDecodes) + .set({ status: "failed", updatedAt: now() }) + .where(eq(rpartstoreDecodes.vin, vin)); + } + throw error; + } +} + +/** Match within the decoded brand first, then the sister brand (Renault ⇄ Dacia + * share platforms and TR badges differ from the WMI). */ +async function matchCatalogVehicle( + db: Database, + decoded: RpartstoreDecoded, +): Promise { + if (!decoded.model) return null; + const sister = decoded.brandName.toLowerCase() === "dacia" ? "renault" : "dacia"; + for (const brand of [decoded.brandName.toLowerCase(), sister]) { + const rows = await db + .select({ + id: catalogVehicles.id, + model: catalogVehicles.model, + categoryCount: sql`( + SELECT count(*)::int FROM categories + WHERE categories.catalog_vehicle_id = ${catalogVehicles.id} + )`, + }) + .from(catalogVehicles) + .where( + and( + sql`lower(${catalogVehicles.brandName}) = ${brand}`, + eq(catalogVehicles.source, "pl24"), + ), + ); + const id = pickRpartstoreCatalogMatch(decoded.model, rows as RpartstoreCatalogCandidate[]); + if (id) return id; + } + return null; +} diff --git a/apps/api/src/jobs/queues/rpartstore-decode.queue.ts b/apps/api/src/jobs/queues/rpartstore-decode.queue.ts new file mode 100644 index 0000000..8acd267 --- /dev/null +++ b/apps/api/src/jobs/queues/rpartstore-decode.queue.ts @@ -0,0 +1,32 @@ +import { Provider } from "@nestjs/common"; +import { type JobsOptions, Queue } from "bullmq"; +import { QUEUE_NAMES, getBullConnection, getBullTelemetry } from "../bull.config"; + +export const RPARTSTORE_DECODE_QUEUE = "RPARTSTORE_DECODE_QUEUE"; + +/** + * `attempts: 1` — the processor records a definitive outcome per job + * (decoded / not_found / capped / failed) and the daily cap must never be + * burned by automatic re-runs. A `failed` or `capped` row is re-enqueued by the + * decode path itself after 24 h (see VehiclesService.tryRpartstoreFallback). + */ +export const RPARTSTORE_DECODE_JOB_OPTIONS: JobsOptions = { + attempts: 1, + removeOnComplete: { count: 50 }, + removeOnFail: { count: 100 }, +}; + +/** RPartStore (Renault/Dacia) VIN-decode fallback queue. One shared dealer + * account → the worker consumes it with concurrency 1 and a 1-job-per-6 s + * limiter (RPartStore allows 2 searches per 10 s). Jobs carry `{ vin }`. */ +export const RpartstoreDecodeQueueProvider: Provider = { + provide: RPARTSTORE_DECODE_QUEUE, + useFactory: () => { + const telemetry = getBullTelemetry(); + return new Queue(QUEUE_NAMES.RPARTSTORE_DECODE, { + connection: getBullConnection(), + ...(telemetry ? { telemetry } : {}), + defaultJobOptions: RPARTSTORE_DECODE_JOB_OPTIONS, + }); + }, +}; diff --git a/apps/api/src/vehicles/vehicles.service.spec.ts b/apps/api/src/vehicles/vehicles.service.spec.ts index d63f462..e4da294 100644 --- a/apps/api/src/vehicles/vehicles.service.spec.ts +++ b/apps/api/src/vehicles/vehicles.service.spec.ts @@ -1,6 +1,11 @@ import { BadRequestException, ForbiddenException, NotFoundException } from "@nestjs/common"; -import { beforeEach, describe, expect, it, vi } from "vitest"; -import { catalogVehicles, queryLogs, vinpinDecodes } from "../database/schema/core"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + catalogVehicles, + queryLogs, + rpartstoreDecodes, + vinpinDecodes, +} from "../database/schema/core"; import { VehiclesService } from "./vehicles.service"; vi.mock("@sase/shared", () => ({ @@ -93,10 +98,15 @@ function createService(dbOrOverrides: any = {}) { add: vi.fn().mockResolvedValue(undefined), }; + const rpartstoreQueue = { + add: vi.fn().mockResolvedValue(undefined), + }; + const service = new VehiclesService( db as any, prefetchQueue as any, vinpinQueue as any, + rpartstoreQueue as any, corgiService as any, pl24Service as any, vinApiService as any, @@ -115,6 +125,7 @@ function createService(dbOrOverrides: any = {}) { partsCatalogsService, redisService, vinpinQueue, + rpartstoreQueue, }; } @@ -340,9 +351,7 @@ describe("VehiclesService", () => { selectChain.limit = vi .fn() .mockImplementation(() => - currentTable === vinpinDecodes - ? [{ vin: "NM435600006123456", status: "pending" }] - : [], + currentTable === vinpinDecodes ? [{ vin: "NM435600006123456", status: "pending" }] : [], ); const insertChain = { values: vi.fn().mockReturnThis(), @@ -992,3 +1001,168 @@ describe("VehiclesService", () => { }); }); }); + +describe("VehiclesService — RPartStore fallback (Renault/Dacia, after the pcat/PL24/emex race)", () => { + const VIN = "VF1RFE00653633190"; + + function noCatalogDb() { + const selectChain = { + from: vi.fn().mockReturnThis(), + where: vi.fn().mockReturnThis(), + limit: vi.fn().mockReturnValue([]), + }; + const insertChain = { + values: vi.fn().mockReturnThis(), + returning: vi.fn().mockReturnValue([]), + onConflictDoNothing: vi.fn().mockReturnThis(), + }; + return { + select: vi.fn().mockReturnValue(selectChain), + insert: vi.fn().mockReturnValue(insertChain), + }; + } + + function renaultIdent(svc: ReturnType) { + svc.corgiService.decodeVin.mockReturnValue({ + isKnown: true, + brandName: "Renault", + modelYear: 2015, + }); + svc.vinApiService.decodeVin.mockResolvedValue({ + make: "RENAULT", + model: "Kadjar", + modelYear: "2015", + }); + } + + let prevRparts: string | undefined; + let prevVinpin: string | undefined; + beforeEach(() => { + prevRparts = process.env.RPARTSTORE_ENABLED; + prevVinpin = process.env.VINPIN_ENABLED; + process.env.VINPIN_ENABLED = "false"; + vi.mocked(isValidVin).mockReturnValue(true); + }); + afterEach(() => { + process.env.RPARTSTORE_ENABLED = prevRparts ?? "false"; + process.env.VINPIN_ENABLED = prevVinpin ?? "false"; + }); + + it("[off] leaves the no-catalog path untouched and never touches the queue", async () => { + process.env.RPARTSTORE_ENABLED = "false"; + const db = noCatalogDb(); + const svc = createService(db); + renaultIdent(svc); + + const result: any = await svc.service.decodeVin(VIN, "u1"); + expect(result).toMatchObject({ noCatalog: { brandName: "Renault" }, vin: VIN }); + expect(svc.rpartstoreQueue.add).not.toHaveBeenCalled(); + expect(db.insert.mock.calls.some((c: unknown[]) => c[0] === rpartstoreDecodes)).toBe(false); + }); + + it("[on] records a pending row, enqueues a day-scoped job and answers `decoding` for an unseen Renault VIN", async () => { + process.env.RPARTSTORE_ENABLED = "true"; + const db = noCatalogDb(); + const svc = createService(db); + renaultIdent(svc); + + const result: any = await svc.service.decodeVin(VIN, "u1"); + expect(result).toMatchObject({ decoding: { vin: VIN }, vin: VIN }); + expect(result.decoding.display).toContain("Renault"); + expect(svc.rpartstoreQueue.add).toHaveBeenCalledTimes(1); + const [name, data, opts] = svc.rpartstoreQueue.add.mock.calls[0]; + expect(name).toBe("rpartstore-decode"); + expect(data).toEqual({ vin: VIN }); + expect(opts.jobId).toMatch(new RegExp(`^rpartstore-${VIN}-\\d{4}-\\d{2}-\\d{2}$`)); + expect(db.insert.mock.calls.some((c: unknown[]) => c[0] === rpartstoreDecodes)).toBe(true); + // Exactly ONE coverage-gap failure row at first sighting. + expect(db.insert.mock.calls.filter((c: unknown[]) => c[0] === queryLogs)).toHaveLength(1); + // Vinpin (disabled) is never consulted. + expect(svc.vinpinQueue.add).not.toHaveBeenCalled(); + }); + + it("[on] does not enqueue a non-Renault VIN", async () => { + process.env.RPARTSTORE_ENABLED = "true"; + const db = noCatalogDb(); + const svc = createService(db); + svc.corgiService.decodeVin.mockReturnValue({ + isKnown: true, + brandName: "Fiat", + modelYear: 2018, + }); + svc.vinApiService.decodeVin.mockResolvedValue({ + make: "FIAT", + model: "Tipo", + modelYear: "2018", + }); + + const result: any = await svc.service.decodeVin("NM435600006123456", "u1"); + expect(result).toMatchObject({ noCatalog: { brandName: "Fiat" } }); + expect(svc.rpartstoreQueue.add).not.toHaveBeenCalled(); + }); + + it("[on] falls through to noCatalog without queueing once today's cap is spent", async () => { + process.env.RPARTSTORE_ENABLED = "true"; + process.env.RPARTSTORE_DAILY_CAP = "10"; + const db = noCatalogDb(); + const svc = createService(db); + renaultIdent(svc); + svc.redisService.get.mockImplementation(async (key: string) => + key.startsWith("rpartstore:daily:") ? "10" : null, + ); + + const result: any = await svc.service.decodeVin(VIN, "u1"); + expect(result).toMatchObject({ noCatalog: { brandName: "Renault" } }); + expect(svc.rpartstoreQueue.add).not.toHaveBeenCalled(); + expect(db.insert.mock.calls.some((c: unknown[]) => c[0] === rpartstoreDecodes)).toBe(false); + process.env.RPARTSTORE_DAILY_CAP = ""; + }); + + it("[on] a decoded row with a catalog match answers `catalogVehicle` and logs one success row", async () => { + process.env.RPARTSTORE_ENABLED = "true"; + const rpartsRow = { + vin: VIN, + status: "decoded", + catalogVehicleId: "cv-1", + createdAt: new Date(), + updatedAt: new Date(), + }; + const cv = { id: "cv-1", brandId: null, brandName: "Renault", model: "KADJAR", year: null }; + let selectCalls = 0; + const selectChain = { + from: vi.fn().mockReturnThis(), + where: vi.fn().mockReturnThis(), + limit: vi.fn().mockImplementation(() => { + // Call order inside decodeVin: vehicles lookup → rpartstore row → catalog vehicle. + selectCalls += 1; + if (selectCalls === 2) return [rpartsRow]; + if (selectCalls === 3) return [cv]; + return []; + }), + }; + const insertChain = { + values: vi.fn().mockReturnThis(), + returning: vi.fn().mockReturnValue([]), + onConflictDoNothing: vi.fn().mockReturnThis(), + }; + const db = { + select: vi.fn().mockReturnValue(selectChain), + insert: vi.fn().mockReturnValue(insertChain), + }; + const svc = createService(db); + renaultIdent(svc); + + const result: any = await svc.service.decodeVin(VIN, "u1"); + expect(result).toEqual({ + catalogVehicle: { id: "cv-1", brandName: "Renault", model: "KADJAR", year: null }, + vin: VIN, + }); + expect(svc.rpartstoreQueue.add).not.toHaveBeenCalled(); + const logRows = db.insert.mock.calls.filter((c: unknown[]) => c[0] === queryLogs); + expect(logRows).toHaveLength(1); + expect(insertChain.values.mock.calls.at(-1)?.[0]).toMatchObject({ + source: "rpartstore", + success: true, + }); + }); +}); diff --git a/apps/api/src/vehicles/vehicles.service.ts b/apps/api/src/vehicles/vehicles.service.ts index 2ee9352..a4ff0a1 100644 --- a/apps/api/src/vehicles/vehicles.service.ts +++ b/apps/api/src/vehicles/vehicles.service.ts @@ -16,6 +16,7 @@ import { parts, plans, queryLogs, + rpartstoreDecodes, userBrands, userSubscriptions, userVehicles, @@ -34,10 +35,13 @@ import { PartsCatalogsService } from "../integrations/parts-catalogs/parts-catal import { PcatCar, PcatVinResult } from "../integrations/parts-catalogs/parts-catalogs.types"; import { PL24Service } from "../integrations/pl24/pl24.service"; import { SERVICE_TO_BRAND } from "../integrations/pl24/pl24.types"; +import { isRpartstoreVin } from "../integrations/rpartstore/rpartstore.routing"; import { VinApiService } from "../integrations/vin-api/vin-api.service"; import { isVinpinBrandAllowed } from "../integrations/vinpin/vinpin.constants"; import { PrefetchSource } from "../jobs/prefetch.types"; +import { istanbulDayKey, rpartstoreDailyKey } from "../jobs/processors/rpartstore-decode.processor"; import { CATALOG_PREFETCH_FAST_QUEUE } from "../jobs/queues/catalog-prefetch.queue"; +import { RPARTSTORE_DECODE_QUEUE } from "../jobs/queues/rpartstore-decode.queue"; import { VINPIN_DECODE_QUEUE } from "../jobs/queues/vinpin-decode.queue"; import { RedisService } from "../redis/redis.service"; import { vinCandidateStashKey, vinResolveCacheKeys } from "./vin-cache-keys"; @@ -99,6 +103,7 @@ export class VehiclesService { @Inject(DATABASE) private db: Database, @Inject(CATALOG_PREFETCH_FAST_QUEUE) private prefetchQueue: Queue, @Inject(VINPIN_DECODE_QUEUE) private vinpinQueue: Queue, + @Inject(RPARTSTORE_DECODE_QUEUE) private rpartstoreQueue: Queue, private corgiService: CorgiService, private pl24Service: PL24Service, private vinApiService: VinApiService, @@ -191,6 +196,19 @@ export class VehiclesService { // The fallback logs the coverage-gap failure row exactly once, at // first sighting (job enqueue); not_found/failed/stale fall through // to the failure log below, unchanged. + // RPartStore decode-oracle fallback for Renault/Dacia (flag-gated, + // hard daily cap). Ordered AFTER the pcat/PL24/emex race (we only get + // here when it returned nothing) and BEFORE Vinpin, which stays wired + // as a last resort behind its own flag. Same accounting contract as + // the Vinpin fallback below: one failure row at first sighting, polls + // log nothing, a resolved decode logs its own success row. + if ( + process.env.RPARTSTORE_ENABLED === "true" && + isRpartstoreVin(vin, ident.browseBrand) + ) { + const rpartsResp = await this.tryRpartstoreFallback(vin, ident, userId, ctx, startTime); + if (rpartsResp) return rpartsResp; + } if (process.env.VINPIN_ENABLED === "true" && isVinpinBrandAllowed(ident.browseBrand)) { const vinpinResp = await this.tryVinpinFallback(vin, ident, userId, ctx, startTime); if (vinpinResp) return vinpinResp; @@ -791,6 +809,154 @@ export class VehiclesService { return null; } + /** `failed` / `capped` RPartStore rows become eligible for one more attempt after this long. */ + private static readonly RPARTSTORE_RETRY_AFTER_MS = 24 * 60 * 60 * 1000; + + /** + * RPartStore decode-oracle fallback for a Renault/Dacia VIN the race couldn't + * identify (feature-flagged, guarded by the caller on RPARTSTORE_ENABLED + + * `isRpartstoreVin`). Mirrors the Vinpin fallback contract: + * + * - decoded + catalog_vehicle_id → resolve that EXISTING PL24 catalog vehicle + * and return `catalogVehicle` (parts come from PL24, not RPartStore). + * - pending → `{ decoding }` (frontend polls). + * - no row → insert 'pending', enqueue a decode job, return `{ decoding }`. + * - capped / failed older than 24 h → re-enqueue (a fresh cap reservation), + * return `{ decoding }`; younger → null. + * - not_found / decoded-without-match / cap exhausted today → null → caller + * falls through to Vinpin (if enabled) and then noCatalog, unchanged. + * + * The API side never talks to RPartStore; it only checks today's counter so a + * VIN is not queued (and left `pending`) when the daily cap is already spent. + */ + private async tryRpartstoreFallback( + vin: string, + ident: { browseBrand: string | null; display: string }, + userId: string, + ctx: ResolveContext, + startTime: number, + // biome-ignore lint/suspicious/noExplicitAny: heterogeneous short-circuit response shapes + ): Promise { + try { + const [row] = await this.db + .select() + .from(rpartstoreDecodes) + .where(eq(rpartstoreDecodes.vin, vin)) + .limit(1); + + if (row) { + if (row.status === "decoded" && row.catalogVehicleId) { + const [cv] = await this.db + .select() + .from(catalogVehicles) + .where(eq(catalogVehicles.id, row.catalogVehicleId)) + .limit(1); + if (cv) { + if (cv.brandId) await this.checkBrandAccess(userId, cv.brandId); + ctx.timings.rpartstore_resolved = 1; + await this.logQuery( + userId, + vin, + cv.brandId, + "rpartstore", + true, + Date.now() - startTime, + undefined, + ctx.timings, + ); + return { + catalogVehicle: { + id: cv.id, + brandName: cv.brandName, + model: cv.model, + year: cv.year, + }, + vin, + }; + } + return null; + } + + if (row.status === "pending") { + ctx.timings.rpartstore_pending = 1; + return { decoding: { vin, display: ident.display }, vin }; + } + + const retryable = row.status === "capped" || row.status === "failed"; + const ageMs = Date.now() - (row.updatedAt ?? row.createdAt).getTime(); + if (!retryable || ageMs < VehiclesService.RPARTSTORE_RETRY_AFTER_MS) { + // not_found / decoded-without-match / recent capped|failed → existing path. + return null; + } + if (await this.isRpartstoreCapSpent()) { + ctx.timings.rpartstore_capped = 1; + return null; + } + await this.db + .update(rpartstoreDecodes) + .set({ status: "pending", updatedAt: new Date() }) + .where(eq(rpartstoreDecodes.vin, vin)); + await this.enqueueRpartstoreDecode(vin); + ctx.timings.rpartstore_requeued = 1; + return { decoding: { vin, display: ident.display }, vin }; + } + + if (await this.isRpartstoreCapSpent()) { + ctx.timings.rpartstore_capped = 1; + return null; + } + await this.db + .insert(rpartstoreDecodes) + .values({ vin, status: "pending" }) + .onConflictDoNothing(); + await this.enqueueRpartstoreDecode(vin); + ctx.timings.rpartstore_enqueued = 1; + // The ONE coverage-gap failure row for this VIN (see the Vinpin fallback). + await this.logQuery( + userId, + vin, + null, + "none", + false, + Date.now() - startTime, + `No catalog — identified as ${ident.display}`, + ctx.timings, + ); + return { decoding: { vin, display: ident.display }, vin }; + } catch (err) { + this.logger.warn(`RPartStore fallback failed for ${vin}: ${(err as Error).message}`); + return null; + } + } + + /** Today's RPartStore search counter (Istanbul day, maintained by the worker) + * is already at the cap → don't queue. Fails open on Redis trouble. */ + private async isRpartstoreCapSpent(): Promise { + const cap = Number(process.env.RPARTSTORE_DAILY_CAP); + const limit = Number.isFinite(cap) && cap >= 0 ? Math.floor(cap) : 10; + try { + const used = Number((await this.redis.get(rpartstoreDailyKey(new Date()))) ?? 0); + return used >= limit; + } catch { + return false; + } + } + + private async enqueueRpartstoreDecode(vin: string): Promise { + // Day-scoped jobId: dedupes concurrent requests for the same VIN today while + // letting a capped/failed VIN be re-queued tomorrow (BullMQ ignores a re-add + // whose jobId still exists among kept completed/failed jobs). + await this.rpartstoreQueue.add( + "rpartstore-decode", + { vin }, + { + jobId: `rpartstore-${vin}-${istanbulDayKey(new Date())}`, + removeOnComplete: true, + removeOnFail: false, + }, + ); + } + /** * Vinpin ePER decode-oracle fallback for a no-catalog VIN (feature-flagged, * guarded by the caller on VINPIN_ENABLED + brand allowlist). diff --git a/apps/api/src/worker.ts b/apps/api/src/worker.ts index 71f5cea..16438b4 100644 --- a/apps/api/src/worker.ts +++ b/apps/api/src/worker.ts @@ -15,6 +15,11 @@ import { processExpertRewards } from "./jobs/processors/expert-rewards.processor import { processLifecycleEmails } from "./jobs/processors/lifecycle-email.processor"; import { processPartPriceRefresh } from "./jobs/processors/part-price-refresh.processor"; import { processQueryCleanup } from "./jobs/processors/query-cleanup.processor"; +import { + buildRpartstoreClient, + processRpartstoreDecode, + rpartstoreDailyCapFromEnv, +} from "./jobs/processors/rpartstore-decode.processor"; import { processSubscriptionExpiry } from "./jobs/processors/subscription-expiry.processor"; import { processTranslation } from "./jobs/processors/translation.processor"; import { processVinpinDecode } from "./jobs/processors/vinpin-decode.processor"; @@ -244,6 +249,54 @@ vinpinDecodeWorker.on("failed", (job, err) => { workers.push(vinpinDecodeWorker); +// RPartStore Decode Worker (Renault/Dacia VINs the pcat/PL24/emex race couldn't +// identify — runs BEFORE Vinpin). One shared dealer account: concurrency 1 plus +// a 1-job-per-6 s limiter (the portal allows 2 VIN searches per 10 s), and the +// processor enforces the hard RPARTSTORE_DAILY_CAP. Strict no-op when +// RPARTSTORE_ENABLED!=true. Its own ioredis client caches the 1 h Okta token +// and the per-day counter (BullMQ's connection is not for app data). +const rpartstoreRedis = 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, +}); +let rpartstoreClient: ReturnType | null = null; +const rpartstoreDecodeWorker = new Worker( + QUEUE_NAMES.RPARTSTORE_DECODE, + async (job) => { + return processRpartstoreDecode(job, { + db, + redis: rpartstoreRedis, + decoder: () => { + rpartstoreClient ??= buildRpartstoreClient(rpartstoreRedis); + return rpartstoreClient; + }, + dailyCap: rpartstoreDailyCapFromEnv(), + }); + }, + { + connection, + concurrency: 1, + limiter: { max: 1, duration: 6_000 }, + ...(telemetry ? { telemetry } : {}), + }, +); + +rpartstoreDecodeWorker.on("completed", (job, result) => { + console.log(`[worker] rpartstore-decode job ${job.id} completed → ${result?.status}`); +}); + +rpartstoreDecodeWorker.on("failed", (job, err) => { + console.error(`[worker] rpartstore-decode job ${job?.id} failed: ${err.message}`); + Sentry.captureException(err, { + tags: { queue: QUEUE_NAMES.RPARTSTORE_DECODE, jobId: job?.id }, + }); +}); + +workers.push(rpartstoreDecodeWorker); + // 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 @@ -355,6 +408,9 @@ async function shutdown(signal: string) { await vinpinDaemon.stop(); console.log("[worker] Vinpin warm daemon stopped"); + // 2c. Release the RPartStore token/counter Redis client. + rpartstoreRedis.disconnect(); + // 3. Close database connection await sql.end(); console.log("[worker] Database connection closed"); diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index e8cb951..0282bc2 100644 --- a/docker-compose.coolify.yml +++ b/docker-compose.coolify.yml @@ -70,6 +70,14 @@ services: # Rpartstore outage → Renault decodes go straight to Dialogys (drops the # per-decode launch-error probe + "Loading application..." stray + budget burn). - VINPIN_RPARTSTORE_ENABLED=${VINPIN_RPARTSTORE_ENABLED:-true} + # RPartStore (Renault/Dacia) VIN-decode fallback — after pcat/PL24/emex, before Vinpin. + # Hard daily cap on searches sent to rpartstore.renault.com (default 10). + - RPARTSTORE_ENABLED=${RPARTSTORE_ENABLED:-false} + - RPARTSTORE_USER=${RPARTSTORE_USER:-} + - RPARTSTORE_PASS=${RPARTSTORE_PASS:-} + - 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} - POSTAL_API_URL=${POSTAL_API_URL:-} - POSTAL_API_KEY=${POSTAL_API_KEY:-} - POSTAL_FROM_ADDRESS=${POSTAL_FROM_ADDRESS:-noreply@sase.tr} @@ -227,6 +235,14 @@ services: - VINPIN_WARM_DAEMON=${VINPIN_WARM_DAEMON:-false} # Skip Rpartstore during a known upstream outage (Renault → straight to Dialogys). - VINPIN_RPARTSTORE_ENABLED=${VINPIN_RPARTSTORE_ENABLED:-true} + # RPartStore (Renault/Dacia) VIN-decode fallback — after pcat/PL24/emex, before Vinpin. + # Hard daily cap on searches sent to rpartstore.renault.com (default 10). + - RPARTSTORE_ENABLED=${RPARTSTORE_ENABLED:-false} + - RPARTSTORE_USER=${RPARTSTORE_USER:-} + - RPARTSTORE_PASS=${RPARTSTORE_PASS:-} + - 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} # 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:-} diff --git a/packages/config/src/index.ts b/packages/config/src/index.ts index c10b1f9..ed30f97 100644 --- a/packages/config/src/index.ts +++ b/packages/config/src/index.ts @@ -76,6 +76,26 @@ export const envSchema = z.object({ .transform((v) => v === "true") .default("false"), + // RPartStore (rpartstore.renault.com) Renault/Dacia VIN-decode fallback — runs + // AFTER the pcat/PL24/emex race, before Vinpin. Off by default; the worker + // needs the dealer credentials. RPARTSTORE_DAILY_CAP is a hard ceiling on VIN + // searches sent per Istanbul day (the portal itself also rate-limits 2/10 s). + RPARTSTORE_ENABLED: z + .string() + .transform((v) => v === "true") + .default("false"), + RPARTSTORE_USER: z.string().optional(), + RPARTSTORE_PASS: z.string().optional(), + RPARTSTORE_DAILY_CAP: z.coerce.number().int().min(0).default(10), + RPARTSTORE_BROKER_URL: z.preprocess( + (v) => (typeof v === "string" && v.trim() === "" ? undefined : v), + z.string().url().default("wss://1po-bff.renault-edh.com/ws"), + ), + RPARTSTORE_APP_VERSION: z.preprocess( + (v) => (typeof v === "string" && v.trim() === "" ? undefined : v), + z.string().default("1.34.0.6"), + ), + // Parts-Catalogs (Playwright JWT capture + DataImpulse proxy) PCAT_USE_PROXY: z.string().default("true"), PCAT_PROXY_HOST: z.string().default("gw.dataimpulse.com"),