import { constantTimeEqual, hashToken, newWorkerToken } from "@autoscholar/auth";
import {
  artifacts,
  checkpoints,
  computeWorkers,
  type Db,
  jobLeases,
  projectInstructionVersions,
  providerUsage,
  researchRuns,
  resourceUsage,
  runAttempts,
  stageExecutions,
  type Tx,
  workerAssignments,
  workerHeartbeats,
} from "@autoscholar/db";
import { redact } from "@autoscholar/observability";
import {
  type ArtifactDescriptor,
  type NodeMetadata,
  type RunEvent,
  type RunManifest,
  STAGES,
  type StageKey,
  WORKER_PROTOCOL_VERSION,
  type WorkerStatus,
} from "@autoscholar/schemas";
import { type FilesystemStorage, runArtifactKey } from "@autoscholar/storage";
import { and, asc, desc, desc as descOrder, eq, inArray, isNull, lt, sql } from "drizzle-orm";
import { audit } from "./audit.js";
import { getEnv } from "./env.js";
import { AppError, conflict, forbidden, invalid, notFound, unauthorized } from "./errors.js";
import { appendEvent, latestCheckpoint, transition } from "./runs.js";
import { getSetting } from "./settings.js";

export type WorkerRow = typeof computeWorkers.$inferSelect;

export async function enrollWorker(
  db: Db,
  enrollmentSecret: string,
  name: string,
  node: NodeMetadata,
  lanIp: string | undefined,
  ip: string | null,
) {
  const env = getEnv();
  if (!env.WORKER_ENROLLMENT_SECRET || !constantTimeEqual(enrollmentSecret, env.WORKER_ENROLLMENT_SECRET)) {
    throw unauthorized("invalid enrollment secret");
  }
  if (node.protocolVersion !== WORKER_PROTOCOL_VERSION)
    throw new AppError(
      "protocol_mismatch",
      `worker protocol ${node.protocolVersion} != ${WORKER_PROTOCOL_VERSION}`,
      426,
    );
  const { token, tokenHash } = newWorkerToken();
  const [w] = await db
    .insert(computeWorkers)
    .values({
      name,
      tokenHash,
      status: "online",
      version: node.workerVersion,
      protocolVersion: node.protocolVersion,
      lanIp: lanIp ?? ip,
      nodeMeta: node,
      profiles: node.profiles,
      lastHeartbeatAt: new Date(),
      tokenRotatedAt: new Date(),
    })
    .onConflictDoUpdate({
      target: computeWorkers.name,
      set: {
        tokenHash,
        status: "online",
        version: node.workerVersion,
        protocolVersion: node.protocolVersion,
        lanIp: lanIp ?? ip,
        nodeMeta: node,
        profiles: node.profiles,
        lastHeartbeatAt: new Date(),
        tokenRotatedAt: new Date(),
        disabledAt: null,
      },
    })
    .returning();
  await audit(
    db,
    { type: "worker", id: w!.id, ip },
    "worker.enroll",
    { type: "worker", id: w!.id },
    { name, machineId: node.machineId, sandboxAvailable: node.sandboxAvailable },
  );
  return { workerId: w!.id, token, protocolVersion: WORKER_PROTOCOL_VERSION };
}

export async function authenticateWorker(db: Db, bearer: string | null | undefined): Promise<WorkerRow> {
  if (!bearer?.startsWith("asw_")) throw unauthorized("worker token required");
  const [w] = await db
    .select()
    .from(computeWorkers)
    .where(eq(computeWorkers.tokenHash, hashToken(bearer)))
    .limit(1);
  if (!w) throw unauthorized("unknown worker token");
  if (w.disabledAt || w.status === "disabled") throw forbidden("worker disabled");
  if (w.status === "quarantined") throw forbidden(`worker quarantined: ${w.quarantineReason ?? ""}`);
  return w;
}

