mnote/wolai-frontend/convex/mediaAssets.ts

import { internalMutation, mutation, query } from "./_generated/server";
import { v } from "convex/values";
import { nowIso } from "./_utils/time";
import { enqueueExtractMediaAssetTextJob, enqueueIngestMediaAssetJob } from "./_utils/ingestJobs";
import type { MutationCtx, QueryCtx } from "./_generated/server";
import { internal } from "./_generated/api";
import { getCanonicalDocumentByBusinessId } from "./_utils/documentRecord";

async function assertWorkspaceMember(ctx: QueryCtx | MutationCtx, userId: string, workspaceId: string) {
  const membership = await ctx.db
    .query("workspace_members")
    .withIndex("by_workspace_user", (q) => q.eq("workspace_id", workspaceId).eq("user_id", userId))
    .first();
  if (!membership) {
    throw new Error("无权访问该工作空间");
  }
  return membership;
}

function resolveGraceSeconds(): number {
  const raw = process.env.DELETE_GRACE_SECONDS ?? process.env.NEXT_PUBLIC_DELETE_GRACE_SECONDS ?? "600";
  const parsed = Number(raw);
  if (!Number.isFinite(parsed) || parsed <= 0) {
    return 600;
  }
  return Math.floor(parsed);
}

function isExpiredForPurge(deletedAt: unknown, expiredDeletedAt: string): boolean {
  if (!deletedAt || typeof deletedAt !== "string") return false;
  // 说明：deleted_at 使用 ISO 字符串，按字典序比较即可满足时间先后。
  return deletedAt <= expiredDeletedAt;
}

async function deleteMediaAssetRow(ctx: any, row: any, deletingIds: Set<string>) {
  const storageId = (row.storage_id as any) ?? null;
  if (storageId) {
    const refs = await ctx.db
      .query("media_assets")
      .withIndex("by_storage_id", (q: any) => q.eq("storage_id", storageId))
      .collect();
    // 说明：只要仍有“未清理”的引用（包括仍在垃圾桶但可恢复的记录），就不要删底层文件。
    const otherAlive = refs.some((r: any) => !deletingIds.has(String(r.id)) && !r.purged_at);
    if (!otherAlive) {
      try {
        await ctx.storage.delete(storageId);
      } catch {
        // ignore
      }
    }
  }
  await ctx.db.delete(row._id);
}

function shouldExtractAttachmentText(args: {
  assetType?: string | null;
  mimeType?: string | null;
  fileName?: string | null;
}): boolean {
  if (String(args.assetType ?? "") !== "file") return false;
  const mime = String(args.mimeType ?? "").toLowerCase().trim();
  if (mime === "application/pdf") return true;
  if (mime === "application/vnd.openxmlformats-officedocument.wordprocessingml.document") return true;
  if (mime === "application/vnd.openxmlformats-officedocument.presentationml.presentation") return true;
  if (mime === "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") return true;

  const name = String(args.fileName ?? "").toLowerCase().trim();
  if (name.endsWith(".pdf")) return true;
  if (name.endsWith(".docx")) return true;
  if (name.endsWith(".pptx")) return true;
  if (name.endsWith(".xlsx")) return true;
  return false;
}

function splitExtension(fileName: string): { base: string; ext: string } {
  const safe = fileName.trim();
  const lastDot = safe.lastIndexOf(".");
  if (lastDot <= 0 || lastDot === safe.length - 1) {
    return { base: safe, ext: "" };
  }
  return { base: safe.slice(0, lastDot), ext: safe.slice(lastDot) };
}

function makeUniqueFileName(fileName: string, existing: Set<string>): string {
  const safe = (fileName.trim() || "附件").replace(/[\\/]/g, "_");
  if (!existing.has(safe)) {
    existing.add(safe);
    return safe;
  }
  const { base, ext } = splitExtension(safe);
  const first = `${base} 副本${ext}`;
  if (!existing.has(first)) {
    existing.add(first);
    return first;
  }
  for (let i = 2; i < 1000; i += 1) {
    const candidate = `${base} 副本 ${i}${ext}`;
    if (!existing.has(candidate)) {
      existing.add(candidate);
      return candidate;
    }
  }
  const fallback = `${base} 副本 ${Date.now()}${ext}`;
  existing.add(fallback);
  return fallback;
}

