2026-09-26 16:25:20 +03:00
7 changed files with 203 additions and 7 deletions
--- a/apps/api/src/integrations/carcatonline/carcatonline.pacing.spec.ts
+++ b/apps/api/src/integrations/carcatonline/carcatonline.pacing.spec.ts
@@ -2,12 +2,15 @@ import { describe, expect, it, vi } from "vitest";
 import {
  CARCAT_KEYS,
  CarcatonlineBudgetError,
+  CarcatonlineHourlyCapError,
  CarcatonlineLockedError,
  CarcatonlineThrottle,
  type RedisLike,
  carcatConfigFromEnv,
  dailyCallsKey,
+  hourlyCallsKey,
  isWithinWindow,
+  msUntilNextHour,
  msUntilWindowOpens,
  redisTokenStore,
 } from "./carcatonline.pacing";
@@ -80,6 +83,7 @@ describe("CarcatonlineThrottle", () => {
  const cfg = {
    minIntervalMs: 2000,
    dailyCallCap: 3,
+    hourlyCallCap: 1000,
    windowStartHour: 20,
    windowEndHour: 7,
    lockoutSeconds: 2400,
@@ -120,6 +124,42 @@ describe("CarcatonlineThrottle", () => {
  });
 });

+describe("hourly request cap", () => {
+  const base = {
+    minIntervalMs: 2000,
+    dailyCallCap: 1000,
+    hourlyCallCap: 2,
+    windowStartHour: 20,
+    windowEndHour: 7,
+    lockoutSeconds: 2400,
+  };
+
+  it("keys calls by Istanbul clock hour and computes the wait to the next hour", () => {
+    expect(hourlyCallsKey(at("2026-09-26T19:40:00Z"))).toBe(
+      "carcatonline:calls-hour:2026-09-26T22",
+    );
+    expect(msUntilNextHour(at("2026-09-26T19:40:00Z"))).toBe(20 * 60_000 + 30_000);
+  });
+
+  it("stops before the upstream hourly limit and reports the wait", async () => {
+    const redis = fakeRedis();
+    const t = at("2026-09-26T19:40:00Z");
+    const throttle = new CarcatonlineThrottle(
+      redis,
+      base,
+      () => t,
+      async () => {
+        redis.store.delete(CARCAT_KEYS.lastCall);
+      },
+    );
+    await throttle.beforeCall();
+    redis.store.delete(CARCAT_KEYS.lastCall);
+    await throttle.beforeCall();
+    expect(await throttle.callsThisHour()).toBe(2);
+    await expect(throttle.beforeCall()).rejects.toBeInstanceOf(CarcatonlineHourlyCapError);
+  });
+});
+
 describe("redisTokenStore", () => {
  it("stores the token with a TTL derived from its expiry", async () => {
    const redis = fakeRedis();
--- a/apps/api/src/integrations/carcatonline/carcatonline.pacing.ts
+++ b/apps/api/src/integrations/carcatonline/carcatonline.pacing.ts
@@ -25,12 +25,16 @@ export const CARCAT_KEYS = {
  lockout: "carcatonline:lockout",
  lastCall: "carcatonline:last-call",
  callsPrefix: "carcatonline:calls:",
+  hourlyPrefix: "carcatonline:calls-hour:",
+  groupsPrefix: "carcatonline:groups:",
  modelsPrefix: "carcatonline:models:",
 } as const;

 export interface CarcatConfig {
  minIntervalMs: number;
  dailyCallCap: number;
+  /** Upstream "Customer Hourly Request Limit" (measured ~100/h) — stay under it instead of eating a lockout. */
+  hourlyCallCap: number;
  windowStartHour: number;
  windowEndHour: number;
  lockoutSeconds: number;
@@ -44,6 +48,7 @@ export function carcatConfigFromEnv(env: NodeJS.ProcessEnv = process.env): Carca
  return {
    minIntervalMs: num(env.CARCATONLINE_MIN_INTERVAL_MS, 7000),
    dailyCallCap: num(env.CARCATONLINE_DAILY_CALL_CAP, 15000),
+    hourlyCallCap: num(env.CARCATONLINE_HOURLY_CALL_CAP, 90),
    windowStartHour: num(env.CARCATONLINE_WINDOW_START, 20),
    windowEndHour: num(env.CARCATONLINE_WINDOW_END, 7),
    lockoutSeconds: num(env.CARCATONLINE_LOCKOUT_SECONDS, 40 * 60),
@@ -84,6 +89,18 @@ export function dailyCallsKey(date: Date): string {
  return `${CARCAT_KEYS.callsPrefix}${istanbulParts(date).day}`;
 }

+/** Clock-hour bucket in Istanbul time ("2026-09-26T21"). */
+export function hourlyCallsKey(date: Date): string {
+  const { day, hour } = istanbulParts(date);
+  return `${CARCAT_KEYS.hourlyPrefix}${day}T${String(hour).padStart(2, "0")}`;
+}
+
+/** Milliseconds until the next clock hour starts (plus a small margin). */
+export function msUntilNextHour(date: Date): number {
+  const ms = date.getTime();
+  return 3600_000 - (ms % 3600_000) + 30_000;
+}
+
 export class CarcatonlineLockedError extends Error {
  constructor(readonly retryAfterMs: number) {
    super(`carcatonline is locked out for ${Math.round(retryAfterMs / 1000)}s`);
@@ -101,6 +118,19 @@ export class CarcatonlineBudgetError extends Error {
  }
 }

+export class CarcatonlineHourlyCapError extends Error {
+  constructor(
+    readonly used: number,
+    readonly cap: number,
+    readonly retryAfterMs: number,
+  ) {
+    super(
+      `carcatonline hourly call cap reached (${used}/${cap}); next hour in ${Math.round(retryAfterMs / 60000)} min`,
+    );
+    this.name = "CarcatonlineHourlyCapError";
+  }
+}
+
 export class CarcatonlineWindowClosedError extends Error {
  constructor(readonly retryAfterMs: number) {
    super(`carcatonline bulk window closed; reopens in ${Math.round(retryAfterMs / 60000)} min`);
@@ -159,12 +189,24 @@ export class CarcatonlineThrottle {
    return Number((await this.redis.get(dailyCallsKey(this.now()))) ?? 0);
  }

+  async callsThisHour(): Promise<number> {
+    return Number((await this.redis.get(hourlyCallsKey(this.now()))) ?? 0);
+  }
+
  async beforeCall(): Promise<void> {
    const locked = await this.lockoutRemainingMs();
    if (locked > 0) throw new CarcatonlineLockedError(locked);
    const used = await this.callsToday();
    if (used >= this.cfg.dailyCallCap)
      throw new CarcatonlineBudgetError(used, this.cfg.dailyCallCap);
+    const usedHour = await this.callsThisHour();
+    if (usedHour >= this.cfg.hourlyCallCap) {
+      throw new CarcatonlineHourlyCapError(
+        usedHour,
+        this.cfg.hourlyCallCap,
+        msUntilNextHour(this.now()),
+      );
+    }
    // Shared min-interval slot: SET NX PX; spin (bounded) until acquired.
    for (let i = 0; i < 50; i += 1) {
      const ok = await this.redis.set(
@@ -180,5 +222,8 @@ export class CarcatonlineThrottle {
    const key = dailyCallsKey(this.now());
    const n = await this.redis.incr(key);
    if (n === 1) await this.redis.expire(key, 3 * 24 * 3600);
+    const hourKey = hourlyCallsKey(this.now());
+    const h = await this.redis.incr(hourKey);
+    if (h === 1) await this.redis.expire(hourKey, 2 * 3600);
  }
 }
--- a/apps/api/src/integrations/carcatonline/carcatonline.tree.ts
+++ b/apps/api/src/integrations/carcatonline/carcatonline.tree.ts
@@ -19,6 +19,11 @@ export interface CrawlHooks {
  beforeCall(): Promise<void>;
  /** Throw to abort (e.g. bulk window closed). */
  checkAbort?(): void;
+  /** Optional response cache so a re-run after a quota stop resumes without re-spending calls. */
+  cache?: {
+    get(key: string): Promise<CarcatGroup[] | null>;
+    set(key: string, groups: CarcatGroup[]): Promise<void>;
+  };
 }

 export interface SelectedParameter {
@@ -132,9 +137,14 @@ export async function crawlGroupTree(
  while (queue.length > 0 && nodes.length < maxNodes) {
    const { groupId, depth } = queue.shift() as { groupId: string | null; depth: number };
    hooks.checkAbort?.();
-    await hooks.beforeCall();
-    const groups = await client.groups(catalogId, carId, groupId ?? undefined);
-    calls += 1;
+    const cacheKey = `${catalogId}:${carId}:${groupId ?? "root"}`;
+    let groups = (await hooks.cache?.get(cacheKey)) ?? null;
+    if (!groups) {
+      await hooks.beforeCall();
+      groups = await client.groups(catalogId, carId, groupId ?? undefined);
+      calls += 1;
+      await hooks.cache?.set(cacheKey, groups);
+    }
    for (const g of groups) {
      if (!g?.id || seen.has(g.id)) continue;
      seen.add(g.id);
--- a/apps/api/src/jobs/processors/carcatonline-backfill.processor.spec.ts
+++ b/apps/api/src/jobs/processors/carcatonline-backfill.processor.spec.ts
@@ -11,6 +11,7 @@ const DAY = new Date("2026-09-26T09:00:00Z"); // 12:00 Istanbul
 const CFG = {
  minIntervalMs: 0,
  dailyCallCap: 1000,
+  hourlyCallCap: 1000,
  windowStartHour: 20,
  windowEndHour: 7,
  lockoutSeconds: 2400,
@@ -255,6 +256,58 @@ describe("processCarcatonlineBackfill", () => {
    expect(j.moveToDelayed.mock.calls[0][0]).toBeGreaterThan(NIGHT.getTime() + 2400 * 1000);
  });

+  it("vehicle: stops at the hourly cap and re-schedules to the next clock hour", async () => {
+    const { db } = fakeDb([[cv], []]);
+    const redis = fakeRedis();
+    const client = fakeClient({
+      root: [{ id: "g1", name: "Engine", hasSubgroups: false, hasParts: true }],
+    });
+    const j = job(CARCAT_JOB.vehicle, { catalogVehicleId: "cv-1" });
+    const deps = { ...baseDeps(db, redis, client, NIGHT), config: { ...CFG, hourlyCallCap: 1 } };
+    await expect(processCarcatonlineBackfill(j, "tok", deps)).rejects.toBeInstanceOf(DelayedError);
+    // models (1 call) fits the cap; the cascade's first call trips it → delayed to the next hour + margin
+    const [ts] = j.moveToDelayed.mock.calls[0];
+    expect(ts).toBe(Date.parse("2026-09-26T20:00:00Z") + 30_000);
+  });
+
+  it("vehicle: a re-run resumes from the matched car and cached group responses without new calls", async () => {
+    const tree: Record<string, CarcatGroup[]> = {
+      root: [{ id: "g1", name: "Engine", hasSubgroups: true, hasParts: false }],
+      g1: [{ id: "g1a", name: "Oil", hasSubgroups: false, hasParts: true }],
+    };
+    const matched = {
+      ...cv,
+      metadata: {
+        carcatonline: {
+          status: "matched",
+          catalogId: "pl_renault",
+          modelId: "m-kadjar",
+          carId: "car-1",
+          parameters: [{ key: "engine", value: "1.5", idx: "e15" }],
+          at: "2026-09-26T19:00:00Z",
+        },
+      },
+    };
+    const { db, inserts } = fakeDb([[matched], []]);
+    const redis = fakeRedis();
+    redis.store.set(
+      `${CARCAT_KEYS.modelsPrefix}pl_renault`,
+      JSON.stringify([{ id: "m-kadjar", name: "KADJAR" }]),
+    );
+    redis.store.set(`${CARCAT_KEYS.groupsPrefix}pl_renault:car-1:root`, JSON.stringify(tree.root));
+    redis.store.set(`${CARCAT_KEYS.groupsPrefix}pl_renault:car-1:g1`, JSON.stringify(tree.g1));
+    const client = fakeClient({});
+    const r = await processCarcatonlineBackfill(
+      job(CARCAT_JOB.vehicle, { catalogVehicleId: "cv-1" }),
+      "tok",
+      baseDeps(db, redis, client, NIGHT),
+    );
+    expect(r).toMatchObject({ status: "done", nodes: 2, calls: 0 });
+    expect(client.carsParameters).not.toHaveBeenCalled();
+    expect(client.groups).not.toHaveBeenCalled();
+    expect(inserts.flat().map((x) => x.name)).toEqual(["Engine", "Oil"]);
+  });
+
  it("vehicle: skips catalogs that already have categories", async () => {
    const { db } = fakeDb([[cv], [{ one: 1 }]]);
    const client = fakeClient({});
--- a/apps/api/src/jobs/processors/carcatonline-backfill.processor.ts
+++ b/apps/api/src/jobs/processors/carcatonline-backfill.processor.ts
@@ -3,6 +3,7 @@ import { and, eq, inArray, notExists, sql } from "drizzle-orm";
 import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
 import { catalogVehicles, categories } from "../../database/schema/core";
 import {
+  type CarcatGroup,
  type CarcatModel,
  CarcatonlineClient,
  CarcatonlineRateLimitError,
@@ -16,6 +17,7 @@ import {
  CARCAT_KEYS,
  type CarcatConfig,
  CarcatonlineBudgetError,
+  CarcatonlineHourlyCapError,
  CarcatonlineLockedError,
  CarcatonlineThrottle,
  CarcatonlineWindowClosedError,
@@ -131,8 +133,8 @@ async function scan(
            .from(categories)
            .where(eq(categories.catalogVehicleId, catalogVehicles.id)),
        ),
-        // never tried, or tried more than RETRY_AFTER_DAYS ago
-        sql`(${catalogVehicles.metadata}->'carcatonline'->>'at') IS NULL OR (${catalogVehicles.metadata}->'carcatonline'->>'at') < ${cutoff}`,
+        // never tried, mid-way (matched → resume), or a terminal outcome older than RETRY_AFTER_DAYS
+        sql`(${catalogVehicles.metadata}->'carcatonline'->>'at') IS NULL OR (${catalogVehicles.metadata}->'carcatonline'->>'status') = 'matched' OR (${catalogVehicles.metadata}->'carcatonline'->>'at') < ${cutoff}`,
      ),
    )
    .orderBy(
@@ -195,6 +197,25 @@ async function fillVehicle(
  const client = deps.client();
  const hooks = {
    beforeCall: () => throttle.beforeCall(),
+    cache: {
+      get: async (key: string) => {
+        const raw = await deps.redis.get(`${CARCAT_KEYS.groupsPrefix}${key}`);
+        if (!raw) return null;
+        try {
+          return JSON.parse(raw) as CarcatGroup[];
+        } catch {
+          return null;
+        }
+      },
+      set: async (key: string, groups: CarcatGroup[]) => {
+        await deps.redis.set(
+          `${CARCAT_KEYS.groupsPrefix}${key}`,
+          JSON.stringify(groups),
+          "EX",
+          24 * 3600,
+        );
+      },
+    },
    checkAbort: () => {
      if (!force && !isWithinWindow(now(), cfg.windowStartHour, cfg.windowEndHour)) {
        throw new CarcatonlineWindowClosedError(
@@ -254,8 +275,29 @@ async function fillVehicle(
      return { status: "unmatched", reason: "model" };
    }

-    // 2. representative car
-    const car = await resolveRepresentativeCar(client, hooks, match.catalogId, match.modelId);
+    // 2. representative car — reuse the one resolved by an earlier (quota-stopped) run
+    const prev = (cv.metadata as { carcatonline?: CarcatMetadata } | null)?.carcatonline;
+    const resumable =
+      prev?.status === "matched" &&
+      prev.catalogId === match.catalogId &&
+      prev.modelId === match.modelId &&
+      prev.carId
+        ? {
+            carId: prev.carId,
+            parameters: (prev.parameters ?? []).map((p) => ({ ...p, name: p.key })),
+          }
+        : null;
+    const car =
+      resumable ?? (await resolveRepresentativeCar(client, hooks, match.catalogId, match.modelId));
+    if (car && !resumable) {
+      await setMeta({
+        status: "matched",
+        ...match,
+        carId: car.carId,
+        parameters: car.parameters.map((p) => ({ key: p.key, value: p.value, idx: p.idx })),
+        at: now().toISOString(),
+      });
+    }
    if (!car) {
      await setMeta({ status: "no_car", ...match, at: now().toISOString() });
      return { status: "no_car" };
@@ -305,6 +347,9 @@ async function fillVehicle(
    }
    if (err instanceof CarcatonlineLockedError)
      await reschedule(err.retryAfterMs + 30_000, "locked out");
+    if (err instanceof CarcatonlineHourlyCapError) {
+      await reschedule(err.retryAfterMs, err.message);
+    }
    if (err instanceof CarcatonlineBudgetError) {
      await reschedule(
        msUntilWindowOpens(now(), cfg.windowStartHour, cfg.windowEndHour) + 3600_000,
--- a/docker-compose.coolify.yml
+++ b/docker-compose.coolify.yml
@@ -84,6 +84,7 @@ services:
      - CARCATONLINE_EMAIL=${CARCATONLINE_EMAIL:-}
      - CARCATONLINE_PASSWORD=${CARCATONLINE_PASSWORD:-}
      - CARCATONLINE_DAILY_CALL_CAP=${CARCATONLINE_DAILY_CALL_CAP:-15000}
+      - CARCATONLINE_HOURLY_CALL_CAP=${CARCATONLINE_HOURLY_CALL_CAP:-90}
      - CARCATONLINE_WINDOW_START=${CARCATONLINE_WINDOW_START:-20}
      - CARCATONLINE_WINDOW_END=${CARCATONLINE_WINDOW_END:-7}
      - CARCATONLINE_MIN_INTERVAL_MS=${CARCATONLINE_MIN_INTERVAL_MS:-7000}
@@ -260,6 +261,7 @@ services:
      - CARCATONLINE_EMAIL=${CARCATONLINE_EMAIL:-}
      - CARCATONLINE_PASSWORD=${CARCATONLINE_PASSWORD:-}
      - CARCATONLINE_DAILY_CALL_CAP=${CARCATONLINE_DAILY_CALL_CAP:-15000}
+      - CARCATONLINE_HOURLY_CALL_CAP=${CARCATONLINE_HOURLY_CALL_CAP:-90}
      - CARCATONLINE_WINDOW_START=${CARCATONLINE_WINDOW_START:-20}
      - CARCATONLINE_WINDOW_END=${CARCATONLINE_WINDOW_END:-7}
      - CARCATONLINE_MIN_INTERVAL_MS=${CARCATONLINE_MIN_INTERVAL_MS:-7000}
--- a/packages/config/src/index.ts
+++ b/packages/config/src/index.ts
@@ -106,6 +106,7 @@ export const envSchema = z.object({
  CARCATONLINE_EMAIL: z.string().optional(),
  CARCATONLINE_PASSWORD: z.string().optional(),
  CARCATONLINE_DAILY_CALL_CAP: z.coerce.number().int().min(0).default(15000),
+  CARCATONLINE_HOURLY_CALL_CAP: z.coerce.number().int().min(0).default(90),
  CARCATONLINE_WINDOW_START: z.coerce.number().int().min(0).max(23).default(20),
  CARCATONLINE_WINDOW_END: z.coerce.number().int().min(0).max(23).default(7),
  CARCATONLINE_MIN_INTERVAL_MS: z.coerce.number().int().min(500).default(7000),