export async function heartbeat(
  db: Db,
  worker: WorkerRow,
  body: {
    status: WorkerStatus;
    capacity: number;
    running: string[];
    load: Record<string, unknown>;
    node?: Partial<NodeMetadata>;
    leases: Array<{ runId: string; fencingToken: number }>;
  },
) {
  const env = getEnv();
  const status: WorkerStatus = worker.drainRequested ? "draining" : body.status;
  await db
    .update(computeWorkers)
    .set({
      status,
      capacity: body.capacity,
      running: body.running.length,
      lastHeartbeatAt: new Date(),
      nodeMeta: body.node ? { ...(worker.nodeMeta as object), ...body.node } : worker.nodeMeta,
    })
    .where(eq(computeWorkers.id, worker.id));
  await db
    .insert(workerHeartbeats)
    .values({ workerId: worker.id, load: body.load, running: body.running.length });
  // leases the worker thinks it holds but which are no longer its own (fencing) → tell it to stop
  const revokedLeases: string[] = [];
  const cancel: string[] = [];
  for (const l of body.leases) {
    const [lease] = await db.select().from(jobLeases).where(eq(jobLeases.runId, l.runId)).limit(1);
    if (!lease || lease.workerId !== worker.id || lease.fencingToken !== l.fencingToken)
      revokedLeases.push(l.runId);
    else {
      await db
        .update(jobLeases)
        .set({ lastHeartbeatAt: new Date(), expiresAt: new Date(Date.now() + env.LEASE_TTL_SECONDS * 1000) })
        .where(eq(jobLeases.runId, l.runId));
      const [run] = await db
        .select({ cancelRequested: researchRuns.cancelRequested })
        .from(researchRuns)
        .where(eq(researchRuns.id, l.runId))
        .limit(1);
      if (run?.cancelRequested) cancel.push(l.runId);
    }
  }
  return {
    ok: true as const,
    drain: !!worker.drainRequested,
    cancel,
    revokedLeases,
    leaseTtlSeconds: env.LEASE_TTL_SECONDS,
  };
}

/** Lease the oldest queued run matching the worker's profiles. SKIP LOCKED makes concurrent workers safe. */
export async function leaseJob(db: Db, worker: WorkerRow, profiles: string[]) {
  const env = getEnv();
  if (worker.drainRequested || worker.status === "draining") return { job: null };
  if (env.DISPATCH_ENABLED !== 1 || (await getSetting(db, "dispatch.paused", false))) return { job: null };
  const node = worker.nodeMeta as Partial<NodeMetadata>;
  if (!node.sandboxAvailable) return { job: null }; // never dispatch to a worker that cannot enforce isolation
  return db.transaction(async (tx) => {
    const [run] = await tx
      .select()
      .from(researchRuns)
      .where(
        and(
          eq(researchRuns.status, "queued"),
          inArray(researchRuns.computeProfile, profiles.length ? profiles : ["standard"]),
        ),
      )
      .orderBy(asc(researchRuns.queuedAt))
      .limit(1)
      .for("update", { skipLocked: true });
    if (!run) return { job: null };
    const [existingLease] = await tx.select().from(jobLeases).where(eq(jobLeases.runId, run.id)).limit(1);
    if (existingLease && existingLease.expiresAt.getTime() > Date.now()) return { job: null };
    if (run.currentAttemptNo >= run.maxAttempts) {
      await transition(tx, run.id, "failed", {
        reason: `attempt limit (${run.maxAttempts}) reached`,
        actor: "scheduler",
      });
      return { job: null };
    }
    const attemptNo = run.currentAttemptNo + 1;
    const fencingToken = run.fencingToken + 1;
    const cp = await latestCheckpoint(tx, run.id);
    const resumeFromStage = (cp?.stage as StageKey | undefined) ?? null;
    const [attempt] = await tx
      .insert(runAttempts)
      .values({
        runId: run.id,
        attemptNo,
        workerId: worker.id,
        status: "leased",
        fencingToken,
        resumeFromStage,
      })
      .returning();
    const expiresAt = new Date(Date.now() + env.LEASE_TTL_SECONDS * 1000);
    await tx.delete(jobLeases).where(eq(jobLeases.runId, run.id));
    await tx
      .insert(jobLeases)
      .values({ runId: run.id, attemptId: attempt!.id, workerId: worker.id, fencingToken, expiresAt });
    await tx.insert(workerAssignments).values({ runId: run.id, attemptId: attempt!.id, workerId: worker.id });
    await tx
      .update(researchRuns)
      .set({ currentAttemptNo: attemptNo, fencingToken, updatedAt: new Date() })
      .where(eq(researchRuns.id, run.id));
    await transition(tx, run.id, "provisioning", {
      actor: `worker:${worker.name}`,
      attemptId: attempt!.id,
      workerId: worker.id,
      data: { attemptNo, fencingToken, resumeFromStage },
    });
    await tx
      .update(computeWorkers)
      .set({ running: sql`${computeWorkers.running} + 1` })
      .where(eq(computeWorkers.id, worker.id));
    const [iv] = await tx
      .select({ contentText: projectInstructionVersions.contentText })
      .from(projectInstructionVersions)
      .where(eq(projectInstructionVersions.id, run.instructionVersionId))
      .limit(1);
    return {
      job: {
        runId: run.id,
        attemptId: attempt!.id,
        attemptNo,
        fencingToken,
        leaseExpiresAt: expiresAt.toISOString(),
        manifest: run.manifest as RunManifest,
        instructionText: iv!.contentText,
        resumeFromStage,
        checkpoint: (cp?.data as Record<string, unknown> | undefined) ?? null,
        cancelRequested: run.cancelRequested,
      },
    };
  });
}