function uniqueStrings(values: string[]): string[] {
  const out: string[] = [];
  for (const value of values) {
    const normalized = String(value ?? "").trim();
    if (!normalized || out.includes(normalized)) {
      continue;
    }
    out.push(normalized);
  }
  return out;
}

function validateResourceTransferPlan(args: {
  action: "copy" | "move";
  assetIds: string[];
  targetDocumentId: string;
  targetSubPath?: string | null;
  resourceTransferPlan?: any;
}) {
  const plan = args.resourceTransferPlan;
  if (!plan) {
    return;
  }

  if (plan.action !== args.action) {
    throw new Error("资源操作计划不一致");
  }
  if (String(plan.targetDocumentId ?? "").trim() !== args.targetDocumentId) {
    throw new Error("资源目标页面计划不一致");
  }
  const plannedSubPath = String(plan.targetSubPath ?? "").trim();
  const actualSubPath = String(args.targetSubPath ?? "").trim();
  if (plannedSubPath !== actualSubPath) {
    throw new Error("资源目标子路径计划不一致");
  }
  const plannedAssetIds = Array.isArray(plan.assetIds)
    ? uniqueStrings(plan.assetIds.map((value: unknown) => String(value ?? "")))
    : [];
  if (plannedAssetIds.length !== args.assetIds.length) {
    throw new Error("资源列表计划不一致");
  }
  for (let i = 0; i < args.assetIds.length; i += 1) {
    if (plannedAssetIds[i] !== args.assetIds[i]) {
      throw new Error("资源列表计划不一致");
    }
  }
}

function validateResourceUploadPlan(args: {
  asset: {
    id: string;
    workspace_id: string;
    document_id: string;
    asset_type: string;
    file_name?: string | null;
    file_size?: number | null;
    mime_type?: string | null;
  };
  targetSubPath?: string | null;
  resourceUploadPlan?: any;
}) {
  const plan = args.resourceUploadPlan;
  if (!plan || typeof plan !== "object") {
    return;
  }
  if (plan.action !== "upload") {
    throw new Error("Rust resource upload plan action 不一致");
  }
  if (String(plan.assetId ?? "").trim() !== args.asset.id) {
    throw new Error("Rust resource upload plan assetId 不一致");
  }
  if (String(plan.workspaceId ?? "").trim() !== args.asset.workspace_id) {
    throw new Error("Rust resource upload plan workspaceId 不一致");
  }
  if (String(plan.targetDocumentId ?? "").trim() !== args.asset.document_id) {
    throw new Error("Rust resource upload plan targetDocumentId 不一致");
  }
  const plannedSubPath = String(plan.targetSubPath ?? "").trim();
  const actualSubPath = String(args.targetSubPath ?? "").trim();
  if (plannedSubPath !== actualSubPath) {
    throw new Error("Rust resource upload plan targetSubPath 不一致");
  }
  if (String(plan.assetType ?? "").trim() !== args.asset.asset_type) {
    throw new Error("Rust resource upload plan assetType 不一致");
  }
  const plannedName = String(plan.fileName ?? "").trim();
  const actualName = String(args.asset.file_name ?? "").trim();
  if (plannedName !== actualName) {
    throw new Error("Rust resource upload plan fileName 不一致");
  }
  if (typeof plan.fileSize === "number" && plan.fileSize !== args.asset.file_size) {
    throw new Error("Rust resource upload plan fileSize 不一致");
  }
  const plannedMime = String(plan.mimeType ?? "").trim();
  const actualMime = String(args.asset.mime_type ?? "").trim();
  if (plannedMime !== actualMime) {
    throw new Error("Rust resource upload plan mimeType 不一致");
  }
}

async function loadTransferAssets(ctx: MutationCtx, userId: string, assetIds: string[]) {
  const assets: any[] = [];
  for (const assetId of assetIds) {
    const row = await ctx.db
      .query("media_assets")
      .withIndex("by_asset_id", (q) => q.eq("id", assetId))
      .first();
    if (!row || row.deleted_at || row.purged_at) {
      continue;
    }
    await assertWorkspaceMember(ctx, userId, row.workspace_id);
    assets.push(row);
  }
  return assets;
}

