Merge pull request 'dev' (#74) from dev into main

Reviewed-on: #74

apps/api/src/integrations/parts-catalogs/parts-catalogs-auth.service.ts

@@ -18,8 +18,17 @@
 import { Injectable, Logger, type OnModuleDestroy, type OnModuleInit } from "@nestjs/common";
 import { ConfigService } from "@nestjs/config";
 import { Browser, BrowserContext } from "playwright";
+import { RedisService } from "../../redis/redis.service";
 import { JwtSlot, PcatJwtToken, PcatSession } from "./parts-catalogs.types";
 
+// Shared single-slot cache so dev + prod (and any restarted container) can
+// inherit a freshly-captured warm slot instead of paying the Playwright
+// capture cost from cold. Token is still IP-bound to its proxyPort; if a
+// reader on a different IP gets 401/403, invalidateSession's existing
+// fallback re-captures locally. So worst case = today's behavior.
+const REDIS_SLOT_KEY = "pcat:jwt:slot";
+const REDIS_SAFETY_BUFFER_S = 60; // don't serve from Redis if <60s of life left
+
 const TOKEN_TTL = 600; // seconds — TWS- has no built-in expiry, refresh aggressively
 const REFRESH_BUFFER = 90; // Refresh 90s before expiry
 const CAPTURE_POLL_INTERVAL = 500; // ms
@@ -29,6 +38,14 @@ const PAGE_TIMEOUT = 10_000; // 10s — healthy partner sites load in <5s throug
 const CONTEXT_CLOSE_TIMEOUT = 5_000;
 const SITE_COOLDOWN = 10 * 60 * 1000; // 10 min per site
 const MAX_POOL_SIZE = 5;
+// Capture-time validation: do one tiny upstream call with the freshly-captured
+// JWT + proxy port BEFORE inserting the slot in the pool. DataImpulse's rotating
+// proxy occasionally hands out IPs that the partner-site widget can load but
+// the upstream API cannot reach (or that get instantly 401/403'd). Catching
+// those at capture time means the bad slot never reaches a real user.
+const VALIDATION_VIN = "WVWZZZ1JZ3W597935"; // public demo VIN from the homepage
+const VALIDATION_TIMEOUT_MS = 5_000;
+const PCAT_API_BASE = "https://gui.parts-catalogs.com/v3/api/proxy";
 const RPM_WINDOW = 60_000; // 1-minute rolling window
 const RPM_PER_SLOT = 6; // 1 token per 6 req/min
 
@@ -95,7 +112,13 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
 
   private browser: Browser | null = null;
   private launching: Promise<void> | null = null;
-  private readonly semaphore = new Semaphore(1); // Max 1 concurrent JWT capture
+  // Concurrent captures capped at JWT_SITES.length — the real bottleneck is
+  // how many distinct partner sites we can drive in parallel (one capture per
+  // site, since each site = a different IP / cookie origin). Ports are
+  // ~unlimited (10000-10999) and Playwright contexts are isolated; the
+  // synchronous siteIndex round-robin hands each concurrent capture a
+  // different site so they don't collide.
+  private readonly semaphore = new Semaphore(JWT_SITES.length);
 
   // Pool state
   private pool: JwtSlot[] = [];
@@ -115,7 +138,10 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
   private readonly proxyUser: string;
   private readonly proxyPass: string;
 
-  constructor(private configService: ConfigService) {
+  constructor(
+    private configService: ConfigService,
+    private readonly redis: RedisService,
+  ) {
     this.useProxy = this.configService.get<string>("PCAT_USE_PROXY", "true") === "true";
     this.proxyHost = this.configService.get<string>("PCAT_PROXY_HOST", DI_HOST);
     this.proxyUser = this.configService.get<string>("PCAT_PROXY_USER", DI_DEFAULT_USER);
@@ -130,8 +156,16 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
       this.logger.error(`Failed to launch browser on init: ${(err as Error).message}`);
     }
 
-    // Start business hours scheduling
-    if (this.isBusinessHours()) {
+    // Try Redis first — if another container has a fresh slot, inherit it.
+    const hydrated = await this.hydrateFromRedis();
+    if (hydrated) {
+      this.pool.push(hydrated);
+      this.scheduleSlotRefresh(hydrated);
+      const ttl = hydrated.jwt.exp - Math.floor(Date.now() / 1000);
+      this.logger.log(
+        `JWT pool: slot hydrated from Redis, TTL: ${ttl}s, pool size: ${this.pool.length}`,
+      );
+    } else if (this.isBusinessHours()) {
       this.captureToPool().catch((err) => {
         this.logger.error(`Initial pool capture failed: ${(err as Error).message}`);
       });
@@ -185,6 +219,11 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
       );
     }
 
+    // The Redis-shared slot may point at a now-401'd JWT (e.g. IP rotated
+    // away on the proxy). Drop it so other containers re-capture instead of
+    // inheriting the same dead slot.
+    this.deleteFromRedis().catch(() => {});
+
     // Capture replacement in background (don't block the caller's retry)
     if (this.isBusinessHours() || this.pool.length === 0) {
       this.captureToPool().catch((err) => {
@@ -253,6 +292,18 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
 
         const jwt = await this.attemptCapture(siteUrl, port);
         if (jwt) {
+          // Validate the JWT+proxy combo against the real upstream before
+          // publishing the slot — see VALIDATION_VIN comment above.
+          const valid = await this.validateSlot(jwt, port);
+          if (!valid) {
+            this.logger.warn(
+              `JWT capture attempt ${attempt + 1}/${maxRetries} validation failed (port ${port}, ${new URL(siteUrl).hostname}), trying next site...`,
+            );
+            // Mark site as used anyway so we rotate away rather than retry it.
+            this.siteLastUsedAt.set(siteUrl, Date.now());
+            continue;
+          }
+
           const slot: JwtSlot = {
             jwt,
             proxyPort: port,
@@ -267,6 +318,9 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
           // Add to pool
           this.pool.push(slot);
 
+          // Publish to Redis so other containers can hydrate from this slot.
+          this.persistToRedis(slot).catch(() => {});
+
           // Schedule refresh if in business hours
           if (this.isBusinessHours()) {
             this.scheduleSlotRefresh(slot);
@@ -290,6 +344,81 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
     }
   }
 
+  /**
+   * Probe the just-captured token + proxy port with a single cheap upstream
+   * call (/car/info with a public demo VIN). Returns true iff the upstream
+   * accepts the combo with 2xx within VALIDATION_TIMEOUT_MS. A failure here
+   * (proxy dead, IP blocked, token 401/403) is far cheaper to absorb at
+   * capture time than to inherit when a real user clicks a category.
+   */
+  private async validateSlot(jwt: PcatJwtToken, port: number): Promise<boolean> {
+    const url = `${PCAT_API_BASE}/car/info?q=${VALIDATION_VIN}`;
+    const fetchOptions: RequestInit & { dispatcher?: any } = {
+      method: "GET",
+      headers: {
+        "x-api-key": jwt.raw,
+        "x-api-path": jwt.apiPath,
+        "x-gui-version": jwt.guiVersion,
+        "x-user-id": jwt.userId,
+        origin: jwt.origin,
+        referer: jwt.referer,
+        Accept: "application/json",
+        "User-Agent":
+          "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36",
+      },
+      signal: AbortSignal.timeout(VALIDATION_TIMEOUT_MS),
+    };
+    if (this.useProxy) {
+      const { ProxyAgent } = await import("undici");
+      fetchOptions.dispatcher = new ProxyAgent({
+        uri: `http://${this.proxyUser}:${this.proxyPass}@${this.proxyHost}:${port}`,
+        connect: { timeout: 6_000 },
+      });
+    }
+    try {
+      const r = await fetch(url, fetchOptions);
+      return r.ok;
+    } catch (err) {
+      this.logger.debug(`Slot validation threw: ${(err as Error).message}`);
+      return false;
+    }
+  }
+
+  // ─── Redis persistence (cross-container slot sharing) ─────
+
+  /**
+   * Read the shared slot from Redis if one exists and still has >REDIS_SAFETY_BUFFER_S
+   * of life. Returns a fresh JwtSlot ready to push to the in-memory pool;
+   * refreshTimer is null (scheduler arms it on insert).
+   */
+  private async hydrateFromRedis(): Promise<JwtSlot | null> {
+    try {
+      const raw = await this.redis.getJson<Omit<JwtSlot, "refreshTimer">>(REDIS_SLOT_KEY);
+      if (!raw) return null;
+      const now = Math.floor(Date.now() / 1000);
+      if (raw.jwt.exp - now < REDIS_SAFETY_BUFFER_S) return null;
+      return { ...raw, refreshTimer: null };
+    } catch {
+      return null;
+    }
+  }
+
+  /**
+   * Publish a freshly-captured slot to Redis under REDIS_SLOT_KEY, TTL aligned
+   * with the JWT's own expiry (minus a safety buffer). refreshTimer is stripped
+   * because it's a non-serializable Node Timer handle.
+   */
+  private async persistToRedis(slot: JwtSlot): Promise<void> {
+    const ttl = slot.jwt.exp - Math.floor(Date.now() / 1000) - REDIS_SAFETY_BUFFER_S;
+    if (ttl <= 0) return;
+    const { refreshTimer: _ignored, ...persisted } = slot;
+    await this.redis.setJson(REDIS_SLOT_KEY, persisted, ttl).catch(() => {});
+  }
+
+  private async deleteFromRedis(): Promise<void> {
+    await this.redis.del(REDIS_SLOT_KEY).catch(() => {});
+  }
+
   // ─── Timer-based refresh per slot ─────────────────────────
 
   private scheduleSlotRefresh(slot: JwtSlot): void {