/** Validates that the worker holds the current lease with the exact fencing token. */
export async function assertLease(tx: Tx | Db, runId: string, worker: WorkerRow, fencingToken: number) {
  const [lease] = await tx.select().from(jobLeases).where(eq(jobLeases.runId, runId)).limit(1);
  if (!lease || lease.workerId !== worker.id || lease.fencingToken !== fencingToken) {
    throw new AppError("lease_lost", "lease is not held by this worker (stale fencing token)", 409);
  }
  return lease;
}

export async function renewLease(db: Db, worker: WorkerRow, runId: string, fencingToken: number) {
  const env = getEnv();
  const lease = await assertLease(db, runId, worker, fencingToken);
  const expiresAt = new Date(Date.now() + env.LEASE_TTL_SECONDS * 1000);
  await db
    .update(jobLeases)
    .set({ expiresAt, lastHeartbeatAt: new Date() })
    .where(eq(jobLeases.runId, lease.runId));
  const [run] = await db
    .select({ cancelRequested: researchRuns.cancelRequested, status: researchRuns.status })
    .from(researchRuns)
    .where(eq(researchRuns.id, runId))
    .limit(1);
  return {
    ok: true,
    cancelRequested: !!run?.cancelRequested || run?.status === "cancelling",
    leaseExpiresAt: expiresAt.toISOString(),
  };
}

export async function ingestEvents(
  db: Db,
  worker: WorkerRow,
  runId: string,
  fencingToken: number,
  events: RunEvent[],
) {
  return db.transaction(async (tx) => {
    const lease = await assertLease(tx, runId, worker, fencingToken);
    let last = 0;
    for (const ev of events)
      last = await appendEvent(tx, runId, { ...redact(ev), attemptId: lease.attemptId, workerId: worker.id });
    return { ok: true as const, lastSeq: last };
  });
}

