clawhub/convex/vt.ts

import { v } from "convex/values";
import { internal } from "./_generated/api";
import type { Doc, Id } from "./_generated/dataModel";
import { internalAction, internalMutation } from "./functions";
import { getPackageReleaseArtifactSha256 } from "./lib/packageArtifacts";
import { sourceSkillVersionFiles } from "./lib/skillCards";
import { buildDeterministicPackageZip, buildDeterministicZip } from "./lib/skillZip";

const SHA256_HASH_PATTERN = /^[a-f0-9]{64}$/i;
const VIRUSTOTAL_FILES_URL = "https://www.virustotal.com/api/v3/files";
const VIRUSTOTAL_UPLOAD_URL = "https://www.virustotal.com/api/v3/files/upload_url";
const VIRUSTOTAL_DIRECT_UPLOAD_LIMIT_BYTES = 32 * 1024 * 1024;

const internalRefs = internal as unknown as {
  packages: {
    getReleaseByIdInternal: unknown;
    getPackageByIdInternal: unknown;
    updateReleaseScanResultsInternal: unknown;
  };
  securityScan: {
    enqueuePackageReleaseScanInternal: unknown;
    enqueueSkillVersionScanInternal: unknown;
  };
  vt: {
    scanPackageReleaseWithVirusTotal: unknown;
    pollPackageReleaseScanResults: unknown;
  };
};

async function runQueryRef<T>(
  ctx: { runQuery: (ref: never, args: never) => Promise<unknown> },
  ref: unknown,
  args: unknown,
): Promise<T> {
  return (await ctx.runQuery(ref as never, args as never)) as T;
}

async function runMutationRef<T>(
  ctx: { runMutation: (ref: never, args: never) => Promise<unknown> },
  ref: unknown,
  args: unknown,
): Promise<T> {
  return (await ctx.runMutation(ref as never, args as never)) as T;
}

async function enqueueSkillCodexForVtSignal(
  ctx: { runMutation: (ref: never, args: never) => Promise<unknown> },
  versionId: Id<"skillVersions">,
) {
  await runMutationRef(ctx, internalRefs.securityScan.enqueueSkillVersionScanInternal, {
    versionId,
    source: "vt-update",
    waitForVtMs: 0,
  });
}

async function enqueuePackageCodexForVtSignal(
  ctx: { runMutation: (ref: never, args: never) => Promise<unknown> },
  releaseId: Id<"packageReleases">,
) {
  await runMutationRef(ctx, internalRefs.securityScan.enqueuePackageReleaseScanInternal, {
    releaseId,
    source: "vt-update",
    waitForVtMs: 0,
  });
}

async function runAfterRef(
  ctx: { scheduler: { runAfter: (delayMs: number, ref: never, args: never) => Promise<unknown> } },
  delayMs: number,
  ref: unknown,
  args: unknown,
) {
  return await ctx.scheduler.runAfter(delayMs, ref as never, args as never);
}

/**
 * Fix skills that have version.vtAnalysis but null skill.moderationReason.
 * This syncs the moderation reason from the cached VT results.
 */
export const fixNullModerationReasons = internalAction({
  args: { batchSize: v.optional(v.number()) },
  handler: async (ctx, args): Promise<FixNullModerationReasonsResult> => {
    const batchSize = args.batchSize ?? 100;
    const skills: UnscannedActiveSkill[] = await ctx.runQuery(
      internal.skills.getUnscannedActiveSkillsInternal,
      { limit: batchSize },
    );

    if (skills.length === 0) {
      console.log("[vt:fixNull] No skills with null reason found");
      return { total: 0, fixed: 0, noVtAnalysis: 0 };
    }

    console.log(`[vt:fixNull] Checking ${skills.length} skills with null moderationReason`);

    let fixed = 0;
    let noVtAnalysis = 0;

    for (const { versionId, slug } of skills) {
      if (!versionId) continue;

      const version = await ctx.runQuery(internal.skills.getVersionByIdInternal, { versionId });
      if (!version?.vtAnalysis || !version.sha256hash) {
        noVtAnalysis++;
        continue;
      }

      await enqueueSkillCodexForVtSignal(ctx, versionId);
      fixed++;
      console.log(`[vt:fixNull] Queued Codex scan for ${slug} from cached VT signal`);
    }

    const result: FixNullModerationReasonsResult = { total: skills.length, fixed, noVtAnalysis };
    console.log("[vt:fixNull] Complete:", result);
    return result;
  },
});

export const logScanResultInternal = internalMutation({
  args: {
    type: v.union(v.literal("daily_rescan"), v.literal("backfill"), v.literal("pending_poll")),
    total: v.number(),
    updated: v.number(),
    unchanged: v.number(),
    errors: v.number(),
    flaggedSkills: v.optional(
      v.array(
        v.object({
          slug: v.string(),
          status: v.string(),
        }),
      ),
    ),
    durationMs: v.number(),
  },
  handler: async (ctx, args) => {
    await ctx.db.insert("vtScanLogs", {
      type: args.type,
      total: args.total,
      updated: args.updated,
      unchanged: args.unchanged,
      errors: args.errors,
      flaggedSkills: args.flaggedSkills,
      durationMs: args.durationMs,
      createdAt: Date.now(),
    });
  },
});

type VTFileResponse = {
  data: {
    attributes: {
      sha256: string;
      last_analysis_stats?: {
        malicious: number;
        suspicious: number;
        undetected: number;
        harmless: number;
      };
    };
  };
};

type VTAnalysisStats = NonNullable<VTFileResponse["data"]["attributes"]["last_analysis_stats"]>;
type PackageReleaseScanDoc = Pick<
  Doc<"packageReleases">,
  "verification" | "llmAnalysis" | "staticScan"
>;
type PackageScanDoc = Pick<Doc<"packages">, "family" | "isOfficial" | "name">;

type VirusTotalUploadResponse = Response;

type PackageScanArtifact =
  | {
      ok: true;
      kind: "legacy-zip" | "clawpack";
      bytes: Uint8Array;
      sha256hash: string;
      fileName: string;
      contentType: string;
    }
  | {
      ok: false;
      missingFiles: number;
      fileCount: number;
    };

function bytesToArrayBuffer(bytes: Uint8Array): ArrayBuffer {
  const copy = new Uint8Array(bytes.byteLength);
  copy.set(bytes);
  return copy.buffer;
}

function normalizeVtEngineStats(stats?: VTAnalysisStats | null) {
  if (!stats) return undefined;
  return {
    malicious: stats.malicious,
    suspicious: stats.suspicious,
    undetected: stats.undetected,
    harmless: stats.harmless,
  };
}

