/**
 * IntakeService — subscriptions, the poll scheduler, ingest, and the search
 * surface (firehose phase 1).
 *
 * Scheduler = the live-monitor pattern (task requirement 1): one serialized
 * sweep, per-source state in the DB (`next_poll_at` / `consecutive_failures`
 * / `health`), exponential backoff 2^n capped at 15 min, +/−20 % jitter on
 * every schedule so a fleet of replicas never thundering-herds one host, and
 * a Redis lease + cooldown per source so replicas never double-poll. Sources
 * self-schedule via next_poll_at; the sweep only has to notice.
 *
 * Failures classify exactly like BackoffPollDriver: transient (429/5xx/
 * budget/DNS) → backoff with Retry-After as a lower bound, health='backoff';
 * policy denials and permanent shapes (404/410, bad params) → health='error'
 * with the much longer cap; push-only kinds (poll_interval_sec = 0) are
 * never due and park in 'unsupported' when first seen (027 note).
 */
import { BadRequestException, Inject, Injectable, Logger, NotFoundException, Optional, type OnApplicationBootstrap, type OnApplicationShutdown } from "@nestjs/common";
import { randomUUID } from "node:crypto";
import { pool, withTenant } from "../db/client.js";
import { S3Storage, s3ConfigFromEnv, type StorageService } from "../storage/storage.js";
import type { ControlRedis } from "../media-control/redis.js";
import { MemoryFeedRedis } from "../feeds/testing.js";
import { intakeEgress, IntakeHttpError, intakeRetryAfterMs, type IntakeEgress } from "./intake-egress.js";
import { DEFAULT_MEDIA_POLICY, DEFAULT_POLL_INTERVAL_SEC, HOST_CLASS_BY_KIND, isIntakeKind, KIND_TELEGRAM_CHANNEL, PROVIDER_BY_KIND, type IntakeKind, type MediaPolicy } from "./intake-kinds.js";
import { parseIntakeParams, paramsHash, storedParams, type IntakeParams } from "./intake-params.js";
import { adapterFor } from "./adapters/index.js";
import type { AdapterState } from "./adapters/adapter.js";
import { MediaHarvester, type HarvestedMedia } from "./media-harvest.js";
import { itemContentHash, normalizeRawItem, upsertItems, type UpsertResult } from "./intake-ingest.js";
import { searchIntake, listIntakeItems, type IntakeQueryFilters } from "./intake-search.js";

export const INTAKE_DEPENDENCIES = Symbol("INTAKE_DEPENDENCIES");

export interface IntakeDependencies {
  egress: IntakeEgress;
  storage: StorageService;
  redis: Pick<ControlRedis, "get" | "set" | "eval">;
  random(): number;
  now(): Date;
  /** Sweep period; defaults to 15 s (min cadence is 60 s, so this is ample). */
  sweepMs?: number;
}

/** Backoff curve caps (live-monitor: 2^n; intake polls are slower, cap 15 min). */
export const BACKOFF_CAP_MS = 15 * 60_000;
export const PERMANENT_ERROR_MS = 6 * 60 * 60_000;
export const LEASE_MS = 60_000;

const RENEW_LEASE = "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end";
const RELEASE_LEASE = "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end";

interface SourceRow {
  id: string;
  workspace_id: string;
  kind: IntakeKind;
  params: Record<string, unknown>;
  poll_interval_sec: number;
  state: AdapterState;
  health: string;
  consecutive_failures: number;
}

export function backoffDelayMs(failures: number, intervalMs: number): number {
  return Math.min(BACKOFF_CAP_MS, Math.max(intervalMs, 1_000) * 2 ** Math.min(6, Math.max(0, failures - 1)));
}

/** +/−20 % jitter (live-monitor jitteredInterval shape). */
export function jitteredIntervalMs(intervalMs: number, random: number): number {
  if (!Number.isFinite(intervalMs) || intervalMs <= 0) throw new Error("interval must be positive");
  if (!Number.isFinite(random) || random < 0 || random > 1) throw new Error("random must be between 0 and 1");
  return Math.round(intervalMs * (0.8 + random * 0.4));
}

