sase.tr/apps/api/src/jobs/prefetch-utils.ts

import { RedisService } from "../redis/redis.service";

/**
 * True ONLY on the real production deployment (sase.tr).
 *
 * NOTE: both dev.sase.tr (staging) and sase.tr (prod) run with
 * `NODE_ENV=production` (hard-coded in the Dockerfile / compose) but have
 * SEPARATE databases. So NODE_ENV cannot tell them apart. The catalog backfill
 * must only sweep the prod DB, so we key off the canonical prod host. An
 * explicit `CATALOG_BACKFILL_ENABLED` env wins (e.g. to force-test on staging).
 * Anything else (local dev, unknown host) → disabled, the safe default.
 */
export function isCatalogBackfillEnabled(): boolean {
  const flag = process.env.CATALOG_BACKFILL_ENABLED;
  if (flag != null && flag !== "") return flag === "true";
  return (
    process.env.COOLIFY_FQDN === "sase.tr" || process.env.BETTER_AUTH_URL === "https://sase.tr"
  );
}

/**
 * True ONLY on the real production deployment (sase.tr).
 *
 * Same rationale as {@link isCatalogBackfillEnabled} — both prod and dev run
 * with `NODE_ENV=production`, so we gate on the prod host. Without this gate
 * the dev deploy would send duplicate trial-ending / win-back e-mails to real
 * users (sase_dev DB is a copy of prod data). An explicit
 * `LIFECYCLE_EMAIL_ENABLED` env wins (e.g. to force-test on staging).
 */
export function isLifecycleEmailEnabled(): boolean {
  const flag = process.env.LIFECYCLE_EMAIL_ENABLED;
  if (flag != null && flag !== "") return flag === "true";
  return (
    process.env.COOLIFY_FQDN === "sase.tr" || process.env.BETTER_AUTH_URL === "https://sase.tr"
  );
}

/**
 * Signals the worker that this job is rate-limited and should be deferred.
 *
 * `cause` distinguishes the deferral strategy:
 *  - "cooldown" — active user on the source. The cooldown key is refreshed by
 *    every user request, so per-job 60s defer livelocks: the job comes back,
 *    cooldown is still set, defers again. Handled by pausing the WHOLE worker
 *    for retryAfterMs (BullMQ `worker.rateLimit()`), so the queue waits once
 *    instead of cycling.
 *  - "time-window" — off-hours; only PL24/parts-catalogs are affected. Use
 *    per-job moveToDelayed so EMEX jobs (no window) keep flowing.
 */
export class RateLimitError extends Error {
  constructor(
    public readonly retryAfterMs: number,
    public readonly cause: "cooldown" | "time-window" | "source-rate",
  ) {
    super(`Rate limited (${cause}) — retry after ${retryAfterMs}ms`);
    this.name = "RateLimitError";
  }
}

/**
 * Check if a *user* is actively using the source (the cooldown key is set only
 * by user-facing requests; worker fetches skip it via the backfill context).
 * Throws RateLimitError (retry after the key's remaining lifetime) if set.
 */
export async function checkCooldown(redis: RedisService, source: string): Promise<void> {
  const key = `prefetch:activity:${source}`;
  // ttl() returns the key's remaining lifetime in seconds (-2 = no key,
  // -1 = no expiry). Pause for exactly that long so the worker waits once —
  // until the real user-activity window ends — instead of re-checking every
  // fixed 60s and burning ~5 empty cycles per key lifetime.
  const ttl = await redis.ttl(key);
  if (ttl > 0) {
    throw new RateLimitError(ttl * 1000, "cooldown");
  }
}

/** Europe/Istanbul hour (0–23) at an arbitrary instant. */
function istanbulHourAt(ms: number): number {
  const hourStr = new Intl.DateTimeFormat("en-US", {
    timeZone: "Europe/Istanbul",
    hour: "numeric",
    hour12: false,
  }).format(new Date(ms));
  // `% 24` because the h24 hour cycle renders midnight as "24", which would put
  // the hour outside every window and silently park the source forever.
  return Number.parseInt(hourStr, 10) % 24;
}

/** Current hour (0–23) in Europe/Istanbul. */
function currentIstanbulHour(): number {
  return istanbulHourAt(Date.now());
}

/**
 * Whether `source` may be scraped right now.
 *
 * Time windows are DISABLED by default — PL24 (and everything else) may be
 * scraped 24/7. The window is still env-tunable: set PREFETCH_PL24_START /
 * PREFETCH_PL24_END (Europe/Istanbul hours) to re-narrow it (e.g.
 * PREFETCH_PL24_START=9 PREFETCH_PL24_END=18). EMEX and parts-catalogs never
 * had a window — their Redis-persisted warm JWT pool keeps captures alive
 * through the night.
 */
const PL24_WINDOW_START = Number(process.env.PREFETCH_PL24_START) || 0;
const PL24_WINDOW_END = Number(process.env.PREFETCH_PL24_END) || 24;