async function loadExistingNames(ctx: MutationCtx, documentId: string) {
  const existingRows = await ctx.db
    .query("media_assets")
    .withIndex("by_document", (q) => q.eq("document_id", documentId))
    .collect();
  return new Set<string>(
    existingRows.map((row) => String(row.file_name ?? "")).filter((name) => name.length > 0),
  );
}

async function resolveTransferTarget(ctx: MutationCtx, userId: string, targetDocumentId: string) {
  const targetDoc = await getCanonicalDocumentByBusinessId<any>(ctx, targetDocumentId);
  if (!targetDoc) {
    throw new Error("目标页面不存在");
  }
  await assertWorkspaceMember(ctx, userId, targetDoc.workspace_id);
  return targetDoc;
}

function buildTransferredAssetResult(row: any) {
  return {
    id: row.id,
    workspace_id: row.workspace_id,
    document_id: row.document_id,
    asset_type: row.asset_type,
    file_url: row.file_url ?? null,
    thumbnail_url: row.thumbnail_url ?? row.file_url ?? null,
    storage_id: row.storage_id ?? null,
    bucket: row.bucket ?? null,
    storage_path: row.storage_path ?? null,
    file_name: row.file_name ?? null,
    file_size: row.file_size ?? null,
    mime_type: row.mime_type ?? null,
    ocr_text: row.ocr_text ?? null,
    ocr_status: row.ocr_status ?? null,
    ocr_payload: row.ocr_payload,
    ocr_strategy: row.ocr_strategy ?? null,
    deleted_at: row.deleted_at ?? null,
    deleted_by: row.deleted_by ?? null,
    purged_at: row.purged_at ?? null,
    signed_url: row.signed_url ?? null,
    created_at: row.created_at,
    updated_at: row.updated_at,
  };
}

export const getById = query({
  args: { userId: v.string(), id: v.string() },
  handler: async (ctx, args) => {
    const row = await ctx.db
      .query("media_assets")
      .withIndex("by_asset_id", (q) => q.eq("id", args.id))
      .first();
    if (!row) return null;
    await assertWorkspaceMember(ctx, args.userId, row.workspace_id);
    return row;
  },
});

export const generateUploadUrl = mutation({
  args: { userId: v.string() },
  handler: async (ctx, args) => {
    // 说明：简单兜底，要求用户至少有一个工作空间 membership。
    const membership = await ctx.db
      .query("workspace_members")
      .withIndex("by_user_id", (q) => q.eq("user_id", args.userId))
      .first();
    if (!membership) {
      throw new Error("尚未初始化工作空间，无法上传");
    }
    return await ctx.storage.generateUploadUrl();
  },
});

export const listByWorkspace = query({
  args: {
    userId: v.string(),
    workspaceId: v.string(),
    assetType: v.optional(v.string()),
    includeDeleted: v.optional(v.boolean()),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    await assertWorkspaceMember(ctx, args.userId, args.workspaceId);

    const includeDeleted = Boolean(args.includeDeleted);
    const limit = Math.max(1, Math.min(200, Math.floor(args.limit ?? 12)));

    let rows = await ctx.db
      .query("media_assets")
      .withIndex("by_workspace", (q) => q.eq("workspace_id", args.workspaceId))
      .order("desc")
      .take(limit * 5);

    if (!includeDeleted) {
      rows = rows.filter((r) => !r.deleted_at);
    }

    if (args.assetType) {
      rows = rows.filter((r) => r.asset_type === args.assetType);
    }

    // 说明：Convex 的 take 以索引排序为准，这里再截一刀保证输出稳定。
    return rows.slice(0, limit);
  },
});