export async function reportStage(
  db: Db,
  worker: WorkerRow,
  runId: string,
  report: {
    fencingToken: number;
    stage: StageKey;
    status: "running" | "completed" | "failed" | "skipped";
    costUsd: number;
    turns: number;
    usage?: Record<string, unknown>;
    error?: string;
    checkpoint?: Record<string, unknown>;
  },
) {
  return db.transaction(async (tx) => {
    const lease = await assertLease(tx, runId, worker, report.fencingToken);
    const [run] = await tx.select().from(researchRuns).where(eq(researchRuns.id, runId)).for("update");
    if (!run) throw notFound("run");
    if (report.status === "running") {
      await tx
        .insert(stageExecutions)
        .values({ runId, attemptId: lease.attemptId, stage: report.stage, status: "running" })
        .onConflictDoUpdate({
          target: [stageExecutions.attemptId, stageExecutions.stage],
          set: { status: "running", startedAt: new Date(), endedAt: null, error: null },
        });
      await tx
        .update(researchRuns)
        .set({ currentStage: report.stage, updatedAt: new Date() })
        .where(eq(researchRuns.id, runId));
      if (run.status === "provisioning")
        await transition(tx, runId, "running", {
          actor: `worker:${worker.name}`,
          attemptId: lease.attemptId,
          workerId: worker.id,
        });
      await tx.update(runAttempts).set({ status: "running" }).where(eq(runAttempts.id, lease.attemptId));
    } else {
      await tx
        .insert(stageExecutions)
        .values({
          runId,
          attemptId: lease.attemptId,
          stage: report.stage,
          status: report.status,
          endedAt: new Date(),
          costUsd: String(report.costUsd),
          turns: report.turns,
          usage: report.usage ?? null,
          error: report.error ?? null,
        })
        .onConflictDoUpdate({
          target: [stageExecutions.attemptId, stageExecutions.stage],
          set: {
            status: report.status,
            endedAt: new Date(),
            costUsd: String(report.costUsd),
            turns: report.turns,
            usage: report.usage ?? null,
            error: report.error ?? null,
          },
        });
      if (report.costUsd > 0) {
        const u = (report.usage ?? {}) as Record<string, number | string>;
        await tx.insert(providerUsage).values({
          runId,
          attemptId: lease.attemptId,
          stage: report.stage,
          model: String(u.model ?? run.model),
          inputTokens: Number(u.inputTokens ?? 0),
          outputTokens: Number(u.outputTokens ?? 0),
          cacheReadTokens: Number(u.cacheReadTokens ?? 0),
          cacheCreateTokens: Number(u.cacheCreateTokens ?? 0),
          costUsd: String(report.costUsd),
          turns: report.turns,
        });
        await tx
          .update(researchRuns)
          .set({ totalCostUsd: sql`${researchRuns.totalCostUsd} + ${String(report.costUsd)}` })
          .where(eq(researchRuns.id, runId));
        await tx
          .update(runAttempts)
          .set({ costUsd: sql`${runAttempts.costUsd} + ${String(report.costUsd)}` })
          .where(eq(runAttempts.id, lease.attemptId));
      }
      if (report.status === "completed" && report.checkpoint) {
        await tx
          .insert(checkpoints)
          .values({ runId, attemptId: lease.attemptId, stage: report.stage, data: report.checkpoint });
      }
    }
    await appendEvent(tx, runId, {
      kind: `stage.${report.status}`,
      level: report.status === "failed" ? "error" : "info",
      stage: report.stage,
      message: `${STAGES.find((s) => s.key === report.stage)?.name ?? report.stage} ${report.status}${report.error ? `: ${report.error}` : ""}`,
      data: { costUsd: report.costUsd, turns: report.turns },
      at: new Date().toISOString(),
      attemptId: lease.attemptId,
      workerId: worker.id,
    });
    return { ok: true as const };
  });
}

