feat(pl24): günlük HTTP bütçesi, kullanıcı rezervi ve proxy_logs telemetrisi

Faz 1 / adım 2 (analiz: /home/s/ss/plv2.md, bulgular consumers_jobs-03/06/07,
deploy_ops-13).

SORUN: İki PL24 hesap banı da günde 5-7 bin kategori yazan (≈10 bin+ istek)
otomatik drill dalgalarının ardından geldi; gerçek kullanıcı decode'u ≤14/gün.
Mevcut korumaların hepsi iş-seviyesinde (iş/dk, iş/gün) — HTTP isteği sayan
hiçbir tavan yoktu. Üstelik PL24 çağrıları `proxy_logs`'a hiç yazılmıyordu
(yalnız pcat/emex), bu yüzden hacim ancak ban'dan SONRA, DB satır sayılarından
geriye dönük çıkarılabildi. Kill switch de yalnız decodeVin'i kapatıyordu; drill,
backfill ve katalog gezinmesi "kapalı" kaynağa istek atmaya devam ediyordu.

DEĞİŞİKLİK:
- Yeni `PL24BudgetService`: tüm PL24 upstream trafiği için tek geçit.
  - Günlük sayaç `pl24:http:<gün>` (Redis, 8 gün TTL); tavan `PL24_HTTP_DAILY_MAX`
    (varsayılan 1200).
  - Kullanıcı rezervi `PL24_HTTP_USER_RESERVE` (varsayılan %40): backfill tavanın
    %60'ında durur, kullanıcı decode'u sonuna kadar akar — kaçak bir backfill
    müşteriyi asla aç bırakamaz. Lane ayrımı mevcut `isBackfillContext()` ile.
  - Reddedilen çağrı ağa hiç çıkmaz ve telemetriye yazılmaz.
  - Redis düşerse fail-open (auth/katalog telemetri yüzünden bloklanmaz).
- Telemetri: `ProxyServiceLeg` += `pl24_http`, `pl24_auth`. Login, authorize ve
  tüm katalog fetch'leri `proxy_logs`'a yazıyor (provider dataimpulse/none,
  sessionKey=hesap, status, süre, 403/429 → banned, taşıma hatası sınıflandırma).
  Artık "hangi gün kaç istek attık, ilk 401/403 ne zaman başladı" sorusu
  ban'dan ÖNCE yanıtlanabilir.
- Kill switch yayıldı: `kill-source-pl24` artık `budgetedFetch` içinde, yani
  decode + drill + backfill + browse + görsel indirme dahil TÜM katalog trafiğini
  kapatıyor (eskiden yalnız decodeVin).
- `fetchWithRetry` 401 çaresi netleşti: yalnız servis token'ı tazelenir; oturumun
  kendisi gerekirse auth katmanında düşürülür.

Test: yeni `pl24-budget.service.spec.ts` (9 test: tavan, kullanıcı rezervi,
reddedilen çağrının ağa çıkmaması, Redis fail-open, telemetri biçimi/leg ayrımı,
ban sinyali, taşıma hatası sınıflandırma). Etkilenen paketler: 197 test geçti.
tsc + biome temiz.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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 ====================
 
   /**