feat(flags): server-side feature flags + upstream kill switches + live backfill config

Add server-side PostHog feature-flag evaluation to the API and wire three
high-leverage uses. The flags live in PostHog (project 127747), dormant and
fail-open, so this commit is a no-op until POSTHOG_PERSONAL_API_KEY is set and
a switch is flipped.

Phase 1 - upstream kill switches:
  PostHogService gains local flag evaluation (onlyEvaluateLocally +
  sendFeatureFlagEvents:false -> zero decode latency, no VIN leak) and
  isSourceLive(). Guards at each source's decode entry (parts-catalogs, emex,
  pl24 whole-source + per-brand via LEGACY_ARCH_SOURCE_TAG) let a flailing
  upstream be disabled from the PostHog UI in ~5s instead of a
  code-fix -> dev -> prod redeploy. Fail-open: any unresolved flag keeps the
  source live, so a PostHog outage can never black out decoding.

Phase 2 - guarded rollout primitive:
  isEnabled()/variant() with VIN bucketing, ready to ramp a new decode/parser
  path 0->100% behind a decode-*-v2 flag (recipe in feature-flags-strategy.md).

Phase 4 - remote-config ops tuning:
  prefetch-worker reads cfg-backfill-tuning to retune backfill
  batchSize/maxBacklog/businessHoursOnly live; malformed/missing -> the
  compiled-in constants.

POSTHOG_PERSONAL_API_KEY wired into the api + worker compose blocks (empty ->
flags inert, no added latency). Tests updated for the new constructor params.

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

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

@@ -17,6 +17,7 @@ function makeConfig(overrides: Record<string, string>): ConfigService {
 
 const browser = {} as never;
 const redis = { set: vi.fn() } as never;
+const posthog = {} as never;
 
 describe("EmexService proxy-port coercion (regression)", () => {
   it("does not throw 'Invalid URL' when ports arrive as strings over a real range", () => {
@@ -27,7 +28,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)).not.toThrow();
+    expect(() => new EmexService(config, browser, redis, posthog)).not.toThrow();
   });
 
   it("constructs with a single-port range (string env)", () => {
@@ -36,7 +37,7 @@ describe("EmexService proxy-port coercion (regression)", () => {
       EMEX_PROXY_PORT_START: "823",
       EMEX_PROXY_PORT_END: "823",
     });
-    expect(() => new EmexService(config, browser, redis)).not.toThrow();
+    expect(() => new EmexService(config, browser, redis, posthog)).not.toThrow();
   });
 
   it("falls back to a valid default when the port env is garbage", () => {
@@ -45,11 +46,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)).not.toThrow();
+    expect(() => new EmexService(config, browser, redis, posthog)).not.toThrow();
   });
 
   it("constructs cleanly with the proxy disabled", () => {
     const config = makeConfig({ EMEX_USE_PROXY: "false" });
-    expect(() => new EmexService(config, browser, redis)).not.toThrow();
+    expect(() => new EmexService(config, browser, redis, posthog)).not.toThrow();
   });
 });

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

@@ -21,6 +21,7 @@ import { ConfigService } from "@nestjs/config";
 
 import { ProxyAgent } from "undici";
 import { isBackfillContext } from "../../jobs/prefetch-context";
+import { PostHogService } from "../../posthog/posthog.service";
 import { RedisService } from "../../redis/redis.service";
 import { parseUnitLeaves, parseVehicleTree } from "./emex-tree.parser";
 import { EmexBrowserService } from "./emex.browser";
