feat(rpartstore): Renault/Dacia için RPartStore VIN-decode fallback kaynağı

rpartstore.renault.com (Renault Group bayi portalı) yeni decode kaynağı olarak eklendi. Portal REST değil STOMP 1.2 over WebSocket konuşuyor; login Okta OIE (IDX JSON API + PKCE), tarayıcı gerekmiyor. Sıralama Renault/Dacia için: pcat + PL24 + emex yarışı → rpartstore → (bayrağı açıksa) Vinpin son çare. - integrations/rpartstore: STOMP codec, Okta login, oturum (trace-id ile çok-mesajlı cevap eşleme), istemci (Redis'te 1 saatlik token), Renault/Dacia yönlendirme, RPartStore → PL24 catalog_vehicles model eşleyici (roman rakam, karoseri niteleyici, yanlış-pazar cezası). - rpartstore_decodes tablosu (migration 0037) + rpartstore-decode BullMQ kuyruğu; worker concurrency 1 + 1 iş / 6 s limiter (portal 2 arama / 10 s). - Günlük sert kota RPARTSTORE_DAILY_CAP (varsayılan 10, İstanbul günü): işlemci göndermeden önce Redis INCR ile rezervasyon yapar, aşan VIN'ler `capped` olur ve 24 saat sonra tekrar denenir; API kota doluysa kuyruğa hiç almaz. Kısa vadeli rate-limit cevabında retryAfter kadar bekleyip bir kez tekrar dener. - VehiclesService.tryRpartstoreFallback: Vinpin fallback ile aynı sözleşme (decoding/catalogVehicle cevapları, tek başarısızlık log satırı). - Env: RPARTSTORE_ENABLED/USER/PASS/DAILY_CAP/BROKER_URL/APP_VERSION (compose api+worker). Vinpin koda dokunulmadan pasif kalır (VINPIN_ENABLED). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-25 13:58:09 +03:00
parent 41147f3f05
commit 71568e2274
23 changed files with 2320 additions and 6 deletions
--- a/apps/api/drizzle/0037_rpartstore_decodes.sql
+++ 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");
--- 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
    }
  ]
-}
+}
--- 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",
--- a/apps/api/src/integrations/rpartstore/rpartstore.auth.spec.ts
+++ 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, unknown>): string =>
+  `${b64url('{"alg":"RS256"}')}.${b64url(JSON.stringify(claims))}.sig`;
+
+function jsonResponse(
+  body: unknown,
+  init: { status?: number; headers?: Record<string, string> } = {},
+) {
+  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<string, string>;
+      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("<script>var stateToken = '02.id.abc\\x2Ddef';</script>", {
+          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);
+  });
+});
--- a/apps/api/src/integrations/rpartstore/rpartstore.auth.ts
+++ 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<string, string>();
+
+  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<RpartstoreToken> {
+  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<Response> => {
+    const res = await doFetch(url, {
+      redirect: "manual",
+      ...init,
+      headers: {
+        "User-Agent": USER_AGENT,
+        Cookie: jar.header(),
+        ...(init.headers as Record<string, string> | 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<string, unknown>,
+    step: string,
+  ): Promise<Record<string, any>> => {
+    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<string, any>;
+    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 };
+}
--- a/apps/api/src/integrations/rpartstore/rpartstore.client.ts
+++ 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<RpartstoreToken | null>;
+  set(token: RpartstoreToken, ttlSeconds: number): Promise<void>;
+  clear(): Promise<void>;
+}
+
+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<typeof RpartstoreSession>[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<RpartstoreClientOptions["sessionFactory"]>;
+
+  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<RpartstoreToken> {
+    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<RpartstoreDecoded | null> {
+    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<SearchVehicleResponsePayload>(
+        "/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<RpartstoreSession> {
+    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,
+  };
+}
--- a/apps/api/src/integrations/rpartstore/rpartstore.matcher.spec.ts
+++ 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");
+  });
+});
--- a/apps/api/src/integrations/rpartstore/rpartstore.matcher.ts
+++ 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<string, string> = {
+  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<string>([
+  "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;
+}
--- a/apps/api/src/integrations/rpartstore/rpartstore.routing.ts
+++ 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<string>(["VF1", "VF2", "VF6", "UU1"]);
+
+const RPARTSTORE_BRANDS = new Set<string>(["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());
+}
--- a/apps/api/src/integrations/rpartstore/rpartstore.session.ts
+++ 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<T = unknown> {
+  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<string>;
+  resolve: (msg: BffMessage) => void;
+  reject: (err: Error) => void;
+  timer: NodeJS.Timeout;
+}
+
+export class RpartstoreSession {
+  private ws: WebSocket | null = null;
+  private readonly pending = new Map<string, Pending>();
+  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<void> {
+    const {
+      brokerUrl,
+      accessToken,
+      profile,
+      appVersion,
+      webLanguage,
+      connectTimeoutMs = 10_000,
+      heartbeatMs = 60_000,
+    } = this.opts;
+    return new Promise<void>((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/<path>` and resolves on the first expected message type. */
+  request<T = unknown>(
+    path: string,
+    payload: unknown,
+    options: RequestOptions,
+  ): Promise<BffMessage<T>> {
+    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<BffMessage<T>>((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<T>),
+        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<string, string>, 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);
+  }
+}
--- a/apps/api/src/integrations/rpartstore/rpartstore.stomp.spec.ts
+++ 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");
+  });
+});
--- a/apps/api/src/integrations/rpartstore/rpartstore.stomp.ts
+++ 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/<path> 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<string, string>;
+  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<string, string | number>,
+  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<string, string> = {};
+  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 };
+}
--- a/apps/api/src/integrations/rpartstore/rpartstore.types.ts
+++ 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;
+}
--- 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;
--- 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,
  ],
 })
--- a/apps/api/src/jobs/processors/rpartstore-decode.processor.spec.ts
+++ 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<string, string>; ttls: Map<string, number> } {
+  const store = new Map<string, string>();
+  const ttls = new Map<string, number>();
+  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<string, unknown>[] = [];
+  const update = vi.fn(() => ({
+    set: vi.fn((payload: Record<string, unknown>) => {
+      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);
+  });
+});
--- a/apps/api/src/jobs/processors/rpartstore-decode.processor.ts
+++ 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<Record<string, unknown>>;
+
+export interface RpartstoreDecodeJobData {
+  vin: string;
+}
+
+/** Minimal ioredis surface the processor needs (also satisfied by RedisService.getClient()). */
+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>;
+}
+
+export interface RpartstoreDecoder {
+  searchVin(vin: string): Promise<RpartstoreDecoded | null>;
+}
+
+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<void>;
+}
+
+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<RpartstoreDecodeJobData>,
+  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<void>((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<string | null> {
+  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<number>`(
+          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;
+}
--- a/apps/api/src/jobs/queues/rpartstore-decode.queue.ts
+++ 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,
+    });
+  },
+};
--- 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<typeof createService>) {
+    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,
+    });
+  });
+});
--- 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<any | null> {
+    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<boolean> {
+    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<void> {
+    // 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).
--- 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<typeof buildRpartstoreClient> | 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");