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<string, string> = {}, redis = makeRedis()) => {
   const cfg: Record<string, string> = {
     "pl24.companyCode": "tr-000000",
@@ -49,7 +62,12 @@ const makeService = (cfgOverrides: Record<string, string> = {}, 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 = {

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<string>("pl24.baseUrl", "https://www.partslink24.com");
     this.companyCode = this.configService.get<string>("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) {

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<string, number>();
+  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<string, unknown>[] = [];
+  return { events, record: (e: Record<string, unknown>) => events.push(e) };
+};
+
+const make = (env: Record<string, string> = {}, 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 = <T>(fn: () => Promise<T>) => 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();
+  });
+});

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<string>();
+
+  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<void> {
+    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<number> {
+    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";
+  }
+}

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,
 })

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(

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<string>("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<Response> {
+    // 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 ====================
 
   /**

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 {