import type { CacheContext } from "@/worker/cache";
import type { Logger } from "@/worker/common/logger";
import type { OrderedPackSnapshot } from "@/worker/git/operations/fetch/types";
import type { RepoDurableObject } from "@/worker/do/repo/repoDO";

import {
  type CompactionDeleteQueueMessage,
  type CompactionQueueMessage,
  type RepoQueueMessageHandle,
} from "./types";

import { getRepoStubByDoId } from "@/worker/common";
import { buildCompactionNeededOids } from "@/worker/git/compaction/plan";
import { type SubrequestLimiter } from "@/worker/git/operations/limits";
import { scanPack, resolveDeltasAndWriteIdx } from "@/worker/git/pack/indexer";
import { rewritePackResult } from "@/worker/git/pack/rewrite";
import { loadOrderedPackSnapshot } from "@/worker/git/pack/snapshot";
import {
  deleteStagedPack,
  stagePackToR2,
  type StagedPackUpload,
} from "@/worker/git/receive/r2Upload";
import { doPrefix, packIndexKey, packRefsKey, r2PackKey } from "@/worker/keys";
import { createQueueTaskContext, logSoftBudgetExhausted, retryQueueMessage } from "./context";

const COMPACTION_SUBREQUEST_BUDGET = 7_500;
const COMPACTION_RETRY_DELAY_SECONDS = 30;
const COMPACTION_CONFLICT_RETRY_DELAY_SECONDS = 10;
const COMPACTION_DELETE_DELAY_SECONDS = 60;

function countCompactionSubrequest(cacheCtx: CacheContext, log: Logger, op: string, n = 1): void {
  logSoftBudgetExhausted({
    cacheCtx,
    log,
    flagPrefix: "compaction-soft-budget",
    op,
    count: n,
  });
}

async function cleanupStagedCompaction(args: {
  stagedUpload: StagedPackUpload | undefined;
  log: Logger;
  reason: string;
}) {
  if (!args.stagedUpload) return;
  try {
    await deleteStagedPack(args.stagedUpload);
  } catch (error) {
    args.log.warn("compaction:cleanup-failed", {
      reason: args.reason,
      packKey: args.stagedUpload.packKey,
      error: String(error),
    });
  }
}

async function abortCompactionLease(args: {
  stub: DurableObjectStub<RepoDurableObject>;
  leaseToken: string | undefined;
  limiter: SubrequestLimiter;
  cacheCtx: CacheContext;
  log: Logger;
  reason: string;
}) {
  const leaseToken = args.leaseToken;
  if (!leaseToken) return;
  try {
    countCompactionSubrequest(args.cacheCtx, args.log, "do:abort-compaction");
    const cleared = await args.limiter.run("do:abort-compaction", async () => {
      return await args.stub.abortCompaction(leaseToken);
    });
    if (!cleared) {
      args.log.warn("compaction:abort-missed", {
        reason: args.reason,
        leaseToken: args.leaseToken,
      });
      return;
    }
    args.log.info("compaction:abort-complete", {
      reason: args.reason,
      leaseToken,
    });
  } catch (error) {
    args.log.warn("compaction:abort-failed", {
      reason: args.reason,
      leaseToken: args.leaseToken,
      error: String(error),
    });
  }
}

async function clearCompactionRequestAfterBlocked(args: {
  stub: DurableObjectStub<RepoDurableObject>;
  limiter: SubrequestLimiter;
  cacheCtx: CacheContext;
  log: Logger;
  reason: string;
}): Promise<void> {
  try {
    countCompactionSubrequest(args.cacheCtx, args.log, "do:clear-compaction-request");
    await args.limiter.run("do:clear-compaction-request", async () => {
      await args.stub.clearCompactionRequest();
    });
    args.log.warn("compaction:blocked-cleared", { reason: args.reason });
  } catch (error) {
    args.log.warn("compaction:blocked-clear-failed", {
      reason: args.reason,
      error: String(error),
    });
  }
}