function buildPackageUndetectedFallbackAnalysis(
  release: PackageReleaseScanDoc,
  pkg: PackageScanDoc,
  stats?: VTAnalysisStats,
) {
  if (!stats) return null;
  if (pkg.family === "skill") return null;

  const tier = release.verification?.tier;
  if (tier !== "source-linked" && tier !== "provenance-verified" && tier !== "rebuild-verified") {
    return null;
  }
  if (release.llmAnalysis?.status !== "clean") return null;
  if (!release.staticScan || release.staticScan.status !== "clean") return null;
  if (stats.malicious !== 0 || stats.suspicious !== 0) return null;
  if ((stats.harmless ?? 0) <= 0 && (stats.undetected ?? 0) <= 0) return null;

  return {
    status: "clean",
    verdict: "undetected-only-fallback",
    analysis:
      "VirusTotal reported no malicious or suspicious engine hits. ClawHub promoted this source-linked package after clean LLM and clean static scans.",
    source: "engines-undetected-fallback",
    checkedAt: Date.now(),
  };
}

function buildPackageScanAnalysisFromVtResult(
  release: PackageReleaseScanDoc,
  pkg: PackageScanDoc,
  vtResult: VTFileResponse,
) {
  const stats = vtResult.data.attributes.last_analysis_stats;
  const status = statusFromAvStats(stats);
  if (status) {
    return {
      status,
      source: "engines",
      engineStats: normalizeVtEngineStats(stats),
      checkedAt: Date.now(),
    };
  }

  return buildPackageUndetectedFallbackAnalysis(release, pkg, stats);
}

type ScanQueueHealth = {
  queueSize: number;
  staleCount: number;
  veryStaleCount: number;
  oldestAgeMinutes: number;
  healthy: boolean;
};

type PendingScanSkill = {
  skillId: Id<"skills">;
  versionId: Id<"skillVersions"> | null;
  sha256hash: string | null;
  checkCount: number;
};

type PollPendingScansResult = {
  processed: number;
  updated: number;
  staled?: number;
  healthy: boolean;
  queueSize?: number;
};

type BackfillPendingScansResult =
  | {
      total: number;
      updated: number;
      rescansRequested: number;
      noHash: number;
      notInVT: number;
      errors: number;
      remaining: number;
    }
  | { error: string };

type UnscannedActiveSkill = {
  skillId: Id<"skills">;
  versionId: Id<"skillVersions">;
  slug: string;
};

type LegacyPendingScanSkill = {
  skillId: Id<"skills">;
  versionId: Id<"skillVersions">;
  slug: string;
  hasHash: boolean;
};

type ActiveSkillsMissingVTCache = {
  skillId: Id<"skills">;
  versionId: Id<"skillVersions">;
  sha256hash: string;
  slug: string;
};

type PendingVTSkill = {
  skillId: Id<"skills">;
  versionId: Id<"skillVersions">;
  sha256hash: string;
  slug: string;
  isLatest?: boolean;
};

type NullModerationStatusSkill = {
  skillId: Id<"skills">;
  slug: string;
  moderationReason: string | undefined;
};

type StaleModerationReasonSkill = {
  skillId: Id<"skills">;
  versionId: Id<"skillVersions">;
  slug: string;
  currentReason: string;
  vtStatus: string | null;
  sha256hash: string | null;
};

type FixNullModerationReasonsResult = {
  total: number;
  fixed: number;
  noVtAnalysis: number;
};

type ScanUnscannedSkillsResult =
  | { total: number; scanned: number; errors: number; durationMs?: number }
  | { error: string };

type ScanLegacySkillsResult =
  | { total: number; scanned: number; errors: number; alreadyHasHash?: number; durationMs?: number }
  | { error: string };

type BackfillActiveSkillsVTCacheResult =
  | { total: number; updated: number; noResults: number; errors: number; done: boolean }
  | { error: string };

type RepairPendingSkillVtAnalysisResult =
  | {
      dryRun: boolean;
      total: number;
      wouldUpdate: number;
      updated: number;
      noResults: number;
      noDecisiveStats: number;
      errors: number;
      done: boolean;
      cursor: string | null;
      statusCounts: Record<string, number>;
      sampleUpdated: Array<{ slug: string; status: string }>;
    }
  | { error: string };

type FixNullModerationStatusResult = { total: number; fixed: number; done: boolean };

type SyncModerationReasonsResult = {
  total: number;
  synced: number;
  noVtAnalysis: number;
  done: boolean;
};

function statusFromAvStats(
  stats?: VTAnalysisStats | null,
): "malicious" | "suspicious" | "clean" | null {
  if (!stats) return null;
  if (stats.malicious > 0) return "malicious";
  if (stats.suspicious > 0) return "suspicious";
  // VirusTotal "undetected" means engines completed and found no detection;
  // it is resolved no-detections telemetry, not an in-progress scan.
  if (stats.harmless > 0 || stats.undetected > 0) return "clean";
  return null;
}

async function sha256Hex(bytes: Uint8Array) {
  const hashBuffer = await crypto.subtle.digest("SHA-256", bytesToArrayBuffer(bytes));
  return Array.from(new Uint8Array(hashBuffer))
    .map((b) => b.toString(16).padStart(2, "0"))
    .join("");
}

async function getVirusTotalUploadUrl(apiKey: string) {
  const response = await fetch(VIRUSTOTAL_UPLOAD_URL, {
    method: "GET",
    headers: {
      "x-apikey": apiKey,
    },
  });

  if (!response.ok) {
    const error = await response.text();
    throw new Error(`VT upload URL error: ${response.status} - ${error}`);
  }

  const result = (await response.json()) as { data?: unknown };
  if (typeof result.data !== "string" || !result.data) {
    throw new Error("VT upload URL response did not include a usable URL");
  }
  return result.data;
}

async function uploadFileToVirusTotal(
  apiKey: string,
  bytes: Uint8Array,
  fileName: string,
  contentType: string,
): Promise<VirusTotalUploadResponse> {
  const uploadUrl =
    bytes.byteLength > VIRUSTOTAL_DIRECT_UPLOAD_LIMIT_BYTES
      ? await getVirusTotalUploadUrl(apiKey)
      : VIRUSTOTAL_FILES_URL;
  const formData = new FormData();
  formData.append("file", new Blob([bytesToArrayBuffer(bytes)], { type: contentType }), fileName);
  return await fetch(uploadUrl, {
    method: "POST",
    headers: {
      "x-apikey": apiKey,
    },
    body: formData,
  });
}

