import { env, 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 { isArtifactsError, openArtifactsRepo, rememberInitialToken } from "./git/artifacts";
import { artifactsCall, ArtifactsUnavailableError } from "./git/artifacts-call";
import { artifactsBinding } from "./git/artifacts-shards";
import { BACKFILL_START, loadRepoByArtifactsName, loadRepoById } from "./repos";
import { refreshRepoStatus } from "./repo-ops";
import { sourceAuthorization, syncRefsFromSource } from "./git/ref-sync";
import { invalidateRefs } from "./git/refs";
import { dropHiddenRefsForRepo, requestHiddenRefCleanup } from "./git/pull-bridge";
import { decryptRepoSecret } from "./secrets";
import { requestCodeIndex } from "./code/search-index";
import { warmLastCommits } from "./code/path-history";
import { notifyProvisioningFailed } from "./notify/import";

const provisionParams = z.discriminatedUnion("kind", [
  z.object({
    repoId: z.string(),
    kind: z.literal("import"),
    sourceUrl: z.string(),
    branch: z.string().nullable(),
    /** Encrypted source credentials for private imports; the Worker relays the pack itself. */
    auth: z.object({ ciphertext: z.string(), iv: z.string() }).nullable().optional(),
  }),
  z.object({ repoId: z.string(), kind: z.literal("fork"), sourceRepoId: z.string() }),
]);

export type ProvisionParams = z.infer<typeof provisionParams>;

/** Small forks finish within seconds; after that the forked event wakes the loop and polling is a fallback. */
const FAST_POLLS = 2;
const MAX_POLLS = 32;
const FAST_INTERVAL = "5 seconds";
const SLOW_INTERVAL = "1 minute";
const BEGIN_TIMEOUT = "15 minutes";
const RELAY_STEP = { retries: { limit: 2, delay: "30 seconds" }, timeout: "15 minutes" } as const;
const RELAY_EVENT_SETTLE = "15 seconds";
/** How long an import's backfill window is trusted even if the workflow never closes it. */
const IMPORT_BACKFILL_WINDOW_MS = 6 * 60 * 60 * 1000;
const MAX_BACKFILL_PASSES = 20;

const ERROR_TEXT: Partial<Record<string, string>> = {
  REMOTE_AUTH_REQUIRED:
    "the source returned 401: it is private or does not exist. Private repositories need an access token",
  NOT_FOUND: "the source repository was not found",
};

export function describeProvisioningError(error: unknown): string {
  if (isArtifactsError(error)) return ERROR_TEXT[error.code] ?? error.code.toLowerCase().replace(/_/g, " ");
  if (error instanceof ArtifactsUnavailableError) return error.message;
  return error instanceof Error && error.message ? error.message.slice(0, 300) : "unknown error";
}

export async function markProvisioningFailed(repoId: string, reason: string): Promise<void> {
  const result = await run(
    "UPDATE repos SET status = 'failed', status_error = ?, updated_at = ? WHERE id = ? AND status != 'failed'",
    reason,
    Date.now(),
    repoId,
  );
  if (result.meta.changes > 0) await notifyProvisioningFailed(repoId, reason);
}

/** Hands an import or fork to a durable Workflow so it survives the browser leaving the page. */
export async function startProvisioning(params: ProvisionParams): Promise<void> {
  try {
    await env.REPO_PROVISION.create({ id: `provision-${params.repoId}`, params });
  } catch (error) {
    await markProvisioningFailed(params.repoId, describeProvisioningError(error));
    throw error;
  }
}

/**
 * Creates the target repository (imports) or starts the Artifacts fork. Imports don't use Artifacts'
 * import, which copies one branch and has hung on large remotes; the Workflow relays the primary branch
 * first and backfills the rest. Idempotent across retries: an existing target repo is reused.
 */
async function begin(params: ProvisionParams): Promise<"started" | "gone" | "failed"> {
  const repo = await loadRepoById(params.repoId);
  if (!repo) return "gone";
  try {
    let created: { remote: string; defaultBranch: string; token?: string };
    if (params.kind === "import") {
      created = await artifactsCall("create the repository", repo.artifacts_name, () =>
        artifactsBinding(repo.artifacts_name).create(repo.artifacts_name, {
          description: repo.description ?? undefined,
        }),
      );
    } else {
      const source = await loadRepoById(params.sourceRepoId);
      if (!source) {
        await markProvisioningFailed(repo.id, "the source repository was deleted");
        return "failed";
      }
      // A full fork copies the source's refs/pull and refs/gitfor objects; keep the want-must-be-visible guard.
      await run(
        "UPDATE repos SET bridged_into = 1 WHERE id = ? AND bridged_into = 0 AND (SELECT bridged_into FROM repos WHERE id = ?) = 1",
        repo.id,
        source.id,
      );
      const handle = await openArtifactsRepo(source.artifacts_name);
      try {
        created = await artifactsCall(`fork ${source.artifacts_name}`, repo.artifacts_name, () =>
          handle.fork(repo.artifacts_name, {
            description: source.description ?? undefined,
            defaultBranchOnly: false,
          }),
        );
      } finally {
        handle[Symbol.dispose]();
      }
    }
    await rememberInitialToken(repo.artifacts_name, created.token);
    await run(
      "UPDATE repos SET artifacts_remote = ?, default_branch = ?, updated_at = ? WHERE id = ?",
      created.remote,
      created.defaultBranch || repo.default_branch,
      Date.now(),
      repo.id,
    );
    return "started";
  } catch (error) {
    if (isArtifactsError(error) && (error.code === "ALREADY_EXISTS" || error.code.endsWith("_IN_PROGRESS"))) {
      return "started";
    }
    if (isArtifactsError(error)) {
      await markProvisioningFailed(repo.id, describeProvisioningError(error));
      return "failed";
    }
    throw error;
  }
}

const forkedEvent = z.object({
  type: z.literal("cf.artifacts.repo.forked"),
  payload: z.object({ repoName: z.string() }),
});

const FORKED_EVENT = "artifacts-forked";

/**
 * `cf.artifacts.repo.forked` (wired in wrangler.jsonc) starts a fresh instance of this Workflow; it only
 * wakes the provisioning instance waiting on that fork, so readiness doesn't wait for the next poll.
 */
async function relayForkedEvent(payload: unknown): Promise<boolean> {
  const parsed = forkedEvent.safeParse(payload);
  if (!parsed.success) return false;
  const repo = await loadRepoByArtifactsName(parsed.data.payload.repoName);
  if (!repo || repo.status !== "forking") return true;
  try {
    const instance = await env.REPO_PROVISION.get(`provision-${repo.id}`);
    await instance.sendEvent({ type: FORKED_EVENT, payload: { repoId: repo.id } });
  } catch (error) {
    console.warn("couldn't wake fork provisioning; it will poll", repo.id, error);
  }
  return true;
}

export class RepoProvisionWorkflow extends WorkflowEntrypoint<Env, unknown> {
  override async run(event: Readonly<WorkflowEvent<unknown>>, step: WorkflowStep): Promise<void> {
    if (await step.do("artifacts-event", () => relayForkedEvent(event.payload))) return;
    const parsed = provisionParams.safeParse(event.payload);
    if (!parsed.success) {
      console.error("RepoProvisionWorkflow started with an unknown payload", parsed.error.message);
      return;
    }
    const params = parsed.data;
    let started: Awaited<ReturnType<typeof begin>>;
    try {
      started = await step.do(
        "begin",
        { retries: { limit: 3, delay: "10 seconds", backoff: "exponential" }, timeout: BEGIN_TIMEOUT },
        () => begin(params),
      );
    } catch (error) {
      await step.do("begin-failed", () => markProvisioningFailed(params.repoId, describeProvisioningError(error)));
      return;
    }
    if (started !== "started") return;
    if (params.kind === "import") {
      await this.relayImport(params, step);
      return;
    }

    for (let poll = 0; poll < MAX_POLLS; poll++) {
      if (poll < FAST_POLLS) await step.sleep(`wait-${poll}`, FAST_INTERVAL);
      else {
        try {
          await step.waitForEvent(`forked-${poll}`, { type: FORKED_EVENT, timeout: SLOW_INTERVAL });
        } catch {
          // Timed out: poll anyway in case the event was missed.
        }
      }
      const status = await step.do(`check-${poll}`, async () => {
        const repo = await loadRepoById(params.repoId);
        if (!repo) return "gone";
        return (await refreshRepoStatus(repo)).status;
      });
      if (status === "ready") {
        await this.finishReady(params, step);
        return;
      }
      if (status !== "importing" && status !== "forking") return;
    }
    await step.do("timeout", () => markProvisioningFailed(params.repoId, "timed out waiting for Artifacts"));
  }

  /**
   * Relays the primary branch, marks the repository ready, then backfills the remaining branches and
   * tags in bounded passes so a remote with hundreds of refs can't stall the import.
   */
  private async relayImport(params: ProvisionParams & { kind: "import" }, step: WorkflowStep): Promise<void> {
    const { sourceUrl, auth, branch } = params;
    const authorization = async () =>
      auth ? sourceAuthorization(await decryptRepoSecret(auth.ciphertext, auth.iv)) : null;
    try {
      await step.do("import-head", RELAY_STEP, async () => {
        const repo = await loadRepoById(params.repoId);
        if (!repo) return;
        const result = await syncRefsFromSource(repo, sourceUrl, await authorization(), { only: "head", branch });
        const now = Date.now();
        await run(
          `UPDATE repos SET status = 'ready', ${BACKFILL_START}, default_branch = ?3, pushed_at = ?1, updated_at = ?1
           WHERE id = ?4 AND status = 'importing'`,
          now,
          now + IMPORT_BACKFILL_WINDOW_MS,
          result.headBranch ?? repo.default_branch,
          repo.id,
        );
        await invalidateRefs(repo.id);
      });
    } catch (error) {
      await step.do("import-failed", () => markProvisioningFailed(params.repoId, describeProvisioningError(error)));
      return;
    }
    await step.do("code-index", () => requestCodeIndex(params.repoId));
    await step
      .do("warm-tree", RELAY_STEP, () => warmLastCommits(params.repoId))
      .catch((error: unknown) => console.error("tree warm failed", params.repoId, error));
    let backfillError: string | null = null;
    let remaining = 0;
    for (let pass = 0; pass < MAX_BACKFILL_PASSES; pass++) {
      try {
        remaining = await step.do(`sync-refs-${pass}`, RELAY_STEP, async () => {
          const repo = await loadRepoById(params.repoId);
          if (!repo) return 0;
          // Each pass keeps the relay window open for as long as a pass may take, however many passes there are.
          await run(
            "UPDATE repos SET backfill_expires_at = MAX(COALESCE(backfill_expires_at, 0), ?) WHERE id = ?",
            Date.now() + IMPORT_BACKFILL_WINDOW_MS,
            repo.id,
          );
          return (await syncRefsFromSource(repo, sourceUrl, await authorization())).skipped;
        });
      } catch (error) {
        console.error("ref backfill failed", params.repoId, error);
        backfillError = describeProvisioningError(error);
        break;
      }
      if (remaining === 0) break;
    }
    if (!backfillError && remaining > 0) backfillError = `${remaining} refs were not imported`;
    // Artifacts push events for the relayed refs trail the relay; PushWorkflow reads `backfilling` to know they
    // are imports rather than pushes, so keep it up until they have landed.
    await step.sleep("relay-settle", RELAY_EVENT_SETTLE);
    await step.do("backfill-done", async () => {
      await run(
        "UPDATE repos SET backfilling = MAX(backfilling - 1, 0), backfill_error = ?, updated_at = ? WHERE id = ?",
        backfillError,
        Date.now(),
        params.repoId,
      );
      await kickRepoInsights(params.repoId);
    });
  }

  private async finishReady(params: ProvisionParams, step: WorkflowStep): Promise<void> {
    if (params.kind === "fork") {
      await step.do("drop-inherited-hidden-refs", async () => {
        await requestHiddenRefCleanup(params.repoId);
        await dropHiddenRefsForRepo(params.repoId);
      });
    }
    await step.do("code-index", () => requestCodeIndex(params.repoId));
  }
}