export async function handleCompactionMessage(
  message: Omit<RepoQueueMessageHandle<CompactionQueueMessage>, "body">,
  body: CompactionQueueMessage,
  env: Env,
  ctx: ExecutionContext
): Promise<void> {
  const repoLabel = body.repoId || `do:${body.doId}`;
  const task = createQueueTaskContext({
    env,
    ctx,
    repoLabel,
    operation: "compaction",
    subrequestBudget: COMPACTION_SUBREQUEST_BUDGET,
  });
  const log = task.logFor({
    service: "CompactionQueue",
    repoId: repoLabel,
    doId: body.doId,
  });
  const stub = getRepoStubByDoId(env, body.doId) as DurableObjectStub<RepoDurableObject>;
  const { cacheCtx, limiter } = task;

  let stagedUpload: StagedPackUpload | undefined;
  let leaseToken: string | undefined;

  try {
    countCompactionSubrequest(cacheCtx, log, "do:begin-compaction");
    const begin = await limiter.run("do:begin-compaction", async () => {
      return await stub.beginCompaction();
    });
    if (!begin.ok) {
      if (begin.status === "busy" && begin.reason === "receive-active") {
        log.info("compaction:busy-retry", { reason: begin.reason });
        retryQueueMessage(message, COMPACTION_CONFLICT_RETRY_DELAY_SECONDS);
        return;
      }

      log.info("compaction:skip", {
        status: begin.status,
        reason: "reason" in begin ? begin.reason : undefined,
      });
      message.ack();
      return;
    }

    leaseToken = begin.lease.token;
    cacheCtx.memo = cacheCtx.memo || {};
    cacheCtx.memo.packCatalog = begin.activeCatalog;

    const snapshotLoad = await loadOrderedPackSnapshot(env, repoLabel, cacheCtx, log);
    if (snapshotLoad.type !== "Ready") {
      log.warn("compaction:snapshot-unavailable", { reason: snapshotLoad.reason });
      await abortCompactionLease({
        stub,
        leaseToken,
        limiter,
        cacheCtx,
        log,
        reason: snapshotLoad.reason,
      });
      retryQueueMessage(message, COMPACTION_RETRY_DELAY_SECONDS);
      return;
    }

    const snapshot = snapshotLoad.snapshot;
    const sourcePackMap = new Map(snapshot.packs.map((pack) => [pack.packKey, pack]));
    const sourcePacks = begin.sourcePacks
      .map((row) => sourcePackMap.get(row.packKey))
      .filter((pack): pack is (typeof snapshot.packs)[number] => pack !== undefined);
    if (sourcePacks.length !== begin.sourcePacks.length) {
      log.warn("compaction:source-pack-missing", {
        expected: begin.sourcePacks.length,
        actual: sourcePacks.length,
      });
      await abortCompactionLease({
        stub,
        leaseToken,
        limiter,
        cacheCtx,
        log,
        reason: "source-pack-missing",
      });
      retryQueueMessage(message, COMPACTION_CONFLICT_RETRY_DELAY_SECONDS);
      return;
    }

    const neededOids = buildCompactionNeededOids(sourcePacks);
    log.info("compaction:rewrite-start", {
      sourceTier: begin.targetTier - 1,
      targetTier: begin.targetTier,
      sourceCount: begin.sourcePacks.length,
      neededCount: neededOids.length,
    });

    // Build a compaction-specific snapshot: source packs first so
    // resolveOrderedEntryByOid picks authoritative source entries for needed
    // OIDs, then remaining active packs in their normal newest-first order
    // for delta base closure. Without this reorder, a duplicate identity
    // REF_DELTA in a newer non-source pack can shadow the source entry and
    // create a self-referential delta cycle in the topology sort.
    const sourceKeySet = new Set(begin.sourcePacks.map((row) => row.packKey));
    const fallbackPacks = snapshot.packs.filter((pack) => !sourceKeySet.has(pack.packKey));
    const compactionSnapshot: OrderedPackSnapshot = {
      packs: [...sourcePacks, ...fallbackPacks],
    };

    const rewriteResult = await rewritePackResult(env, compactionSnapshot, neededOids, {
      limiter,
      countSubrequest: (n) => countCompactionSubrequest(cacheCtx, log, "r2:rewrite-pack", n),
    });
    if (rewriteResult.status !== "ok") {
      log.warn("compaction:rewrite-unavailable", {
        reason: rewriteResult.failure.reason,
        retryable: rewriteResult.failure.retryable,
        details: rewriteResult.failure.details,
      });
      await abortCompactionLease({
        stub,
        leaseToken,
        limiter,
        cacheCtx,
        log,
        reason: rewriteResult.failure.reason,
      });

      if (rewriteResult.failure.retryable) {
        retryQueueMessage(message, COMPACTION_RETRY_DELAY_SECONDS);
        return;
      }

      leaseToken = undefined;
      await clearCompactionRequestAfterBlocked({
        stub,
        limiter,
        cacheCtx,
        log,
        reason: rewriteResult.failure.reason,
      });
      log.error("compaction:blocked", {
        reason: rewriteResult.failure.reason,
        sourceCount: begin.sourcePacks.length,
        sourceSeqLo: begin.sourcePacks[0]?.seqLo,
        sourceSeqHi: begin.sourcePacks[begin.sourcePacks.length - 1]?.seqHi,
      });
      message.ack();
      return;
    }

    const packKey = r2PackKey(doPrefix(body.doId), `pack-cmp-${begin.lease.token}.pack`);
    stagedUpload = await stagePackToR2({
      env,
      request: new Request(`https://queue.internal/${encodeURIComponent(repoLabel)}/compact-pack`),
      packStream: rewriteResult.stream,
      packKey,
      bytesConsumed: 0,
      limiter,
      countSubrequest: (op, n = 1) => countCompactionSubrequest(cacheCtx, log, op, n),
    });

    const scanResult = await scanPack({
      env,
      packKey: stagedUpload.packKey,
      packSize: stagedUpload.packBytes,
      limiter,
      countSubrequest: (n = 1) => countCompactionSubrequest(cacheCtx, log, "r2:scan-pack", n),
      log,
    });
    const resolveResult = await resolveDeltasAndWriteIdx({
      env,
      packKey: stagedUpload.packKey,
      packSize: stagedUpload.packBytes,
      limiter,
      countSubrequest: (n = 1) => countCompactionSubrequest(cacheCtx, log, "r2:resolve-pack", n),
      log,
      scanResult,
      activeCatalog: begin.activeCatalog,
      cacheCtx,
      repoId: repoLabel,
    });

    countCompactionSubrequest(cacheCtx, log, "do:commit-compaction");
    const committedUpload = stagedUpload;
    const commit = await limiter.run("do:commit-compaction", async () => {
      return await stub.commitCompaction({
        token: begin.lease.token,
        sourcePacks: begin.sourcePacks,
        targetTier: begin.targetTier,
        packsetVersion: begin.packsetVersion,
        stagedPack: {
          packKey: committedUpload.packKey,
          packBytes: committedUpload.packBytes,
          idxBytes: resolveResult.idxBytes,
          objectCount: resolveResult.objectCount,
        },
      });
    });

    if (commit.status === "retry") {
      await cleanupStagedCompaction({
        stagedUpload,
        log,
        reason: commit.reason,
      });
      leaseToken = undefined;
      log.info("compaction:retry", { reason: commit.reason });
      retryQueueMessage(message, COMPACTION_CONFLICT_RETRY_DELAY_SECONDS);
      return;
    }

    leaseToken = undefined;
    if (commit.shouldRequeue) {
      ctx.waitUntil(
        env.REPO_TASKS_QUEUE.send({
          kind: "compaction",
          doId: body.doId,
          repoId: body.repoId,
        }).catch((error) => {
          log.warn("compaction:follow-up-enqueue-failed", { error: String(error) });
        })
      );
    }

    if (commit.supersededPackKeys.length > 0) {
      ctx.waitUntil(
        env.REPO_TASKS_QUEUE.send(
          {
            kind: "compaction-delete",
            doId: body.doId,
            repoId: body.repoId,
            packKeys: commit.supersededPackKeys,
          },
          { delaySeconds: COMPACTION_DELETE_DELAY_SECONDS }
        ).catch((error) => {
          log.warn("compaction:delete-enqueue-failed", { error: String(error) });
        })
      );
    }

    log.info("compaction:done", {
      targetPackKey: commit.targetPackKey,
      supersededCount: commit.supersededPackKeys.length,
      shouldRequeue: commit.shouldRequeue,
    });
    message.ack();
  } catch (error) {
    log.error("compaction:error", { error: String(error) });
    await cleanupStagedCompaction({
      stagedUpload,
      log,
      reason: "error",
    });
    await abortCompactionLease({
      stub,
      leaseToken,
      limiter,
      cacheCtx,
      log,
      reason: "error",
    });
    retryQueueMessage(message, COMPACTION_RETRY_DELAY_SECONDS);
  }
}