export const fetchResults = internalAction({
  args: {
    sha256hash: v.optional(v.string()),
  },
  handler: async (_ctx, args) => {
    if (!args.sha256hash) {
      return { status: "not_found" };
    }
    if (!SHA256_HASH_PATTERN.test(args.sha256hash)) {
      return { status: "error", message: "Invalid SHA-256 hash" };
    }

    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      return { status: "error", message: "VT_API_KEY not configured" };
    }

    try {
      const response = await fetch(`https://www.virustotal.com/api/v3/files/${args.sha256hash}`, {
        method: "GET",
        headers: {
          "x-apikey": apiKey,
        },
      });

      if (response.status === 404) {
        return { status: "not_found" };
      }

      if (!response.ok) {
        return { status: "error" };
      }

      const data = (await response.json()) as VTFileResponse;
      const stats = data.data.attributes.last_analysis_stats;
      const status = statusFromAvStats(stats) ?? "pending";

      return {
        status,
        source: "engines",
        url: `https://www.virustotal.com/gui/file/${args.sha256hash}`,
        metadata: {
          stats: stats,
        },
      };
    } catch (error) {
      console.error("Error fetching VT results:", error);
      return { status: "error" };
    }
  },
});

export const scanWithVirusTotal = internalAction({
  args: {
    versionId: v.id("skillVersions"),
  },
  handler: async (ctx, args) => {
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("VT_API_KEY not configured, skipping skill scan without activation");
      return;
    }

    // Get the version details and files
    const version = (await ctx.runQuery(internal.skills.getVersionByIdInternal, {
      versionId: args.versionId,
    })) as Doc<"skillVersions"> | null;

    if (!version) {
      console.error(`Version ${args.versionId} not found for scanning`);
      return;
    }

    // Fetch skill info for _meta.json
    const skill = (await ctx.runQuery(internal.skills.getSkillByIdInternal, {
      skillId: version.skillId,
    })) as Doc<"skills"> | null;
    if (!skill) {
      console.error(`Skill ${version.skillId} not found for scanning`);
      return;
    }

    const fingerprintEntries = (await ctx.runQuery(
      internal.skills.listVersionFingerprintsInternal,
      {
        skillVersionId: version._id,
      },
    )) as Array<{ fingerprint: string; kind?: "source" | "generated-bundle" }>;
    const generatedBundleFingerprints = fingerprintEntries
      .filter((entry) => entry.kind === "generated-bundle")
      .map((entry) => entry.fingerprint);

    // Build deterministic ZIP with stable meta (no version history).
    const entries: Array<{ path: string; bytes: Uint8Array }> = [];
    for (const file of sourceSkillVersionFiles(version.files, { generatedBundleFingerprints })) {
      const content = await ctx.storage.get(file.storageId);
      if (content) {
        const buffer = new Uint8Array(await content.arrayBuffer());
        entries.push({ path: file.path, bytes: buffer });
      }
    }

    if (entries.length === 0) {
      console.warn(`No files found for version ${args.versionId}, skipping scan`);
      return;
    }

    const zipArray = buildDeterministicZip(entries, {
      ownerId: String(skill.ownerUserId),
      slug: skill.slug,
      version: version.version,
      publishedAt: version.createdAt,
    });

    // Calculate SHA-256 of the ZIP (this hash includes _meta.json)
    const sha256hash = await sha256Hex(zipArray);

    // Update version with hash
    await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
      versionId: args.versionId,
      sha256hash,
    });

    // Check if file already exists in VT and has engine analysis.
    try {
      const existingFile = await checkExistingFile(apiKey, sha256hash);

      if (existingFile) {
        const stats = existingFile.data.attributes.last_analysis_stats;
        const status = statusFromAvStats(stats);
        if (status) {
          console.log(
            `Version ${args.versionId} found in VT with engine analysis. Hash: ${sha256hash}. Status: ${status}`,
          );

          // Cache VT analysis in version
          await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
            versionId: args.versionId,
            vtAnalysis: {
              status,
              source: "engines",
              engineStats: normalizeVtEngineStats(stats),
              checkedAt: Date.now(),
            },
          });

          await enqueueSkillCodexForVtSignal(ctx, args.versionId);
          return;
        }

        console.log(
          `Version ${args.versionId} found in VT but no decisive engine analysis. Hash: ${sha256hash}. Uploading...`,
        );
      } else {
        console.log(`Version ${args.versionId} not found in VT. Hash: ${sha256hash}. Uploading...`);
      }
    } catch (error) {
      console.error("Error checking existing file in VT:", error);
      // Continue to upload even if check fails
    }

    try {
      const response = await uploadFileToVirusTotal(
        apiKey,
        zipArray,
        "skill.zip",
        "application/zip",
      );

      if (!response.ok) {
        const error = await response.text();
        console.error("VirusTotal upload error:", error);
        return;
      }

      const result = (await response.json()) as { data: { id: string } };
      console.log(
        `Successfully uploaded version ${args.versionId} to VT. Hash: ${sha256hash}. Analysis ID: ${result.data.id}`,
      );

      // Don't set moderation state to scanner.vt.pending here — the LLM eval
      // runs concurrently and will set the initial moderation state. VT only
      // updates moderation when it has an actual verdict (clean/suspicious/malicious).
    } catch (error) {
      console.error("Failed to upload to VirusTotal:", error);
    }
  },
});

const PACKAGE_SCAN_RETRY_DELAY_MS = 5 * 60 * 1000;
const PACKAGE_SCAN_MAX_ATTEMPTS = 10;

