import { db } from "@/app/db/db";
import { tasks } from "@/app/db/schema/tasks/tasks";
import { subTasks } from "@/app/db/schema/tasks/subtask";
import { scenario } from "@/app/db/schema/recommendations";
import { and, desc, eq, inArray, sql } from "drizzle-orm";
import {
  RunFilters,
  RunStatus,
  buildRunRecord,
  deriveRunStatus,
  parseRunConfig,
  selectionSummary,
} from "./model";
import {
  builtFormOverrideJoin,
  propertyTypeOverrideJoin,
  resolvedBuiltFormSql,
  resolvedPropertyTypeSql,
} from "@/lib/services/epcSources";

// Distributor entry point on the Model backend. It fans the run out to
// workers, attaching execution sub-tasks to the task_id we pass (ADR-0008).
// NOTE: sub-task granularity is BATCHES, not properties — progress counts are
// internally consistent but never comparable to the previewed property count.
const DISTRIBUTOR_ENDPOINT = "/v1/modelling/trigger-run";

type BackendResult<T> = { ok: true; data: T } | { ok: false; status: number; message: string };

async function callBackend<T>(
  endpoint: string,
  payload: Record<string, unknown>,
  sessionToken: string | undefined,
): Promise<BackendResult<T>> {
  const url = process.env.FASTAPI_API_URL;
  const key = process.env.FASTAPI_API_KEY;
  if (!url || !key) {
    console.error("FASTAPI_API_URL or FASTAPI_API_KEY not set");
    return { ok: false, status: 500, message: "Server misconfiguration" };
  }
  try {
    const res = await fetch(`${url}${endpoint}`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": key,
        Authorization: `Bearer ${sessionToken}`,
      },
      body: JSON.stringify(payload),
    });
    if (!res.ok) {
      const errText = await res.text().catch(() => "");
      console.error(`FastAPI ${endpoint} failed:`, res.status, errText);
      return { ok: false, status: 502, message: "Modelling backend request failed" };
    }
    return { ok: true, data: (await res.json()) as T };
  } catch (err) {
    console.error(`Failed to reach FastAPI ${endpoint}:`, err);
    return { ok: false, status: 502, message: "Modelling backend request failed" };
  }
}

export type TriggerRunOutcome =
  | { kind: "ok"; taskId: string }
  | { kind: "unknown_scenarios"; missing: string[] }
  | { kind: "dispatch_failed"; status: number; message: string };

/**
 * Record + dispatch a Modelling run: one app-authored task whose `inputs`
 * carries the run's config (BulkUpload trigger convention), then hand
 * task_id to the distributor — sub_tasks are its, one per property. A
 * failed dispatch marks the task failed so history never shows a ghost run
 * as pending (ADR-0008).
 */
export async function triggerModellingRun(args: {
  portfolioId: string;
  scenarioIds: string[];
  filters: RunFilters;
  previewedPropertyCount: number;
  triggeredBy: string;
  sessionToken: string | undefined;
}): Promise<TriggerRunOutcome> {
  const known = await db
    .select({ id: scenario.id })
    .from(scenario)
    .where(
      and(
        eq(scenario.portfolioId, BigInt(args.portfolioId)),
        inArray(scenario.id, args.scenarioIds.map(BigInt)),
      ),
    );
  const knownIds = new Set(known.map((r) => r.id.toString()));
  const missing = args.scenarioIds.filter((id) => !knownIds.has(id));
  if (missing.length > 0) return { kind: "unknown_scenarios", missing };

  const record = buildRunRecord(args);
  const now = new Date();
  const [task] = await db
    .insert(tasks)
    .values({ ...record.task, jobStarted: now })
    .returning();

  // Minimal dispatch payload — previewed count and triggered_by are app
  // provenance, already durable on the task's inputs.
  const dispatch = await callBackend(
    DISTRIBUTOR_ENDPOINT,
    { task_id: task.id, ...record.dispatchPayload },
    args.sessionToken,
  );
  if (!dispatch.ok) {
    await db.update(tasks).set({ status: "failed" }).where(eq(tasks.id, task.id));
    return { kind: "dispatch_failed", status: dispatch.status, message: dispatch.message };
  }

  await db.update(tasks).set({ status: "in progress" }).where(eq(tasks.id, task.id));
  return { kind: "ok", taskId: task.id };
}

export interface PreviewCounts {
  matchedProperties: number;
  perScenario: { scenarioId: string; alreadyModelled: number }[];
}

/**
 * Count the properties a Run filter matches, plus how many of them already
 * hold plans per selected scenario. Resolution precedence per ADR-0008:
 * landlord-override snapshot → EPC-derived (new-approach graph, lodged over
 * predicted, or legacy property-row columns) → Unknown. The distributor
 * implements the same documented rule; both read this database.
 * One grouped, portfolio-scoped statement — no per-scenario probes.
 */
