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) { 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 { 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( 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(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(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), updated_at: nowIso() }; await ctx.db.patch(row._id, next); // 说明:当 OCR 写回完成时,自动触发 LightRAG 入库任务(Convex jobs/actions)。 const patch = args.patch as Record; const prev = row as Record; 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(); 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 }; }, });