async function readPackageScanArtifact(
  ctx: { storage: { get: (id: Id<"_storage">) => Promise<Blob | null> } },
  release: Doc<"packageReleases">,
  packageName: string,
): Promise<PackageScanArtifact> {
  if (release.artifactKind === "npm-pack") {
    if (!release.clawpackStorageId) {
      return { ok: false, missingFiles: 1, fileCount: 1 };
    }

    const content = await ctx.storage.get(release.clawpackStorageId);
    if (!content) {
      return { ok: false, missingFiles: 1, fileCount: 1 };
    }

    const bytes = new Uint8Array(await content.arrayBuffer());
    return {
      ok: true,
      kind: "clawpack",
      bytes,
      sha256hash: await sha256Hex(bytes),
      fileName:
        release.npmTarballName ??
        `${packageName.replace(/^@/, "").replaceAll("/", "-")}-${release.version}.tgz`,
      contentType: "application/gzip",
    };
  }

  const entries: Array<{ path: string; bytes: Uint8Array }> = [];
  let missingFiles = 0;
  for (const file of release.files) {
    const content = await ctx.storage.get(file.storageId);
    if (!content) {
      missingFiles += 1;
      continue;
    }
    entries.push({
      path: file.path,
      bytes: new Uint8Array(await content.arrayBuffer()),
    });
  }

  if (entries.length === 0 || missingFiles > 0) {
    return { ok: false, missingFiles, fileCount: release.files.length };
  }

  const bytes = buildDeterministicPackageZip(entries);
  return {
    ok: true,
    kind: "legacy-zip",
    bytes,
    sha256hash: await sha256Hex(bytes),
    fileName: "package.zip",
    contentType: "application/zip",
  };
}

export const scanPackageReleaseWithVirusTotal = internalAction({
  args: {
    releaseId: v.id("packageReleases"),
    attempt: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("[vt:package] VT_API_KEY not configured, skipping package release scan");
      return;
    }

    const release = (await runQueryRef(ctx, internalRefs.packages.getReleaseByIdInternal, {
      releaseId: args.releaseId,
    })) as Doc<"packageReleases"> | null;
    if (!release || release.softDeletedAt) {
      console.error(`[vt:package] Release ${args.releaseId} not found for scanning`);
      return;
    }

    const pkg = (await runQueryRef(ctx, internalRefs.packages.getPackageByIdInternal, {
      packageId: release.packageId,
    })) as Doc<"packages"> | null;
    if (!pkg) {
      console.error(`[vt:package] Package ${release.packageId} not found for scanning`);
      return;
    }

    const attempt = args.attempt ?? 1;
    const artifact = await readPackageScanArtifact(ctx, release, pkg.name);
    if (!artifact.ok) {
      console.warn(
        `[vt:package] Release ${args.releaseId} missing ${artifact.missingFiles}/${artifact.fileCount} scan artifact file(s), retrying`,
      );
      if (attempt < PACKAGE_SCAN_MAX_ATTEMPTS) {
        await runAfterRef(
          ctx,
          PACKAGE_SCAN_RETRY_DELAY_MS,
          internalRefs.vt.scanPackageReleaseWithVirusTotal,
          {
            releaseId: args.releaseId,
            attempt: attempt + 1,
          },
        );
      }
      return;
    }

    try {
      const existingFile = await checkExistingFile(apiKey, artifact.sha256hash);
      const vtAnalysis = existingFile
        ? buildPackageScanAnalysisFromVtResult(release, pkg, existingFile)
        : null;

      if (vtAnalysis) {
        await runMutationRef(ctx, internalRefs.packages.updateReleaseScanResultsInternal, {
          releaseId: args.releaseId,
          vtAnalysis,
        });
        await enqueuePackageCodexForVtSignal(ctx, args.releaseId);
        return;
      }
    } catch (error) {
      console.error("[vt:package] Error checking existing file in VT:", error);
    }

    try {
      const response = await uploadFileToVirusTotal(
        apiKey,
        artifact.bytes,
        artifact.fileName,
        artifact.contentType,
      );

      if (!response.ok) {
        const error = await response.text();
        console.error("[vt:package] VirusTotal upload error:", error);
        if (attempt < PACKAGE_SCAN_MAX_ATTEMPTS) {
          await runAfterRef(
            ctx,
            PACKAGE_SCAN_RETRY_DELAY_MS,
            internalRefs.vt.scanPackageReleaseWithVirusTotal,
            {
              releaseId: args.releaseId,
              attempt: attempt + 1,
            },
          );
        }
        return;
      }

      await runAfterRef(
        ctx,
        PACKAGE_SCAN_RETRY_DELAY_MS,
        internalRefs.vt.pollPackageReleaseScanResults,
        {
          releaseId: args.releaseId,
          attempt: 1,
        },
      );

      console.log(
        `[vt:package] Uploaded ${pkg.name}@${release.version} ${artifact.kind} for scanning (${artifact.sha256hash})`,
      );
    } catch (error) {
      console.error("[vt:package] Failed to upload to VirusTotal:", error);
      if (attempt < PACKAGE_SCAN_MAX_ATTEMPTS) {
        await runAfterRef(
          ctx,
          PACKAGE_SCAN_RETRY_DELAY_MS,
          internalRefs.vt.scanPackageReleaseWithVirusTotal,
          {
            releaseId: args.releaseId,
            attempt: attempt + 1,
          },
        );
      }
    }
  },
});

export const pollPackageReleaseScanResults = internalAction({
  args: {
    releaseId: v.id("packageReleases"),
    attempt: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) return;

    const release = (await runQueryRef(ctx, internalRefs.packages.getReleaseByIdInternal, {
      releaseId: args.releaseId,
    })) as Doc<"packageReleases"> | null;
    if (!release || release.softDeletedAt) return;
    const artifactSha256 = getPackageReleaseArtifactSha256(release);
    if (!artifactSha256) return;
    const pkg = (await runQueryRef(ctx, internalRefs.packages.getPackageByIdInternal, {
      packageId: release.packageId,
    })) as Doc<"packages"> | null;
    if (!pkg || pkg.softDeletedAt) return;

    const attempt = args.attempt ?? 1;
    try {
      const vtResult = await checkExistingFile(apiKey, artifactSha256);
      if (!vtResult) {
        if (attempt < PACKAGE_SCAN_MAX_ATTEMPTS) {
          await runAfterRef(
            ctx,
            PACKAGE_SCAN_RETRY_DELAY_MS,
            internalRefs.vt.pollPackageReleaseScanResults,
            {
              releaseId: args.releaseId,
              attempt: attempt + 1,
            },
          );
        } else {
          await runMutationRef(ctx, internalRefs.packages.updateReleaseScanResultsInternal, {
            releaseId: args.releaseId,
            vtAnalysis: { status: "stale", checkedAt: Date.now() },
          });
          await enqueuePackageCodexForVtSignal(ctx, args.releaseId);
        }
        return;
      }

      const vtAnalysis = buildPackageScanAnalysisFromVtResult(release, pkg, vtResult);
      if (vtAnalysis) {
        await runMutationRef(ctx, internalRefs.packages.updateReleaseScanResultsInternal, {
          releaseId: args.releaseId,
          vtAnalysis,
        });
        await enqueuePackageCodexForVtSignal(ctx, args.releaseId);
        return;
      }

      if (attempt < PACKAGE_SCAN_MAX_ATTEMPTS) {
        await runAfterRef(
          ctx,
          PACKAGE_SCAN_RETRY_DELAY_MS,
          internalRefs.vt.pollPackageReleaseScanResults,
          {
            releaseId: args.releaseId,
            attempt: attempt + 1,
          },
        );
      } else {
        await runMutationRef(ctx, internalRefs.packages.updateReleaseScanResultsInternal, {
          releaseId: args.releaseId,
          vtAnalysis: { status: "stale", checkedAt: Date.now() },
        });
        await enqueuePackageCodexForVtSignal(ctx, args.releaseId);
      }
    } catch (error) {
      console.error(`[vt:package] Error polling ${artifactSha256}:`, error);
      if (attempt < PACKAGE_SCAN_MAX_ATTEMPTS) {
        await runAfterRef(
          ctx,
          PACKAGE_SCAN_RETRY_DELAY_MS,
          internalRefs.vt.pollPackageReleaseScanResults,
          {
            releaseId: args.releaseId,
            attempt: attempt + 1,
          },
        );
      } else {
        await runMutationRef(ctx, internalRefs.packages.updateReleaseScanResultsInternal, {
          releaseId: args.releaseId,
          vtAnalysis: { status: "error", checkedAt: Date.now() },
        });
        await enqueuePackageCodexForVtSignal(ctx, args.releaseId);
      }
    }
  },
});