function defaults(): IntakeDependencies {
  return {
    egress: undefined as never, // bound lazily by ensureDeps
    storage: new S3Storage(s3ConfigFromEnv()),
    redis: new MemoryFeedRedis(),
    random: Math.random,
    now: () => new Date(),
    sweepMs: 15_000,
  };
}

@Injectable()
export class IntakeService implements OnApplicationBootstrap, OnApplicationShutdown {
  private readonly logger = new Logger(IntakeService.name);
  private readonly deps: IntakeDependencies;
  private readonly egress: IntakeEgress | undefined;
  private running = false;
  private timer?: ReturnType<typeof setTimeout>;
  private active?: Promise<void>;
  private harvester?: MediaHarvester;

  constructor(@Optional() @Inject(INTAKE_DEPENDENCIES) dependencies?: Partial<IntakeDependencies>) {
    const base = defaults();
    this.deps = { ...base, ...dependencies };
    // The egress client binds lazily (native ESM import); tests inject a fake.
    if (dependencies?.egress !== undefined) this.egress = dependencies.egress;
    if (this.egress !== undefined) {
      const nowMs = (): number => this.deps.now().getTime();
      this.harvester = new MediaHarvester(this.egress, this.deps.storage, nowMs);
    }
  }

  onApplicationBootstrap(): void {
    this.running = true;
    this.schedule(0);
  }

  async onApplicationShutdown(): Promise<void> {
    this.running = false;
    if (this.timer) clearTimeout(this.timer);
    await this.active;
  }

  get isRunning(): boolean { return this.running; }

  private schedule(delayMs: number): void {
    if (!this.running) return;
    const sweep = this.deps.sweepMs ?? 15_000;
    this.timer = setTimeout(() => {
      this.timer = undefined;
      this.active = this.pollDueSources().catch((error: unknown) => {
        this.logger.error(`intake sweep failed: ${String(error)}`);
      }).then(() => undefined).finally(() => {
        this.active = undefined;
        this.schedule(jitteredIntervalMs(sweep, this.deps.random()));
      });
    }, delayMs);
    this.timer.unref();
  }

  // -------------------------------------------------------------------------
  // Subscriptions
  // -------------------------------------------------------------------------

  async createSubscription(workspaceId: string, kind: unknown, paramsRaw: unknown,
    options: { pollIntervalSec?: unknown; mediaPolicy?: unknown } = {}): Promise<{ id: string; kind: IntakeKind; health: string }> {
    const parsed = parseIntakeParams(kind, paramsRaw);
    const hash = paramsHash(parsed);
    const kindDefaults = DEFAULT_POLL_INTERVAL_SEC[parsed.kind];
    const mediaPolicy = parseMediaPolicy(options.mediaPolicy, DEFAULT_MEDIA_POLICY[parsed.kind]);
    let interval = kindDefaults;
    const rawInterval: unknown = options.pollIntervalSec;
    if (rawInterval !== undefined && rawInterval !== null && rawInterval !== "") {
      const n = typeof rawInterval === "number" ? rawInterval : typeof rawInterval === "string" ? Number(rawInterval) : NaN;
      if (!Number.isInteger(n) || n < 0 || n > 86_400) throw new BadRequestException("pollIntervalSec must be an integer 0..86400 (0 = push-only)");
      if (kindDefaults === 0 && n > 0) throw new BadRequestException("this kind is push-only; pollIntervalSec must be 0");
      interval = n > 0 ? Math.max(60, Math.min(n, 86_400)) : 0;
    }
    return withTenant(workspaceId, async (client) => {
      const result = await client.query<{ id: string; health: string }>(
        `INSERT INTO intake_sources (workspace_id, kind, params, params_hash, poll_interval_sec, state, health)
         VALUES ($1,$2,$3,$4,$5,$6,$7)
         ON CONFLICT (workspace_id, kind, params_hash) DO UPDATE SET updated_at = now()
         RETURNING id, health`,
        [workspaceId, parsed.kind, JSON.stringify({ ...parsed.params, mediaPolicy }), hash, interval, JSON.stringify(mediaPolicy ? { mediaPolicy } : {}),
          kindDefaults === 0 ? "unsupported" : "active"]);
      const row = result.rows[0];
      if (!row) throw new Error("intake source insert failed");
      return { id: String(row.id), kind: parsed.kind, health: row.health };
    });
  }