export async function handleCompactionDeleteMessage(
  message: Omit<RepoQueueMessageHandle<CompactionDeleteQueueMessage>, "body">,
  body: CompactionDeleteQueueMessage,
  env: Env,
  ctx: ExecutionContext
): Promise<void> {
  const repoLabel = body.repoId || `do:${body.doId}`;
  const task = createQueueTaskContext({
    env,
    ctx,
    repoLabel,
    operation: "compaction-delete",
    subrequestBudget: 25,
  });
  const log = task.logFor({
    service: "CompactionDeleteQueue",
    repoId: repoLabel,
    doId: body.doId,
  });
  const { limiter } = task;

  try {
    const keysToDelete: string[] = [];
    for (const packKey of body.packKeys) {
      // Each superseded pack has three derived immutable artifacts in R2:
      // the pack bytes, the idx, and the logical-reference sidecar.
      keysToDelete.push(packKey, packIndexKey(packKey), packRefsKey(packKey));
    }

    await limiter.run("r2:delete-superseded-packs", async () => {
      await env.REPO_BUCKET.delete(keysToDelete);
    });
    log.info("compaction:delete-complete", {
      packCount: body.packKeys.length,
      artifactCount: keysToDelete.length,
    });
    message.ack();
  } catch (error) {
    log.warn("compaction:delete-failed", { error: String(error) });
    retryQueueMessage(message, COMPACTION_RETRY_DELAY_SECONDS);
  }
}