diff --git a/apps/api/src/integrations/pl24/pl24-auth.service.spec.ts b/apps/api/src/integrations/pl24/pl24-auth.service.spec.ts index 4f9d378..1d1d037 100644 --- a/apps/api/src/integrations/pl24/pl24-auth.service.spec.ts +++ b/apps/api/src/integrations/pl24/pl24-auth.service.spec.ts @@ -37,6 +37,19 @@ const makeRedis = (opts: { lockTaken?: boolean } = {}): RedisStub => { }; }; +/** Bütçe servisi stub'ı: sayar ama engellemez, telemetriyi yutar. */ +const makeBudget = () => ({ + consumed: [] as string[], + recorded: [] as unknown[], + consume: async function (kind: string) { + this.consumed.push(kind); + }, + record: function (e: unknown) { + this.recorded.push(e); + }, + spentToday: async () => 0, +}); + const makeService = (cfgOverrides: Record = {}, redis = makeRedis()) => { const cfg: Record = { "pl24.companyCode": "tr-000000", @@ -49,7 +62,12 @@ const makeService = (cfgOverrides: Record = {}, redis = makeRedi ...cfgOverrides, }; const configService = { get: (k: string, d?: unknown) => cfg[k] ?? d } as never; - return { svc: new PL24AuthService(configService, redis as never), redis }; + const budget = makeBudget(); + return { + svc: new PL24AuthService(configService, redis as never, budget as never), + redis, + budget, + }; }; type Exposed = { diff --git a/apps/api/src/integrations/pl24/pl24-auth.service.ts b/apps/api/src/integrations/pl24/pl24-auth.service.ts index a5222a2..2a2c455 100644 --- a/apps/api/src/integrations/pl24/pl24-auth.service.ts +++ b/apps/api/src/integrations/pl24/pl24-auth.service.ts @@ -25,6 +25,7 @@ import { Injectable, Logger, type OnModuleInit, UnauthorizedException } from "@nestjs/common"; import { ConfigService } from "@nestjs/config"; import { RedisService } from "../../redis/redis.service"; +import { PL24BudgetService } from "./pl24-budget.service"; import { PL24_DEFAULTS, PL24_ENDPOINTS, PL24_USER_AGENT } from "./pl24.constants"; import { PL24AuthorizeRequest, @@ -96,6 +97,7 @@ export class PL24AuthService implements OnModuleInit { constructor( private configService: ConfigService, private readonly redis: RedisService, + private readonly budget: PL24BudgetService, ) { this.baseUrl = this.configService.get("pl24.baseUrl", "https://www.partslink24.com"); this.companyCode = this.configService.get("pl24.companyCode", ""); @@ -405,7 +407,29 @@ export class PL24AuthService implements OnModuleInit { if (dispatcher) fetchOpts.dispatcher = dispatcher; const url = `${this.baseUrl}${legacy ? PL24_ENDPOINTS.LOGIN_LEGACY : PL24_ENDPOINTS.LOGIN}`; - const response = await fetch(url, fetchOpts); + await this.budget.consume("auth"); + const startedAt = Date.now(); + const response = await fetch(url, fetchOpts).catch((err) => { + this.budget.record({ + kind: "auth", + url, + proxied: Boolean(dispatcher), + account, + success: false, + startedAt, + error: err, + }); + throw err; + }); + this.budget.record({ + kind: "auth", + url, + proxied: Boolean(dispatcher), + account, + statusCode: response.status, + success: response.ok, + startedAt, + }); const cookieToken = this.extractSessionToken(response); // 400/412 carry an RFC7807 problem body naming the reason. @@ -538,7 +562,30 @@ export class PL24AuthService implements OnModuleInit { }; if (dispatcher) fetchOpts.dispatcher = dispatcher; - const response = await fetch(`${this.baseUrl}${PL24_ENDPOINTS.AUTHORIZE}`, fetchOpts); + const authorizeUrl = `${this.baseUrl}${PL24_ENDPOINTS.AUTHORIZE}`; + await this.budget.consume("auth"); + const startedAt = Date.now(); + const response = await fetch(authorizeUrl, fetchOpts).catch((err) => { + this.budget.record({ + kind: "auth", + url: authorizeUrl, + proxied: Boolean(dispatcher), + account, + success: false, + startedAt, + error: err, + }); + throw err; + }); + this.budget.record({ + kind: "auth", + url: authorizeUrl, + proxied: Boolean(dispatcher), + account, + statusCode: response.status, + success: response.ok, + startedAt, + }); // A dead session shows up here, not at login time. if (response.status === 401 || response.status === 403) { diff --git a/apps/api/src/integrations/pl24/pl24-budget.service.spec.ts b/apps/api/src/integrations/pl24/pl24-budget.service.spec.ts new file mode 100644 index 0000000..3794149 --- /dev/null +++ b/apps/api/src/integrations/pl24/pl24-budget.service.spec.ts @@ -0,0 +1,141 @@ +import { describe, expect, it, vi } from "vitest"; +import { backfillContext } from "../../jobs/prefetch-context"; +import { PL24BudgetExceededError, PL24BudgetService } from "./pl24-budget.service"; + +// Günlük PL24 HTTP tavanı + kullanıcı rezervi + proxy_logs telemetrisi. +// Redis ve telemetri stub'lanır; ağ yok. + +const makeRedis = (opts: { fail?: boolean } = {}) => { + const counters = new Map(); + return { + counters, + incr: async (k: string) => { + if (opts.fail) throw new Error("redis down"); + const n = (counters.get(k) ?? 0) + 1; + counters.set(k, n); + return n; + }, + expire: async () => {}, + get: async (k: string) => (counters.has(k) ? String(counters.get(k)) : null), + }; +}; + +const makeTelemetry = () => { + const events: Record[] = []; + return { events, record: (e: Record) => events.push(e) }; +}; + +const make = (env: Record = {}, redis = makeRedis()) => { + for (const [k, v] of Object.entries(env)) vi.stubEnv(k, v); + const telemetry = makeTelemetry(); + return { svc: new PL24BudgetService(redis as never, telemetry as never), redis, telemetry }; +}; + +const inBackfill = (fn: () => Promise) => backfillContext.run(true, fn); + +describe("PL24BudgetService — günlük tavan", () => { + it("tavan altında çağrılara izin verir", async () => { + const { svc } = make({ PL24_HTTP_DAILY_MAX: "5" }); + for (let i = 0; i < 5; i++) await svc.consume("catalog"); + expect(await svc.spentToday()).toBe(5); + }); + + it("tavan aşılınca kullanıcı çağrısını da reddeder", async () => { + const { svc } = make({ PL24_HTTP_DAILY_MAX: "3" }); + for (let i = 0; i < 3; i++) await svc.consume("catalog"); + await expect(svc.consume("catalog")).rejects.toBeInstanceOf(PL24BudgetExceededError); + }); + + it("backfill kullanıcı rezervine dokunamaz, kullanıcı trafiği devam eder", async () => { + // tavan 10, rezerv %40 → backfill 6'da durur, kullanıcı 10'a kadar sürer + const { svc } = make({ PL24_HTTP_DAILY_MAX: "10", PL24_HTTP_USER_RESERVE: "0.4" }); + + for (let i = 0; i < 6; i++) await inBackfill(() => svc.consume("catalog")); + await expect(inBackfill(() => svc.consume("catalog"))).rejects.toBeInstanceOf( + PL24BudgetExceededError, + ); + + // Aynı gün, aynı sayaç: kullanıcı hâlâ geçebilmeli (7., 8. … 10. istek) + await expect(svc.consume("catalog")).resolves.toBeUndefined(); + await expect(svc.consume("catalog")).resolves.toBeUndefined(); + }); + + it("reddedilen çağrı yalnız reddedildiği yerde sayılır (ağ isteği yok)", async () => { + const { svc, telemetry } = make({ PL24_HTTP_DAILY_MAX: "1" }); + await svc.consume("catalog"); + await expect(svc.consume("catalog")).rejects.toBeInstanceOf(PL24BudgetExceededError); + expect(telemetry.events).toHaveLength(0); + }); + + it("Redis düşükse PL24'ü bloklamaz (fail-open)", async () => { + const { svc } = make({ PL24_HTTP_DAILY_MAX: "1" }, makeRedis({ fail: true })); + await expect(svc.consume("catalog")).resolves.toBeUndefined(); + await expect(svc.consume("catalog")).resolves.toBeUndefined(); + }); +}); + +describe("PL24BudgetService — telemetri", () => { + it("katalog çağrısını proxy_logs biçiminde kaydeder", () => { + const { svc, telemetry } = make(); + svc.record({ + kind: "catalog", + url: "https://www.partslink24.com/p5psa/extern/group/vin/scope?vin=X", + proxied: true, + account: "de", + statusCode: 200, + success: true, + startedAt: Date.now() - 120, + }); + const e = telemetry.events[0]; + expect(e.service).toBe("pl24_http"); + expect(e.provider).toBe("dataimpulse"); + expect(e.targetHost).toBe("www.partslink24.com"); + expect(e.success).toBe(true); + expect(e.errorKind).toBeNull(); + }); + + it("login çağrısını pl24_auth olarak ayırır", () => { + const { svc, telemetry } = make(); + svc.record({ + kind: "auth", + url: "https://www.partslink24.com/auth/ext/api/1.1/login", + proxied: false, + account: "tr", + statusCode: 400, + success: false, + startedAt: Date.now(), + }); + expect(telemetry.events[0].service).toBe("pl24_auth"); + expect(telemetry.events[0].provider).toBe("none"); + }); + + it("403/429'u ban sinyali olarak işaretler", () => { + const { svc, telemetry } = make(); + svc.record({ + kind: "catalog", + url: "https://www.partslink24.com/x", + proxied: true, + account: "de", + statusCode: 403, + success: false, + startedAt: Date.now(), + }); + expect(telemetry.events[0].errorKind).toBe("banned"); + }); + + it("taşıma hatasını sınıflandırır", () => { + const { svc, telemetry } = make(); + const err = Object.assign(new Error("fetch failed"), { name: "TimeoutError" }); + svc.record({ + kind: "catalog", + url: "https://www.partslink24.com/x", + proxied: true, + account: "de", + success: false, + startedAt: Date.now(), + error: err, + }); + expect(telemetry.events[0].errorKind).toBe("timeout"); + expect(telemetry.events[0].statusCode).toBeNull(); + }); +}); diff --git a/apps/api/src/integrations/pl24/pl24-budget.service.ts b/apps/api/src/integrations/pl24/pl24-budget.service.ts new file mode 100644 index 0000000..714b1cb --- /dev/null +++ b/apps/api/src/integrations/pl24/pl24-budget.service.ts @@ -0,0 +1,137 @@ +import { Injectable, Logger } from "@nestjs/common"; +import { isBackfillContext } from "../../jobs/prefetch-context"; +import { RedisService } from "../../redis/redis.service"; +import { + ProxyTelemetryService, + classifyTransportError, + isBanStatus, +} from "../proxy-telemetry/proxy-telemetry.service"; + +/** + * PL24 HTTP budget + telemetry. + * + * WHY (see /home/s/ss/plv2.md §2.2, findings consumers_jobs-03/07): + * Both PL24 account bans (tr 2026-07-24, de 2026-09-04) followed days that wrote + * 5-7k new category rows — roughly 10k+ upstream calls/day — while real user + * decodes never exceeded 14/day. The existing guards are all job-level + * (jobs/min, jobs/day); nothing counted actual HTTP requests, and PL24 calls + * were invisible in `proxy_logs` (only pcat/emex were logged), so the volume was + * only reconstructable after the fact from DB row counts. + * + * This service is the one choke point every PL24 upstream call passes through: + * - counts requests per UTC day in Redis and refuses new ones past the cap, + * - reserves a share of that cap for real users so a runaway backfill can + * never starve a paying customer's decode, + * - records every attempt into `proxy_logs` (service `pl24_http`/`pl24_auth`). + * + * The cap is deliberately low to start (PL24_HTTP_DAILY_MAX, default 1200); + * raise it only with telemetry in hand. + */ +@Injectable() +export class PL24BudgetService { + private readonly logger = new Logger(PL24BudgetService.name); + /** Log "budget exhausted" once per day per lane instead of on every call. */ + private warnedFor = new Set(); + + constructor( + private readonly redis: RedisService, + private readonly telemetry: ProxyTelemetryService, + ) {} + + private get dailyMax(): number { + const raw = Number(process.env.PL24_HTTP_DAILY_MAX); + return Number.isFinite(raw) && raw > 0 ? raw : 1200; + } + + /** Fraction of the daily cap that only user-facing calls may spend. */ + private get userReserve(): number { + const raw = Number(process.env.PL24_HTTP_USER_RESERVE); + return Number.isFinite(raw) && raw > 0 && raw < 1 ? raw : 0.4; + } + + private dayKey(): string { + return `pl24:http:${new Date().toISOString().slice(0, 10)}`; + } + + /** + * Count one upstream call and decide whether it may proceed. + * Backfill stops at (1 - userReserve) of the cap; user traffic gets the rest. + */ + async consume(kind: "auth" | "catalog"): Promise { + const backfill = isBackfillContext(); + let spent: number; + try { + const key = this.dayKey(); + spent = await this.redis.incr(key); + if (spent === 1) await this.redis.expire(key, 8 * 86_400); + } catch { + return; // Redis down → never block PL24 on telemetry + } + + const max = this.dailyMax; + const limit = backfill ? Math.floor(max * (1 - this.userReserve)) : max; + if (spent > limit) { + const lane = backfill ? "backfill" : "user"; + if (!this.warnedFor.has(`${lane}:${this.dayKey()}`)) { + this.warnedFor.add(`${lane}:${this.dayKey()}`); + this.logger.warn( + `PL24 daily HTTP budget exhausted for the ${lane} lane (${spent}/${limit}, cap ${max}, kind=${kind}) — refusing further calls today`, + ); + } + throw new PL24BudgetExceededError(lane, spent, limit); + } + } + + /** Current spend (for /health and the Süper Panel). */ + async spentToday(): Promise { + try { + return Number((await this.redis.get(this.dayKey())) ?? 0); + } catch { + return 0; + } + } + + record(opts: { + kind: "auth" | "catalog"; + url: string; + proxied: boolean; + account: string; + statusCode?: number | null; + success: boolean; + startedAt: number; + error?: unknown; + }): void { + let targetHost: string | null = null; + try { + targetHost = new URL(opts.url).hostname; + } catch { + /* keep null */ + } + this.telemetry.record({ + service: opts.kind === "auth" ? "pl24_auth" : "pl24_http", + provider: opts.proxied ? "dataimpulse" : "none", + sessionKey: opts.account, + targetHost, + statusCode: opts.statusCode ?? null, + errorKind: opts.error + ? classifyTransportError(opts.error) + : isBanStatus(opts.statusCode) + ? "banned" + : null, + success: opts.success, + durationMs: Date.now() - opts.startedAt, + }); + } +} + +/** Thrown when the daily PL24 HTTP budget is used up for this lane. */ +export class PL24BudgetExceededError extends Error { + constructor( + readonly lane: "user" | "backfill", + readonly spent: number, + readonly limit: number, + ) { + super(`PL24 daily HTTP budget exhausted (${lane} lane: ${spent}/${limit})`); + this.name = "PL24BudgetExceededError"; + } +} diff --git a/apps/api/src/integrations/pl24/pl24.module.ts b/apps/api/src/integrations/pl24/pl24.module.ts index 3a07b2a..f28b2e3 100644 --- a/apps/api/src/integrations/pl24/pl24.module.ts +++ b/apps/api/src/integrations/pl24/pl24.module.ts @@ -1,5 +1,7 @@ import { Module } from "@nestjs/common"; +import { ProxyTelemetryModule } from "../proxy-telemetry/proxy-telemetry.module"; import { PL24AuthService } from "./pl24-auth.service"; +import { PL24BudgetService } from "./pl24-budget.service"; import { PL24FordLegacyService } from "./pl24-ford-legacy.service"; import { PL24FordService } from "./pl24-ford.service"; import { PL24HyundaiKiaService } from "./pl24-hyundai-kia.service"; @@ -11,6 +13,7 @@ import { PL24Service } from "./pl24.service"; const PL24_PROVIDERS = [ PL24Service, PL24AuthService, + PL24BudgetService, PL24FordLegacyService, PL24PsaService, PL24VolvoService, @@ -20,6 +23,7 @@ const PL24_PROVIDERS = [ ]; @Module({ + imports: [ProxyTelemetryModule], providers: PL24_PROVIDERS, exports: PL24_PROVIDERS, }) diff --git a/apps/api/src/integrations/pl24/pl24.service.spec.ts b/apps/api/src/integrations/pl24/pl24.service.spec.ts index 7593bec..04fa884 100644 --- a/apps/api/src/integrations/pl24/pl24.service.spec.ts +++ b/apps/api/src/integrations/pl24/pl24.service.spec.ts @@ -15,6 +15,7 @@ const svc = new PL24Service( {} as never, // redis {} as never, // storage {} as never, // posthog + {} as never, // budget ); const p = svc as unknown as { parseVehicleResponse( diff --git a/apps/api/src/integrations/pl24/pl24.service.ts b/apps/api/src/integrations/pl24/pl24.service.ts index b382d40..cb79ead 100644 --- a/apps/api/src/integrations/pl24/pl24.service.ts +++ b/apps/api/src/integrations/pl24/pl24.service.ts @@ -20,6 +20,7 @@ import { PostHogService } from "../../posthog/posthog.service"; import { RedisService } from "../../redis/redis.service"; import { StorageService } from "../../storage/storage.service"; import { PL24AuthService } from "./pl24-auth.service"; +import { PL24BudgetService } from "./pl24-budget.service"; import { PL24FordLegacyService } from "./pl24-ford-legacy.service"; import { PL24FordService } from "./pl24-ford.service"; import { PL24HyundaiKiaService } from "./pl24-hyundai-kia.service"; @@ -74,6 +75,7 @@ export class PL24Service { private redis: RedisService, private storage: StorageService, private posthog: PostHogService, + private readonly budget: PL24BudgetService, ) { this.baseUrl = this.configService.get("pl24.baseUrl", "https://www.partslink24.com"); this.timeout = 30000; @@ -944,15 +946,28 @@ export class PL24Service { return opts; }; - const response = await fetch(url, buildOpts(headers)); + // Every PL24 upstream call is counted and logged here (the one choke point). + const response = await this.budgetedFetch( + url, + buildOpts(headers), + account, + Boolean(dispatcher), + ); if (response.status === 401) { - this.logger.warn(`Got 401 (account=${account}), refreshing token...`); + this.logger.warn(`Got 401 (account=${account}), refreshing service token...`); + // A stale *service* token is the cheap explanation; the session itself is + // only dropped if the re-authorize also fails (handled in the auth service). this.authService.clearTokensForAccount(account); await this.authService.authorizeServiceForAccount(serviceName, account); const newHeaders = await this.authService.buildAuthHeadersForAccount(account, serviceName); - const retry = await fetch(url, buildOpts(newHeaders)); + const retry = await this.budgetedFetch( + url, + buildOpts(newHeaders), + account, + Boolean(dispatcher), + ); if (!retry.ok) { throw new Error(`HTTP ${retry.status}: ${retry.statusText}`); @@ -971,6 +986,49 @@ export class PL24Service { return response; } + /** + * Single gate for PL24 upstream traffic: daily budget first (a refusal costs + * no request at all), then the call, then telemetry into `proxy_logs`. + */ + private async budgetedFetch( + url: string, + opts: RequestInit & { dispatcher?: any }, + account: "tr" | "de", + proxied: boolean, + ): Promise { + // The kill switch used to gate decodeVin only, so drills, backfill and + // catalog browse kept hammering PL24 after the source was "killed". + if (!(await this.posthog.isSourceLive("pl24"))) { + throw new ServiceUnavailableException("PL24 kapali (kill-source-pl24)"); + } + await this.budget.consume("catalog"); + const startedAt = Date.now(); + try { + const response = await fetch(url, opts); + this.budget.record({ + kind: "catalog", + url, + proxied, + account, + statusCode: response.status, + success: response.ok, + startedAt, + }); + return response; + } catch (error) { + this.budget.record({ + kind: "catalog", + url, + proxied, + account, + success: false, + startedAt, + error, + }); + throw error; + } + } + // ==================== PRIVATE: Account routing ==================== /** diff --git a/apps/api/src/integrations/proxy-telemetry/proxy-telemetry.service.ts b/apps/api/src/integrations/proxy-telemetry/proxy-telemetry.service.ts index a154226..ec8e9f5 100644 --- a/apps/api/src/integrations/proxy-telemetry/proxy-telemetry.service.ts +++ b/apps/api/src/integrations/proxy-telemetry/proxy-telemetry.service.ts @@ -4,7 +4,7 @@ import { proxyLogs } from "../../database/schema/core"; /** * Proxy telemetry — fire-and-forget logging of every proxied upstream attempt - * (pcat call/capture/validate, emex http) into `proxy_logs`, so the Süper Panel + * (pcat call/capture/validate, emex http, pl24 catalog/auth) into `proxy_logs`, so the Süper Panel * can grade proxy providers per service and track banned exit IPs / sessions. * * Design constraints: @@ -13,7 +13,13 @@ import { proxyLogs } from "../../database/schema/core"; * - Bounded memory: the buffer is capped; when full, oldest rows are dropped. */ -export type ProxyServiceLeg = "pcat_call" | "pcat_capture" | "pcat_validate" | "emex_http"; +export type ProxyServiceLeg = + | "pcat_call" + | "pcat_capture" + | "pcat_validate" + | "emex_http" + | "pl24_http" + | "pl24_auth"; export type ProxyProviderName = "floxy" | "dataimpulse" | "none"; export interface ProxyLogEvent {