  async listSources(workspaceId: string): Promise<Record<string, unknown>[]> {
    return withTenant(workspaceId, async (client) => {
      const result = await client.query(
        `SELECT id, kind, params, poll_interval_sec, health, last_error, last_polled_at, next_poll_at,
                consecutive_failures, created_at, updated_at
           FROM intake_sources WHERE workspace_id = $1 ORDER BY created_at`, [workspaceId]);
      return result.rows.map((row) => ({
        id: String(row.id), kind: row.kind, params: row.params,
        pollIntervalSec: Number(row.poll_interval_sec), health: row.health,
        lastError: row.last_error, lastPolledAt: row.last_polled_at?.toISOString() ?? null,
        nextPollAt: row.next_poll_at?.toISOString() ?? null,
        consecutiveFailures: Number(row.consecutive_failures),
        createdAt: row.created_at.toISOString(), updatedAt: row.updated_at.toISOString(),
      }));
    });
  }

  async getSource(workspaceId: string, sourceId: string): Promise<Record<string, unknown>> {
    const rows = await withTenant(workspaceId, async (client) => client.query(
      "SELECT * FROM intake_sources WHERE workspace_id = $1 AND id = $2", [workspaceId, sourceId]));
    const row = rows.rows[0];
    if (!row) throw new NotFoundException("intake source not found");
    return {
      id: String(row.id), kind: row.kind, params: row.params,
      pollIntervalSec: Number(row.poll_interval_sec), health: row.health,
      lastError: row.last_error, lastPolledAt: row.last_polled_at?.toISOString() ?? null,
      nextPollAt: row.next_poll_at?.toISOString() ?? null, state: row.state,
      consecutiveFailures: Number(row.consecutive_failures),
      createdAt: row.created_at.toISOString(), updatedAt: row.updated_at.toISOString(),
    };
  }

  async setSourceHealth(workspaceId: string, sourceId: string, health: "active" | "paused"): Promise<void> {
    if (health !== "active" && health !== "paused") throw new BadRequestException("health must be active or paused");
    await withTenant(workspaceId, async (client) => {
      const result = await client.query(
        `UPDATE intake_sources SET health = $3, updated_at = now(),
           next_poll_at = CASE WHEN $3 = 'active' THEN now() ELSE next_poll_at END
         WHERE workspace_id = $1 AND id = $2`, [workspaceId, sourceId, health]);
      if (result.rowCount === 0) throw new NotFoundException("intake source not found");
    });
  }

  async deleteSource(workspaceId: string, sourceId: string): Promise<void> {
    await withTenant(workspaceId, async (client) => {
      const result = await client.query(
        "DELETE FROM intake_sources WHERE workspace_id = $1 AND id = $2", [workspaceId, sourceId]);
      if (result.rowCount === 0) throw new NotFoundException("intake source not found");
    });
  }

  // -------------------------------------------------------------------------
  // Polling
  // -------------------------------------------------------------------------

  /** One sweep over every workspace's due sources; per-source failures never abort the sweep. */
  async pollDueSources(): Promise<{ polled: number; skipped: number; failed: number }> {
    if (this.egress === undefined) return { polled: 0, skipped: 0, failed: 0 };
    const workspaces = await pool.query<{ id: string }>("SELECT id FROM workspaces ORDER BY id");
    const totals = { polled: 0, skipped: 0, failed: 0 };
    for (const { id } of workspaces.rows) {
      const pass = await this.pollWorkspaceDue(id);
      totals.polled += pass.polled;
      totals.skipped += pass.skipped;
      totals.failed += pass.failed;
    }
    return totals;
  }