/**
 * Poll for pending scans and update skill moderation status
 * Called by cron job to check VT results for skills awaiting scan
 */
export const pollPendingScans = internalAction({
  args: {
    batchSize: v.optional(v.number()),
  },
  handler: async (ctx, args): Promise<PollPendingScansResult> => {
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("[vt:pollPendingScans] VT_API_KEY not configured, skipping");
      return { processed: 0, updated: 0, healthy: false };
    }

    const batchSize = args.batchSize ?? 10;

    // Check queue health
    // TODO: Setup webhook/notification (Slack, Discord, email) when queue is unhealthy
    const health: ScanQueueHealth = await ctx.runQuery(
      internal.skills.getScanQueueHealthInternal,
      {},
    );
    if (!health.healthy) {
      console.warn(
        `[vt:pollPendingScans] QUEUE UNHEALTHY: ${health.queueSize} pending, ${health.veryStaleCount} stale >24h, oldest ${health.oldestAgeMinutes}m`,
      );
    }

    // Get skills pending scan (randomized selection)
    const pendingSkills: PendingScanSkill[] = await ctx.runQuery(
      internal.skills.getPendingScanSkillsInternal,
      {
        limit: batchSize,
      },
    );

    if (pendingSkills.length === 0) {
      return { processed: 0, updated: 0, healthy: health.healthy, queueSize: health.queueSize };
    }

    console.log(
      `[vt:pollPendingScans] Checking ${pendingSkills.length} pending skills (queue: ${health.queueSize})`,
    );

    const MAX_CHECK_COUNT = 10; // After this many checks, mark as stale

    let updated = 0;
    let staled = 0;
    for (const { skillId, versionId, sha256hash, checkCount } of pendingSkills) {
      if (!versionId) {
        console.log(`[vt:pollPendingScans] Skill ${skillId} missing versionId, skipping`);
        continue;
      }
      if (!sha256hash) {
        console.log(
          `[vt:pollPendingScans] Skill ${skillId} version ${versionId} has no hash, skipping`,
        );
        continue;
      }

      // Track this check attempt
      await ctx.runMutation(internal.skills.updateScanCheckInternal, { skillId });

      try {
        const vtResult = await checkExistingFile(apiKey, sha256hash);
        if (!vtResult) {
          console.log(`[vt:pollPendingScans] Hash ${sha256hash} not found in VT yet`);
          // Check if we've exceeded max attempts — write stale vtAnalysis so it
          // drops out of the poll query without overwriting LLM moderationReason
          if (checkCount + 1 >= MAX_CHECK_COUNT) {
            console.warn(
              `[vt:pollPendingScans] Skill ${skillId} exceeded max checks, marking stale`,
            );
            await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
              versionId,
              vtAnalysis: { status: "stale", checkedAt: Date.now() },
            });
            await enqueueSkillCodexForVtSignal(ctx, versionId);
            staled++;
          }
          continue;
        }

        const stats = vtResult.data.attributes.last_analysis_stats;
        const status = statusFromAvStats(stats);

        if (status) {
          console.log(
            `[vt:pollPendingScans] Hash ${sha256hash} verdict from AV engines: ${status}`,
          );

          await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
            versionId,
            vtAnalysis: {
              status,
              source: "engines",
              engineStats: normalizeVtEngineStats(stats),
              checkedAt: Date.now(),
            },
          });

          await enqueueSkillCodexForVtSignal(ctx, versionId);
          updated++;
          continue;
        }

        if (checkCount + 1 >= MAX_CHECK_COUNT) {
          console.log(`[vt:pollPendingScans] Hash ${sha256hash} has no decisive engine stats`);
          console.warn(`[vt:pollPendingScans] Skill ${skillId} exceeded max checks, marking stale`);
          await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
            versionId,
            vtAnalysis: { status: "stale", checkedAt: Date.now() },
          });
          await enqueueSkillCodexForVtSignal(ctx, versionId);
          staled++;
        }
      } catch (error) {
        console.error(`[vt:pollPendingScans] Error checking hash ${sha256hash}:`, error);
      }
    }

    console.log(
      `[vt:pollPendingScans] Processed ${pendingSkills.length}, updated ${updated}, staled ${staled}`,
    );
    return {
      processed: pendingSkills.length,
      updated,
      staled,
      healthy: health.healthy,
      queueSize: health.queueSize,
    };
  },
});

