369 lines
11 KiB
TypeScript
369 lines
11 KiB
TypeScript
import { makeFunctionReference } from "convex/server";
|
|
import type { FunctionReference } from "convex/server";
|
|
import { ConvexError, v } from "convex/values";
|
|
|
|
import type { Doc, Id } from "./_generated/dataModel";
|
|
import type { MutationCtx, QueryCtx } from "./_generated/server";
|
|
import { internalMutation, mutation, query } from "./_generated/server";
|
|
import { requireCurrentOrganization, requireWorkMember } from "./authz";
|
|
|
|
type TimelineEventView = Doc<"events"> & {
|
|
readonly artifacts: readonly Doc<"artifacts">[];
|
|
};
|
|
|
|
const promptHandlerRef = makeFunctionReference<
|
|
"mutation",
|
|
{ eventId: Id<"events"> },
|
|
undefined
|
|
>("timeline:handlePrompt") as unknown as FunctionReference<
|
|
"mutation",
|
|
"internal",
|
|
{ eventId: Id<"events"> },
|
|
undefined
|
|
>;
|
|
|
|
const promptPayload = (event: Doc<"events">): { readonly text: string } => {
|
|
if (
|
|
typeof event.payload !== "object" ||
|
|
event.payload === null ||
|
|
!("text" in event.payload) ||
|
|
typeof event.payload.text !== "string"
|
|
) {
|
|
throw new ConvexError("Prompt event payload is invalid");
|
|
}
|
|
return { text: event.payload.text };
|
|
};
|
|
|
|
const appendEvent = async (
|
|
ctx: MutationCtx,
|
|
input: {
|
|
readonly actor: Doc<"events">["actor"];
|
|
readonly artifactIds?: readonly Id<"artifacts">[];
|
|
readonly causationId?: Id<"events">;
|
|
readonly correlationId: string;
|
|
readonly idempotencyKey?: string;
|
|
readonly organizationId: Id<"organizations">;
|
|
readonly payload: unknown;
|
|
readonly projectId?: Id<"projects">;
|
|
readonly replyToEventId?: Id<"events">;
|
|
readonly scopeKind: "global" | "work";
|
|
readonly threadId?: Id<"threads">;
|
|
readonly type: string;
|
|
readonly visibility: "timeline" | "compact" | "internal";
|
|
readonly workId?: Id<"works">;
|
|
}
|
|
): Promise<Id<"events">> =>
|
|
await ctx.db.insert("events", {
|
|
actor: input.actor,
|
|
artifactIds: input.artifactIds ? [...input.artifactIds] : undefined,
|
|
causationId: input.causationId,
|
|
correlationId: input.correlationId,
|
|
idempotencyKey:
|
|
input.idempotencyKey ??
|
|
`${input.type}:${input.causationId ?? input.correlationId}`,
|
|
occurredAt: Date.now(),
|
|
organizationId: input.organizationId,
|
|
payload: input.payload,
|
|
projectId: input.projectId,
|
|
recordedAt: Date.now(),
|
|
replyToEventId: input.replyToEventId,
|
|
scopeKind: input.scopeKind,
|
|
threadId: input.threadId,
|
|
type: input.type,
|
|
visibility: input.visibility,
|
|
workId: input.workId,
|
|
});
|
|
|
|
const toEventView = async (
|
|
ctx: QueryCtx | MutationCtx,
|
|
event: Doc<"events">
|
|
): Promise<TimelineEventView> => {
|
|
const artifacts = await Promise.all(
|
|
(event.artifactIds ?? []).map((artifactId) => ctx.db.get(artifactId))
|
|
);
|
|
return {
|
|
...event,
|
|
artifacts: artifacts.filter(
|
|
(artifact): artifact is Doc<"artifacts"> => artifact !== null
|
|
),
|
|
};
|
|
};
|
|
|
|
interface EventQuery {
|
|
readonly order: (direction: "asc" | "desc") => {
|
|
readonly take: (limit: number) => Promise<Doc<"events">[]>;
|
|
};
|
|
}
|
|
|
|
const takeVisibleEvents = async (
|
|
baseQuery: EventQuery,
|
|
limit: number
|
|
): Promise<Doc<"events">[]> => {
|
|
const candidates = await baseQuery.order("desc").take(limit * 3);
|
|
return candidates
|
|
.filter((event) => event.visibility !== "internal")
|
|
.slice(0, limit);
|
|
};
|
|
|
|
export const sendPrompt = mutation({
|
|
args: {
|
|
clientIdempotencyKey: v.string(),
|
|
projectId: v.optional(v.id("projects")),
|
|
replyToEventId: v.optional(v.id("events")),
|
|
scope: v.union(
|
|
v.object({ kind: v.literal("global") }),
|
|
v.object({ kind: v.literal("work"), workId: v.id("works") })
|
|
),
|
|
text: v.string(),
|
|
},
|
|
handler: async (ctx, args): Promise<TimelineEventView> => {
|
|
const { organizationId, userId } = await requireCurrentOrganization(ctx);
|
|
const text = args.text.trim();
|
|
const clientIdempotencyKey = args.clientIdempotencyKey.trim();
|
|
if (!text) {
|
|
throw new ConvexError("Prompt text is required");
|
|
}
|
|
if (!clientIdempotencyKey) {
|
|
throw new ConvexError("Client idempotency key is required");
|
|
}
|
|
|
|
const existing = await ctx.db
|
|
.query("events")
|
|
.withIndex("by_organizationId_and_idempotencyKey", (q) =>
|
|
q
|
|
.eq("organizationId", organizationId)
|
|
.eq("idempotencyKey", clientIdempotencyKey)
|
|
)
|
|
.unique();
|
|
if (existing) {
|
|
return await toEventView(ctx, existing);
|
|
}
|
|
|
|
let workId: Id<"works"> | undefined;
|
|
const { scope } = args;
|
|
let { projectId } = args;
|
|
if (scope.kind === "work") {
|
|
const work = await ctx.db.get(scope.workId);
|
|
if (!work || work.organizationId !== organizationId) {
|
|
throw new ConvexError("Work not found");
|
|
}
|
|
const { _id: resolvedWorkId, primaryProjectId } = work;
|
|
workId = resolvedWorkId;
|
|
projectId ??= primaryProjectId;
|
|
}
|
|
if (projectId) {
|
|
const project = await ctx.db.get(projectId);
|
|
if (!project || project.organizationId !== organizationId) {
|
|
throw new ConvexError("Project not found");
|
|
}
|
|
}
|
|
|
|
const eventId = await ctx.db.insert("events", {
|
|
actor: { kind: "user", userId },
|
|
causationId: undefined,
|
|
correlationId: clientIdempotencyKey,
|
|
idempotencyKey: clientIdempotencyKey,
|
|
occurredAt: Date.now(),
|
|
organizationId,
|
|
payload: { text },
|
|
projectId,
|
|
recordedAt: Date.now(),
|
|
replyToEventId: args.replyToEventId,
|
|
scopeKind: scope.kind,
|
|
type: "client.prompt",
|
|
visibility: "timeline",
|
|
workId,
|
|
});
|
|
|
|
await ctx.scheduler.runAfter(0, promptHandlerRef, { eventId });
|
|
const event = await ctx.db.get(eventId);
|
|
if (!event) {
|
|
throw new ConvexError("Prompt event could not be read after insert");
|
|
}
|
|
return await toEventView(ctx, event);
|
|
},
|
|
});
|
|
|
|
export const getGlobal = query({
|
|
args: {
|
|
before: v.optional(v.number()),
|
|
limit: v.optional(v.number()),
|
|
},
|
|
handler: async (ctx, args): Promise<readonly TimelineEventView[]> => {
|
|
const { organizationId } = await requireCurrentOrganization(ctx);
|
|
const limit = Math.min(Math.max(Math.floor(args.limit ?? 50), 1), 100);
|
|
const baseQuery = ctx.db
|
|
.query("events")
|
|
.withIndex("by_organizationId_and_scopeKind_and_recordedAt", (q) => {
|
|
const scoped = q
|
|
.eq("organizationId", organizationId)
|
|
.eq("scopeKind", "global");
|
|
return args.before === undefined
|
|
? scoped
|
|
: scoped.lt("recordedAt", args.before);
|
|
});
|
|
const events = await takeVisibleEvents(baseQuery, limit);
|
|
return await Promise.all(events.map((event) => toEventView(ctx, event)));
|
|
},
|
|
});
|
|
|
|
export const getWork = query({
|
|
args: {
|
|
before: v.optional(v.number()),
|
|
limit: v.optional(v.number()),
|
|
workId: v.id("works"),
|
|
},
|
|
handler: async (ctx, args): Promise<readonly TimelineEventView[]> => {
|
|
const { organizationId } = await requireWorkMember(ctx, args.workId);
|
|
const limit = Math.min(Math.max(Math.floor(args.limit ?? 50), 1), 100);
|
|
const baseQuery = ctx.db
|
|
.query("events")
|
|
.withIndex(
|
|
"by_organizationId_and_scopeKind_and_workId_and_recordedAt",
|
|
(q) => {
|
|
const scoped = q
|
|
.eq("organizationId", organizationId)
|
|
.eq("scopeKind", "work")
|
|
.eq("workId", args.workId);
|
|
return args.before === undefined
|
|
? scoped
|
|
: scoped.lt("recordedAt", args.before);
|
|
}
|
|
);
|
|
const events = await takeVisibleEvents(baseQuery, limit);
|
|
return await Promise.all(events.map((event) => toEventView(ctx, event)));
|
|
},
|
|
});
|
|
|
|
export const handlePrompt = internalMutation({
|
|
args: { eventId: v.id("events") },
|
|
handler: async (ctx, args): Promise<void> => {
|
|
const prompt = await ctx.db.get(args.eventId);
|
|
if (!prompt || prompt.type !== "client.prompt") {
|
|
return;
|
|
}
|
|
const existingResponses = await ctx.db
|
|
.query("events")
|
|
.withIndex("by_organizationId_and_causationId", (q) =>
|
|
q
|
|
.eq("organizationId", prompt.organizationId)
|
|
.eq("causationId", args.eventId)
|
|
)
|
|
.collect();
|
|
if (
|
|
existingResponses.some((event) => event.type === "assistant.acknowledged")
|
|
) {
|
|
return;
|
|
}
|
|
|
|
const { text } = promptPayload(prompt);
|
|
const now = Date.now();
|
|
await appendEvent(ctx, {
|
|
actor: { kind: "system", service: "timeline" },
|
|
causationId: args.eventId,
|
|
correlationId: prompt.correlationId,
|
|
organizationId: prompt.organizationId,
|
|
payload: { text: "I have it." },
|
|
projectId: prompt.projectId,
|
|
replyToEventId: args.eventId,
|
|
scopeKind: prompt.scopeKind,
|
|
threadId: prompt.threadId,
|
|
type: "assistant.acknowledged",
|
|
visibility: "timeline",
|
|
workId: prompt.workId,
|
|
});
|
|
|
|
const { actor, projectId, workId: promptWorkId } = prompt;
|
|
let workId = promptWorkId;
|
|
if (!workId) {
|
|
workId = await ctx.db.insert("works", {
|
|
createdAt: now,
|
|
createdBy: actor.kind === "user" ? actor.userId : "system",
|
|
createdFromEventId: args.eventId,
|
|
organizationId: prompt.organizationId,
|
|
primaryProjectId: projectId,
|
|
summary: text,
|
|
title: text.slice(0, 80),
|
|
});
|
|
const workCreatedEventId = await appendEvent(ctx, {
|
|
actor: { kind: "system", service: "timeline" },
|
|
causationId: args.eventId,
|
|
correlationId: prompt.correlationId,
|
|
organizationId: prompt.organizationId,
|
|
payload: { title: text.slice(0, 80), workId },
|
|
projectId,
|
|
scopeKind: "global",
|
|
type: "work.created",
|
|
visibility: "compact",
|
|
workId,
|
|
});
|
|
const threadId = await ctx.db.insert("threads", {
|
|
createdAt: now,
|
|
createdFromEventId: workCreatedEventId,
|
|
organizationId: prompt.organizationId,
|
|
projectId,
|
|
purpose: "Initial exploration",
|
|
title: "Initial exploration",
|
|
workId,
|
|
});
|
|
await appendEvent(ctx, {
|
|
actor: { kind: "thread", threadId },
|
|
causationId: workCreatedEventId,
|
|
correlationId: prompt.correlationId,
|
|
organizationId: prompt.organizationId,
|
|
payload: { title: "Initial exploration" },
|
|
projectId,
|
|
scopeKind: "work",
|
|
threadId,
|
|
type: "thread.started",
|
|
visibility: "compact",
|
|
workId,
|
|
});
|
|
}
|
|
|
|
const previousArtifacts = await ctx.db
|
|
.query("artifacts")
|
|
.withIndex("by_workId_and_createdAt", (q) => q.eq("workId", workId))
|
|
.order("desc")
|
|
.take(1);
|
|
const version = (previousArtifacts[0]?.version ?? 0) + 1;
|
|
const artifactId = await ctx.db.insert("artifacts", {
|
|
card: {
|
|
component: "summary",
|
|
label: "Summary",
|
|
status: "ready",
|
|
summary: text,
|
|
title: text.slice(0, 80),
|
|
},
|
|
content: {
|
|
body: `# ${text.slice(0, 80)}\n\n${text}`,
|
|
type: "markdown",
|
|
},
|
|
createdAt: now,
|
|
createdByEventId: args.eventId,
|
|
kind: "summary",
|
|
logicalKey: `work:${workId}:summary`,
|
|
organizationId: prompt.organizationId,
|
|
projectId,
|
|
status: "ready",
|
|
summary: text,
|
|
title: text.slice(0, 80),
|
|
version,
|
|
workId,
|
|
});
|
|
await appendEvent(ctx, {
|
|
actor: { kind: "system", service: "timeline" },
|
|
artifactIds: [artifactId],
|
|
causationId: args.eventId,
|
|
correlationId: prompt.correlationId,
|
|
organizationId: prompt.organizationId,
|
|
payload: { artifactId, title: text.slice(0, 80) },
|
|
projectId,
|
|
scopeKind: "work",
|
|
type: "artifact.published",
|
|
visibility: "timeline",
|
|
workId,
|
|
});
|
|
},
|
|
});
|