  async pollWorkspaceDue(workspaceId: string): Promise<{ polled: number; skipped: number; failed: number }> {
    const due = await withTenant(workspaceId, async (client) => client.query<SourceRow>(
      `SELECT id, workspace_id, kind, params, poll_interval_sec, state, health, consecutive_failures
         FROM intake_sources
        WHERE workspace_id = $1
          AND health NOT IN ('paused','unsupported')
          AND poll_interval_sec > 0
          AND (next_poll_at IS NULL OR next_poll_at <= now())
        ORDER BY COALESCE(next_poll_at, created_at), created_at`, [workspaceId]));
    const totals = { polled: 0, skipped: 0, failed: 0 };
    for (const row of due.rows) {
      const outcome = await this.pollSource(workspaceId, row);
      totals[outcome] += 1;
    }
    return totals;
  }

  /**
   * Poll one source under its Redis lease. The lease makes a multi-replica
   * deployment safe (one owner at a time; renewed on a heartbeat, released on
   * completion, expires on crash). Spacing between polls is enforced solely
   * by next_poll_at in the DB — the single source of truth the scheduler
   * reads — so a manual due-reset (or an ops force-poll) is honored rather
   * than silently eaten by a side-channel cooldown.
   */
  async pollSource(workspaceId: string, row: SourceRow): Promise<"polled" | "skipped" | "failed"> {
    if (this.egress === undefined || this.harvester === undefined) return "skipped";
    const intervalMs = Math.max(60_000, row.poll_interval_sec * 1_000);
    const leaseKey = `intake:${workspaceId}:${row.id}:lease`;
    const owner = randomUUID();
    if (!await this.deps.redis.set(leaseKey, owner, { nx: true, pxMs: LEASE_MS })) return "skipped";
    let renew: Promise<unknown> = Promise.resolve();
    let heartbeat: ReturnType<typeof setInterval> | undefined;
    try {
      heartbeat = setInterval(() => {
        renew = this.deps.redis.eval(RENEW_LEASE, [leaseKey], [owner, String(LEASE_MS)]).catch(() => {
          this.logger.warn("intake poll lease renewal unavailable");
        });
      }, LEASE_MS / 3);
      heartbeat.unref();

      const params = storedParams(row.kind, row.params);
      const adapter = adapterFor(row.kind);
      const fetcher = {
        fetch: (url: string | URL, options: { headers?: Record<string, string>; hostClass: string }) =>
          this.egress!.egressFetch(url, { headers: options.headers ?? {} }, { hostClass: options.hostClass }),
      };
      const result = await adapter.poll({
        workspaceId, sourceId: row.id, params, kind: row.kind, state: row.state ?? {}, fetch: fetcher,
      });

      // Harvest media per the source's policy, then normalize + upsert.
      const policy = sourceMediaPolicy(row);
      const harvested: HarvestedMedia[][] = [];
      for (const item of result.items) {
        harvested.push(await this.harvester.harvestAll(policy, item.media ?? []));
      }
      const rows = result.items.map((raw, index) => ({
        raw,
        outcome: normalizeRawItem(raw, row.kind, params, workspaceId, row.id),
        harvested: harvested[index] ?? [],
      }));
      const upsert: UpsertResult = await withTenant(workspaceId, async (client) =>
        upsertItems(client, workspaceId, row.id, row.kind, rows));
      void upsert;
      await withTenant(workspaceId, async (client) => client.query(
        `UPDATE intake_sources SET state = $3, health = 'active', last_error = NULL, last_polled_at = now(),
           consecutive_failures = 0, next_poll_at = now() + ($4::bigint * interval '1 millisecond'), updated_at = now()
         WHERE workspace_id = $1 AND id = $2`,
        [workspaceId, row.id, JSON.stringify(result.state), jitteredIntervalMs(intervalMs, this.deps.random())]));
      return "polled";
    } catch (error) {
      await this.recordFailure(workspaceId, row, error, intervalMs);
      return "failed";
    } finally {
      if (heartbeat) clearInterval(heartbeat);
      await renew;
      await this.deps.redis.eval(RELEASE_LEASE, [leaseKey], [owner]).catch(() => undefined);
    }
  }

