import { WorkflowEntrypoint } from "cloudflare:workers";
import { kickRepoInsights } from "../insights/repo-insights-do";
import type { WorkflowEvent, WorkflowStep } from "cloudflare:workers";
import { z } from "zod";
import { run } from "../db";
import { repoDb } from "../scoped-db";
import { isBackfilling, loadRepoById, loadRepoByArtifactsName } from "../repos";
import { loadUserById } from "../auth";
import { refreshRefs } from "../git/refs";
import { ZERO_SHA, withStore } from "../git/store";
import { compareCommits } from "../git/history";
import { diffTreesBounded } from "../git/diff";
import { checkMergeability, locatePullByIssueId, syncPullsForPush } from "../pulls";
import { scheduleWorkflows } from "../ci/scheduler";
import { indexRepoWorkflows } from "../ci/workflow-index";
import { forkPushNeedsApproval } from "./fork-approval";
import { emitWebhook, publicUrl, recordActivity } from "./emit";
import { pushEventId } from "./push-trigger";
import type { PushParams } from "./push-trigger";
import { closeIssuesFromCommits } from "../issues";
import { requestCodeIndex } from "../code/search-index";
import { warmLastCommits } from "../code/path-history";
import { isHiddenRef } from "../git/hidden-refs";
import { deploySitesForPush } from "../sites/triggers";
import { queuePushSecurity } from "../security/queue";
import type { CommitSummary } from "~/lib/types";
import { notifyMirrorOfPush } from "../mirror/notify";

const MAX_PUSH_COMMITS = 50;
const MAX_CHANGED_PATHS = 3000;
const MAX_MERGEABILITY_CHECKS = 10;
const MAX_PULL_CI_TRIGGERS = 20;

const internalPush = z.object({
  source: z.enum(["proxy", "internal", "artifacts"]),
  repoId: z.string().nullable(),
  artifactsName: z.string().nullable(),
  ref: z.string(),
  before: z.string(),
  after: z.string(),
  pusherId: z.string().nullable(),
});

const eventIdentity = z.object({ name: z.string(), email: z.string() });
const eventCommit = z.object({
  id: z.string(),
  message: z.string(),
  messageTruncated: z.boolean().optional(),
  timestamp: z.union([z.string(), z.number()]),
  author: eventIdentity,
});

const ARTIFACTS_EVENT_GRACE = "20 seconds";

const artifactsEvent = z.object({
  type: z.literal("cf.artifacts.repo.pushed"),
  source: z.object({ namespace: z.string(), repoName: z.string() }),
  payload: z.object({
    ref: z.string(),
    before: z.string(),
    after: z.string(),
    commits: z.array(z.unknown()).optional(),
    commitsTruncated: z.boolean().optional(),
  }),
});

/** Numeric timestamps below this are Unix seconds, not milliseconds. */
const MILLISECOND_EPOCH_THRESHOLD = 1e12;
const MS_PER_SECOND = 1000;

function eventTime(value: string | number): number {
  if (typeof value === "number") return value < MILLISECOND_EPOCH_THRESHOLD ? value * MS_PER_SECOND : value;
  return Date.parse(value);
}

/**
 * The pushed event already lists the pushed commits; when it is complete (nothing truncated, every entry
 * well-formed) it replaces the Artifacts history walk. Anything unexpected returns null and the walk runs.
 */
export function commitsFromEvent(payload: { commits?: unknown[]; commitsTruncated?: boolean }): PushCommit[] | null {
  if (!payload.commits || payload.commitsTruncated) return null;
  const commits: PushCommit[] = [];
  for (const raw of payload.commits) {
    const parsed = eventCommit.safeParse(raw);
    if (!parsed.success || parsed.data.messageTruncated) return null;
    const committedAt = eventTime(parsed.data.timestamp);
    if (!Number.isFinite(committedAt)) return null;
    commits.push({
      sha: parsed.data.id,
      subject: parsed.data.message.split("\n", 1)[0] ?? "",
      message: parsed.data.message,
      authorName: parsed.data.author.name,
      authorEmail: parsed.data.author.email,
      committedAt,
    });
  }
  return commits.sort((left, right) => right.committedAt - left.committedAt);
}

