import { db, sensors, sources, sql, type Sensor, type Source } from "@websensor/db";
import { config, log } from "./config";
import { m } from "./metrics";
import { runSensor } from "./pipeline";
import { publishEngineStatus } from "./redis";

/**
 * Scheduler: claims due sensors with `FOR UPDATE SKIP LOCKED` (safe with several engine
 * processes), enforces global + per-host concurrency, and runs the pipeline. A claim moves
 * `next_check_at` 10 minutes ahead as a lease; the pipeline then sets the real next time.
 */
export class Scheduler {
  private inflight = 0;
  private perHost = new Map<string, number>();
  private stopped = false;
  private sourceCache = new Map<string, Source>();
  private sourceCacheAt = 0;
  private timer: NodeJS.Timeout | null = null;

  async start(): Promise<void> {
    log.info({ concurrency: config.fetchConcurrency, perHost: config.perHostConcurrency }, "scheduler started");
    const tick = async (): Promise<void> => {
      if (this.stopped) return;
      try {
        await this.tick();
      } catch (e) {
        log.error({ err: (e as Error).message }, "scheduler tick failed");
      }
      this.timer = setTimeout(tick, this.inflight >= config.fetchConcurrency ? 500 : 800);
    };
    void tick();
  }

  async stop(): Promise<void> {
    this.stopped = true;
    if (this.timer) clearTimeout(this.timer);
    const deadline = Date.now() + 30_000;
    while (this.inflight > 0 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 200));
  }

  private async refreshSources(): Promise<void> {
    if (Date.now() - this.sourceCacheAt < 60_000 && this.sourceCache.size) return;
    const rows = await db.select().from(sources);
    this.sourceCache = new Map(rows.map((r) => [r.id, r]));
    this.sourceCacheAt = Date.now();
  }

  private lastStatusAt = 0;
  private hostFailures = new Map<string, { count: number; until: number }>();

  /** Snapshot for the ops dashboard. */
  status(): Record<string, unknown> {
    const busyHosts = [...this.perHost.entries()].filter(([, n]) => n > 0).map(([h, n]) => ({ host: h, inflight: n }));
    const tripped = [...this.hostFailures.entries()].filter(([, f]) => f.until > Date.now()).map(([h, f]) => ({ host: h, failures: f.count, until: new Date(f.until).toISOString() }));
    return { inflight: this.inflight, concurrency: config.fetchConcurrency, perHostConcurrency: config.perHostConcurrency, busyHosts, circuitOpen: tripped, sources: this.sourceCache.size };
  }

  /** Domain circuit breaker (spec §86): after 5 consecutive failures on a host, pause it for 10 minutes. */
  noteHostOutcome(host: string, failed: boolean): void {
    if (!failed) {
      this.hostFailures.delete(host);
      return;
    }
    const f = this.hostFailures.get(host) ?? { count: 0, until: 0 };
    f.count++;
    if (f.count >= 5) f.until = Date.now() + 10 * 60e3;
    this.hostFailures.set(host, f);
  }

  private async tick(): Promise<void> {
    await this.refreshSources();
    const due = await db.execute<{ n: string }>(sql`select count(*)::text as n from sensors where enabled and next_check_at <= now()`);
    const dueN = Number(due.rows[0]?.n ?? 0);
    m.queueDue.set(dueN);
    if (Date.now() - this.lastStatusAt > 10_000) {
      this.lastStatusAt = Date.now();
      void publishEngineStatus({ ...this.status(), due: dueN, version: config.version });
    }
    const slots = config.fetchConcurrency - this.inflight;
    if (slots <= 0) return;
    // Priority-aware claim: critical sensors (priority 0) go first among what is due; within a priority the most
    // overdue first. We claim more candidates than free slots because many sensors share a few big hosts
    // (github.com, data.sec.gov…) and the per-host limit would otherwise leave slots idle.
    const claimed = await db.execute<Sensor>(sql`
      update sensors set next_check_at = now() + interval '10 minutes'
      where id in (select id from sensors where enabled and next_check_at <= now() order by priority asc, next_check_at asc limit ${slots * 4} for update skip locked)
      returning *`);
    let started = 0;
    for (const row of claimed.rows) {
      const sensor = normalizeRow(row as unknown as Record<string, unknown>);
      const host = safeHost(sensor.url);
      const hostLimit = config.perHostOverrides[host] ?? config.perHostConcurrency;
      const hostBusy = started >= slots || (this.perHost.get(host) ?? 0) >= hostLimit;
      if (hostBusy) {
        // give back the lease quickly (5–20 s, jittered so a big host drains smoothly)
        await db.update(sensors).set({ nextCheckAt: new Date(Date.now() + 5000 + Math.random() * 15000) }).where(sql`id = ${sensor.id}`);
        continue;
      }
      started++;
      const trip = this.hostFailures.get(host);
      if (trip && trip.until > Date.now()) {
        await db.update(sensors).set({ nextCheckAt: new Date(trip.until + Math.random() * 30_000) }).where(sql`id = ${sensor.id}`);
        continue;
      }
      const source = this.sourceCache.get(sensor.sourceId);
      if (!source || !source.enabled) {
        await db.update(sensors).set({ nextCheckAt: new Date(Date.now() + 3600e3) }).where(sql`id = ${sensor.id}`);
        continue;
      }
      this.inflight++;
      this.perHost.set(host, (this.perHost.get(host) ?? 0) + 1);
      m.inflight.set(this.inflight);
      void runSensor(sensor, source)
        .then((outcome) => this.noteHostOutcome(host, outcome === "error" || outcome === "rate_limited"))
        .catch(async (e) => {
          log.error({ sensor: sensor.id, err: (e as Error).stack ?? (e as Error).message }, "pipeline crashed");
          this.noteHostOutcome(host, true);
          await db.update(sensors).set({ nextCheckAt: new Date(Date.now() + 15 * 60e3), lastError: `pipeline: ${(e as Error).message}`.slice(0, 500), consecutiveErrors: sensor.consecutiveErrors + 1, health: "DEGRADED" }).where(sql`id = ${sensor.id}`).catch(() => undefined);
        })
        .finally(() => {
          this.inflight--;
          this.perHost.set(host, Math.max(0, (this.perHost.get(host) ?? 1) - 1));
          m.inflight.set(this.inflight);
        });
    }
  }
}