export const listSearchDataByWorkspace = query({
  args: {
    userId: v.string(),
    workspaceId: v.string(),
    includeDeleted: v.optional(v.boolean()),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    await assertWorkspaceMember(ctx, args.userId, args.workspaceId);

    const includeDeleted = Boolean(args.includeDeleted);
    const limit = Math.max(1, Math.min(5000, Math.floor(args.limit ?? 5000)));

    const rows = await ctx.db
      .query("media_assets")
      .withIndex("by_workspace", (q) => q.eq("workspace_id", args.workspaceId))
      .order("desc")
      .take(limit * 2);

    return rows
      .filter((r) => (includeDeleted ? true : !r.deleted_at))
      .filter((r) => !r.purged_at)
      .slice(0, limit)
      .map((r) => ({
        id: r.id,
        workspace_id: r.workspace_id,
        document_id: r.document_id,
        asset_type: r.asset_type,
        file_name: r.file_name ?? null,
        mime_type: r.mime_type ?? null,
        ocr_text: r.ocr_text ?? null,
        ocr_status: r.ocr_status ?? null,
        created_at: r.created_at ?? null,
        updated_at: r.updated_at ?? null,
      }));
  },
});

export const listByDocument = query({
  args: {
    userId: v.string(),
    documentId: v.string(),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const limit = Math.max(1, Math.min(500, Math.floor(args.limit ?? 200)));
    const rows = await ctx.db
      .query("media_assets")
      .withIndex("by_document", (q) => q.eq("document_id", args.documentId))
      .order("desc")
      .take(limit * 2);

    const filtered = rows.filter((r) => !r.deleted_at);
    const ws = filtered[0]?.workspace_id ?? rows[0]?.workspace_id ?? null;
    if (ws) await assertWorkspaceMember(ctx, args.userId, ws);
    return filtered.slice(0, limit);
  },
});

export const listDeletedByWorkspace = query({
  args: {
    userId: v.string(),
    workspaceId: v.string(),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    await assertWorkspaceMember(ctx, args.userId, args.workspaceId);
    const limit = Math.max(1, Math.min(2000, Math.floor(args.limit ?? 2000)));

    // 说明：Convex 目前不支持“deleted_at is not null”这种索引条件，先全取再过滤（对练手项目足够）。
    const rows = await ctx.db
      .query("media_assets")
      .withIndex("by_workspace", (q) => q.eq("workspace_id", args.workspaceId))
      .order("desc")
      .take(limit * 3);

    return rows.filter((r) => Boolean(r.deleted_at) && !r.purged_at).slice(0, limit);
  },
});

export const listByIds = query({
  args: { userId: v.string(), ids: v.array(v.string()) },
  handler: async (ctx, args) => {
    const ids = Array.from(new Set(args.ids.filter(Boolean))).slice(0, 200);
    const out: any[] = [];
    for (const id of ids) {
      const row = await ctx.db
        .query("media_assets")
        .withIndex("by_asset_id", (q) => q.eq("id", id))
        .first();
      if (!row) continue;
      await assertWorkspaceMember(ctx, args.userId, row.workspace_id);
      out.push(row);
    }
    return out;
  },
});

export const create = mutation({
  args: {
    userId: v.string(),
    asset: v.object({
      id: v.string(),
      workspace_id: v.string(),
      document_id: v.string(),
      asset_type: v.string(),
      file_url: v.union(v.string(), v.null()),
      thumbnail_url: v.union(v.string(), v.null()),
      storage_id: v.optional(v.union(v.id("_storage"), v.null())),
      bucket: v.union(v.string(), v.null()),
      storage_path: v.union(v.string(), v.null()),
      file_name: v.union(v.string(), v.null()),
      file_size: v.union(v.number(), v.null()),
      mime_type: v.union(v.string(), v.null()),
    }),
  },
  handler: async (ctx, args) => {
    await assertWorkspaceMember(ctx, args.userId, args.asset.workspace_id);
    const ts = nowIso();

    await ctx.db.insert("media_assets", {
      ...args.asset,
      storage_id: args.asset.storage_id ?? null,
      ocr_text: null,
      ocr_status: shouldExtractAttachmentText({
        assetType: args.asset.asset_type,
        mimeType: args.asset.mime_type,
        fileName: args.asset.file_name,
      })
        ? "queued"
        : null,
      deleted_at: null,
      deleted_by: null,
      purged_at: null,
      created_by: args.userId,
      created_at: ts,
      updated_at: ts,
    });

    if (
      shouldExtractAttachmentText({
        assetType: args.asset.asset_type,
        mimeType: args.asset.mime_type,
        fileName: args.asset.file_name,
      })
    ) {
      await enqueueExtractMediaAssetTextJob(ctx, { userId: args.userId, assetId: args.asset.id, debounceMs: 800 });
    }

    return args.asset;
  },
});