function normalize(payload: unknown): (PushParams & { eventCommits?: PushCommit[] | null }) | null {
  const internal = internalPush.safeParse(payload);
  if (internal.success) return internal.data;
  const event = artifactsEvent.safeParse(payload);
  if (!event.success) return null;
  return {
    source: "artifacts",
    repoId: null,
    artifactsName: event.data.source.repoName,
    ref: event.data.payload.ref,
    before: event.data.payload.before,
    after: event.data.payload.after,
    pusherId: null,
    eventCommits: commitsFromEvent(event.data.payload),
  };
}

type PushContext = {
  pushId: string;
  repoId: string;
  ref: string;
  before: string;
  after: string;
  pusherId: string | null;
  pusherName: string;
  branch: string | null;
  tag: string | null;
  isDefaultBranch: boolean;
  eventCommits: PushCommit[] | null;
  /** The forge itself relayed this ref from an import source; nobody "pushed" it. */
  importRelay: boolean;
};

type PushCommit = Pick<CommitSummary, "sha" | "subject" | "message" | "authorName" | "authorEmail" | "committedAt">;

/**
 * Whether an anonymous Artifacts event is the forge relaying refs from an import source rather than a
 * push. Decided when the event arrives, before any grace sleep, since the backfilling flag clears once
 * the relay finishes.
 */
async function isImportRelay(params: PushParams): Promise<boolean> {
  if (params.pusherId !== null || !params.artifactsName) return false;
  const repo = await loadRepoByArtifactsName(params.artifactsName);
  return repo?.import_source_url != null && (repo.status === "importing" || isBackfilling(repo));
}

async function recordPush(
  params: PushParams & { eventCommits?: PushCommit[] | null; importRelay?: boolean },
): Promise<PushContext | null> {
  const repo = params.repoId
    ? await loadRepoById(params.repoId)
    : params.artifactsName
      ? await loadRepoByArtifactsName(params.artifactsName)
      : null;
  if (!repo) return null;
  const pushId = await pushEventId(repo.artifacts_name, params.ref, params.before, params.after);
  const inserted = await repoDb(repo.id).run(
    `INSERT OR IGNORE INTO push_events (id, repo_id, ref, before_sha, after_sha, pusher_id, created_at)
     VALUES (?, ?, ?, ?, ?, ?, ?)`,
    pushId,
    repo.id,
    params.ref,
    params.before,
    params.after,
    params.pusherId,
    Date.now(),
  );
  if (inserted.meta.changes === 0) {
    if (params.pusherId) await claimAnonymousPush(repo.id, pushId, params);
    return null;
  }
  const pusher = params.pusherId ? await loadUserById(params.pusherId) : null;
  const branch = params.ref.startsWith("refs/heads/") ? params.ref.slice("refs/heads/".length) : null;
  const tag = params.ref.startsWith("refs/tags/") ? params.ref.slice("refs/tags/".length) : null;
  return {
    pushId,
    repoId: repo.id,
    ref: params.ref,
    before: params.before,
    after: params.after,
    pusherId: params.pusherId,
    pusherName: pusher?.name ?? "git",
    branch,
    tag,
    isDefaultBranch: branch === repo.default_branch,
    eventCommits: params.eventCommits ?? null,
    importRelay: params.importRelay ?? false,
  };
}

/**
 * The Artifacts push event carries no forge user, and it can beat the proxy's own trigger (which only
 * fires once the receive-pack response has streamed back). When the proxy's attributed run arrives
 * second, give the push and its activity rows the pusher instead of leaving them as a deleted user.
 */
