import { WorkflowEntrypoint } from "cloudflare:workers";
import type { WorkflowEvent, WorkflowStep } from "cloudflare:workers";
import { z } from "zod";
import { run } from "../db";
import { 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, loadPullByIssueId, syncPullsForPush } from "../pulls";
import { scheduleWorkflows } from "../ci/scheduler";
import { indexRepoWorkflows } from "../ci/workflow-index";
import { forkRunNeedsApproval } from "../pull-events";
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 { isHiddenRef } from "../git/hidden-refs";
import { deployPagesForPush } from "../pages/triggers";
import { deploySitesForPush } from "../sites/triggers";
import { queuePushSecurity } from "../security/queue";
import type { CommitSummary } from "~/lib/types";

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 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() }),
});

function normalize(payload: unknown): PushParams | 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,
  };
}

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;
};

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

async function recordPush(params: PushParams): 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 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) 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,
  };
}

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: CommitSummary[];
    let changedPaths: string[] | null = null;
    if (context.before !== ZERO_SHA) {
      const comparison = await compareCommits(store, context.before, store, context.after);
      commits = comparison.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,
    };
  });
}

async function announcePush(context: PushContext, commits: PushCommit[]): Promise<void> {
  const repo = await loadRepoById(context.repoId);
  if (!repo) return;
  const created = context.before === ZERO_SHA;
  const deleted = context.after === ZERO_SHA;
  const refName = context.branch ?? context.tag ?? context.ref;
  await recordActivity(context.pusherId, 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,
  });
  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 loadPullByIssueId(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 forkRunNeedsApproval(baseRepo, headRepo, 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;
    const context = await step.do("record", () => recordPush(params));
    if (!context) return;

    await step.do("refs", () => updateRepoRefs(context));
    const { commits, changedPaths } = await step.do("commits", () => pushCommits(context));
    await step.do("announce", () => announcePush(context, commits));
    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));
    }
    const headline = commits[0]?.subject ?? `${context.branch ?? context.tag ?? context.ref} updated`;
    await step.do("ci-push", () => scheduleCi(context, changedPaths, headline));
    await step.do("pages", () => deployPagesForPush(context.repoId, context.branch, context.after, context.pusherId));
    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 loadPullByIssueId(check.issueId);
        if (!pull || pull.state !== "open") return "skipped";
        return checkMergeability(pull);
      });
    }
  }
}