mnote/recycle/20260522-convex-runtime-retirement/convex/aiSessions.ts

import { mutation, query } from "./_generated/server";
import { v } from "convex/values";

const runtimeRunArgs = {
  schema: v.literal("mnote.acp_runtime_run.v1"),
  source: v.literal("acp"),
  userId: v.string(),
  workspaceId: v.optional(v.union(v.string(), v.null())),
  documentId: v.string(),
  sessionId: v.string(),
  runId: v.string(),
  title: v.optional(v.union(v.string(), v.null())),
  profile: v.string(),
  acpRuntime: v.string(),
  traceId: v.string(),
  status: v.string(),
  runtime: v.any(),
  payload: v.any(),
  retention: v.any(),
  createdAt: v.number(),
  updatedAt: v.number(),
};

const runtimeEventArgs = {
  schema: v.literal("mnote.acp_runtime_event.v1"),
  source: v.literal("acp"),
  eventId: v.string(),
  userId: v.string(),
  workspaceId: v.optional(v.union(v.string(), v.null())),
  documentId: v.string(),
  sessionId: v.string(),
  runId: v.string(),
  profile: v.string(),
  acpRuntime: v.string(),
  eventType: v.string(),
  payload: v.any(),
  createdAt: v.number(),
};

function isoFromMillis(value: number) {
  return new Date(value).toISOString();
}

async function requireScopedUser(ctx: any, expectedUserId: string) {
  const identity = await ctx.auth.getUserIdentity();
  const subject = String(identity?.subject ?? "");
  const identityUserId = subject.split("|")[0] || "";
  if (identityUserId && identityUserId !== expectedUserId) {
    throw new Error("无权限");
  }
  if (!identityUserId && !expectedUserId.trim()) {
    throw new Error("未登录");
  }
  return expectedUserId;
}

function rowToClient(row: any) {
  return {
    id: row.id,
    userId: row.user_id,
    workspaceId: row.workspace_id,
    documentId: row.document_id,
    sessionId: row.session_id,
    runId: row.run_id,
    profile: row.profile,
    title: row.title,
    acpRuntime: row.acp_runtime,
    source: row.source,
    status: row.status,
    traceId: row.trace_id,
    runtime: row.runtime,
    usage: row.usage ?? null,
    payload: row.payload,
    retention: row.retention,
    createdAt: row.created_at,
    updatedAt: row.updated_at,
  };
}

export const upsertRuntimeRun = mutation({
  args: runtimeRunArgs,
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const workspaceId = args.workspaceId ?? null;
    const nowIso = isoFromMillis(args.updatedAt);
    const existing = await ctx.db
      .query("acp_runtime_runs")
      .withIndex("by_run_id", (q: any) => q.eq("run_id", args.runId))
      .first();

    const patch = {
      id: `acp_run:${args.runId}`,
      user_id: userId,
      workspace_id: workspaceId,
      document_id: args.documentId,
      session_id: args.sessionId,
      run_id: args.runId,
      title: args.title ?? null,
      profile: args.profile,
      acp_runtime: args.acpRuntime,
      source: args.source,
      status: args.status,
      trace_id: args.traceId,
      runtime: args.runtime,
      usage: null,
      payload: args.payload,
      retention: args.retention,
      updated_at: nowIso,
      deleted_at: null,
    };

    if (existing) {
      if (String(existing.user_id) !== userId) {
        throw new Error("无权限");
      }
      await ctx.db.patch(existing._id, patch);
      return { ok: true, id: existing.id, runId: args.runId, updated: true };
    }

    await ctx.db.insert("acp_runtime_runs", {
      ...patch,
      created_at: isoFromMillis(args.createdAt),
    });
    return { ok: true, id: patch.id, runId: args.runId, created: true };
  },
});

export const getRuntimeRun = query({
  args: {
    userId: v.string(),
    runId: v.string(),
  },
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const row = await ctx.db
      .query("acp_runtime_runs")
      .withIndex("by_run_id", (q: any) => q.eq("run_id", args.runId))
      .first();
    if (!row || row.deleted_at !== null || String(row.user_id) !== userId) {
      return null;
    }
    return rowToClient(row);
  },
});

