feat(proxy): proxy_logs telemetry — per-attempt provider/ban/latency tracking

One row per proxied upstream attempt (pcat call/capture/validate, emex http)
written fire-and-forget by the new ProxyTelemetryService (buffered, capped,
errors swallowed — telemetry can never hurt the request path).

- banned = upstream 403/429 (IP-block signal), distinct from auth/data errors
- sticky legs (pcat capture, emex floxy) carry a session_key; pcat capture
  also resolves the actual residential exit IP via a parallel ipify probe
  through the same sticky session → concrete banned-IP tracking
- rotating legs log provider + outcome (ban *rate* instead of per-IP)
- 30-day retention piggybacked on the query-cleanup job

Feeds the Süper Panel /analytics/proxy page (provider grading + banned IPs).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

apps/api/drizzle/0014_proxy_logs.sql

@@ -0,0 +1,23 @@
+-- Proxy telemetry: one row per proxied upstream attempt (pcat call/capture/
+-- validate, emex http). Lets the Süper Panel grade proxy providers per service
+-- (success rate, ban rate, latency) and track banned residential exit IPs /
+-- sticky sessions. Written fire-and-forget by ProxyTelemetryService; 30-day
+-- retention enforced by the query-cleanup job.
+CREATE TABLE "proxy_logs" (
+	"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+	"service" varchar(30) NOT NULL,
+	"provider" varchar(20) NOT NULL,
+	"session_key" varchar(60),
+	"exit_ip" varchar(45),
+	"target_host" varchar(120),
+	"status_code" integer,
+	"error_kind" varchar(30),
+	"success" boolean DEFAULT false NOT NULL,
+	"banned" boolean DEFAULT false NOT NULL,
+	"duration_ms" integer,
+	"created_at" timestamp with time zone DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+CREATE INDEX "proxy_logs_created_at_idx" ON "proxy_logs" ("created_at");
+--> statement-breakpoint
+CREATE INDEX "proxy_logs_banned_created_at_idx" ON "proxy_logs" ("banned","created_at");

apps/api/drizzle/meta/_journal.json

@@ -99,6 +99,13 @@
       "when": 1780600000000,
       "tag": "0013_fix_brand_casing",
       "breakpoints": true
+    },
+    {
+      "idx": 14,
+      "version": "7",
+      "when": 1781136000000,
+      "tag": "0014_proxy_logs",
+      "breakpoints": true
     }
   ]
 }

apps/api/src/database/schema/core.ts

@@ -617,3 +617,41 @@ export const emexCategoryTranslations = pgTable(
   },
   (table) => [uniqueIndex("emex_translations_original_name_idx").on(table.originalName)],
 );
+
+// ─── Proxy Logs ──────────────────────────────────────
+// One row per proxied upstream attempt (pcat call/capture/validate, emex http).
+// Written fire-and-forget by ProxyTelemetryService so the Süper Panel can grade
+// proxy providers (success/ban/latency per service) and track banned exit IPs.
+// `banned` = HTTP 403/429 from the upstream (IP-level block signal); `session_key`
+// identifies a sticky Floxy session or a DataImpulse port so bans are attributable
+// even when the exact exit IP is unknown (rotating call legs).
+export const proxyLogs = pgTable(
+  "proxy_logs",
+  {
+    id: uuid("id").primaryKey().defaultRandom(),
+    // Which integration leg made the request: pcat_call | pcat_capture |
+    // pcat_validate | emex_http
+    service: varchar("service", { length: 30 }).notNull(),
+    // floxy | dataimpulse | none (direct)
+    provider: varchar("provider", { length: 20 }).notNull(),
+    // Sticky Floxy session id ("s-ab12cd34") or DataImpulse port ("di:10042");
+    // null on per-request rotating legs (exit IP changes every call).
+    sessionKey: varchar("session_key", { length: 60 }),
+    // Resolved residential exit IP (sticky sessions only — one cheap ipify probe
+    // per session). Null when rotation makes the IP unknowable.
+    exitIp: varchar("exit_ip", { length: 45 }),
+    targetHost: varchar("target_host", { length: 120 }),
+    statusCode: integer("status_code"),
+    // Transport failure class when no HTTP answer arrived:
+    // timeout | connect | reset | dns | aborted | transport
+    errorKind: varchar("error_kind", { length: 30 }),
+    success: boolean("success").default(false).notNull(),
+    banned: boolean("banned").default(false).notNull(),
+    durationMs: integer("duration_ms"),
+    createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
+  },
+  (table) => [
+    index("proxy_logs_created_at_idx").on(table.createdAt),
+    index("proxy_logs_banned_created_at_idx").on(table.banned, table.createdAt),
+  ],
+);