export const repairPendingSkillVtAnalysis = internalAction({
  args: {
    dryRun: v.boolean(),
    batchSize: v.optional(v.number()),
    concurrency: v.optional(v.number()),
    cursor: v.optional(v.union(v.string(), v.null())),
  },
  handler: async (ctx, args): Promise<RepairPendingSkillVtAnalysisResult> => {
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("[vt:repairPendingSkillVt] VT_API_KEY not configured");
      return { error: "VT_API_KEY not configured" };
    }
    const vtApiKey = apiKey;

    const batchSize = Math.max(1, Math.min(Math.floor(args.batchSize ?? 100), 500));
    const concurrency = Math.max(1, Math.min(Math.floor(args.concurrency ?? 16), 32));
    const pendingPage: {
      skills: PendingVTSkill[];
      cursor: string | null;
      done: boolean;
    } = await ctx.runQuery(internal.skills.getPendingVTSkillsInternal, {
      limit: batchSize,
      cursor: args.cursor ?? null,
    });
    const skills = pendingPage.skills;

    let wouldUpdate = 0;
    let updated = 0;
    let noResults = 0;
    let noDecisiveStats = 0;
    let errors = 0;
    const statusCounts: Record<string, number> = {};
    const sampleUpdated: Array<{ slug: string; status: string }> = [];

    async function repairSkill({
      skillId,
      versionId,
      sha256hash,
      slug,
      isLatest = true,
    }: PendingVTSkill) {
      try {
        const vtResult = await checkExistingFile(vtApiKey, sha256hash);
        if (!vtResult) {
          noResults++;
          return;
        }

        const stats = vtResult.data.attributes.last_analysis_stats;
        const status = statusFromAvStats(stats);
        if (!status) {
          noDecisiveStats++;
          return;
        }

        wouldUpdate++;
        statusCounts[status] = (statusCounts[status] ?? 0) + 1;
        if (sampleUpdated.length < 20) sampleUpdated.push({ slug, status });
        if (args.dryRun) return;

        await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
          versionId,
          sha256hash,
          vtAnalysis: {
            status,
            source: "engines",
            engineStats: normalizeVtEngineStats(stats),
            checkedAt: Date.now(),
          },
        });
        if (!isLatest) {
          updated++;
          return;
        }

        if (status === "malicious" || status === "suspicious") {
          await enqueueSkillCodexForVtSignal(ctx, versionId);
        } else {
          await ctx.runMutation(internal.skills.recomputeLatestSkillModerationInternal, {
            skillId,
          });
        }
        updated++;
      } catch (error) {
        console.error(`[vt:repairPendingSkillVt] Error for ${slug}:`, error);
        errors++;
      }
    }

    for (let index = 0; index < skills.length; index += concurrency) {
      await Promise.all(skills.slice(index, index + concurrency).map(repairSkill));
    }

    return {
      dryRun: args.dryRun,
      total: skills.length,
      wouldUpdate,
      updated,
      noResults,
      noDecisiveStats,
      errors,
      done: pendingPage.done,
      cursor: pendingPage.cursor,
      statusCounts,
      sampleUpdated,
    };
  },
});

/**
 * Check if a file already exists in VirusTotal by hash
 */
async function checkExistingFile(
  apiKey: string,
  sha256hash: string,
): Promise<VTFileResponse | null> {
  const response = await fetch(`https://www.virustotal.com/api/v3/files/${sha256hash}`, {
    method: "GET",
    headers: {
      "x-apikey": apiKey,
    },
  });

  if (response.status === 404) {
    // File not found in VT
    return null;
  }

  if (!response.ok) {
    const error = await response.text();
    throw new Error(`VT API error: ${response.status} - ${error}`);
  }

  return (await response.json()) as VTFileResponse;
}

export const __test = {
  VIRUSTOTAL_DIRECT_UPLOAD_LIMIT_BYTES,
  normalizeVtEngineStats,
  sha256Hex,
  statusFromAvStats,
  uploadFileToVirusTotal,
};

/**
 * Backfill function to process ALL pending skills at once
 * Run manually to clear backlog
 */
export const backfillPendingScans = internalAction({
  args: {
    triggerRescans: v.optional(v.boolean()),
  },
  handler: async (ctx): Promise<BackfillPendingScansResult> => {
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("[vt:backfill] VT_API_KEY not configured");
      return { error: "VT_API_KEY not configured" };
    }

    // Get ALL pending skills (no limit)
    const pendingSkills: PendingScanSkill[] = await ctx.runQuery(
      internal.skills.getPendingScanSkillsInternal,
      {
        limit: 10000,
        exhaustive: true,
        skipRecentMinutes: 0,
      },
    );

    console.log(`[vt:backfill] Found ${pendingSkills.length} pending skills`);

    let updated = 0;
    let rescansRequested = 0;
    let noHash = 0;
    let notInVT = 0;
    let errors = 0;

    for (const { versionId, sha256hash } of pendingSkills) {
      if (!versionId || !sha256hash) {
        noHash++;
        continue;
      }

      try {
        const vtResult = await checkExistingFile(apiKey, sha256hash);

        if (!vtResult) {
          notInVT++;
          continue;
        }

        const stats = vtResult.data.attributes.last_analysis_stats;
        const status = statusFromAvStats(stats);

        if (!status) {
          continue;
        }

        console.log(`[vt:backfill] Hash ${sha256hash} verdict from AV engines: ${status}`);
        await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
          versionId,
          vtAnalysis: {
            status,
            source: "engines",
            engineStats: normalizeVtEngineStats(stats),
            checkedAt: Date.now(),
          },
        });
        await enqueueSkillCodexForVtSignal(ctx, versionId);
        updated++;
      } catch (error) {
        console.error(`[vt:backfill] Error for ${sha256hash}:`, error);
        errors++;
      }
    }

    const result: BackfillPendingScansResult = {
      total: pendingSkills.length,
      updated,
      rescansRequested,
      noHash,
      notInVT,
      errors,
      remaining: pendingSkills.length - updated,
    };

    console.log("[vt:backfill] Complete:", result);
    return result;
  },
});

/**
 * Daily re-scan of ALL active skills to detect verdict changes.
 * Cursor-based: processes one batch per invocation and self-schedules the next.
 * Cron calls with {} to start from the beginning; subsequent batches pass accumulated totals.
 * API budget: 25k hourly / 100k daily calls.
 */
