wehub-resource-sync
144 lines
3.6 KiB
TypeScript
144 lines
3.6 KiB
TypeScript
import { json } from "@remix-run/server-runtime";
|
|
import { tryCatch } from "@trigger.dev/core/utils";
|
|
import { nanoid } from "nanoid";
|
|
import { z } from "zod";
|
|
import { $replica, prisma } from "~/db.server";
|
|
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
|
|
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
|
|
import { ServiceValidationError } from "~/v3/services/common.server";
|
|
import { runStore } from "~/v3/runStore.server";
|
|
|
|
const ParamsSchema = z.object({
|
|
runId: z.string(),
|
|
target: z.enum(["self", "parent", "root"]),
|
|
streamId: z.string(),
|
|
});
|
|
|
|
// S2 enforces a 1 MiB per-record limit (metered as
|
|
// `8 + 2*H + Σ(header name+value) + body`). Cap the raw HTTP body at
|
|
// 512 KiB so the JSON wrapper, string escaping, and any future per-record
|
|
// header additions all stay well under S2's ceiling.
|
|
// See https://s2.dev/docs/limits.
|
|
const MAX_APPEND_BODY_BYTES = 1024 * 512;
|
|
|
|
const { action } = createActionApiRoute(
|
|
{
|
|
params: ParamsSchema,
|
|
maxContentLength: MAX_APPEND_BODY_BYTES,
|
|
},
|
|
async ({ request, params, authentication }) => {
|
|
const run = await runStore.findRun(
|
|
{
|
|
friendlyId: params.runId,
|
|
runtimeEnvironmentId: authentication.environment.id,
|
|
},
|
|
{
|
|
select: {
|
|
id: true,
|
|
friendlyId: true,
|
|
parentTaskRun: {
|
|
select: {
|
|
friendlyId: true,
|
|
},
|
|
},
|
|
rootTaskRun: {
|
|
select: {
|
|
friendlyId: true,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
$replica
|
|
);
|
|
|
|
if (!run) {
|
|
return new Response("Run not found", { status: 404 });
|
|
}
|
|
|
|
const targetId =
|
|
params.target === "self"
|
|
? run.friendlyId
|
|
: params.target === "parent"
|
|
? run.parentTaskRun?.friendlyId
|
|
: run.rootTaskRun?.friendlyId;
|
|
|
|
if (!targetId) {
|
|
return new Response("Target not found", { status: 404 });
|
|
}
|
|
|
|
const targetRun = await runStore.findRun(
|
|
{
|
|
friendlyId: targetId,
|
|
runtimeEnvironmentId: authentication.environment.id,
|
|
},
|
|
{
|
|
select: {
|
|
realtimeStreams: true,
|
|
realtimeStreamsVersion: true,
|
|
completedAt: true,
|
|
id: true,
|
|
streamBasinName: true,
|
|
},
|
|
},
|
|
prisma
|
|
);
|
|
|
|
if (!targetRun) {
|
|
return new Response("Run not found", { status: 404 });
|
|
}
|
|
|
|
if (targetRun.completedAt) {
|
|
return new Response("Cannot append to a realtime stream on a completed run", {
|
|
status: 400,
|
|
});
|
|
}
|
|
|
|
if (!targetRun.realtimeStreams.includes(params.streamId)) {
|
|
await runStore.pushRealtimeStream(targetRun.id, params.streamId, prisma);
|
|
}
|
|
|
|
const part = await request.text();
|
|
|
|
const realtimeStream = getRealtimeStreamInstance(
|
|
authentication.environment,
|
|
targetRun.realtimeStreamsVersion,
|
|
{ run: targetRun }
|
|
);
|
|
|
|
const partId = request.headers.get("X-Part-Id") ?? nanoid(7);
|
|
|
|
const [appendError] = await tryCatch(
|
|
realtimeStream.appendPart(part, partId, targetId, params.streamId)
|
|
);
|
|
|
|
if (appendError) {
|
|
if (appendError instanceof ServiceValidationError) {
|
|
return json(
|
|
{
|
|
ok: false,
|
|
error: appendError.message,
|
|
},
|
|
{ status: appendError.status ?? 422 }
|
|
);
|
|
} else {
|
|
return json(
|
|
{
|
|
ok: false,
|
|
error: appendError.message,
|
|
},
|
|
{ status: 500 }
|
|
);
|
|
}
|
|
}
|
|
|
|
return json(
|
|
{
|
|
ok: true,
|
|
},
|
|
{ status: 200 }
|
|
);
|
|
}
|
|
);
|
|
|
|
export { action };
|