evOo-local/packages/db/src/plan-messages.ts

54 lines
1.9 KiB
TypeScript

import type { MessageBlock, PlanBlock } from "@rakazo/contracts";
import type { PlanStepUpdateInput } from "@rakazo/core";
import { updatePlanInBlocks } from "@rakazo/core";
import type { Prisma } from "./client.js";
import { appendEventInTransaction } from "./events.js";
/**
* Update a bot's execution plan in place inside the thread and emit `thread.message.updated`.
* When `planId` is omitted, updates the most recent plan block posted by `botId` in `threadId`.
*/
export async function updatePlanInTransaction(
tx: Prisma.TransactionClient,
input: {
spaceId: string;
threadId: string;
botId: string;
planId?: string;
updates: readonly PlanStepUpdateInput[];
},
): Promise<{ seq: number; plan: PlanBlock; messageId: string } | null> {
await tx.$queryRaw`SELECT id FROM threads WHERE id = ${input.threadId} FOR UPDATE`;
const candidates = await tx.message.findMany({
where: {
threadId: input.threadId,
role: "bot",
botId: input.botId,
blocks: {
array_contains: [input.planId ? { kind: "plan", planId: input.planId } : { kind: "plan" }],
},
},
orderBy: { seq: "desc" },
take: 5,
select: { id: true, blocks: true, botId: true },
});
for (const message of candidates) {
const blocks = (Array.isArray(message.blocks) ? message.blocks : []) as MessageBlock[];
const applied = updatePlanInBlocks(blocks, input.planId, input.updates);
if (!applied) continue;
await tx.message.update({
where: { id: message.id },
data: { blocks: applied.blocks as Prisma.InputJsonValue },
});
const event = await appendEventInTransaction(tx, {
spaceId: input.spaceId,
threadId: input.threadId,
botId: input.botId,
type: "thread.message.updated",
payload: { messageId: message.id, role: "bot", blocks: applied.blocks },
});
return { seq: event.seq, plan: applied.plan, messageId: message.id };
}
return null;
}