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;
}

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"),
    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);

    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: 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 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 };
  },
});