export const listRuntimeRuns = query({
  args: {
    userId: v.string(),
    workspaceId: v.optional(v.union(v.string(), v.null())),
    documentId: v.optional(v.union(v.string(), v.null())),
    sessionId: v.optional(v.union(v.string(), v.null())),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const workspaceId = args.workspaceId ?? null;
    const limit = Math.min(Math.max(args.limit ?? 50, 1), 100);
    let rows;
    if (args.sessionId) {
      rows = await ctx.db
        .query("acp_runtime_runs")
        .withIndex("by_user_workspace_session", (q: any) =>
          q.eq("user_id", userId).eq("workspace_id", workspaceId).eq("session_id", args.sessionId),
        )
        .order("desc")
        .take(limit);
    } else if (args.documentId) {
      rows = await ctx.db
        .query("acp_runtime_runs")
        .withIndex("by_user_workspace_document", (q: any) =>
          q.eq("user_id", userId).eq("workspace_id", workspaceId).eq("document_id", args.documentId),
        )
        .order("desc")
        .take(limit);
    } else {
      rows = await ctx.db
        .query("acp_runtime_runs")
        .withIndex("by_user_created_at", (q: any) => q.eq("user_id", userId))
        .order("desc")
        .take(limit);
    }

    return rows
      .filter((row: any) => row.deleted_at === null)
      .map(rowToClient);
  },
});

export const appendRuntimeEvent = mutation({
  args: runtimeEventArgs,
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const workspaceId = args.workspaceId ?? null;
    const existingRun = await ctx.db
      .query("acp_runtime_runs")
      .withIndex("by_run_id", (q: any) => q.eq("run_id", args.runId))
      .first();
    if (existingRun && String(existingRun.user_id) !== userId) {
      throw new Error("无权限");
    }

    await ctx.db.insert("acp_runtime_events", {
      id: args.eventId,
      user_id: userId,
      workspace_id: workspaceId,
      document_id: args.documentId,
      session_id: args.sessionId,
      run_id: args.runId,
      profile: args.profile,
      acp_runtime: args.acpRuntime,
      source: args.source,
      event_type: args.eventType,
      payload: args.payload,
      created_at: isoFromMillis(args.createdAt),
    });
    const usage = usageFromRuntimeEvent(args.eventType, args.payload);
    if (existingRun && usage) {
      await ctx.db.patch(existingRun._id, {
        usage,
        updated_at: isoFromMillis(args.createdAt),
      });
    }
    return { ok: true, id: args.eventId, runId: args.runId };
  },
});

function usageFromRuntimeEvent(eventType: string, payload: any) {
  if (eventType === "usage.updated") {
    return {
      source: "usage_update",
      contextUsed: Number(payload?.used ?? payload?.contextUsed ?? 0),
      contextSize: Number(payload?.size ?? payload?.contextSize ?? 0),
    };
  }
  if (eventType === "run.completed" && payload?.usage && typeof payload.usage === "object") {
    return {
      source: "run_completed",
      ...payload.usage,
    };
  }
  return null;
}

export const listRuntimeEvents = query({
  args: {
    userId: v.string(),
    runId: v.string(),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const limit = Math.min(Math.max(args.limit ?? 100, 1), 500);
    const rows = await ctx.db
      .query("acp_runtime_events")
      .withIndex("by_run_created_at", (q: any) => q.eq("run_id", args.runId))
      .order("asc")
      .take(limit);
    return rows
      .filter((row: any) => String(row.user_id) === userId)
      .map((row: any) => ({
        id: row.id,
        userId: row.user_id,
        workspaceId: row.workspace_id,
        documentId: row.document_id,
        sessionId: row.session_id,
        runId: row.run_id,
        profile: row.profile,
        acpRuntime: row.acp_runtime,
        source: row.source,
        eventType: row.event_type,
        payload: row.payload,
        createdAt: row.created_at,
      }));
  },
});

function titleFromRun(row: any) {
  const payload = row?.payload ?? {};
  const raw = String(payload.message ?? payload.input ?? payload.title ?? "").trim();
  if (!raw) return "当前页问答";
  return raw.length > 32 ? `${raw.slice(0, 32)}...` : raw;
}

async function sessionRows(ctx: any, userId: string, workspaceId: string | null, sessionId: string) {
  return await ctx.db
    .query("acp_runtime_runs")
    .withIndex("by_user_workspace_session", (q: any) =>
      q.eq("user_id", userId).eq("workspace_id", workspaceId).eq("session_id", sessionId),
    )
    .collect();
}

