333 lines
11 KiB
TypeScript
333 lines
11 KiB
TypeScript
import { mutation, query } from "./_generated/server";
|
|
import { v } from "convex/values";
|
|
import { requireUserId } from "./_utils/auth";
|
|
import { nowIso } from "./_utils/time";
|
|
|
|
async function requireWorkspaceMember(ctx: any, workspaceId: string, userId: string) {
|
|
const member = await ctx.db
|
|
.query("workspace_members")
|
|
.withIndex("by_workspace_user", (q: any) => q.eq("workspace_id", workspaceId).eq("user_id", userId))
|
|
.first();
|
|
if (!member) {
|
|
throw new Error("无权限");
|
|
}
|
|
return member;
|
|
}
|
|
|
|
function sortByNewest<T extends Record<string, any>>(rows: T[]) {
|
|
return [...rows].sort((left, right) => {
|
|
const leftTime = String(left.created_at ?? left.finished_at ?? "");
|
|
const rightTime = String(right.created_at ?? right.finished_at ?? "");
|
|
return rightTime.localeCompare(leftTime);
|
|
});
|
|
}
|
|
|
|
function decodeCursor(raw: string | null | undefined) {
|
|
if (!raw) return null;
|
|
try {
|
|
const decoded = JSON.parse(raw) as {
|
|
createdAt?: string | null;
|
|
id?: string | null;
|
|
};
|
|
const createdAt = typeof decoded.createdAt === "string" ? decoded.createdAt : "";
|
|
const id = typeof decoded.id === "string" ? decoded.id : "";
|
|
if (!createdAt || !id) return null;
|
|
return { createdAt, id };
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function encodeCursor(row: { created_at?: string | null; id?: string | null } | null) {
|
|
if (!row?.created_at || !row?.id) return null;
|
|
return JSON.stringify({
|
|
createdAt: row.created_at,
|
|
id: row.id,
|
|
});
|
|
}
|
|
|
|
function matchesCursor<T extends Record<string, any>>(
|
|
row: T,
|
|
cursor: { createdAt: string; id: string } | null,
|
|
) {
|
|
if (!cursor) return true;
|
|
const createdAt = String(row.created_at ?? row.finished_at ?? "");
|
|
const id = String(row.id ?? "");
|
|
if (!createdAt || !id) return false;
|
|
if (createdAt < cursor.createdAt) return true;
|
|
if (createdAt > cursor.createdAt) return false;
|
|
return id < cursor.id;
|
|
}
|
|
|
|
function normalizeStatusFilter(raw: string | null | undefined) {
|
|
const normalized = String(raw ?? "").trim();
|
|
return normalized.length > 0 ? normalized : null;
|
|
}
|
|
|
|
function normalizeObjectFilter(raw: string | null | undefined) {
|
|
const normalized = String(raw ?? "").trim();
|
|
return normalized.length > 0 ? normalized : null;
|
|
}
|
|
|
|
export const recordCommandLog = mutation({
|
|
args: {
|
|
workspaceId: v.string(),
|
|
id: v.string(),
|
|
requestId: v.string(),
|
|
traceId: v.string(),
|
|
commandId: v.string(),
|
|
commandName: v.string(),
|
|
actorId: v.string(),
|
|
actorType: v.string(),
|
|
sourceChannel: v.string(),
|
|
sourceClient: v.string(),
|
|
status: v.union(v.literal("pending"), v.literal("succeeded"), v.literal("failed"), v.literal("rolled_back")),
|
|
targetPageId: v.optional(v.union(v.string(), v.null())),
|
|
targetBlockId: v.optional(v.union(v.string(), v.null())),
|
|
payload: v.any(),
|
|
payloadSummary: v.string(),
|
|
refs: v.array(v.string()),
|
|
idempotencyKey: v.optional(v.union(v.string(), v.null())),
|
|
error: v.optional(v.union(v.string(), v.null())),
|
|
createdAt: v.string(),
|
|
finishedAt: v.optional(v.union(v.string(), v.null())),
|
|
},
|
|
handler: async (ctx, args) => {
|
|
const userId = await requireUserId(ctx);
|
|
await requireWorkspaceMember(ctx, args.workspaceId, userId);
|
|
|
|
const existing = await ctx.db
|
|
.query("command_logs")
|
|
.withIndex("by_command_log_id", (q) => q.eq("id", args.id))
|
|
.first();
|
|
if (existing) {
|
|
return { ok: true, id: existing.id, duplicated: true };
|
|
}
|
|
|
|
await ctx.db.insert("command_logs", {
|
|
id: args.id,
|
|
workspace_id: args.workspaceId,
|
|
request_id: args.requestId,
|
|
trace_id: args.traceId,
|
|
command_id: args.commandId,
|
|
command_name: args.commandName,
|
|
actor_id: args.actorId,
|
|
actor_type: args.actorType,
|
|
source_channel: args.sourceChannel,
|
|
source_client: args.sourceClient,
|
|
status: args.status,
|
|
target_page_id: args.targetPageId ?? null,
|
|
target_block_id: args.targetBlockId ?? null,
|
|
payload: args.payload,
|
|
payload_summary: args.payloadSummary,
|
|
refs: args.refs,
|
|
idempotency_key: args.idempotencyKey ?? null,
|
|
error: args.error ?? null,
|
|
created_at: args.createdAt,
|
|
finished_at: args.finishedAt ?? null,
|
|
});
|
|
return { ok: true, id: args.id, duplicated: false };
|
|
},
|
|
});
|
|
|
|
export const recordDomainEvent = mutation({
|
|
args: {
|
|
workspaceId: v.string(),
|
|
id: v.string(),
|
|
requestId: v.string(),
|
|
traceId: v.string(),
|
|
commandId: v.string(),
|
|
commandLogId: v.string(),
|
|
eventType: v.string(),
|
|
aggregateType: v.string(),
|
|
aggregateId: v.string(),
|
|
eventVersion: v.number(),
|
|
status: v.union(v.literal("pending"), v.literal("committed"), v.literal("rejected"), v.literal("failed")),
|
|
actorType: v.string(),
|
|
payload: v.any(),
|
|
createdAt: v.string(),
|
|
},
|
|
handler: async (ctx, args) => {
|
|
const userId = await requireUserId(ctx);
|
|
await requireWorkspaceMember(ctx, args.workspaceId, userId);
|
|
|
|
const existing = await ctx.db
|
|
.query("domain_events")
|
|
.withIndex("by_domain_event_id", (q) => q.eq("id", args.id))
|
|
.first();
|
|
if (existing) {
|
|
return { ok: true, id: existing.id, duplicated: true };
|
|
}
|
|
|
|
await ctx.db.insert("domain_events", {
|
|
id: args.id,
|
|
workspace_id: args.workspaceId,
|
|
request_id: args.requestId,
|
|
trace_id: args.traceId,
|
|
command_id: args.commandId,
|
|
command_log_id: args.commandLogId,
|
|
event_type: args.eventType,
|
|
aggregate_type: args.aggregateType,
|
|
aggregate_id: args.aggregateId,
|
|
event_version: args.eventVersion,
|
|
status: args.status,
|
|
actor_type: args.actorType,
|
|
payload: args.payload,
|
|
created_at: args.createdAt,
|
|
});
|
|
return { ok: true, id: args.id, duplicated: false };
|
|
},
|
|
});
|
|
|
|
export const listByTrace = query({
|
|
args: { workspaceId: v.string(), traceId: v.string() },
|
|
handler: async (ctx, args) => {
|
|
const userId = await requireUserId(ctx);
|
|
await requireWorkspaceMember(ctx, args.workspaceId, userId);
|
|
|
|
const commandLogs = await ctx.db
|
|
.query("command_logs")
|
|
.withIndex("by_workspace_trace", (q) => q.eq("workspace_id", args.workspaceId).eq("trace_id", args.traceId))
|
|
.collect();
|
|
const domainEvents = await ctx.db
|
|
.query("domain_events")
|
|
.withIndex("by_workspace_trace", (q) => q.eq("workspace_id", args.workspaceId).eq("trace_id", args.traceId))
|
|
.collect();
|
|
|
|
return {
|
|
trace_id: args.traceId,
|
|
command_logs: commandLogs,
|
|
domain_events: domainEvents,
|
|
generated_at: nowIso(),
|
|
};
|
|
},
|
|
});
|
|
|
|
export const listByRequest = query({
|
|
args: { workspaceId: v.string(), requestId: v.string() },
|
|
handler: async (ctx, args) => {
|
|
const userId = await requireUserId(ctx);
|
|
await requireWorkspaceMember(ctx, args.workspaceId, userId);
|
|
|
|
const commandLogs = await ctx.db
|
|
.query("command_logs")
|
|
.withIndex("by_workspace_request", (q) => q.eq("workspace_id", args.workspaceId).eq("request_id", args.requestId))
|
|
.collect();
|
|
const domainEvents = await ctx.db
|
|
.query("domain_events")
|
|
.withIndex("by_workspace_request", (q) => q.eq("workspace_id", args.workspaceId).eq("request_id", args.requestId))
|
|
.collect();
|
|
|
|
return {
|
|
request_id: args.requestId,
|
|
command_logs: commandLogs,
|
|
domain_events: domainEvents,
|
|
generated_at: nowIso(),
|
|
};
|
|
},
|
|
});
|
|
|
|
export const listByCommand = query({
|
|
args: { workspaceId: v.string(), commandId: v.string() },
|
|
handler: async (ctx, args) => {
|
|
const userId = await requireUserId(ctx);
|
|
await requireWorkspaceMember(ctx, args.workspaceId, userId);
|
|
|
|
const commandLogs = await ctx.db
|
|
.query("command_logs")
|
|
.withIndex("by_workspace_command", (q) =>
|
|
q.eq("workspace_id", args.workspaceId).eq("command_id", args.commandId),
|
|
)
|
|
.collect();
|
|
const domainEvents = await ctx.db
|
|
.query("domain_events")
|
|
.withIndex("by_workspace_command", (q) =>
|
|
q.eq("workspace_id", args.workspaceId).eq("command_id", args.commandId),
|
|
)
|
|
.collect();
|
|
|
|
return {
|
|
command_id: args.commandId,
|
|
command_logs: commandLogs,
|
|
domain_events: domainEvents,
|
|
generated_at: nowIso(),
|
|
};
|
|
},
|
|
});
|
|
|
|
export const listWorkspaceOverview = query({
|
|
args: {
|
|
workspaceId: v.string(),
|
|
limit: v.optional(v.number()),
|
|
cursor: v.optional(v.union(v.string(), v.null())),
|
|
commandStatus: v.optional(v.union(v.string(), v.null())),
|
|
eventStatus: v.optional(v.union(v.string(), v.null())),
|
|
targetPageId: v.optional(v.union(v.string(), v.null())),
|
|
targetBlockId: v.optional(v.union(v.string(), v.null())),
|
|
aggregateType: v.optional(v.union(v.string(), v.null())),
|
|
aggregateId: v.optional(v.union(v.string(), v.null())),
|
|
},
|
|
handler: async (ctx, args) => {
|
|
const userId = await requireUserId(ctx);
|
|
await requireWorkspaceMember(ctx, args.workspaceId, userId);
|
|
|
|
const limit = Math.max(1, Math.min(100, Math.floor(args.limit ?? 50)));
|
|
const cursor = decodeCursor(args.cursor ?? null);
|
|
const commandStatus = normalizeStatusFilter(args.commandStatus);
|
|
const eventStatus = normalizeStatusFilter(args.eventStatus);
|
|
const targetPageId = normalizeObjectFilter(args.targetPageId);
|
|
const targetBlockId = normalizeObjectFilter(args.targetBlockId);
|
|
const aggregateType = normalizeObjectFilter(args.aggregateType);
|
|
const aggregateId = normalizeObjectFilter(args.aggregateId);
|
|
|
|
const allCommandLogs = await ctx.db
|
|
.query("command_logs")
|
|
.withIndex("by_workspace_request", (q) => q.eq("workspace_id", args.workspaceId))
|
|
.collect();
|
|
const allDomainEvents = await ctx.db
|
|
.query("domain_events")
|
|
.withIndex("by_workspace_request", (q) => q.eq("workspace_id", args.workspaceId))
|
|
.collect();
|
|
|
|
const filteredCommandLogs = sortByNewest(
|
|
allCommandLogs.filter((row: any) => {
|
|
if (commandStatus && row.status !== commandStatus) return false;
|
|
if (targetPageId && String(row.target_page_id ?? "") !== targetPageId) return false;
|
|
if (targetBlockId && String(row.target_block_id ?? "") !== targetBlockId) return false;
|
|
return matchesCursor(row, cursor);
|
|
}),
|
|
);
|
|
|
|
const pageCommandLogs = filteredCommandLogs.slice(0, limit);
|
|
const nextCursor = encodeCursor(pageCommandLogs[pageCommandLogs.length - 1] ?? null);
|
|
const commandIds = new Set(pageCommandLogs.map((row: any) => String(row.command_id)));
|
|
|
|
const domainEvents = sortByNewest(
|
|
allDomainEvents.filter((row: any) => {
|
|
if (commandIds.size > 0 && !commandIds.has(String(row.command_id ?? ""))) return false;
|
|
if (eventStatus && row.status !== eventStatus) return false;
|
|
if (aggregateType && String(row.aggregate_type ?? "") !== aggregateType) return false;
|
|
if (aggregateId && String(row.aggregate_id ?? "") !== aggregateId) return false;
|
|
return true;
|
|
}),
|
|
);
|
|
|
|
return {
|
|
workspace_id: args.workspaceId,
|
|
command_logs: pageCommandLogs,
|
|
domain_events: domainEvents,
|
|
next_cursor: nextCursor,
|
|
has_more: filteredCommandLogs.length > pageCommandLogs.length,
|
|
filters: {
|
|
command_status: commandStatus,
|
|
event_status: eventStatus,
|
|
target_page_id: targetPageId,
|
|
target_block_id: targetBlockId,
|
|
aggregate_type: aggregateType,
|
|
aggregate_id: aggregateId,
|
|
},
|
|
generated_at: nowIso(),
|
|
};
|
|
},
|
|
});
|