export async function ingestArtifact(
  db: Db,
  storage: FilesystemStorage,
  worker: WorkerRow,
  runId: string,
  fencingToken: number,
  artifact: ArtifactDescriptor,
  body: Buffer,
) {
  const desc = artifact;
  if (body.length !== desc.bytes) throw invalid(`byte length mismatch: ${body.length} != ${desc.bytes}`);
  if (body.length > 512 * 1024 * 1024) throw invalid("artifact exceeds 512 MiB");
  const lease = await assertLease(db, runId, worker, fencingToken);
  const [attempt] = await db.select().from(runAttempts).where(eq(runAttempts.id, lease.attemptId)).limit(1);
  // existing versions of this path (any attempt): identical content is idempotent, changed content gets a new version
  const [prev] = await db
    .select({ version: artifacts.version, sha256: artifacts.sha256, id: artifacts.id })
    .from(artifacts)
    .where(and(eq(artifacts.runId, runId), eq(artifacts.path, desc.path)))
    .orderBy(descOrder(artifacts.version))
    .limit(1);
  if (prev && prev.sha256 === desc.sha256)
    return { ok: true as const, artifactId: prev.id, sha256: prev.sha256, bytes: body.length };
  const version = (prev?.version ?? 0) + 1;
  const key = runArtifactKey(
    runId,
    attempt!.attemptNo,
    version === 1 ? desc.path : `${desc.path}.v${version}`,
  );
  const stored = await storage.put(key, body, desc.sha256);
  return db.transaction(async (tx) => {
    await assertLease(tx, runId, worker, fencingToken);
    const [row] = await tx
      .insert(artifacts)
      .values({
        runId,
        attemptId: lease.attemptId,
        stage: desc.stage,
        path: desc.path,
        version,
        kind: desc.kind,
        sha256: stored.sha256,
        bytes: stored.bytes,
        contentType: desc.contentType,
        storageKey: key,
        synthetic: desc.synthetic,
      })
      .onConflictDoNothing()
      .returning({ id: artifacts.id });
    if (!row) {
      const [again] = await tx
        .select({ id: artifacts.id })
        .from(artifacts)
        .where(and(eq(artifacts.runId, runId), eq(artifacts.path, desc.path), eq(artifacts.version, version)))
        .limit(1);
      return { ok: true as const, artifactId: again!.id, sha256: stored.sha256, bytes: stored.bytes };
    }
    return { ok: true as const, artifactId: row.id, sha256: stored.sha256, bytes: stored.bytes };
  });
}

export async function completeAttempt(
  db: Db,
  worker: WorkerRow,
  runId: string,
  body: {
    fencingToken: number;
    outcome: "completed" | "failed" | "cancelled" | "budget_exceeded" | "timeout";
    error?: string;
    summary?: Record<string, unknown>;
    costUsd: number;
    resourceUsage?: { cpuSeconds?: number; wallSeconds?: number; diskBytes?: number; peakRssBytes?: number };
  },
) {
  return db.transaction(async (tx) => {
    const lease = await assertLease(tx, runId, worker, body.fencingToken);
    const [run] = await tx.select().from(researchRuns).where(eq(researchRuns.id, runId)).for("update");
    if (!run) throw notFound("run");
    await tx
      .update(runAttempts)
      .set({
        status: body.outcome,
        endedAt: new Date(),
        error: body.error ?? null,
        summary: body.summary ?? null,
      })
      .where(eq(runAttempts.id, lease.attemptId));
    if (body.resourceUsage)
      await tx.insert(resourceUsage).values({
        runId,
        attemptId: lease.attemptId,
        workerId: worker.id,
        cpuSeconds: body.resourceUsage.cpuSeconds?.toFixed(2),
        wallSeconds: body.resourceUsage.wallSeconds?.toFixed(2),
        diskBytes: body.resourceUsage.diskBytes,
        peakRssBytes: body.resourceUsage.peakRssBytes,
      });
    await tx.delete(jobLeases).where(eq(jobLeases.runId, runId));
    await tx
      .update(workerAssignments)
      .set({ releasedAt: new Date(), releaseReason: body.outcome })
      .where(eq(workerAssignments.attemptId, lease.attemptId));
    await tx
      .update(computeWorkers)
      .set({ running: sql`greatest(${computeWorkers.running} - 1, 0)` })
      .where(eq(computeWorkers.id, worker.id));
    const base = { actor: `worker:${worker.name}`, attemptId: lease.attemptId, workerId: worker.id };
    if (body.outcome === "completed") {
      await transition(tx, runId, "quality_check", {
        ...base,
        reason: "worker reported completion; awaiting deterministic quality evaluation",
      });
    } else if (body.outcome === "cancelled" || run.status === "cancelling") {
      await transition(tx, runId, "cancelled", { ...base, reason: body.error ?? "cancelled" });
    } else {
      const reason = `${body.outcome}${body.error ? `: ${body.error}` : ""}`;
      const noRetry = /^(INFEASIBLE|POLICY)\b/.test(body.error ?? "") || /INFEASIBLE:/.test(body.error ?? "");
      if (body.outcome === "budget_exceeded" || noRetry || run.currentAttemptNo >= run.maxAttempts) {
        await transition(tx, runId, "failed", { ...base, reason });
      } else {
        await transition(tx, runId, "retrying", {
          ...base,
          reason: `${reason}; attempt ${run.currentAttemptNo}/${run.maxAttempts}`,
        });
      }
    }
    return { ok: true as const };
  });
}