export const createWithStorage = mutation({
  args: {
    userId: v.string(),
    storageId: v.id("_storage"),
    targetSubPath: v.optional(v.union(v.string(), v.null())),
    resourceUploadPlan: v.optional(v.any()),
    asset: v.object({
      id: v.string(),
      workspace_id: v.string(),
      document_id: v.string(),
      asset_type: v.string(),
      file_name: v.union(v.string(), v.null()),
      file_size: v.union(v.number(), v.null()),
      mime_type: v.union(v.string(), v.null()),
    }),
  },
  handler: async (ctx, args) => {
    await assertWorkspaceMember(ctx, args.userId, args.asset.workspace_id);
    validateResourceUploadPlan({
      asset: args.asset,
      targetSubPath: args.targetSubPath,
      resourceUploadPlan: args.resourceUploadPlan,
    });

    const doc = await getCanonicalDocumentByBusinessId<any>(ctx, args.asset.document_id);

    if (!doc || doc.workspace_id !== args.asset.workspace_id) {
      throw new Error("目标页面不存在或不属于该工作空间");
    }

    const url = await ctx.storage.getUrl(args.storageId);
    if (!url) {
      throw new Error("文件不存在或已过期");
    }

    const ts = nowIso();
    const row = {
      id: args.asset.id,
      workspace_id: args.asset.workspace_id,
      document_id: args.asset.document_id,
      asset_type: args.asset.asset_type,
      file_url: url,
      thumbnail_url: url,
      storage_id: args.storageId,
      bucket: null,
      storage_path: args.targetSubPath ? `${args.targetSubPath}/${args.asset.file_name ?? args.asset.id}` : null,
      file_name: args.asset.file_name,
      file_size: args.asset.file_size,
      mime_type: args.asset.mime_type,

      ocr_text: null,
      ocr_status: shouldExtractAttachmentText({
        assetType: args.asset.asset_type,
        mimeType: args.asset.mime_type,
        fileName: args.asset.file_name,
      })
        ? "queued"
        : null,
      deleted_at: null,
      deleted_by: null,
      purged_at: null,
      created_by: args.userId,
      created_at: ts,
      updated_at: ts,
    };

    await ctx.db.insert("media_assets", row);

    if (
      shouldExtractAttachmentText({
        assetType: args.asset.asset_type,
        mimeType: args.asset.mime_type,
        fileName: args.asset.file_name,
      })
    ) {
      await enqueueExtractMediaAssetTextJob(ctx, { userId: args.userId, assetId: args.asset.id, debounceMs: 800 });
    }
    return row;
  },
});