/** Raw `returning *` rows come back snake_case; map to the Drizzle camelCase shape. */
function normalizeRow(r: Record<string, unknown>): Sensor {
  const d = (v: unknown): Date | null => (v ? new Date(v as string) : null);
  return {
    id: r.id as string,
    sourceId: r.source_id as string,
    name: r.name as string,
    url: r.url as string,
    type: r.type as string,
    connector: r.connector as string,
    tier: r.tier as string,
    importanceWeight: Number(r.importance_weight),
    config: (r.config ?? {}) as Record<string, unknown>,
    baseIntervalSeconds: (r.base_interval_seconds as number | null) ?? null,
    enabled: Boolean(r.enabled),
    health: r.health as string,
    nextCheckAt: d(r.next_check_at) ?? new Date(),
    lastCheckAt: d(r.last_check_at),
    lastChangeAt: d(r.last_change_at),
    lastEventAt: d(r.last_event_at),
    lastStatus: (r.last_status as number | null) ?? null,
    lastError: (r.last_error as string | null) ?? null,
    etag: (r.etag as string | null) ?? null,
    lastModified: (r.last_modified as string | null) ?? null,
    state: (r.state as Record<string, unknown> | null) ?? null,
    lastSnapshotId: (r.last_snapshot_id as string | null) ?? null,
    consecutiveErrors: Number(r.consecutive_errors ?? 0),
    totalRuns: Number(r.total_runs ?? 0),
    totalNotModified: Number(r.total_not_modified ?? 0),
    rawChanges: Number(r.raw_changes ?? 0),
    meaningfulChanges: Number(r.meaningful_changes ?? 0),
    avgLatencyMs: (r.avg_latency_ms as number | null) ?? null,
    status: (r.status as string) ?? "ACTIVE",
    validatedAt: d(r.validated_at),
    priority: Number(r.priority ?? 2),
    createdAt: d(r.created_at) ?? new Date(),
    updatedAt: d(r.updated_at) ?? new Date(),
  };
}