  private async recordFailure(workspaceId: string, row: SourceRow, error: unknown, intervalMs: number): Promise<void> {
    const now = this.deps.now().getTime();
    const http = error instanceof IntakeHttpError ? error : undefined;
    const egress = this.egress?.isEgressError(error) === true ? error : undefined;
    const retryAfter = http !== undefined
      ? intakeRetryAfterMs(http.headers, now)
      : egress !== undefined ? egress.retryAfterMs ?? 0 : 0;
    const transient = (http !== undefined && http.transient)
      || egress !== undefined && ["BUDGET_EXHAUSTED", "BUDGET_UNAVAILABLE", "TIMEOUT", "NETWORK_ERROR", "ABORTED", "DNS_FAILED"].includes(egress.code)
      || http === undefined && egress === undefined; // network shape unknown → benefit of the doubt
    const permanent = (http !== undefined && http.gone)
      || egress !== undefined && ["DENIED_PRIVATE_ADDRESS", "SCHEME_DENIED", "PORT_DENIED", "ORIGIN_DENIED", "INVALID_URL"].includes(egress.code);
    const failures = row.consecutive_failures + 1;
    const base = permanent ? PERMANENT_ERROR_MS : transient ? backoffDelayMs(failures, intervalMs) : PERMANENT_ERROR_MS;
    const delay = Math.max(retryAfter, jitteredIntervalMs(base, this.deps.random()));
    const reason = permanent ? "permanent" : transient ? "transient" : "unknown";
    const message = http !== undefined ? `http ${http.status}`
      : egress !== undefined ? `egress:${egress.code}`
        : error instanceof Error ? error.message.slice(0, 240) : "poll failed";
    await withTenant(workspaceId, async (client) => client.query(
      `UPDATE intake_sources SET health = $3, last_error = $4, last_polled_at = now(),
         consecutive_failures = $5, next_poll_at = now() + ($6::bigint * interval '1 millisecond'), updated_at = now()
       WHERE workspace_id = $1 AND id = $2`,
      [workspaceId, row.id, permanent ? "error" : "backoff", `${reason}: ${message}`, failures, delay],
    )).catch(() => this.logger.warn("intake source state write-back unavailable"));
  }

  // -------------------------------------------------------------------------
  // Telegram archive bridge (task requirement: reuse the existing harvest)
  // -------------------------------------------------------------------------

  /**
   * Bridges an archived_feed_items row into intake_items. The archive webhook
   * path (apps/api/src/archive) already normalized the post, harvested media
   * into S3 content-addressed keys and enforced revisions — intake stores the
   * reference, it never re-fetches.
   */
  async ingestArchiveItem(workspaceId: string, archiveItemId: string): Promise<void> {
    const found = await withTenant(workspaceId, async (client) => client.query<{
      id: string; remote_ref: string; channel_id: string; message_id: string; text_content: string;
      media_refs: Array<{ storageKey?: string; contentType?: string; sizeBytes?: number | null }>;
      attribution: Record<string, unknown>; posted_at: Date;
    }>("SELECT * FROM archived_feed_items WHERE id = $1", [archiveItemId]));
    const item = found.rows[0];
    if (!item) throw new NotFoundException("archive item not found");
    const channelId = String(item.channel_id);
    await withTenant(workspaceId, async (client) => {
      const source = await client.query<{ id: string; params: Record<string, unknown> }>(
        `SELECT id, params FROM intake_sources WHERE workspace_id = $1 AND kind = $2
           AND params->>'channelId' = $3`, [workspaceId, KIND_TELEGRAM_CHANNEL, channelId]);
      const sourceRow = source.rows[0];
      if (!sourceRow) return; // not subscribed in intake — nothing to do
      const mediaKeys = (item.media_refs ?? []).map((media) => media.storageKey ?? "").filter((key) => key !== "");
      const bridgeHash = itemContentHash(item.text_content, mediaKeys);
      await client.query(
        `INSERT INTO intake_items (workspace_id, source_id, remote_ref, content_hash, published_at,
           text, content, media_refs, attribution)
         VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
         ON CONFLICT (workspace_id, content_hash) DO NOTHING`,
        [workspaceId, sourceRow.id, item.remote_ref, bridgeHash,
          item.posted_at.toISOString(), item.text_content,
          JSON.stringify({ type: "doc", version: 1, children: [{ type: "paragraph", children: [{ type: "text", text: item.text_content }] }], truncated: false }),
          JSON.stringify(item.media_refs.map((media) => ({ kind: "image", proxyUrl: "", storageKey: media.storageKey ?? null, contentType: media.contentType ?? null, sizeBytes: media.sizeBytes ?? null }))),
          JSON.stringify({ ...item.attribution, provider: PROVIDER_BY_KIND[KIND_TELEGRAM_CHANNEL], archiveItemId: item.id })]);
    });
  }