/** Scheduler: expire stale leases and recover runs (retry or fail). Returns the run IDs touched. */
export async function expireStaleLeases(db: Db): Promise<string[]> {
  const stale = await db.select().from(jobLeases).where(lt(jobLeases.expiresAt, new Date()));
  const touched: string[] = [];
  for (const lease of stale) {
    await db.transaction(async (tx) => {
      const [run] = await tx
        .select()
        .from(researchRuns)
        .where(eq(researchRuns.id, lease.runId))
        .for("update");
      if (!run) return;
      const [still] = await tx
        .select()
        .from(jobLeases)
        .where(and(eq(jobLeases.runId, lease.runId), eq(jobLeases.fencingToken, lease.fencingToken)))
        .limit(1);
      if (!still || still.expiresAt.getTime() > Date.now()) return;
      await tx.delete(jobLeases).where(eq(jobLeases.runId, lease.runId));
      await tx
        .update(runAttempts)
        .set({ status: "expired", endedAt: new Date(), error: "lease expired (worker heartbeat lost)" })
        .where(eq(runAttempts.id, lease.attemptId));
      await tx
        .update(workerAssignments)
        .set({ releasedAt: new Date(), releaseReason: "lease_expired" })
        .where(eq(workerAssignments.attemptId, lease.attemptId));
      await tx
        .update(computeWorkers)
        .set({ running: sql`greatest(${computeWorkers.running} - 1, 0)` })
        .where(eq(computeWorkers.id, lease.workerId));
      const base = { actor: "scheduler", attemptId: lease.attemptId, workerId: lease.workerId };
      if (["provisioning", "running", "cancelling"].includes(run.status)) {
        if (run.status === "cancelling")
          await transition(tx, run.id, "cancelled", { ...base, reason: "lease expired during cancellation" });
        else if (run.currentAttemptNo >= run.maxAttempts)
          await transition(tx, run.id, "failed", {
            ...base,
            reason: "lease expired and attempt limit reached",
          });
        else
          await transition(tx, run.id, "retrying", {
            ...base,
            reason: `lease expired (attempt ${run.currentAttemptNo}/${run.maxAttempts}); will resume from last checkpoint`,
          });
      }
      touched.push(run.id);
    });
  }
  return touched;
}