function safeHost(u: string): string {
  try {
    return new URL(u).hostname;
  } catch {
    return u;
  }
}

/** Roll up connector health from the last 24 h of runs. */
export async function rollupConnectorHealth(): Promise<void> {
  await db.execute(sql`
    insert into connector_health (connector, status, runs_24h, errors_24h, success_rate, avg_latency_ms, changes_24h, events_24h, last_success_at, last_error_at, last_error, http_codes, updated_at)
    select s.connector,
      case when count(r.id) = 0 then 'UP'
           when sum(case when r.outcome in ('error','parse_error') then 1 else 0 end)::float / count(r.id) > 0.5 then 'ERROR'
           when bool_or(r.outcome = 'rate_limited') and max(r.started_at) filter (where r.outcome = 'rate_limited') > now() - interval '1 hour' then 'RATE_LIMITED'
           when sum(case when r.outcome in ('error','parse_error') then 1 else 0 end)::float / count(r.id) > 0.15 then 'DEGRADED'
           else 'UP' end,
      count(r.id)::int,
      sum(case when r.outcome in ('error','parse_error','rate_limited') then 1 else 0 end)::int,
      case when count(r.id) = 0 then null else 1 - sum(case when r.outcome in ('error','parse_error','rate_limited') then 1 else 0 end)::float / count(r.id) end,
      avg(r.duration_ms)::int,
      (select count(*) from changes c join sensors s2 on s2.id = c.sensor_id where s2.connector = s.connector and c.detected_at >= now() - interval '24 hours')::int,
      (select count(*) from events e join sensors s3 on s3.id = e.sensor_id where s3.connector = s.connector and e.detected_at >= now() - interval '24 hours')::int,
      max(r.started_at) filter (where r.outcome not in ('error','parse_error','rate_limited')),
      max(r.started_at) filter (where r.outcome in ('error','parse_error','rate_limited')),
      (array_agg(r.error order by r.started_at desc) filter (where r.error is not null))[1],
      coalesce((select jsonb_object_agg(code, n) from (select coalesce(r2.http_status, 0)::text as code, count(*) as n from sensor_runs r2 join sensors s4 on s4.id = r2.sensor_id where s4.connector = s.connector and r2.started_at >= now() - interval '24 hours' group by 1) x), '{}'::jsonb),
      now()
    from sensors s left join sensor_runs r on r.sensor_id = s.id and r.started_at >= now() - interval '24 hours'
    group by s.connector
    on conflict (connector) do update set status = excluded.status, runs_24h = excluded.runs_24h, errors_24h = excluded.errors_24h, success_rate = excluded.success_rate, avg_latency_ms = excluded.avg_latency_ms, changes_24h = excluded.changes_24h, events_24h = excluded.events_24h, last_success_at = excluded.last_success_at, last_error_at = excluded.last_error_at, last_error = excluded.last_error, http_codes = excluded.http_codes, updated_at = now()`);
}

/** Retention: raw fetch logs are short-lived; snapshots/events are kept. */
export async function pruneOldRuns(): Promise<void> {
  await db.execute(sql`delete from sensor_runs where started_at < now() - interval '14 days' and outcome in ('unchanged','not_modified')`);
  await db.execute(sql`delete from sensor_runs where started_at < now() - interval '60 days'`);
}