export const batchCopy = mutation({
  args: {
    userId: v.string(),
    assetIds: v.array(v.string()),
    targetDocumentId: v.string(),
    targetSubPath: v.optional(v.union(v.string(), v.null())),
    resourceTransferPlan: v.optional(v.any()),
  },
  handler: async (ctx, args) => {
    const assetIds = uniqueStrings(args.assetIds);
    if (assetIds.length === 0) {
      throw new Error("缺少附件");
    }

    const targetDoc = await resolveTransferTarget(ctx, args.userId, args.targetDocumentId);
    validateResourceTransferPlan({
      action: "copy",
      assetIds,
      targetDocumentId: args.targetDocumentId,
      targetSubPath: args.targetSubPath,
      resourceTransferPlan: args.resourceTransferPlan,
    });

    const assets = await loadTransferAssets(ctx, args.userId, assetIds);
    const existingNames = await loadExistingNames(ctx, args.targetDocumentId);
    const items: any[] = [];
    const ts = nowIso();

    for (const asset of assets) {
      const storageId = (asset.storage_id as any) ?? null;
      if (!storageId) {
        continue;
      }
      const id =
        typeof crypto.randomUUID === "function"
          ? crypto.randomUUID()
          : `${Date.now()}_${Math.random().toString(16).slice(2)}`;
      const fileName = makeUniqueFileName(String(asset.file_name ?? "附件"), existingNames);
      const row = {
        id,
        workspace_id: String(targetDoc.workspace_id),
        document_id: String(args.targetDocumentId),
        asset_type: String(asset.asset_type ?? "file"),
        file_url: asset.file_url ?? null,
        thumbnail_url: asset.thumbnail_url ?? asset.file_url ?? null,
        storage_id: storageId,
        bucket: asset.bucket ?? null,
        storage_path: asset.storage_path ?? null,
        file_name: fileName,
        file_size: typeof asset.file_size === "number" ? asset.file_size : null,
        mime_type: (asset.mime_type ?? null) as any,
        ocr_text: null,
        ocr_status: shouldExtractAttachmentText({
          assetType: asset.asset_type,
          mimeType: asset.mime_type,
          fileName,
        })
          ? "queued"
          : null,
        ocr_payload: undefined,
        ocr_strategy: null,
        deleted_at: null,
        deleted_by: null,
        purged_at: null,
        created_by: args.userId,
        created_at: ts,
        updated_at: ts,
      };
      await ctx.db.insert("media_assets", row);
      if (row.ocr_status === "queued") {
        await enqueueExtractMediaAssetTextJob(ctx, { userId: args.userId, assetId: id, debounceMs: 800 });
      }
      items.push(buildTransferredAssetResult(row));
    }

    return { items };
  },
});

export const batchMove = mutation({
  args: {
    userId: v.string(),
    assetIds: v.array(v.string()),
    targetDocumentId: v.string(),
    targetSubPath: v.optional(v.union(v.string(), v.null())),
    resourceTransferPlan: v.optional(v.any()),
  },
  handler: async (ctx, args) => {
    const assetIds = uniqueStrings(args.assetIds);
    if (assetIds.length === 0) {
      throw new Error("缺少附件");
    }

    const targetDoc = await resolveTransferTarget(ctx, args.userId, args.targetDocumentId);
    validateResourceTransferPlan({
      action: "move",
      assetIds,
      targetDocumentId: args.targetDocumentId,
      targetSubPath: args.targetSubPath,
      resourceTransferPlan: args.resourceTransferPlan,
    });

    const assets = await loadTransferAssets(ctx, args.userId, assetIds);
    const existingNames = await loadExistingNames(ctx, args.targetDocumentId);
    const items: any[] = [];

    for (const asset of assets) {
      const fileName = makeUniqueFileName(String(asset.file_name ?? "附件"), existingNames);
      const patch = {
        workspace_id: String(targetDoc.workspace_id),
        document_id: String(args.targetDocumentId),
        file_name: fileName,
        updated_at: nowIso(),
      };
      await ctx.db.patch(asset._id, patch);
      items.push(buildTransferredAssetResult({ ...asset, ...patch }));
    }

    return { items };
  },
});

export const patchById = mutation({
  args: {
    userId: v.string(),
    id: v.string(),
    patch: v.any(),
  },
  handler: async (ctx, args) => {
    const row = await ctx.db
      .query("media_assets")
      .withIndex("by_asset_id", (q) => q.eq("id", args.id))
      .first();
    if (!row) throw new Error("资源不存在");
    await assertWorkspaceMember(ctx, args.userId, row.workspace_id);

    const prevDeletedAt = (row as any).deleted_at ?? null;

    const next = { ...(args.patch as Record<string, unknown>), updated_at: nowIso() };
    await ctx.db.patch(row._id, next);

    // 说明：当 OCR 写回完成时，自动触发 LightRAG 入库任务（Convex jobs/actions）。
    const patch = args.patch as Record<string, unknown>;
    const prev = row as Record<string, unknown>;
    const nextOcrStatus = typeof patch.ocr_status === "string" ? patch.ocr_status : prev.ocr_status;
    const nextOcrText = typeof patch.ocr_text === "string" ? patch.ocr_text : prev.ocr_text;
    if (nextOcrStatus === "completed" && typeof nextOcrText === "string" && nextOcrText.trim()) {
      await enqueueIngestMediaAssetJob(ctx, { userId: args.userId, assetId: args.id, debounceMs: 1200 });
    }

    // 说明：当附件进入垃圾桶时，自动调度“到期彻底删除”，避免长期堆积垃圾数据。
    const nextDeletedAt =
      patch.deleted_at === null
        ? null
        : typeof patch.deleted_at === "string"
          ? patch.deleted_at
          : ((row as any).deleted_at ?? null);
    if (prevDeletedAt == null && typeof nextDeletedAt === "string" && nextDeletedAt.trim()) {
      const graceMs = (resolveGraceSeconds() + 5) * 1000;
      await ctx.scheduler.runAfter(graceMs, internal.mediaAssets.purgeIfExpired, {
        id: args.id,
        deletedAt: nextDeletedAt,
      });
    }
    return { ok: true };
  },
});