export const renameRuntimeSession = mutation({
  args: {
    userId: v.string(),
    workspaceId: v.optional(v.union(v.string(), v.null())),
    sessionId: v.string(),
    title: v.string(),
  },
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const workspaceId = args.workspaceId ?? null;
    const title = args.title.trim();
    if (!title) throw new Error("标题不能为空");
    const rows = await sessionRows(ctx, userId, workspaceId, args.sessionId);
    for (const row of rows) {
      if (row.deleted_at === null) {
        await ctx.db.patch(row._id, {
          title,
          updated_at: new Date().toISOString(),
        });
      }
    }
    return { ok: true, sessionId: args.sessionId, title, updated: rows.length };
  },
});

export const autoTitleRuntimeSession = mutation({
  args: {
    userId: v.string(),
    workspaceId: v.optional(v.union(v.string(), v.null())),
    sessionId: v.string(),
  },
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const workspaceId = args.workspaceId ?? null;
    const rows = await sessionRows(ctx, userId, workspaceId, args.sessionId);
    const title = titleFromRun(rows.find((row: any) => row.deleted_at === null));
    for (const row of rows) {
      if (row.deleted_at === null) {
        await ctx.db.patch(row._id, {
          title,
          updated_at: new Date().toISOString(),
        });
      }
    }
    return { ok: true, sessionId: args.sessionId, title, updated: rows.length };
  },
});

export const deleteRuntimeSession = mutation({
  args: {
    userId: v.string(),
    workspaceId: v.optional(v.union(v.string(), v.null())),
    sessionId: v.string(),
  },
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const workspaceId = args.workspaceId ?? null;
    const rows = await sessionRows(ctx, userId, workspaceId, args.sessionId);
    const deletedAt = new Date().toISOString();
    for (const row of rows) {
      if (row.deleted_at === null) {
        await ctx.db.patch(row._id, {
          deleted_at: deletedAt,
          updated_at: deletedAt,
        });
      }
    }
    return { ok: true, sessionId: args.sessionId, deleted: rows.length };
  },
});

function searchText(row: any) {
  const payload = row?.payload ?? {};
  return [
    row.title,
    row.session_id,
    row.run_id,
    row.profile,
    row.status,
    payload.message,
    payload.input,
    payload.title,
  ]
    .map((value) => String(value ?? ""))
    .join(" ")
    .toLowerCase();
}

function snippetFor(row: any, q: string) {
  const payload = row?.payload ?? {};
  const raw = String(row.title ?? payload.message ?? payload.input ?? row.session_id ?? "").trim();
  if (!raw) return "";
  const index = raw.toLowerCase().indexOf(q.toLowerCase());
  if (index < 0) return raw.length > 80 ? `${raw.slice(0, 80)}...` : raw;
  const start = Math.max(0, index - 24);
  const end = Math.min(raw.length, index + q.length + 56);
  return `${start > 0 ? "..." : ""}${raw.slice(start, end)}${end < raw.length ? "..." : ""}`;
}

export const searchRuntimeSessions = query({
  args: {
    userId: v.string(),
    workspaceId: v.optional(v.union(v.string(), v.null())),
    q: v.string(),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const userId = await requireScopedUser(ctx, args.userId);
    const workspaceId = args.workspaceId ?? null;
    const q = args.q.trim().toLowerCase();
    if (!q) return [];
    const limit = Math.min(Math.max(args.limit ?? 20, 1), 50);
    const rows = await ctx.db
      .query("acp_runtime_runs")
      .withIndex("by_user_created_at", (query: any) => query.eq("user_id", userId))
      .order("desc")
      .take(500);
    const seen = new Set<string>();
    const results = [];
    for (const row of rows) {
      if (row.deleted_at !== null) continue;
      if ((row.workspace_id ?? null) !== workspaceId) continue;
      if (!searchText(row).includes(q)) continue;
      if (seen.has(row.session_id)) continue;
      seen.add(row.session_id);
      results.push({
        sessionId: row.session_id,
        runId: row.run_id,
        workspaceId: row.workspace_id,
        documentId: row.document_id,
        title: row.title ?? titleFromRun(row),
        profile: row.profile,
        status: row.status,
        snippet: snippetFor(row, args.q),
        score: 1,
      });
      if (results.length >= limit) break;
    }
    return results;
  },
});