apps/api/src/integrations/emex/emex.module.ts

@@ -1,8 +1,10 @@
 import { Module } from "@nestjs/common";
+import { ProxyTelemetryModule } from "../proxy-telemetry/proxy-telemetry.module";
 import { EmexBrowserService } from "./emex.browser";
 import { EmexService } from "./emex.service";
 
 @Module({
+  imports: [ProxyTelemetryModule],
   providers: [EmexBrowserService, EmexService],
   exports: [EmexService],
 })

apps/api/src/integrations/emex/emex.service.spec.ts

@@ -18,6 +18,7 @@ function makeConfig(overrides: Record<string, string>): ConfigService {
 const browser = {} as never;
 const redis = { set: vi.fn() } as never;
 const posthog = {} as never;
+const proxyTelemetry = { record: vi.fn() } as never;
 
 describe("EmexService proxy-port coercion (regression)", () => {
   it("does not throw 'Invalid URL' when ports arrive as strings over a real range", () => {
@@ -28,7 +29,7 @@ describe("EmexService proxy-port coercion (regression)", () => {
       EMEX_PROXY_PORT_END: "10099",
     });
     // Pre-fix: threw "Invalid URL" here (port "4510001" > 65535).
-    expect(() => new EmexService(config, browser, redis, posthog)).not.toThrow();
+    expect(() => new EmexService(config, browser, redis, posthog, proxyTelemetry)).not.toThrow();
   });
 
   it("constructs with a single-port range (string env)", () => {
@@ -37,7 +38,7 @@ describe("EmexService proxy-port coercion (regression)", () => {
       EMEX_PROXY_PORT_START: "823",
       EMEX_PROXY_PORT_END: "823",
     });
-    expect(() => new EmexService(config, browser, redis, posthog)).not.toThrow();
+    expect(() => new EmexService(config, browser, redis, posthog, proxyTelemetry)).not.toThrow();
   });
 
   it("falls back to a valid default when the port env is garbage", () => {
@@ -46,11 +47,11 @@ describe("EmexService proxy-port coercion (regression)", () => {
       EMEX_PROXY_PORT_START: "not-a-number",
       EMEX_PROXY_PORT_END: "999999",
     });
-    expect(() => new EmexService(config, browser, redis, posthog)).not.toThrow();
+    expect(() => new EmexService(config, browser, redis, posthog, proxyTelemetry)).not.toThrow();
   });
 
   it("constructs cleanly with the proxy disabled", () => {
     const config = makeConfig({ EMEX_USE_PROXY: "false" });
-    expect(() => new EmexService(config, browser, redis, posthog)).not.toThrow();
+    expect(() => new EmexService(config, browser, redis, posthog, proxyTelemetry)).not.toThrow();
   });
 });

apps/api/src/integrations/emex/emex.service.ts

@@ -23,6 +23,10 @@ import { ProxyAgent } from "undici";
 import { isBackfillContext } from "../../jobs/prefetch-context";
 import { PostHogService } from "../../posthog/posthog.service";
 import { RedisService } from "../../redis/redis.service";
+import {
+  ProxyTelemetryService,
+  classifyTransportError,
+} from "../proxy-telemetry/proxy-telemetry.service";
 import { parseUnitLeaves, parseVehicleTree } from "./emex-tree.parser";
 import { EmexBrowserService } from "./emex.browser";
 import { createEmptyDecodedVehicle, mapEmexResponse } from "./emex.mapper";
@@ -142,6 +146,7 @@ export class EmexService {
     private browserService: EmexBrowserService,
     private redis: RedisService,
     private posthog: PostHogService,
+    private proxyTelemetry: ProxyTelemetryService,
   ) {
     // __dirname is apps/api/src/integrations/emex/ or dist/integrations/emex/
     // Scraper lives at <monorepo-root>/scripts/emex-vin-scraper.js
@@ -389,12 +394,25 @@ export class EmexService {
     for (let attempt = 1; attempt <= maxAttempts; attempt++) {
       const provider = schedule[attempt - 1]; // undefined when no proxy → direct
       const agent = provider ? this.newProxyAgent(provider) : null;
+      const attemptStart = Date.now();
+      // Sticky Floxy sessions are the attributable identity here; DataImpulse
+      // rotates a random port per agent, so there is nothing stable to pin.
+      const sessionKey = provider === "floxy" ? `s-${this.floxySessionId}` : null;
       try {
         const res = await fetch(url, {
           headers: { "User-Agent": EMEX_UA, Accept: "text/html,application/xhtml+xml" },
           signal: AbortSignal.timeout(this.timeout),
           ...(agent ? { dispatcher: agent } : {}),
         } as RequestInit);
+        this.proxyTelemetry.record({
+          service: "emex_http",
+          provider: provider ?? "none",
+          sessionKey,
+          targetHost: new URL(url).hostname,
+          statusCode: res.status,
+          success: res.ok,
+          durationMs: Date.now() - attemptStart,
+        });
         if (!res.ok) {
           throw new Error(`EMEX HTTP ${res.status} for ${url}`);
         }
@@ -407,6 +425,19 @@ export class EmexService {
         const e = err as Error & { cause?: unknown };
         // A real HTTP response ("EMEX HTTP 404") is a definitive answer — never retry.
         const httpAnswer = /^EMEX HTTP \d/.test(e.message);
+        // HTTP answers were already recorded right after fetch; only transport
+        // failures (no answer at all) still need a row.
+        if (!httpAnswer) {
+          this.proxyTelemetry.record({
+            service: "emex_http",
+            provider: provider ?? "none",
+            sessionKey,
+            targetHost: new URL(url).hostname,
+            errorKind: classifyTransportError(e),
+            success: false,
+            durationMs: Date.now() - attemptStart,
+          });
+        }
         const transient =
           !httpAnswer &&
           (e.name === "TimeoutError" ||

apps/api/src/integrations/parts-catalogs/parts-catalogs-auth.service.ts

@@ -23,6 +23,11 @@ import { Injectable, Logger, type OnModuleDestroy, type OnModuleInit } from "@ne
 import { ConfigService } from "@nestjs/config";
 import { Browser, BrowserContext } from "playwright";
 import { RedisService } from "../../redis/redis.service";
+import {
+  ProxyTelemetryService,
+  classifyTransportError,
+  resolveExitIp,
+} from "../proxy-telemetry/proxy-telemetry.service";
 import { JwtSlot, PcatJwtToken, PcatSession } from "./parts-catalogs.types";
 
 // Shared single-slot cache so dev + prod (and any restarted container) can
@@ -184,6 +189,7 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
   constructor(
     private configService: ConfigService,
     private readonly redis: RedisService,
+    private readonly proxyTelemetry: ProxyTelemetryService,
   ) {
     const cfg = this.configService;
 
@@ -245,8 +251,12 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
   private buildProxy(leg: ProxyLeg): {
     proxyUrl: string | null;
     proxyConfig: { server: string; username: string; password: string } | null;
+    // Telemetry identity: sticky Floxy session id or DataImpulse port. Null on
+    // rotating legs (exit IP changes per request — nothing stable to attribute).
+    sessionKey: string | null;
   } {
-    if (this.proxyProvider === "none") return { proxyUrl: null, proxyConfig: null };
+    if (this.proxyProvider === "none")
+      return { proxyUrl: null, proxyConfig: null, sessionKey: null };
 
     if (this.proxyProvider === "dataimpulse") {
       const port = this.allocatePort();
@@ -254,19 +264,23 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
       return {
         proxyUrl: `http://${this.diUser}:${this.diPass}@${this.diHost}:${port}`,
         proxyConfig: { server, username: this.diUser, password: this.diPass },
+        sessionKey: `di:${port}`,
       };
     }
 
     // floxy
     let password = this.floxyPass;
+    let sessionKey: string | null = null;
     if (leg === "capture" && this.floxyCaptureLifetime > 0) {
       const sid = Math.random().toString(36).slice(2, 10);
       password = `${this.floxyPass}_session-${sid}_lifetime-${this.floxyCaptureLifetime}`;
+      sessionKey = `s-${sid}`;
     }
     const server = `http://${this.floxyHost}:${this.floxyPort}`;
     return {
       proxyUrl: `http://${this.floxyUser}:${password}@${this.floxyHost}:${this.floxyPort}`,
       proxyConfig: { server, username: this.floxyUser, password },
+      sessionKey,
     };
   }
 
@@ -509,10 +523,27 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
         connect: { timeout: 6_000 },
       });
     }
+    const start = Date.now();
     try {
       const r = await fetch(url, fetchOptions);
+      this.proxyTelemetry.record({
+        service: "pcat_validate",
+        provider: this.proxyProvider,
+        targetHost: "gui.parts-catalogs.com",
+        statusCode: r.status,
+        success: r.ok,
+        durationMs: Date.now() - start,
+      });
       return r.ok;
     } catch (err) {
+      this.proxyTelemetry.record({
+        service: "pcat_validate",
+        provider: this.proxyProvider,
+        targetHost: "gui.parts-catalogs.com",
+        errorKind: classifyTransportError(err),
+        success: false,
+        durationMs: Date.now() - start,
+      });
       this.logger.debug(`Slot validation threw: ${(err as Error).message}`);
       return false;
     }
@@ -704,11 +735,31 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
     let context: BrowserContext | null = null;
     const startTime = Date.now();
 
+    // Sticky capture proxy: one residential exit IP pinned for the lifetime of
+    // this capture so the partner site + widget see a coherent session.
+    const { proxyConfig, proxyUrl, sessionKey } = this.buildProxy("capture");
+
+    // The session is sticky, so a parallel ipify probe exits from the SAME IP
+    // the capture will use — that's what makes banned-IP tracking concrete.
+    // Resolves in ~1-2s while the capture itself takes 7s+; never throws.
+    const exitIpPromise = proxyUrl ? resolveExitIp(proxyUrl) : Promise.resolve(null);
+    const logCapture = (ok: boolean, errorKind?: string): void => {
+      void exitIpPromise.then((exitIp) =>
+        this.proxyTelemetry.record({
+          service: "pcat_capture",
+          provider: this.proxyProvider,
+          sessionKey,
+          exitIp,
+          targetHost: new URL(siteUrl).hostname,
+          errorKind: errorKind ?? null,
+          success: ok,
+          durationMs: Date.now() - startTime,
+        }),
+      );
+    };
+
     try {
-      // Sticky capture proxy: one residential exit IP pinned for the lifetime of
-      // this capture so the partner site + widget see a coherent session.
       const contextOptions: Record<string, unknown> = {};
-      const { proxyConfig } = this.buildProxy("capture");
       if (proxyConfig) {
         contextOptions.proxy = proxyConfig;
       }
@@ -795,13 +846,18 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
       const elapsed = Date.now() - startTime;
 
       if (capturedToken) {
+        logCapture(true);
         this.logger.log(`Token captured in ${elapsed}ms from ${new URL(siteUrl).hostname}`);
         return capturedToken;
       }
 
+      // Widget never fired its token call — on a sticky residential IP this is
+      // the classic "partner site blocked/ratelimited this exit IP" signature.
+      logCapture(false, "capture_timeout");
       this.logger.debug(`No token after ${elapsed}ms from ${siteUrl}`);
       return null;
     } catch (err) {
+      logCapture(false, classifyTransportError(err));
       this.logger.warn(`Token capture error: ${(err as Error).message}`);
       return null;
     } finally {

apps/api/src/integrations/parts-catalogs/parts-catalogs.module.ts

@@ -1,8 +1,10 @@
 import { Module } from "@nestjs/common";
+import { ProxyTelemetryModule } from "../proxy-telemetry/proxy-telemetry.module";
 import { PartsCatalogsAuthService } from "./parts-catalogs-auth.service";
 import { PartsCatalogsService } from "./parts-catalogs.service";
 
 @Module({
+  imports: [ProxyTelemetryModule],
   providers: [PartsCatalogsAuthService, PartsCatalogsService],
   exports: [PartsCatalogsService],
 })

apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts

@@ -11,6 +11,11 @@ import { Injectable, Logger } from "@nestjs/common";
 import { isBackfillContext } from "../../jobs/prefetch-context";
 import { PostHogService } from "../../posthog/posthog.service";
 import { RedisService } from "../../redis/redis.service";
+import {
+  ProxyTelemetryService,
+  classifyTransportError,
+  describeProxyUrl,
+} from "../proxy-telemetry/proxy-telemetry.service";
 import { PartsCatalogsAuthService } from "./parts-catalogs-auth.service";
 import {
   PcatCar,
@@ -62,6 +67,7 @@ export class PartsCatalogsService {
     private authService: PartsCatalogsAuthService,
     private redis: RedisService,
     private posthog: PostHogService,
+    private proxyTelemetry: ProxyTelemetryService,
   ) {}
 
   /** Mark parts-catalogs as actively used (5min TTL) to defer prefetch worker */
@@ -273,6 +279,7 @@ export class PartsCatalogsService {
         }
       }
 
+      const attemptStart = Date.now();
       try {
         const signals = [AbortSignal.timeout(timeoutMs)];
         if (externalSignal) signals.push(externalSignal);
@@ -303,6 +310,17 @@ export class PartsCatalogsService {
 
         const response = await fetch(url.toString(), fetchOptions);
 
+        // An HTTP answer arrived — the proxy leg worked at transport level;
+        // log the upstream's verdict (403/429 are the IP-ban signals).
+        this.proxyTelemetry.record({
+          service: "pcat_call",
+          ...describeProxyUrl(session.proxyUrl),
+          targetHost: url.hostname,
+          statusCode: response.status,
+          success: response.ok,
+          durationMs: Date.now() - attemptStart,
+        });
+
         if (response.ok) {
           // Token proven good on this IP — clear any accrued failure streak.
           session._slot.failCount = 0;
@@ -323,6 +341,18 @@ export class PartsCatalogsService {
         throw new Error(`HTTP ${response.status} from ${endpoint}: ${text.slice(0, 200)}`);
       } catch (err) {
         const e = err as Error & { cause?: unknown };
+        // Transport failure (no HTTP answer) — the try block already logged any
+        // real HTTP response before rethrowing it as `Error: HTTP <code> …`.
+        if (!/^HTTP \d/.test(e.message ?? "")) {
+          this.proxyTelemetry.record({
+            service: "pcat_call",
+            ...describeProxyUrl(session.proxyUrl),
+            targetHost: url.hostname,
+            errorKind: classifyTransportError(e),
+            success: false,
+            durationMs: Date.now() - attemptStart,
+          });
+        }
         // Retry transient TRANSPORT failures only: request timeouts and undici
         // network errors ("TypeError: fetch failed" — a dropped/reset DataImpulse
         // proxy connection). A definitive HTTP response (e.g. 400 "list of parts

apps/api/src/integrations/proxy-telemetry/proxy-telemetry.module.ts

@@ -0,0 +1,10 @@
+import { Module } from "@nestjs/common";
+import { ProxyTelemetryService } from "./proxy-telemetry.service";
+
+// Imported by PartsCatalogsModule and EmexModule; Nest module caching makes the
+// service a singleton (one shared flush buffer) across both.
+@Module({
+  providers: [ProxyTelemetryService],
+  exports: [ProxyTelemetryService],
+})
+export class ProxyTelemetryModule {}

apps/api/src/integrations/proxy-telemetry/proxy-telemetry.service.ts

@@ -0,0 +1,151 @@
+import { Inject, Injectable, Logger, OnModuleDestroy } from "@nestjs/common";
+import { DATABASE, type Database } from "../../database/database.provider";
+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
+ * can grade proxy providers per service and track banned exit IPs / sessions.
+ *
+ * Design constraints:
+ * - MUST never affect the request path: record() only pushes to an in-memory
+ *   buffer; a timer flushes batches, and every DB error is swallowed.
+ * - 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 ProxyProviderName = "floxy" | "dataimpulse" | "none";
+
+export interface ProxyLogEvent {
+  service: ProxyServiceLeg;
+  provider: ProxyProviderName;
+  sessionKey?: string | null;
+  exitIp?: string | null;
+  targetHost?: string | null;
+  statusCode?: number | null;
+  errorKind?: string | null;
+  success: boolean;
+  durationMs?: number | null;
+}
+
+const FLUSH_INTERVAL_MS = 5_000;
+const FLUSH_BATCH_SIZE = 50;
+const BUFFER_CAP = 500;
+
+// Upstream statuses that signal an IP-level block (not an auth/data problem):
+// 403 = forbidden (WAF/geo/IP ban), 429 = rate limited. 402 is the proxy
+// provider itself rejecting (balance exhausted — Floxy did this 2026-06-10);
+// counted as banned=false but worth surfacing via status_code.
+export function isBanStatus(status: number | null | undefined): boolean {
+  return status === 403 || status === 429;
+}
+
+/** Classify a transport error (no HTTP answer) into a coarse error_kind. */
+export function classifyTransportError(err: unknown): string {
+  const e = err as Error & { cause?: unknown };
+  const text = `${e?.name ?? ""} ${e?.message ?? ""} ${String(e?.cause ?? "")}`;
+  if (/TimeoutError|UND_ERR_CONNECT_TIMEOUT|UND_ERR_HEADERS_TIMEOUT/i.test(text)) return "timeout";
+  if (/AbortError|aborted/i.test(text)) return "aborted";
+  if (/ECONNREFUSED|UND_ERR_SOCKET|connect/i.test(text)) return "connect";
+  if (/ECONNRESET|socket hang up|other side closed|terminated/i.test(text)) return "reset";
+  if (/EAI_AGAIN|ENOTFOUND/i.test(text)) return "dns";
+  return "transport";
+}
+
+/**
+ * Derive provider + session key from a proxy URL the way the integrations
+ * build them: Floxy passwords may embed `_session-<id>_lifetime-<n>` (sticky);
+ * DataImpulse rotates by port, so the port IS the session identity.
+ */
+export function describeProxyUrl(proxyUrl: string | null | undefined): {
+  provider: ProxyProviderName;
+  sessionKey: string | null;
+} {
+  if (!proxyUrl) return { provider: "none", sessionKey: null };
+  try {
+    const u = new URL(proxyUrl);
+    if (/floxy/i.test(u.hostname)) {
+      const m = decodeURIComponent(u.password).match(/_session-([A-Za-z0-9]+)/);
+      return { provider: "floxy", sessionKey: m ? `s-${m[1]}` : null };
+    }
+    return { provider: "dataimpulse", sessionKey: `di:${u.port}` };
+  } catch {
+    return { provider: "none", sessionKey: null };
+  }
+}
+
+/**
+ * Resolve the residential exit IP of a (sticky) proxy session with one cheap
+ * probe through the same proxy. Only meaningful for sticky sessions — on a
+ * rotating leg the probe and the real request exit from different IPs.
+ */
+export async function resolveExitIp(
+  proxyUrl: string,
+  timeoutMs = 5_000,
+): Promise<string | null> {
+  try {
+    const { ProxyAgent } = await import("undici");
+    const res = await fetch("https://api.ipify.org?format=text", {
+      dispatcher: new ProxyAgent({ uri: proxyUrl, connect: { timeout: timeoutMs } }),
+      signal: AbortSignal.timeout(timeoutMs),
+    } as unknown as RequestInit);
+    if (!res.ok) return null;
+    const ip = (await res.text()).trim();
+    return /^[0-9a-fA-F.:]{3,45}$/.test(ip) ? ip : null;
+  } catch {
+    return null;
+  }
+}
+
+@Injectable()
+export class ProxyTelemetryService implements OnModuleDestroy {
+  private readonly logger = new Logger(ProxyTelemetryService.name);
+  private buffer: ProxyLogEvent[] = [];
+  private flushTimer: NodeJS.Timeout | null = null;
+  private flushing = false;
+
+  constructor(@Inject(DATABASE) private readonly db: Database) {
+    this.flushTimer = setInterval(() => void this.flush(), FLUSH_INTERVAL_MS);
+    // Never keep the process alive just to flush telemetry.
+    this.flushTimer.unref?.();
+  }
+
+  /** Queue one proxied-attempt record. Never throws, never blocks. */
+  record(event: ProxyLogEvent): void {
+    if (this.buffer.length >= BUFFER_CAP) this.buffer.shift();
+    this.buffer.push(event);
+    if (this.buffer.length >= FLUSH_BATCH_SIZE) void this.flush();
+  }
+
+  private async flush(): Promise<void> {
+    if (this.flushing || this.buffer.length === 0) return;
+    this.flushing = true;
+    const batch = this.buffer.splice(0, FLUSH_BATCH_SIZE * 2);
+    try {
+      await this.db.insert(proxyLogs).values(
+        batch.map((e) => ({
+          service: e.service,
+          provider: e.provider,
+          sessionKey: e.sessionKey ?? null,
+          exitIp: e.exitIp ?? null,
+          targetHost: e.targetHost?.slice(0, 120) ?? null,
+          statusCode: e.statusCode ?? null,
+          errorKind: e.errorKind ?? null,
+          success: e.success,
+          banned: isBanStatus(e.statusCode),
+          durationMs: e.durationMs != null ? Math.round(e.durationMs) : null,
+        })),
+      );
+    } catch (err) {
+      // Telemetry must never matter more than the product. Drop the batch.
+      this.logger.debug(`proxy_logs flush failed (${batch.length} rows): ${(err as Error).message}`);
+    } finally {
+      this.flushing = false;
+    }
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    if (this.flushTimer) clearInterval(this.flushTimer);
+    await this.flush();
+  }
+}

apps/api/src/jobs/processors/query-cleanup.processor.ts

@@ -1,11 +1,14 @@
 import { Job } from "bullmq";
 import { lt } from "drizzle-orm";
 import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
-import { queryLogs } from "../../database/schema/core";
+import { proxyLogs, queryLogs } from "../../database/schema/core";
 
 type Database = PostgresJsDatabase<Record<string, unknown>>;
 
 const RETENTION_DAYS = 90;
+// proxy_logs is high-volume per-attempt telemetry; 30 days is plenty for
+// grading providers and spotting ban waves.
+const PROXY_LOG_RETENTION_DAYS = 30;
 
 export async function processQueryCleanup(
   job: Job,
@@ -27,5 +30,15 @@ export async function processQueryCleanup(
     `[query-cleanup] Deleted ${deletedCount} query log(s) older than ${RETENTION_DAYS} days (before ${cutoffDate.toISOString()})`,
   );
 
+  const proxyCutoff = new Date();
+  proxyCutoff.setDate(proxyCutoff.getDate() - PROXY_LOG_RETENTION_DAYS);
+  const proxyDeleted = await db
+    .delete(proxyLogs)
+    .where(lt(proxyLogs.createdAt, proxyCutoff))
+    .returning({ id: proxyLogs.id });
+  console.log(
+    `[query-cleanup] Deleted ${proxyDeleted.length} proxy log(s) older than ${PROXY_LOG_RETENTION_DAYS} days`,
+  );
+
   return { deletedCount };
 }