diff --git a/apps/api/src/integrations/carcatonline/carcatonline.pacing.spec.ts b/apps/api/src/integrations/carcatonline/carcatonline.pacing.spec.ts index 781da08..49ecab5 100644 --- 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(); diff --git a/apps/api/src/integrations/carcatonline/carcatonline.pacing.ts b/apps/api/src/integrations/carcatonline/carcatonline.pacing.ts index 5c52923..d14a5ed 100644 --- 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 { + return Number((await this.redis.get(hourlyCallsKey(this.now()))) ?? 0); + } + async beforeCall(): Promise { 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); } } diff --git a/apps/api/src/integrations/carcatonline/carcatonline.tree.ts b/apps/api/src/integrations/carcatonline/carcatonline.tree.ts index 93dcadb..cbf8c9b 100644 --- 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; /** 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; + set(key: string, groups: CarcatGroup[]): Promise; + }; } 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); diff --git a/apps/api/src/jobs/processors/carcatonline-backfill.processor.spec.ts b/apps/api/src/jobs/processors/carcatonline-backfill.processor.spec.ts index 312205d..2d8ac73 100644 --- 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 = { + 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({}); diff --git a/apps/api/src/jobs/processors/carcatonline-backfill.processor.ts b/apps/api/src/jobs/processors/carcatonline-backfill.processor.ts index 02905ee..ae52d37 100644 --- 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, diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index 66ec938..0ae546e 100644 --- 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} diff --git a/packages/config/src/index.ts b/packages/config/src/index.ts index 3bd8d8d..c9de884 100644 --- 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),