export const refreshUrl = mutation({
  args: { userId: v.string(), id: v.string() },
  handler: async (ctx, args) => {
    const row = await ctx.db
      .query("media_assets")
      .withIndex("by_asset_id", (q) => q.eq("id", args.id))
      .first();
    if (!row) throw new Error("资源不存在");
    await assertWorkspaceMember(ctx, args.userId, row.workspace_id);

    const storageId = (row.storage_id as any) ?? null;
    if (!storageId) {
      // 外链资源：直接回传现有 URL
      return { signedUrl: row.file_url };
    }

    const url = await ctx.storage.getUrl(storageId);
    if (!url) {
      throw new Error("文件不存在或已被删除");
    }

    await ctx.db.patch(row._id, { file_url: url, thumbnail_url: url, updated_at: nowIso() });
    return { signedUrl: url };
  },
});

export const replaceStorageFromUpload = mutation({
  args: { userId: v.string(), id: v.string(), storageId: v.id("_storage") },
  handler: async (ctx, args) => {
    const row = await ctx.db
      .query("media_assets")
      .withIndex("by_asset_id", (q) => q.eq("id", args.id))
      .first();
    if (!row) throw new Error("资源不存在");
    await assertWorkspaceMember(ctx, args.userId, row.workspace_id);

    const url = await ctx.storage.getUrl(args.storageId);
    if (!url) {
      throw new Error("文件不存在或已过期");
    }

    await ctx.db.patch(row._id, {
      storage_id: args.storageId,
      file_url: url,
      thumbnail_url: url,
      bucket: null,
      storage_path: null,
      ocr_text: null,
      ocr_status: shouldExtractAttachmentText({
        assetType: row.asset_type,
        mimeType: row.mime_type ?? null,
        fileName: row.file_name ?? null,
      })
        ? "queued"
        : row.ocr_status ?? null,
      updated_at: nowIso(),
    });

    if (
      shouldExtractAttachmentText({
        assetType: row.asset_type,
        mimeType: row.mime_type ?? null,
        fileName: row.file_name ?? null,
      })
    ) {
      await enqueueExtractMediaAssetTextJob(ctx, { userId: args.userId, assetId: args.id, debounceMs: 800 });
    }

    return { ok: true, fileUrl: url };
  },
});

export const enqueueExtractText = mutation({
  args: { userId: v.string(), id: v.string() },
  handler: async (ctx, args) => {
    const row = await ctx.db
      .query("media_assets")
      .withIndex("by_asset_id", (q) => q.eq("id", args.id))
      .first();
    if (!row) throw new Error("资源不存在");
    await assertWorkspaceMember(ctx, args.userId, row.workspace_id);

    if (
      !shouldExtractAttachmentText({
        assetType: row.asset_type,
        mimeType: row.mime_type ?? null,
        fileName: row.file_name ?? null,
      })
    ) {
      return { ok: false, reason: "unsupported" as const };
    }

    await ctx.db.patch(row._id, { ocr_status: "queued", updated_at: nowIso() });
    await enqueueExtractMediaAssetTextJob(ctx, { userId: args.userId, assetId: args.id, debounceMs: 0 });
    return { ok: true };
  },
});