export async function previewModellingRun(args: {
  portfolioId: string;
  scenarioIds: string[];
  filters: RunFilters;
}): Promise<PreviewCounts> {
  const pid = BigInt(args.portfolioId);
  const inList = (values: (string | number)[]) =>
    sql.join(values.map((v) => sql`${v}`), sql`, `);
  const conditions = [sql`TRUE`];
  if (args.filters.postcodes) {
    conditions.push(sql`resolved.postcode IN (${inList(args.filters.postcodes)})`);
  }
  if (args.filters.propertyTypes) {
    conditions.push(sql`resolved.ptype IN (${inList(args.filters.propertyTypes)})`);
  }
  if (args.filters.builtForms) {
    conditions.push(sql`resolved.bform IN (${inList(args.filters.builtForms)})`);
  }
  if (args.filters.tagIds) {
    // Tags resolve by direct membership (any-of), not the override→EPC rule —
    // ADR-0013. The distributor mirrors this join against the shared DB.
    conditions.push(
      sql`resolved.id IN (SELECT pt.property_id FROM property_tag pt WHERE pt.tag_id IN (${inList(args.filters.tagIds)}))`,
    );
  }

  const result = await db.execute<{
    matched: number;
    per_scenario: { scenario_id: string; n: number }[] | null;
  }>(sql`
    WITH resolved AS (
      -- Resolution rule shared with the portfolio list via epcSources fragments
      -- so the preview and the list can never drift (ADR-0008, ADR-0012).
      SELECT p.id, p.postcode,
        ${resolvedPropertyTypeSql} AS ptype,
        ${resolvedBuiltFormSql} AS bform
      FROM property p
      ${propertyTypeOverrideJoin}
      ${builtFormOverrideJoin}
      LEFT JOIN epc_property epl ON epl.property_id = p.id AND epl.source = 'lodged'
      LEFT JOIN epc_property epp ON epp.property_id = p.id AND epp.source = 'predicted'
      WHERE p.portfolio_id = ${pid} AND p.marked_for_deletion = false
    ),
    matched AS (
      SELECT resolved.id FROM resolved
      WHERE ${sql.join(conditions, sql` AND `)}
    ),
    already AS (
      SELECT scenario_id, COUNT(DISTINCT property_id)::int AS n
      FROM plan
      WHERE portfolio_id = ${pid}
        AND scenario_id IN (${inList(args.scenarioIds.map(Number))})
        AND property_id IN (SELECT id FROM matched)
      GROUP BY scenario_id
    )
    SELECT
      (SELECT COUNT(*)::int FROM matched) AS matched,
      (SELECT json_agg(json_build_object('scenario_id', scenario_id::text, 'n', n)) FROM already) AS per_scenario
  `);

  const row = result.rows[0];
  return {
    matchedProperties: row?.matched ?? 0,
    perScenario: (row?.per_scenario ?? []).map((p) => ({
      scenarioId: p.scenario_id,
      alreadyModelled: p.n,
    })),
  };
}

export interface RunHistoryRow {
  taskId: string;
  triggeredAt: string;
  triggeredBy: string;
  scenarios: { id: string; name: string | null }[];
  selection: string;
  previewedPropertyCount: number;
  planCount: number;
  status: RunStatus;
}

/**
 * Run history for a portfolio: its modelling_run tasks with the app-authored
 * config parsed from task inputs and progress from sub-task counts (one
 * sub-task per property, distributor-attached). One grouped query —
 * portfolio-scoped, never per-run probes.
 */
export async function listModellingRuns(
  portfolioId: string,
  limit = 20,
): Promise<RunHistoryRow[]> {
  const rows = await db
    .select({
      taskId: tasks.id,
      jobStarted: tasks.jobStarted,
      jobCompleted: tasks.jobCompleted,
      status: tasks.status,
      configInputs: tasks.inputs,
      total: sql<number>`count(${subTasks.id})::int`,
      completed: sql<number>`count(case when lower(${subTasks.status}) in ('completed','complete') then 1 end)::int`,
      failed: sql<number>`count(case when lower(${subTasks.status}) in ('failed','failure','error') then 1 end)::int`,
    })
    .from(tasks)
    .leftJoin(subTasks, eq(subTasks.taskId, tasks.id))
    .where(and(eq(tasks.service, "modelling_run"), eq(tasks.sourceId, portfolioId)))
    .groupBy(tasks.id)
    .orderBy(desc(tasks.jobStarted))
    .limit(limit);

  const configs = rows.map((r) => parseRunConfig(r.configInputs));
  const scenarioIds = [...new Set(configs.flatMap((c) => c?.scenarioIds ?? []))];
  const names = scenarioIds.length
    ? await db
        .select({ id: scenario.id, name: scenario.name })
        .from(scenario)
        .where(inArray(scenario.id, scenarioIds.map(BigInt)))
    : [];
  const nameById = new Map(names.map((n) => [n.id.toString(), n.name]));

  return rows.map((r, i) => {
    const config = configs[i];
    const scenarios = (config?.scenarioIds ?? []).map((id) => ({
      id,
      name: nameById.get(id) ?? null,
    }));
    return {
      taskId: r.taskId,
      triggeredAt: (r.jobStarted ?? new Date(0)).toISOString(),
      triggeredBy: config?.triggeredBy ?? "unknown",
      scenarios,
      selection: config ? selectionSummary(config.filters) : "—",
      previewedPropertyCount: config?.previewedPropertyCount ?? 0,
      planCount: (config?.previewedPropertyCount ?? 0) * scenarios.length,
      status: deriveRunStatus({
        taskStatus: r.status,
        jobCompleted: r.jobCompleted,
        total: r.total,
        completed: r.completed,
        failed: r.failed,
      }),
    };
  });
}

/** Scenario ids with a run currently in flight — drives the third badge state. */
export async function scenarioIdsWithActiveRuns(portfolioId: string): Promise<Set<string>> {
  const runs = await listModellingRuns(portfolioId, 50);
  const active = new Set<string>();
  for (const run of runs) {
    if (run.status.state === "dispatched" || run.status.state === "in_progress") {
      run.scenarios.forEach((s) => active.add(s.id));
    }
  }
  return active;
}
