292 lines
10 KiB
TypeScript
292 lines
10 KiB
TypeScript
import { env } from "@code/env/convex";
|
|
import { makeFunctionReference } from "convex/server";
|
|
import { ConvexError } from "convex/values";
|
|
|
|
import type { Id } from "./_generated/dataModel";
|
|
import { httpAction } from "./_generated/server";
|
|
import type { ActionCtx } from "./_generated/server";
|
|
// never call Convex mutations directly; they POST to these HTTP endpoints and
|
|
// we run the validated mutations server-side.
|
|
const appendToolEventRef = makeFunctionReference<
|
|
"mutation",
|
|
{
|
|
actorTool: string;
|
|
artifactIds?: Id<"artifacts">[];
|
|
correlationId: string;
|
|
idempotencyKey: string;
|
|
organizationId: Id<"organizations">;
|
|
payload: unknown;
|
|
projectId?: Id<"projects">;
|
|
replyToEventId?: Id<"events">;
|
|
scopeKind: "global" | "work";
|
|
threadId?: Id<"threads">;
|
|
type: string;
|
|
visibility: "timeline" | "compact" | "internal";
|
|
workId?: Id<"works">;
|
|
},
|
|
Id<"events">
|
|
>("events:appendToolEvent");
|
|
const publishRevisionRef = makeFunctionReference<
|
|
"mutation",
|
|
{
|
|
card: unknown;
|
|
content: unknown;
|
|
createdByEventId: Id<"events">;
|
|
kind: string;
|
|
logicalKey: string;
|
|
organizationId: Id<"organizations">;
|
|
projectId?: Id<"projects">;
|
|
status: string;
|
|
storageId?: Id<"_storage">;
|
|
summary: string;
|
|
threadId?: Id<"threads">;
|
|
title: string;
|
|
workId?: Id<"works">;
|
|
},
|
|
{ _id: Id<"artifacts">; _creationTime: number }
|
|
>("artifacts:publishRevision");
|
|
const getAgentForProjectRef = makeFunctionReference<
|
|
"query",
|
|
{ projectId: Id<"projects"> },
|
|
{
|
|
_id: Id<"conversationAgents">;
|
|
id: string;
|
|
organizationId: Id<"organizations">;
|
|
projectId: Id<"projects">;
|
|
} | null
|
|
>("conversationAgents:getForProject");
|
|
|
|
/**
|
|
* Validate the `Authorization: Bearer <FLUE_DB_TOKEN>` header. Returns the
|
|
* token on success; throws a ConvexError (mapped to 401 by the httpAction).
|
|
*/
|
|
const requireServiceToken = (request: Request): string => {
|
|
if (!env.FLUE_DB_TOKEN) {
|
|
throw new ConvexError("FLUE_DB_TOKEN is not configured");
|
|
}
|
|
const header = request.headers.get("authorization") ?? "";
|
|
if (!header.startsWith("Bearer ")) {
|
|
throw new ConvexError("Missing bearer token");
|
|
}
|
|
const token = header.slice("Bearer ".length).trim();
|
|
if (token.length === 0 || token !== env.FLUE_DB_TOKEN) {
|
|
throw new ConvexError("Invalid service token");
|
|
}
|
|
return token;
|
|
};
|
|
|
|
interface BoundaryContext {
|
|
agentId: string;
|
|
organizationId: Id<"organizations">;
|
|
projectId: Id<"projects">;
|
|
}
|
|
|
|
/** Shape returned by the conversationAgents:getForProject internal query. */
|
|
interface AgentRow {
|
|
_id: Id<"conversationAgents">;
|
|
id: string;
|
|
organizationId: Id<"organizations">;
|
|
projectId: Id<"projects">;
|
|
}
|
|
|
|
/**
|
|
* Enforce the agent -> project -> organization boundary. The caller asserts an
|
|
* `agentId` of the form `conversation:<org>:<project>:v1`; we verify the
|
|
* conversationAgents row exists for that project and belongs to the claimed
|
|
* organization. This prevents a project-A agent from writing into project-B.
|
|
*/
|
|
const resolveBoundary = async (
|
|
ctx: ActionCtx,
|
|
body: { agentId?: unknown; organizationId?: unknown; projectId?: unknown }
|
|
): Promise<BoundaryContext> => {
|
|
if (typeof body.agentId !== "string" || body.agentId.length === 0) {
|
|
throw new ConvexError("agentId is required");
|
|
}
|
|
if (typeof body.projectId !== "string") {
|
|
throw new ConvexError("projectId is required");
|
|
}
|
|
if (typeof body.organizationId !== "string") {
|
|
throw new ConvexError("organizationId is required");
|
|
}
|
|
const projectId = body.projectId as Id<"projects">;
|
|
const organizationId = body.organizationId as Id<"organizations">;
|
|
const agent = (await ctx.runQuery(getAgentForProjectRef, {
|
|
projectId,
|
|
})) as AgentRow | null;
|
|
if (
|
|
!agent ||
|
|
agent.id !== body.agentId ||
|
|
agent.organizationId !== organizationId
|
|
) {
|
|
throw new ConvexError("Agent is not bound to this project/organization");
|
|
}
|
|
return { agentId: body.agentId, organizationId, projectId };
|
|
};
|
|
|
|
const json = (status: number, payload: unknown): Response =>
|
|
Response.json(payload, { status });
|
|
|
|
const asString = (value: unknown, field: string): string => {
|
|
if (typeof value !== "string" || value.length === 0) {
|
|
throw new ConvexError(`${field} is required`);
|
|
}
|
|
return value;
|
|
};
|
|
|
|
const asOptionalId = (value: unknown): string | undefined => {
|
|
if (typeof value !== "string" || value.length === 0) {
|
|
return undefined;
|
|
}
|
|
return value;
|
|
};
|
|
|
|
/**
|
|
* POST /api/agents/events/message
|
|
*
|
|
* Service-authenticated endpoint for a project conversation agent to append a
|
|
* timeline event (an assistant message, acknowledgement, etc.). Enforces the
|
|
* agent->project->organization boundary and exact-once semantics via the
|
|
* supplied idempotency key.
|
|
*
|
|
* Body: {
|
|
* agentId, organizationId, projectId,
|
|
* text, type, visibility, idempotencyKey, correlationId,
|
|
* replyToEventId?, workId?, threadId?
|
|
* }
|
|
*/
|
|
export const postAgentMessage = httpAction(async (ctx, request) => {
|
|
try {
|
|
requireServiceToken(request);
|
|
const body = (await request.json()) as Record<string, unknown>;
|
|
const boundary = await resolveBoundary(ctx, body);
|
|
const text = asString(body.text, "text");
|
|
const type = asString(body.type, "type");
|
|
const visibility = asString(body.visibility, "visibility");
|
|
const idempotencyKey = asString(body.idempotencyKey, "idempotencyKey");
|
|
const correlationId = asString(body.correlationId, "correlationId");
|
|
if (!["timeline", "compact", "internal"].includes(visibility)) {
|
|
return json(400, { error: "Invalid visibility" });
|
|
}
|
|
const eventId = (await ctx.runMutation(appendToolEventRef, {
|
|
actorTool: boundary.agentId,
|
|
artifactIds: undefined,
|
|
correlationId,
|
|
idempotencyKey,
|
|
organizationId: boundary.organizationId,
|
|
payload: { text },
|
|
projectId: boundary.projectId,
|
|
replyToEventId: asOptionalId(body.replyToEventId) as
|
|
| Id<"events">
|
|
| undefined,
|
|
scopeKind: "global",
|
|
threadId: asOptionalId(body.threadId) as Id<"threads"> | undefined,
|
|
type,
|
|
visibility: visibility as "timeline" | "compact" | "internal",
|
|
workId: asOptionalId(body.workId) as Id<"works"> | undefined,
|
|
})) as Id<"events">;
|
|
return json(200, { eventId });
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
const status = message.startsWith("Unauthorized") ? 401 : 400;
|
|
return json(status, { error: message });
|
|
}
|
|
});
|
|
|
|
/**
|
|
* POST /api/agents/events/artifact
|
|
*
|
|
* Service-authenticated endpoint for a project conversation agent to publish a
|
|
* typed artifact revision (with an optional Convex Storage id) and emit the
|
|
* `artifact.published` event referencing it. Enforces the boundary and is
|
|
* idempotent through the artifact's logicalKey+version and the event key.
|
|
*
|
|
* Body: {
|
|
* agentId, organizationId, projectId,
|
|
* createdByEventId, kind, logicalKey, title, summary, status,
|
|
* card, content, storageId?, idempotencyKey, correlationId,
|
|
* workId?, threadId?
|
|
* }
|
|
*/
|
|
export const publishAgentArtifact = httpAction(async (ctx, request) => {
|
|
try {
|
|
requireServiceToken(request);
|
|
const body = (await request.json()) as Record<string, unknown>;
|
|
const boundary = await resolveBoundary(ctx, body);
|
|
const createdByEventId = asString(
|
|
body.createdByEventId,
|
|
"createdByEventId"
|
|
) as Id<"events">;
|
|
const kind = asString(body.kind, "kind");
|
|
const logicalKey = asString(body.logicalKey, "logicalKey");
|
|
const title = asString(body.title, "title");
|
|
const summary = asString(body.summary, "summary");
|
|
const status = asString(body.status, "status");
|
|
const idempotencyKey = asString(body.idempotencyKey, "idempotencyKey");
|
|
const correlationId = asString(body.correlationId, "correlationId");
|
|
const card = body.card ?? {};
|
|
const content = body.content ?? {};
|
|
|
|
const artifact = (await ctx.runMutation(publishRevisionRef, {
|
|
card,
|
|
content,
|
|
createdByEventId,
|
|
kind,
|
|
logicalKey,
|
|
organizationId: boundary.organizationId,
|
|
projectId: boundary.projectId,
|
|
status,
|
|
storageId: asOptionalId(body.storageId) as Id<"_storage"> | undefined,
|
|
summary,
|
|
threadId: asOptionalId(body.threadId) as Id<"threads"> | undefined,
|
|
title,
|
|
workId: asOptionalId(body.workId) as Id<"works"> | undefined,
|
|
})) as { _id: Id<"artifacts"> };
|
|
|
|
// Emit the artifact.published timeline event (exact-once).
|
|
const eventId = (await ctx.runMutation(appendToolEventRef, {
|
|
actorTool: boundary.agentId,
|
|
artifactIds: [artifact._id],
|
|
correlationId,
|
|
idempotencyKey: `${idempotencyKey}:event`,
|
|
organizationId: boundary.organizationId,
|
|
payload: { artifactId: artifact._id, title },
|
|
projectId: boundary.projectId,
|
|
scopeKind: "global",
|
|
threadId: asOptionalId(body.threadId) as Id<"threads"> | undefined,
|
|
type: "artifact.published",
|
|
visibility: "timeline",
|
|
workId: asOptionalId(body.workId) as Id<"works"> | undefined,
|
|
})) as Id<"events">;
|
|
|
|
return json(200, { artifactId: artifact._id, eventId });
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
const status = message.startsWith("Unauthorized") ? 401 : 400;
|
|
return json(status, { error: message });
|
|
}
|
|
});
|
|
|
|
/**
|
|
* POST /api/agents/storage/generate-upload-url
|
|
*
|
|
* Service-authenticated endpoint for an agent to obtain a one-time Convex
|
|
* Storage upload URL for a binary artifact (e.g. rendered HTML). The agent
|
|
* uploads the bytes to the returned URL and then references the resulting
|
|
* storageId in {@link publishAgentArtifact}.
|
|
*
|
|
* Body: { agentId, organizationId, projectId }
|
|
*/
|
|
export const generateAgentUploadUrl = httpAction(async (ctx, request) => {
|
|
try {
|
|
requireServiceToken(request);
|
|
const body = (await request.json()) as Record<string, unknown>;
|
|
await resolveBoundary(ctx, body);
|
|
const url = await ctx.storage.generateUploadUrl();
|
|
return json(200, { url });
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
const status = message.startsWith("Unauthorized") ? 401 : 400;
|
|
return json(status, { error: message });
|
|
}
|
|
});
|