export const purgeById = mutation({
  args: { userId: v.string(), id: v.string(), expiredDeletedAt: v.string() },
  handler: async (ctx, args) => {
    const row = await ctx.db
      .query("media_assets")
      .withIndex("by_asset_id", (q) => q.eq("id", args.id))
      .first();
    if (!row) throw new Error("未找到附件");
    await assertWorkspaceMember(ctx, args.userId, row.workspace_id);

    // 仅允许清理“已进入垃圾桶且超出宽限期”的附件；清空垃圾桶会传入一个很靠后的 expiredDeletedAt 以强制清理。
    if (!isExpiredForPurge((row as any).deleted_at, args.expiredDeletedAt)) {
      return { ok: false, reason: "not_expired" as const };
    }
    await deleteMediaAssetRow(ctx, row, new Set([String(row.id)]));
    return { ok: true, deleted: 1 };
  },
});

export const purgeIfExpired = internalMutation({
  args: { id: v.string(), deletedAt: v.string() },
  handler: async (ctx, args) => {
    const row = await ctx.db
      .query("media_assets")
      .withIndex("by_asset_id", (q) => q.eq("id", args.id))
      .first();

    if (!row) {
      return { ok: true, deleted: false, reason: "not_found" as const };
    }

    if ((row as any).deleted_at == null) {
      return { ok: true, deleted: false, reason: "restored" as const };
    }

    if ((row as any).deleted_at !== args.deletedAt) {
      // 删除时间不一致：说明已被恢复/重新删除过，旧调度作废
      return { ok: true, deleted: false, reason: "changed" as const };
    }

    const graceSeconds = resolveGraceSeconds();
    const deletedTs = Date.parse(String((row as any).deleted_at));
    if (Number.isFinite(deletedTs)) {
      const elapsedMs = Date.now() - deletedTs;
      const graceMs = graceSeconds * 1000;
      if (elapsedMs < graceMs) {
        const remainingMs = Math.max(1000, graceMs - elapsedMs + 5000);
        await ctx.scheduler.runAfter(remainingMs, internal.mediaAssets.purgeIfExpired, {
          id: args.id,
          deletedAt: args.deletedAt,
        });
        return { ok: true, deleted: false, rescheduled: true, remainingMs };
      }
    }

    await deleteMediaAssetRow(ctx, row, new Set([String(row.id)]));
    return { ok: true, deleted: true };
  },
});

export const emptyTrashByWorkspace = mutation({
  args: { userId: v.string(), workspaceId: v.string(), expiredDeletedAt: v.string() },
  handler: async (ctx, args) => {
    await assertWorkspaceMember(ctx, args.userId, args.workspaceId);

    const rows = await ctx.db
      .query("media_assets")
      .withIndex("by_workspace", (q) => q.eq("workspace_id", args.workspaceId))
      .order("desc")
      .take(5000);

    const targets = rows.filter((r) => {
      // 说明：
      // - 清空垃圾桶是强制操作，route 会传一个很靠后的 expiredDeletedAt
      // - 这里也顺手清掉历史 purged_at 墓碑（避免长期堆积）
      if (r.purged_at) return true;
      return isExpiredForPurge((r as any).deleted_at, args.expiredDeletedAt);
    });
    if (targets.length === 0) {
      return { ok: true, updated: 0 };
    }

    // 说明：按 storage_id 分组，只有当该 storage_id 没有任何“未清理”的引用时才删除底层文件。
    const byStorage = new Map<string, string[]>();
    for (const r of targets) {
      const sid = (r.storage_id as any) ?? null;
      if (!sid) continue;
      const list = byStorage.get(sid) ?? [];
      list.push(r.id);
      byStorage.set(sid, list);
    }

    for (const [sid] of byStorage.entries()) {
      const refs = await ctx.db
        .query("media_assets")
        .withIndex("by_storage_id", (q) => q.eq("storage_id", sid as any))
        .collect();
      const deletingIds = new Set(byStorage.get(sid) ?? []);
      // 说明：只要仍有“未清理”的引用（包括仍在垃圾桶但可恢复的记录），就不要删底层文件。
      const otherAlive = refs.some((r) => !deletingIds.has(r.id) && !r.purged_at);
      if (!otherAlive) {
        try {
          await ctx.storage.delete(sid as any);
        } catch {
          // ignore
        }
      }
    }

    for (const r of targets) {
      await ctx.db.delete(r._id);
    }

    return { ok: true, updated: targets.length, deleted: targets.length };
  },
});