export const rescanActiveSkills = internalAction({
  args: {
    cursor: v.optional(v.number()),
    batchSize: v.optional(v.number()),
    accTotal: v.optional(v.number()),
    accUpdated: v.optional(v.number()),
    accUnchanged: v.optional(v.number()),
    accErrors: v.optional(v.number()),
    accFlaggedSkills: v.optional(v.array(v.object({ slug: v.string(), status: v.string() }))),
    startTime: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const startTime = args.startTime ?? Date.now();
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("[vt:rescan] VT_API_KEY not configured");
      return { error: "VT_API_KEY not configured" };
    }

    const batchSize = args.batchSize ?? 100;
    const cursor = args.cursor ?? 0;
    let accTotal = args.accTotal ?? 0;
    let accUpdated = args.accUpdated ?? 0;
    let accUnchanged = args.accUnchanged ?? 0;
    let accErrors = args.accErrors ?? 0;
    const accFlaggedSkills = [...(args.accFlaggedSkills ?? [])];

    const batch = await ctx.runQuery(internal.skills.getActiveSkillBatchForRescanInternal, {
      cursor,
      batchSize,
    });

    if (batch.skills.length === 0 && accTotal === 0) {
      console.log("[vt:rescan] No active skills to re-scan");
      return { total: 0, updated: 0, unchanged: 0, errors: 0 };
    }

    console.log(
      `[vt:rescan] Processing batch of ${batch.skills.length} skills (cursor=${cursor}, accumulated=${accTotal})`,
    );

    for (const { versionId, sha256hash, slug, wasFlagged } of batch.skills) {
      try {
        const vtResult = await checkExistingFile(apiKey, sha256hash);

        if (!vtResult) {
          accErrors++;
          continue;
        }

        const stats = vtResult.data.attributes.last_analysis_stats;
        const status = statusFromAvStats(stats);

        if (!status) {
          await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
            versionId,
            vtAnalysis: {
              status: "pending",
              checkedAt: Date.now(),
            },
          });
          accUnchanged++;
          continue;
        }

        console.log(`[vt:rescan] ${slug} verdict from AV engines: ${status}`);

        await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
          versionId,
          vtAnalysis: {
            status,
            source: "engines",
            engineStats: normalizeVtEngineStats(stats),
            checkedAt: Date.now(),
          },
        });

        if (status === "malicious" || status === "suspicious") {
          console.warn(`[vt:rescan] ${slug}: verdict changed to ${status}!`);
          accFlaggedSkills.push({ slug, status });
          await enqueueSkillCodexForVtSignal(ctx, versionId);
          accUpdated++;
        } else if (wasFlagged && status === "clean") {
          console.log(`[vt:rescan] ${slug}: VT verdict improved to clean`);
          await enqueueSkillCodexForVtSignal(ctx, versionId);
          accUpdated++;
        } else {
          accUnchanged++;
        }
      } catch (error) {
        console.error(`[vt:rescan] Error for ${slug}:`, error);
        accErrors++;
      }
    }

    accTotal += batch.skills.length;

    if (!batch.done) {
      // Schedule next batch
      console.log(
        `[vt:rescan] Scheduling next batch (cursor=${batch.nextCursor}, total so far=${accTotal})`,
      );
      await ctx.scheduler.runAfter(0, internal.vt.rescanActiveSkills, {
        cursor: batch.nextCursor,
        batchSize,
        accTotal,
        accUpdated,
        accUnchanged,
        accErrors,
        accFlaggedSkills: accFlaggedSkills.length > 0 ? accFlaggedSkills : undefined,
        startTime,
      });
      return { status: "continuing", totalSoFar: accTotal };
    }

    // Final batch — log results
    const durationMs = Date.now() - startTime;

    await ctx.runMutation(internal.vt.logScanResultInternal, {
      type: "daily_rescan",
      total: accTotal,
      updated: accUpdated,
      unchanged: accUnchanged,
      errors: accErrors,
      flaggedSkills: accFlaggedSkills.length > 0 ? accFlaggedSkills : undefined,
      durationMs,
    });

    const result = {
      total: accTotal,
      updated: accUpdated,
      unchanged: accUnchanged,
      errors: accErrors,
      durationMs,
    };
    console.log("[vt:rescan] Complete:", result);
    return result;
  },
});

/**
 * Scan all unscanned skills (active with null moderationReason).
 * These completely bypassed VT and need immediate scanning.
 */
export const scanUnscannedSkills = internalAction({
  args: { batchSize: v.optional(v.number()) },
  handler: async (ctx, args): Promise<ScanUnscannedSkillsResult> => {
    const startTime = Date.now();
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("[vt:scanUnscanned] VT_API_KEY not configured");
      return { error: "VT_API_KEY not configured" };
    }

    const batchSize = args.batchSize ?? 50;
    const skills: UnscannedActiveSkill[] = await ctx.runQuery(
      internal.skills.getUnscannedActiveSkillsInternal,
      { limit: batchSize },
    );

    if (skills.length === 0) {
      console.log("[vt:scanUnscanned] No unscanned skills found");
      return { total: 0, scanned: 0, errors: 0 };
    }

    console.log(`[vt:scanUnscanned] Scanning ${skills.length} unscanned skills`);

    let scanned = 0;
    let errors = 0;

    for (const { versionId, slug } of skills) {
      if (!versionId) {
        errors++;
        continue;
      }

      try {
        await ctx.runAction(internal.vt.scanWithVirusTotal, { versionId });
        scanned++;
        console.log(`[vt:scanUnscanned] Scanned ${slug} (${scanned}/${skills.length})`);
      } catch (error) {
        console.error(`[vt:scanUnscanned] Error scanning ${slug}:`, error);
        errors++;
      }
    }

    const durationMs = Date.now() - startTime;

    await ctx.runMutation(internal.vt.logScanResultInternal, {
      type: "backfill",
      total: skills.length,
      updated: scanned,
      unchanged: 0,
      errors,
      durationMs,
    });

    const result: ScanUnscannedSkillsResult = { total: skills.length, scanned, errors, durationMs };
    console.log("[vt:scanUnscanned] Complete:", result);
    return result;
  },
});

/**
 * Scan all legacy skills (active but still have pending.scan reason).
 * These are skills approved before VT integration that need proper scanning.
 */
export const scanLegacySkills = internalAction({
  args: { batchSize: v.optional(v.number()) },
  handler: async (ctx, args): Promise<ScanLegacySkillsResult> => {
    const startTime = Date.now();
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("[vt:scanLegacy] VT_API_KEY not configured");
      return { error: "VT_API_KEY not configured" };
    }

    const batchSize = args.batchSize ?? 100;
    const skills: LegacyPendingScanSkill[] = await ctx.runQuery(
      internal.skills.getLegacyPendingScanSkillsInternal,
      { limit: batchSize },
    );

    if (skills.length === 0) {
      console.log("[vt:scanLegacy] No legacy skills to scan");
      return { total: 0, scanned: 0, errors: 0 };
    }

    console.log(`[vt:scanLegacy] Scanning ${skills.length} legacy skills`);

    let scanned = 0;
    let alreadyHasHash = 0;
    let errors = 0;

    for (const { versionId, slug, hasHash } of skills) {
      if (!versionId) {
        errors++;
        continue;
      }

      try {
        if (hasHash) {
          // Already has hash, just need to check VT and update reason
          alreadyHasHash++;
        }

        // Trigger VT scan (will upload if needed, check for results)
        await ctx.runAction(internal.vt.scanWithVirusTotal, { versionId });
        scanned++;
        console.log(`[vt:scanLegacy] Scanned ${slug} (${scanned}/${skills.length})`);
      } catch (error) {
        console.error(`[vt:scanLegacy] Error scanning ${slug}:`, error);
        errors++;
      }
    }

    const durationMs = Date.now() - startTime;

    await ctx.runMutation(internal.vt.logScanResultInternal, {
      type: "backfill",
      total: skills.length,
      updated: scanned,
      unchanged: alreadyHasHash,
      errors,
      durationMs,
    });

    const result: ScanLegacySkillsResult = {
      total: skills.length,
      scanned,
      alreadyHasHash,
      errors,
      durationMs,
    };
    console.log("[vt:scanLegacy] Complete:", result);
    return result;
  },
});