  // -------------------------------------------------------------------------
  // Read surface (search + per-source listing)
  // -------------------------------------------------------------------------

  search(workspaceId: string, query: Record<string, unknown>): Promise<{ items: Record<string, unknown>[]; nextCursor: string | null }> {
    return searchIntake(workspaceId, parseSearchQuery(query), this.deps.storage);
  }

  listItems(workspaceId: string, query: Record<string, unknown>): Promise<{ items: Record<string, unknown>[]; nextCursor: string | null }> {
    return listIntakeItems(workspaceId, parseSearchQuery(query), this.deps.storage);
  }
}

function parseMediaPolicy(value: unknown, fallback: MediaPolicy): MediaPolicy {
  if (value === undefined || value === null || value === "") return fallback;
  if (value !== "metadata" && value !== "images" && value !== "full") {
    throw new BadRequestException("mediaPolicy must be metadata, images, or full");
  }
  return value;
}

function sourceMediaPolicy(row: SourceRow): MediaPolicy {
  const fromParams = row.params.mediaPolicy;
  if (fromParams === "metadata" || fromParams === "images" || fromParams === "full") return fromParams;
  return DEFAULT_MEDIA_POLICY[row.kind];
}

/** Shared query parsing for /intake/search and /intake/items. */
export function parseSearchQuery(query: Record<string, unknown>): IntakeQueryFilters {
  const rawQ = query.q;
  const q = typeof rawQ === "string" && rawQ.trim() !== "" ? rawQ.trim() : null;
  if (q !== null && q.length > 256) throw new BadRequestException("q must be at most 256 characters");
  const rawSource = query.sourceId;
  const sourceId = rawSource === undefined || rawSource === null || rawSource === "" ? null
    : typeof rawSource === "string" && /^[0-9a-f-]{36}$/i.test(rawSource) ? rawSource : (() => { throw new BadRequestException("sourceId must be a UUID"); })();
  const rawKind = query.kind;
  const kind = rawKind === undefined || rawKind === null || rawKind === "" ? null
    : isIntakeKind(rawKind) ? rawKind : (() => { throw new BadRequestException("kind is not a supported intake kind"); })();
  const cursor = typeof query.cursor === "string" && query.cursor !== "" ? query.cursor : undefined;
  if (cursor !== undefined && (cursor.length > 256 || !/^[A-Za-z0-9_-]+$/.test(cursor))) throw new BadRequestException("invalid cursor");
  let limit = 20;
  if (query.limit !== undefined && query.limit !== "") {
    const n = typeof query.limit === "string" ? Number(query.limit)
      : typeof query.limit === "number" ? query.limit : NaN;
    if (!Number.isInteger(n) || n < 1 || n > 100) throw new BadRequestException("limit must be an integer between 1 and 100");
    limit = n;
  }
  return { q, sourceId, kind, cursor, limit };
}