@@ -136,6 +137,7 @@ export class EmexService {
     private configService: ConfigService,
     private browserService: EmexBrowserService,
     private redis: RedisService,
+    private posthog: PostHogService,
   ) {
     // __dirname is apps/api/src/integrations/emex/ or dist/integrations/emex/
     // Scraper lives at <monorepo-root>/scripts/emex-vin-scraper.js
@@ -618,6 +620,10 @@ export class EmexService {
     | { type: "notFound" }
     | { type: "error" }
   > {
+    if (!(await this.posthog.isSourceLive("emex"))) {
+      this.logger.warn("EMEX disabled by kill switch (kill-source-emex)");
+      return { type: "notFound" };
+    }
     try {
       const vinUrl = `${EMEX_BASE_URL}/Vehicles.aspx?ft=findByVIN&c=&ssd=&vin=${vin}`;
       const html = await this.fetchEmexHtml(vinUrl);
@@ -708,6 +714,10 @@ export class EmexService {
    * Used after the user selects a vehicle from the multi-candidate modal.
    */
   async decodeVinByIndex(vin: string, index: number): Promise<DecodedVehicle | null> {
+    if (!(await this.posthog.isSourceLive("emex"))) {
+      this.logger.warn("EMEX disabled by kill switch (kill-source-emex)");
+      return null;
+    }
     try {
       const vinUrl = `${EMEX_BASE_URL}/Vehicles.aspx?ft=findByVIN&c=&ssd=&vin=${vin}`;
       const vinHtml = await this.fetchEmexHtml(vinUrl);

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

@@ -8,6 +8,7 @@
 
 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 { PartsCatalogsAuthService } from "./parts-catalogs-auth.service";
 import {
@@ -40,6 +41,7 @@ export class PartsCatalogsService {
   constructor(
     private authService: PartsCatalogsAuthService,
     private redis: RedisService,
+    private posthog: PostHogService,
   ) {}
 
   /** Mark parts-catalogs as actively used (5min TTL) to defer prefetch worker */
@@ -62,6 +64,10 @@ export class PartsCatalogsService {
     signal?: AbortSignal,
     outcome?: { transient: boolean },
   ): Promise<PcatVinResult | null> {
+    if (!(await this.posthog.isSourceLive("parts-catalogs"))) {
+      this.logger.warn("parts-catalogs disabled by kill switch (kill-source-parts-catalogs)");
+      return null;
+    }
     try {
       const data = await this.fetchWithAuth("/car/info", { q: vin }, signal, {
         timeoutMs: DECODE_REQUEST_TIMEOUT,

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

@@ -14,6 +14,7 @@ const svc = new PL24Service(
   configStub, // configService
   {} as never, // redis
   {} as never, // storage
+  {} as never, // posthog
 );
 const p = svc as unknown as {
   parseVehicleResponse(

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

@@ -16,6 +16,7 @@ import {
 import { ConfigService } from "@nestjs/config";
 import { extractModelYear } from "@sase/shared";
 import { isBackfillContext } from "../../jobs/prefetch-context";
+import { PostHogService } from "../../posthog/posthog.service";
 import { RedisService } from "../../redis/redis.service";
 import { StorageService } from "../../storage/storage.service";
 import { PL24AuthService } from "./pl24-auth.service";
@@ -41,6 +42,19 @@ import {
   isP5Modern,
 } from "./pl24.types";
 
+/**
+ * Legacy (P4) architecture → kill-switch source tag. Lets one fragile brand be
+ * disabled (`kill-source-psa`, `kill-source-volvo`, …) without taking all of PL24
+ * down. Unmapped legacy brands fall back to the shared `pl24-legacy` switch.
+ */
+const LEGACY_ARCH_SOURCE_TAG: Record<string, string> = {
+  LEGACY_PSA: "psa",
+  LEGACY_VOLVO: "volvo",
+  LEGACY_FORD: "ford",
+  LEGACY_OPEL: "opel",
+  LEGACY_HYUNDAI_KIA: "hyundai-kia",
+};
+
 @Injectable()
 export class PL24Service {
   private readonly logger = new Logger(PL24Service.name);
@@ -59,6 +73,7 @@ export class PL24Service {
     private configService: ConfigService,
     private redis: RedisService,
     private storage: StorageService,
+    private posthog: PostHogService,
   ) {
     this.baseUrl = this.configService.get<string>("pl24.baseUrl", "https://www.partslink24.com");
     this.timeout = 30000;
@@ -70,6 +85,10 @@ export class PL24Service {
    * Only P5 Modern architecture is supported here; legacy is dispatched to fordLegacyService.
    */
   async decodeVin(vin: string, userId?: string): Promise<PL24DecodedVehicle | null> {
+    if (!(await this.posthog.isSourceLive("pl24"))) {
+      this.logger.warn("PL24 disabled by kill switch (kill-source-pl24)");
+      return null;
+    }
     const cleanVin = vin.toUpperCase().replace(/[^A-HJ-NPR-Z0-9]/g, "");
     this.validateVin(cleanVin);
 
@@ -88,6 +107,15 @@ export class PL24Service {
 
     // Dispatch P4 legacy architectures to the generic legacy service
     if (!isP5Modern(serviceName)) {
+      // Per-brand kill switch: disable one fragile legacy brand without taking all
+      // of PL24 down. PSA/Volvo/Ford/etc. break independently upstream (cf. the
+      // PSA illustration-dispatch outage that broke 41/43 vehicles for months).
+      const legacyArch = getServiceConfig(serviceName)?.architecture ?? "";
+      const brandTag = LEGACY_ARCH_SOURCE_TAG[legacyArch] ?? "pl24-legacy";
+      if (!(await this.posthog.isSourceLive(brandTag))) {
+        this.logger.warn(`PL24 ${brandTag} disabled by kill switch (kill-source-${brandTag})`);
+        return null;
+      }
       // PSA (Peugeot/Citroën/DS) uses a dedicated FI/VIN-indexed decode service.
       if (getServiceConfig(serviceName)?.architecture === "LEGACY_PSA") {
         return this.psaService.decodeVinForService(cleanVin, serviceName, userId);

apps/api/src/jobs/prefetch-worker.service.ts

@@ -10,6 +10,7 @@ import { and, asc, eq, gt, inArray, isNull, notExists, sql } from "drizzle-orm";
 import { CategoriesService } from "../categories/categories.service";
 import { DATABASE, type Database } from "../database/database.provider";
 import { categories, parts, vehicles } from "../database/schema/core";
+import { PostHogService } from "../posthog/posthog.service";
 import { RedisService } from "../redis/redis.service";
 import { QUEUE_NAMES, getBullConnection } from "./bull.config";
 import { backfillContext } from "./prefetch-context";
@@ -79,6 +80,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     @Inject(CATALOG_PREFETCH_QUEUE) private queue: Queue,
     private categoriesService: CategoriesService,
     private redis: RedisService,
+    private posthog: PostHogService,
     @Inject(DATABASE) private db: Database,
   ) {}
 
@@ -361,11 +363,26 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
+    // Live-tunable knobs via the `cfg-backfill-tuning` remote-config flag — change
+    // batch size / backlog ceiling / business-hours gating from the PostHog UI
+    // without a redeploy. Falls back to the compiled-in defaults on any miss.
+    const cfg = await this.posthog.payload<{
+      batchSize?: number;
+      maxBacklog?: number;
+      businessHoursOnly?: boolean;
+    }>("cfg-backfill-tuning", {});
+    const batchSize =
+      typeof cfg.batchSize === "number" && cfg.batchSize > 0 ? cfg.batchSize : BACKFILL_BATCH_SIZE;
+    const maxBacklog =
+      typeof cfg.maxBacklog === "number" && cfg.maxBacklog > 0
+        ? cfg.maxBacklog
+        : BACKFILL_MAX_BACKLOG;
+
     // Self-throttle: don't pile on if the queue is already deep — let it drain.
     const counts = await this.queue.getJobCounts("waiting", "delayed", "active");
     const backlog = (counts.waiting ?? 0) + (counts.delayed ?? 0) + (counts.active ?? 0);
-    if (backlog > BACKFILL_MAX_BACKLOG) {
-      this.logger.log(`[backfill] Skip — queue backlog ${backlog} > ${BACKFILL_MAX_BACKLOG}`);
+    if (backlog > maxBacklog) {
+      this.logger.log(`[backfill] Skip — queue backlog ${backlog} > ${maxBacklog}`);
       return;
     }
 
@@ -374,7 +391,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     const eligible: string[] = [];
     for (const s of BACKFILL_SOURCES) {
       if (await this.redis.exists(`prefetch:activity:${s}`)) continue;
-      if (!isWithinTimeWindow(s)) continue;
+      if (cfg.businessHoursOnly !== false && !isWithinTimeWindow(s)) continue;
       eligible.push(s);
     }
     if (eligible.length === 0) {
@@ -384,10 +401,10 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
     const picked: Array<{ id: string; source: string }> = [];
     const seen = new Set<string>();
-    const overfetch = BACKFILL_BATCH_SIZE * 4; // headroom for in-flight skips
+    const overfetch = batchSize * 4; // headroom for in-flight skips
 
     const tryPick = async (v: { id: string; source: string | null }): Promise<void> => {
-      if (picked.length >= BACKFILL_BATCH_SIZE || seen.has(v.id) || !v.source) return;
+      if (picked.length >= batchSize || seen.has(v.id) || !v.source) return;
       if (await this.redis.exists(`prefetch:scheduled:${v.id}`)) return; // already in flight
       // Skip exhausted residue: vehicles whose prefetch keeps finishing with zero
       // parts (no catalog data). They'd otherwise be re-picked every wave forever.
@@ -416,7 +433,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
     // Phase 2 — rolling rescan of ALL decoded vehicles to gap-fill partially-fetched
     // ones. A createdAt cursor walks forward and wraps around at the end.
-    if (picked.length < BACKFILL_BATCH_SIZE) {
+    if (picked.length < batchSize) {
       const cursorObj = await this.redis.getJson<{ ts: string }>(BACKFILL_CURSOR_KEY);
       const cursor = cursorObj?.ts ? new Date(cursorObj.ts) : new Date(0);
 

apps/api/src/posthog/posthog.service.ts

@@ -7,6 +7,8 @@ export class PostHogService implements OnModuleDestroy {
   private readonly logger = new Logger(PostHogService.name);
   private client: PostHogClient | null = null;
   private enabled = false;
+  /** True once a personal API key is configured → local flag evaluation works. */
+  private flagsEnabled = false;
 
   constructor(private configService: ConfigService) {
     const apiKey = this.configService.get<string>("POSTHOG_API_KEY");
@@ -23,8 +25,29 @@ export class PostHogService implements OnModuleDestroy {
     try {
       const { PostHog } = await import("posthog-node");
       const host = this.configService.get<string>("POSTHOG_HOST") ?? "https://t.sase.tr";
-      this.client = new PostHog(apiKey, { host, flushAt: 1, flushInterval: 1000 });
-      this.logger.log("PostHog server-side client initialized");
+      // A personal API key (or feature-flags secure key) unlocks LOCAL flag
+      // evaluation: the SDK polls + caches flag definitions, so isEnabled() and
+      // payload() resolve with zero network I/O. That matters because the source
+      // kill switches sit on the VIN-decode hot path — a per-call network hop
+      // would be unacceptable. Without the key, server-side flags fail open (see
+      // isEnabled) and add no latency.
+      // Empty string (an unset compose `${VAR:-}`) must not be passed as a key, or
+      // the SDK polls flag definitions with a bogus credential — coalesce to undefined.
+      const personalApiKey =
+        this.configService.get<string>("POSTHOG_PERSONAL_API_KEY") || undefined;
+      this.client = new PostHog(apiKey, {
+        host,
+        personalApiKey,
+        featureFlagsPollingInterval: 30_000,
+        flushAt: 1,
+        flushInterval: 1000,
+      });
+      this.flagsEnabled = Boolean(personalApiKey);
+      this.logger.log(
+        `PostHog server-side client initialized (local flag evaluation: ${
+          this.flagsEnabled ? "enabled" : "disabled — set POSTHOG_PERSONAL_API_KEY"
+        })`,
+      );
     } catch (err) {
       this.logger.error("Failed to initialize PostHog client", (err as Error).stack);
       this.enabled = false;
@@ -61,4 +84,82 @@ export class PostHogService implements OnModuleDestroy {
   captureForUser(userId: string, event: string, properties?: Record<string, unknown>): void {
     this.capture(event, { ...properties, $user_id: userId }, userId);
   }
+
+  // ──────────────────────────────────────────────────────────────────────────
+  // Feature flags — server-side, local evaluation
+  // ──────────────────────────────────────────────────────────────────────────
+
+  /**
+   * Evaluate a boolean feature flag server-side, LOCALLY.
+   *
+   * `onlyEvaluateLocally` guarantees zero network I/O (no latency on the request
+   * being guarded) and that `distinctId` — which may be a VIN — is NEVER sent to
+   * PostHog. Returns `fallback` whenever the flag can't be resolved locally (client
+   * not ready, no personal API key, PostHog unreachable, or flag undefined), so
+   * every caller MUST pass a fail-safe default.
+   */
+  async isEnabled(key: string, distinctId = "server", fallback = false): Promise<boolean> {
+    if (!this.client) return fallback;
+    try {
+      const value = await this.client.isFeatureEnabled(key, distinctId, {
+        onlyEvaluateLocally: true,
+        sendFeatureFlagEvents: false,
+      });
+      return value ?? fallback;
+    } catch (err) {
+      this.logger.warn(`Feature flag "${key}" evaluation failed: ${(err as Error).message}`);
+      return fallback;
+    }
+  }
+
+  /**
+   * Resolve a multivariant flag to its variant key (e.g. "control" / "test") for
+   * guarded / percentage rollout of new code paths. Pass the VIN as `distinctId`
+   * for stable per-VIN bucketing. Local-only; never leaks the id. Returns
+   * `fallback` when unresolved.
+   */
+  async variant(
+    key: string,
+    distinctId = "server",
+    fallback?: string,
+  ): Promise<string | undefined> {
+    if (!this.client) return fallback;
+    try {
+      const value = await this.client.getFeatureFlag(key, distinctId, {
+        onlyEvaluateLocally: true,
+        sendFeatureFlagEvents: false,
+      });
+      return typeof value === "string" ? value : fallback;
+    } catch {
+      return fallback;
+    }
+  }
+
+  /**
+   * Read a remote-config / flag payload (JSON). Used OFF the hot path (e.g. the
+   * hourly backfill cron), so remote evaluation is acceptable and the personal
+   * key is not required. Returns `fallback` on any failure.
+   */
+  async payload<T>(key: string, fallback: T, distinctId = "server"): Promise<T> {
+    if (!this.client) return fallback;
+    try {
+      const value = await this.client.getFeatureFlagPayload(key, distinctId);
+      return (value as T | undefined) ?? fallback;
+    } catch (err) {
+      this.logger.warn(`Feature flag payload "${key}" failed: ${(err as Error).message}`);
+      return fallback;
+    }
+  }
+
+  /**
+   * Is an upstream source allowed to run? Returns false ONLY when its kill switch
+   * `kill-source-<source>` is explicitly enabled in PostHog. Fail-OPEN: any
+   * inability to evaluate (no key, flag absent, PostHog down) leaves the source
+   * live, so an analytics outage can never black out VIN decoding.
+   *
+   * @example if (!(await this.posthog.isSourceLive("emex"))) return { type: "notFound" };
+   */
+  async isSourceLive(source: string, distinctId = "server"): Promise<boolean> {
+    return !(await this.isEnabled(`kill-source-${source}`, distinctId, false));
+  }
 }