export function isWithinTimeWindow(source: string): boolean {
  if (source !== "pl24") return true;
  // Default (0–24) means the window is disabled → PL24 flows 24/7.
  if (PL24_WINDOW_START <= 0 && PL24_WINDOW_END >= 24) return true;
  const h = currentIstanbulHour();
  return h >= PL24_WINDOW_START && h < PL24_WINDOW_END;
}

/**
 * Check PL24 business hours (09:00–18:00 Europe/Istanbul).
 * Throws RateLimitError with delay until next 09:00 if outside window.
 */
export function checkTimeWindow(source: string): void {
  if (!isWithinTimeWindow(source)) {
    throw new RateLimitError(msUntilNext9AM(), "time-window");
  }
}

/**
 * The first instant at or after `fromMs` that falls inside `source`'s scrape
 * window. Returns `fromMs` unchanged when the window is disabled (the default
 * 0–24) or when `fromMs` is already inside it.
 *
 * WHY (plv2.md, finding consumers_jobs-03 — the budget/window deadlock):
 * `checkSourceDailyBudget` defers a spent source to the next UTC midnight. With
 * PREFETCH_PL24_START=9 that midnight lands at 03:00 Europe/Istanbul — six hours
 * BEFORE the window opens. The woken job therefore did no work, threw
 * `time-window`, and was deferred again to 09:00 — by which point the fresh
 * daily budget had already been spent by the same stampede of no-op wake-ups.
 * Measured on prod 2026-09-20: 600/600 pl24 budget consumed, 0 catalog requests
 * and 0 new categories for the whole day. Landing the deferral inside the window
 * breaks the cycle.
 */
export function alignToWindow(source: string, fromMs: number): number {
  if (source !== "pl24") return fromMs;
  if (PL24_WINDOW_START <= 0 && PL24_WINDOW_END >= 24) return fromMs;
  let t = fromMs;
  // Step by the hour rather than constructing a local-midnight date: DST-safe and
  // free of month/year rollover edge cases. 48 steps covers any window shape.
  for (let i = 0; i < 48; i++) {
    const h = istanbulHourAt(t);
    if (h >= PL24_WINDOW_START && h < PL24_WINDOW_END) return t;
    t += 3_600_000;
  }
  return t;
}

/**
 * Milliseconds until the next 09:00 Europe/Istanbul.
 */
export function msUntilNext9AM(): number {
  const now = new Date();

  // Get current Istanbul time components
  const istParts = new Intl.DateTimeFormat("en-US", {
    timeZone: "Europe/Istanbul",
    year: "numeric",
    month: "2-digit",
    day: "2-digit",
    hour: "2-digit",
    minute: "2-digit",
    second: "2-digit",
    hour12: false,
  }).formatToParts(now);

  const get = (type: string) =>
    Number.parseInt(istParts.find((p) => p.type === type)?.value || "0", 10);

  const hour = get("hour");
  const minute = get("minute");
  const second = get("second");

  // Before the window opens → wait until it opens today; otherwise → tomorrow.
  let hoursToWait: number;
  if (hour < PL24_WINDOW_START) {
    hoursToWait = PL24_WINDOW_START - hour;
  } else {
    hoursToWait = 24 - hour + PL24_WINDOW_START;
  }

  const ms = hoursToWait * 3600_000 - minute * 60_000 - second * 1000;

  // At least 1 minute, at most 15 hours
  return Math.max(60_000, Math.min(ms, 15 * 3600_000));
}

/** Redis hash key for prefetch progress */
export function progressKey(vehicleId: string): string {
  return `prefetch:progress:${vehicleId}`;
}

export interface PrefetchProgress {
  status: "running" | "completed" | "error";
  total: number;
  completed: number;
  errors: number;
  startedAt: string;
  updatedAt: string;
}

export async function initProgress(redis: RedisService, vehicleId: string): Promise<void> {
  const now = new Date().toISOString();
  await redis.setJson(
    progressKey(vehicleId),
    {
      status: "running",
      total: 0,
      completed: 0,
      errors: 0,
      startedAt: now,
      updatedAt: now,
    } satisfies PrefetchProgress,
    86400,
  );
}

export async function updateProgress(
  redis: RedisService,
  vehicleId: string,
  update: Partial<PrefetchProgress>,
): Promise<void> {
  const key = progressKey(vehicleId);
  const current = await redis.getJson<PrefetchProgress>(key);
  if (!current) return;

  const updated: PrefetchProgress = {
    ...current,
    ...update,
    updatedAt: new Date().toISOString(),
  };
  await redis.setJson(key, updated, 86400);
}

export async function getProgress(
  redis: RedisService,
  vehicleId: string,
): Promise<PrefetchProgress | null> {
  return redis.getJson<PrefetchProgress>(progressKey(vehicleId));
}