/** Scheduler: retrying → queued (immediate re-queue; backoff is applied through queuedAt ordering). */
export async function requeueRetrying(db: Db): Promise<number> {
  const rows = await db
    .select({ id: researchRuns.id })
    .from(researchRuns)
    .where(eq(researchRuns.status, "retrying"));
  for (const r of rows) await db.transaction((tx) => transition(tx, r.id, "queued", { actor: "scheduler" }));
  return rows.length;
}

export async function markOfflineWorkers(db: Db, staleAfterMs = 90_000): Promise<number> {
  const cutoff = new Date(Date.now() - staleAfterMs);
  const rows = await db
    .update(computeWorkers)
    .set({ status: "offline" })
    .where(
      and(inArray(computeWorkers.status, ["online", "draining"]), lt(computeWorkers.lastHeartbeatAt, cutoff)),
    )
    .returning({ id: computeWorkers.id });
  return rows.length;
}

export async function listWorkers(db: Db) {
  const rows = await db.select().from(computeWorkers).orderBy(asc(computeWorkers.name));
  return rows.map((w) => ({ ...w, tokenHash: undefined }));
}

export async function setWorkerDrain(db: Db, workerId: string, drain: boolean, actor: { id: string }) {
  const [w] = await db
    .update(computeWorkers)
    .set({ drainRequested: drain ? new Date() : null, status: drain ? "draining" : "online" })
    .where(eq(computeWorkers.id, workerId))
    .returning();
  if (!w) throw notFound("worker");
  await audit(db, { type: "admin", id: actor.id }, drain ? "worker.drain" : "worker.undrain", {
    type: "worker",
    id: workerId,
  });
  return w;
}

export async function setWorkerQuarantine(
  db: Db,
  workerId: string,
  reason: string | null,
  actor: { id: string },
) {
  const [w] = await db
    .update(computeWorkers)
    .set({ status: reason ? "quarantined" : "online", quarantineReason: reason })
    .where(eq(computeWorkers.id, workerId))
    .returning();
  if (!w) throw notFound("worker");
  await audit(
    db,
    { type: "admin", id: actor.id },
    reason ? "worker.quarantine" : "worker.unquarantine",
    { type: "worker", id: workerId },
    { reason },
  );
  return w;
}

export async function workerTelemetry(db: Db, workerId: string, limit = 60) {
  return db
    .select()
    .from(workerHeartbeats)
    .where(eq(workerHeartbeats.workerId, workerId))
    .orderBy(desc(workerHeartbeats.createdAt))
    .limit(limit);
}

export { conflict, isNull };

/** Latest version of every artifact of a run, for a worker that holds the lease (resume on another node). */
export async function listRunArtifactsForWorker(
  db: Db,
  worker: WorkerRow,
  runId: string,
  fencingToken: number,
) {
  await assertLease(db, runId, worker, fencingToken);
  const rows = await db
    .select({
      path: artifacts.path,
      version: artifacts.version,
      sha256: artifacts.sha256,
      bytes: artifacts.bytes,
      contentType: artifacts.contentType,
      storageKey: artifacts.storageKey,
      kind: artifacts.kind,
    })
    .from(artifacts)
    .where(eq(artifacts.runId, runId))
    .orderBy(descOrder(artifacts.version));
  const latest = new Map<string, (typeof rows)[number]>();
  for (const r of rows) if (!latest.has(r.path)) latest.set(r.path, r);
  return [...latest.values()].map(({ storageKey: _k, ...rest }) => rest);
}

export async function readRunArtifactForWorker(
  db: Db,
  storage: FilesystemStorage,
  worker: WorkerRow,
  runId: string,
  fencingToken: number,
  path: string,
) {
  await assertLease(db, runId, worker, fencingToken);
  const [row] = await db
    .select()
    .from(artifacts)
    .where(and(eq(artifacts.runId, runId), eq(artifacts.path, path)))
    .orderBy(descOrder(artifacts.version))
    .limit(1);
  if (!row) throw notFound("artifact");
  return { row, buf: await storage.get(row.storageKey) };
}