async function claimAnonymousPush(repoId: string, pushId: string, params: PushParams): Promise<void> {
  const db = repoDb(repoId);
  const claimed = await db.run(
    "UPDATE push_events SET pusher_id = ? WHERE id = ? AND pusher_id IS NULL",
    params.pusherId,
    pushId,
  );
  if (claimed.meta.changes === 0) return;
  const refName = params.ref.replace(/^refs\/(heads|tags)\//, "");
  await db.run(
    `UPDATE activity SET actor_id = ? WHERE repo_id = ? AND actor_id IS NULL AND kind IN ('push', 'ref_created', 'ref_deleted')
       AND json_extract(data, '$.ref') = ? AND json_extract(data, '$.before') = ? AND json_extract(data, '$.after') = ?`,
    params.pusherId,
    repoId,
    refName,
    params.before,
    params.after,
  );
}

async function updateRepoRefs(context: PushContext): Promise<void> {
  const repo = await loadRepoById(context.repoId);
  if (!repo) return;
  const refs = await refreshRefs(repo);
  const now = Date.now();
  const hasDefault = refs.branches.some((branch) => branch.name === repo.default_branch);
  const fallback = refs.head ?? refs.branches[0]?.name ?? null;
  if (!hasDefault && fallback) {
    await run(
      "UPDATE repos SET default_branch = ?, pushed_at = ?, updated_at = ? WHERE id = ?",
      fallback,
      now,
      now,
      repo.id,
    );
  } else {
    await run("UPDATE repos SET pushed_at = ?, updated_at = ? WHERE id = ?", now, now, repo.id);
  }
}

async function pushCommits(context: PushContext): Promise<{ commits: PushCommit[]; changedPaths: string[] | null }> {
  if (context.after === ZERO_SHA) return { commits: [], changedPaths: null };
  const repo = await loadRepoById(context.repoId);
  if (!repo) return { commits: [], changedPaths: null };
  return withStore(repo.artifacts_name, async (store) => {
    let commits: PushCommit[];
    let changedPaths: string[] | null = null;
    if (context.before !== ZERO_SHA) {
      commits = context.eventCommits ?? (await compareCommits(store, context.before, store, context.after)).commits;
      const [beforeCommit, afterCommit] = await Promise.all([
        store.commit(context.before),
        store.commit(context.after),
      ]);
      if (beforeCommit && afterCommit) {
        const { changes, truncated } = await diffTreesBounded(store, beforeCommit.treeSha, afterCommit.treeSha);
        if (!truncated && changes.length <= MAX_CHANGED_PATHS) {
          changedPaths = changes.flatMap((change) => (change.oldPath ? [change.path, change.oldPath] : [change.path]));
        }
      }
    } else {
      commits = context.tag ? [] : await store.log(context.after, MAX_PUSH_COMMITS, 0);
      if (!context.isDefaultBranch) commits = commits.slice(0, 1);
    }
    return {
      commits: commits.slice(0, MAX_PUSH_COMMITS).map((commit) => ({
        sha: commit.sha,
        subject: commit.subject,
        message: commit.message,
        authorName: commit.authorName,
        authorEmail: commit.authorEmail,
        committedAt: commit.committedAt,
      })),
      changedPaths,
    };
  });
}

/**
 * The proxy's attributed trigger may claim this push after the anonymous Artifacts run started, so the
 * stored pusher is re-read before writing rows that would otherwise land on a deleted user.
 */
async function resolvePusherId(context: PushContext): Promise<string | null> {
  if (context.pusherId) return context.pusherId;
  const row = await repoDb(context.repoId).first<{ pusher_id: string | null }>(
    "SELECT pusher_id FROM push_events WHERE id = ?",
    context.pushId,
  );
  return row?.pusher_id ?? null;
}

/** Covers a claim that lands between `resolvePusherId` and the activity insert. */
async function adoptClaimedPusher(context: PushContext, refName: string): Promise<void> {
  await repoDb(context.repoId).run(
    `UPDATE activity SET actor_id = (SELECT pusher_id FROM push_events WHERE id = ?)
     WHERE repo_id = ? AND actor_id IS NULL AND kind IN ('push', 'ref_created', 'ref_deleted')
       AND json_extract(data, '$.ref') = ? AND json_extract(data, '$.before') = ? AND json_extract(data, '$.after') = ?`,
    context.pushId,
    context.repoId,
    refName,
    context.before,
    context.after,
  );
}

async function announcePush(context: PushContext, commits: PushCommit[]): Promise<void> {
  const repo = await loadRepoById(context.repoId);
  if (!repo) return;
  const actorId = await resolvePusherId(context);
  const created = context.before === ZERO_SHA;
  const deleted = context.after === ZERO_SHA;
  const refName = context.branch ?? context.tag ?? context.ref;
  await recordActivity(actorId, repo.id, deleted ? "ref_deleted" : created ? "ref_created" : "push", {
    ref: refName,
    refType: context.tag ? "tag" : "branch",
    before: context.before,
    after: context.after,
    commits: commits.slice(0, 5).map((commit) => ({ sha: commit.sha, subject: commit.subject })),
    total: commits.length,
  });
  if (!actorId) await adoptClaimedPusher(context, refName);
  const repository = {
    id: repo.id,
    name: repo.name,
    full_name: `${repo.owner_name}/${repo.name}`,
    html_url: publicUrl(`/${repo.owner_name}/${repo.name}`),
    clone_url: publicUrl(`/${repo.owner_name}/${repo.name}.git`),
    default_branch: repo.default_branch,
    private: repo.visibility === "private",
  };
  const sender = { login: context.pusherName, id: context.pusherId };
  if (created || deleted) {
    await emitWebhook(repo.id, created ? "create" : "delete", {
      ref: refName,
      ref_type: context.tag ? "tag" : "branch",
      repository,
      sender,
    });
  }
  await emitWebhook(repo.id, "push", {
    ref: context.ref,
    before: context.before,
    after: context.after,
    created,
    deleted,
    commits: commits.map((commit) => ({
      id: commit.sha,
      message: commit.message,
      timestamp: new Date(commit.committedAt).toISOString(),
      author: { name: commit.authorName, email: commit.authorEmail },
      url: publicUrl(`/${repo.owner_name}/${repo.name}/commit/${commit.sha}`),
    })),
    repository,
    pusher: { name: context.pusherName },
    sender,
  });
}

type PullCheck = { issueId: string };

async function syncPulls(context: PushContext): Promise<{ checks: PullCheck[]; ciPulls: Array<{ issueId: string }> }> {
  if (!context.branch) return { checks: [], ciPulls: [] };
  const repo = await loadRepoById(context.repoId);
  if (!repo) return { checks: [], ciPulls: [] };
  const result = await syncPullsForPush(repo, context.branch, context.after, context.pusherId);
  return {
    checks: result.synchronized.slice(0, MAX_MERGEABILITY_CHECKS).map((item) => ({ issueId: item.pull.issue_id })),
    ciPulls: result.synchronized.slice(0, MAX_PULL_CI_TRIGGERS).map((item) => ({ issueId: item.pull.issue_id })),
  };
}

async function scheduleCi(context: PushContext, changedPaths: string[] | null, headline: string): Promise<number> {
  if (context.after === ZERO_SHA) return 0;
  const repo = await loadRepoById(context.repoId);
  if (!repo) return 0;
  const runs = await scheduleWorkflows(repo, {
    event: "push",
    action: null,
    ref: context.ref,
    sha: context.after,
    actorId: context.pusherId,
    actorName: context.pusherName,
    title: headline,
    prNumber: null,
    headRef: null,
    baseRef: null,
    changedPaths,
    inputs: {},
    workflowPath: null,
    sourceRepo: repo,
  });
  return runs.length;
}

async function schedulePullCi(issueId: string, context: PushContext): Promise<number> {
  const pull = await locatePullByIssueId(issueId);
  if (!pull) return 0;
  const [baseRepo, headRepo] = await Promise.all([loadRepoById(pull.repo_id), loadRepoById(pull.head_repo_id)]);
  if (!baseRepo || !headRepo) return 0;
  const runs = await scheduleWorkflows(
    baseRepo,
    {
      event: "pull_request",
      action: "synchronize",
      ref: `refs/pull/${pull.number}/head`,
      sha: pull.head_sha,
      actorId: context.pusherId,
      actorName: context.pusherName,
      title: pull.title,
      prNumber: pull.number,
      headRef: pull.head_ref,
      baseRef: pull.base_ref,
      changedPaths: null,
      inputs: {},
      workflowPath: null,
      sourceRepo: headRepo,
    },
    { requireApproval: await forkPushNeedsApproval(baseRepo, headRepo, pull.author_id, context.pusherId) },
  );
  return runs.length;
}

/** Durable post-receive pipeline: refs cache, activity, webhooks, pull request sync, issue closing, CI. */
export class PushWorkflow extends WorkflowEntrypoint<Env, unknown> {
  override async run(event: Readonly<WorkflowEvent<unknown>>, step: WorkflowStep): Promise<void> {
    const params = normalize(event.payload);
    if (!params || isHiddenRef(params.ref)) return;
    // Pushes through the forge proxy also raise an anonymous Artifacts event; let the attributed
    // proxy run record the push first so the activity feed and webhooks name the pusher.
    const relay = params.source === "artifacts" && (await step.do("classify", () => isImportRelay(params)));
    if (params.source === "artifacts" && !relay) await step.sleep("yield-to-proxy", ARTIFACTS_EVENT_GRACE);
    const context = await step.do("record", () => recordPush({ ...params, importRelay: relay }));
    if (!context) return;

    await step.do("refs", () => updateRepoRefs(context));
    await step.do("insights", () => kickRepoInsights(context.repoId));
    const { commits, changedPaths } = await step.do("commits", () => pushCommits(context));
    if (context.importRelay) {
      // Imported refs are not pushes: the "imported from" activity already covers them, and the imported
      // history must not fire webhooks, close issues or run CI as if someone had just pushed it.
      await step.do("security", () =>
        queuePushSecurity({ ...context, commitShas: commits.map((commit) => commit.sha), changedPaths }),
      );
      if (context.isDefaultBranch) await step.do("ci-index", () => indexRepoWorkflows(context.repoId, context.after));
      return;
    }
    await step.do("announce", () => announcePush(context, commits));
    await step.do("mirror", () => notifyMirrorOfPush(context.repoId, context.ref, context.after));
    const pulls = await step.do("pulls", () => syncPulls(context));
    if (context.isDefaultBranch && commits.length > 0) {
      await step.do("close-issues", () =>
        closeIssuesFromCommits(
          context.repoId,
          context.pusherId,
          commits.map((commit) => ({ sha: commit.sha, message: commit.message })),
        ),
      );
    }
    if (context.isDefaultBranch && context.after !== ZERO_SHA) {
      await step.do("code-index", () => requestCodeIndex(context.repoId));
      await step
        .do("warm-tree", () => warmLastCommits(context.repoId))
        .catch((error: unknown) => console.warn("tree warm failed", context.repoId, error));
    }
    const headline = commits[0]?.subject ?? `${context.branch ?? context.tag ?? context.ref} updated`;
    await step.do("ci-push", () => scheduleCi(context, changedPaths, headline));
    await step.do("sites", () => deploySitesForPush(context.repoId, context.branch, context.after, context.pusherId));
    await step.do("security", () =>
      queuePushSecurity({ ...context, commitShas: commits.map((commit) => commit.sha), changedPaths }),
    );
    if (context.isDefaultBranch) await step.do("ci-index", () => indexRepoWorkflows(context.repoId, context.after));
    for (const pull of pulls.ciPulls) {
      await step.do(`ci-pull-${pull.issueId}`, () => schedulePullCi(pull.issueId, context));
    }
    for (const check of pulls.checks) {
      await step.do(`mergeable-${check.issueId}`, { retries: { limit: 2, delay: "10 seconds" } }, async () => {
        const pull = await locatePullByIssueId(check.issueId);
        if (!pull || pull.state !== "open") return "skipped";
        return checkMergeability(pull);
      });
    }
  }
}