/**
 * Backfill vtAnalysis for active skills that have VT results but no cached data.
 * This covers highlighted skills and others approved before VT integration.
 * Processes in batches with self-scheduling to drain the backlog.
 */
export const backfillActiveSkillsVTCache = internalAction({
  args: { batchSize: v.optional(v.number()) },
  handler: async (ctx, args): Promise<BackfillActiveSkillsVTCacheResult> => {
    const apiKey = process.env.VT_API_KEY;
    if (!apiKey) {
      console.log("[vt:backfillActive] VT_API_KEY not configured");
      return { error: "VT_API_KEY not configured" };
    }

    const batchSize = args.batchSize ?? 100;

    const skills: ActiveSkillsMissingVTCache[] = await ctx.runQuery(
      internal.skills.getActiveSkillsMissingVTCacheInternal,
      { limit: batchSize },
    );

    console.log(`[vt:backfillActive] Found ${skills.length} active skills missing VT cache`);

    if (skills.length === 0) {
      return { total: 0, updated: 0, noResults: 0, errors: 0, done: true };
    }

    let updated = 0;
    let noResults = 0;
    let errors = 0;

    for (const { versionId, sha256hash, slug } of skills) {
      try {
        const vtResult = await checkExistingFile(apiKey, sha256hash);

        if (!vtResult) {
          console.log(`[vt:backfillActive] ${slug}: not in VT`);
          noResults++;
          continue;
        }

        const stats = vtResult.data.attributes.last_analysis_stats;
        const status = statusFromAvStats(stats);

        if (!status) {
          console.log(`[vt:backfillActive] ${slug}: no decisive engine stats yet`);
          noResults++;
          continue;
        }

        console.log(`[vt:backfillActive] ${slug}: updated with ${status} (from AV engines)`);

        await ctx.runMutation(internal.skills.updateVersionScanResultsInternal, {
          versionId,
          sha256hash,
          vtAnalysis: {
            status,
            source: "engines",
            engineStats: normalizeVtEngineStats(stats),
            checkedAt: Date.now(),
          },
        });
        await enqueueSkillCodexForVtSignal(ctx, versionId);

        console.log(`[vt:backfillActive] ${slug}: updated with ${status}`);
        updated++;
      } catch (error) {
        console.error(`[vt:backfillActive] Error for ${slug}:`, error);
        errors++;
      }
    }

    const done = skills.length < batchSize;
    const result: BackfillActiveSkillsVTCacheResult = {
      total: skills.length,
      updated,
      noResults,
      errors,
      done,
    };
    console.log("[vt:backfillActive] Complete:", result);

    // Self-schedule next batch if there are more skills to process
    if (!done) {
      console.log("[vt:backfillActive] Scheduling next batch...");
      await ctx.scheduler.runAfter(0, internal.vt.backfillActiveSkillsVTCache, { batchSize });
    }

    return result;
  },
});

/**
 * Fix skills with null moderationStatus by setting them to 'active'.
 */
export const fixNullModerationStatus = internalAction({
  args: { batchSize: v.optional(v.number()) },
  handler: async (ctx, args): Promise<FixNullModerationStatusResult> => {
    const batchSize = args.batchSize ?? 100;

    const skills: NullModerationStatusSkill[] = await ctx.runQuery(
      internal.skills.getSkillsWithNullModerationStatusInternal,
      { limit: batchSize },
    );

    if (skills.length === 0) {
      console.log("[vt:fixNullStatus] No skills with null status found");
      return { total: 0, fixed: 0, done: true };
    }

    console.log(`[vt:fixNullStatus] Found ${skills.length} skills with null moderationStatus`);

    for (const { skillId, slug: _slug } of skills) {
      await ctx.runMutation(internal.skills.setSkillModerationStatusActiveInternal, { skillId });
    }

    console.log(`[vt:fixNullStatus] Fixed ${skills.length} skills`);
    return { total: skills.length, fixed: skills.length, done: skills.length < batchSize };
  },
});

/**
 * Queue Codex scans for skills with cached VT telemetry but stale scanner state.
 * VT is telemetry only; Codex owns visibility changes.
 */
export const syncModerationReasons = internalAction({
  args: { batchSize: v.optional(v.number()) },
  handler: async (ctx, args): Promise<SyncModerationReasonsResult> => {
    const batchSize = args.batchSize ?? 100;

    const skills: StaleModerationReasonSkill[] = await ctx.runQuery(
      internal.skills.getSkillsWithStaleModerationReasonInternal,
      { limit: batchSize },
    );

    if (skills.length === 0) {
      console.log("[vt:syncModeration] No stale skills found");
      return { total: 0, synced: 0, noVtAnalysis: 0, done: true };
    }

    console.log(`[vt:syncModeration] Found ${skills.length} skills with stale moderationReason`);

    let synced = 0;
    let noVtAnalysis = 0;

    for (const { skillId, slug, currentReason, vtStatus } of skills) {
      if (!vtStatus) {
        noVtAnalysis++;
        continue;
      }

      const skill = await ctx.runQuery(internal.skills.getSkillByIdInternal, { skillId });
      if (skill?.latestVersionId) {
        await enqueueSkillCodexForVtSignal(ctx, skill.latestVersionId);
      }

      console.log(`[vt:syncModeration] ${slug}: queued Codex for ${currentReason}/${vtStatus}`);
      synced++;
    }

    const result: SyncModerationReasonsResult = {
      total: skills.length,
      synced,
      noVtAnalysis,
      done: skills.length < batchSize,
    };
    console.log("[vt:syncModeration] Complete:", result);
    return result;
  },
});