84f3e1b39c
## What Routes the webapp write path through the run-ops split seam: trigger/batch minting, idempotency-key resolution, and the run-lifecycle services now determine residency and dispatch writes to the correct store. - **Trigger & batch** (`runEngine/services/triggerTask.server.ts`, `batchTrigger.server.ts`, `createBatch.server.ts`, `streamBatchItems.server.ts`, `v3/services/batchTriggerV3.server.ts`): mint ids with the run-ops-aware minting and route creation/streaming through the store; batch children inherit the parent's residency. - **Idempotency** (`runEngine/concerns/idempotencyKeys.server.ts` + new `idempotencyResidency.server.ts`): idempotency-key lookup/dedup is residency-aware so a keyed retrigger resolves against the store that owns the original run. - **Run lifecycle services** (`createCheckpoint`, `createTaskRunAttempt`, `enqueueDelayedRun`, `expireEnqueuedRun`, `finalizeTaskRun`, `resumeBatchRun`, `cancelDevSessionRuns`, `executeTasksWaitingForDeploy`, `triggerFailedTask`): resolve their target run through the store rather than a fixed client. - **Reads that fan out from writes** (`runsRepository` + `clickhouseRunsRepository`, `BulkActionV2` + batch read-through, realtime `sessions`/`runReader`, alerts `deliverAlert`/`performTaskRunAlerts`): route through the read-through resolver. - `9535ae63d` — resolves the parent run through an injectable run store in `TriggerFailedTaskService`. - `bf8f7c881` — drops the "known-migrated" concept from write-path and read repos; residency is id-shape only. - `515b897ea` — self-defaults `resolveWaitpointThroughReadThrough` to the safe run-ops clients. ## Why PR6 of the run-ops split stack. This is the write-path counterpart to the read foundation in the previous PRs: with it in place, both reads and writes route through the seam. Additive when the split is disabled (id-shape resolution collapses to the control-plane client); behavior-changing on the minting, idempotency, and lifecycle paths when enabled. ## Tests Large new/expanded vitest suite under `apps/webapp/test/` and colocated service tests: trigger-task and batch-trigger store routing, residency inheritance, idempotency dedup residency + legacy-authority, bulk-action read routing, cancel-dev-session routing, alerts store routing, runs-repository read-through, realtime session/run-reader read-through and stream-registration routing, and the waitpoint read-through default. Testcontainers-backed; no mocks. ## Notes Draft, **stacked on #4117** (`runops/pr05-webapp-foundation`). Review that first; this diff is against it. Server-change / changeset note to be added at stack-assembly time. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
473 lines
13 KiB
TypeScript
473 lines
13 KiB
TypeScript
import type { AuthenticatedEnvironment } from "@internal/run-engine";
|
|
import type { Prisma, PrismaClientOrTransaction, RuntimeEnvironment } from "@trigger.dev/database";
|
|
import { $replica, prisma } from "~/db.server";
|
|
import { runStore } from "~/v3/runStore.server";
|
|
import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
|
|
import { logger } from "~/services/logger.server";
|
|
import { getUsername } from "~/utils/username";
|
|
import { isDefaultDevBranch, sanitizeBranchName } from "@trigger.dev/core/v3/utils/gitBranch";
|
|
|
|
export type { RuntimeEnvironment };
|
|
|
|
// Prisma include shape that maps cleanly to the slim AuthenticatedEnvironment.
|
|
// Use this everywhere we fetch an env that flows to handlers — keeps the
|
|
// returned shape consistent (and the Decimal coercion in toAuthenticated()
|
|
// strips Prisma's Decimal class from the public surface).
|
|
export const authIncludeBase = {
|
|
project: true,
|
|
organization: true,
|
|
orgMember: {
|
|
select: {
|
|
userId: true,
|
|
user: { select: { id: true, displayName: true, name: true } },
|
|
},
|
|
},
|
|
} satisfies Prisma.RuntimeEnvironmentInclude;
|
|
|
|
export const authIncludeWithParent = {
|
|
...authIncludeBase,
|
|
parentEnvironment: { select: { id: true, apiKey: true } },
|
|
} satisfies Prisma.RuntimeEnvironmentInclude;
|
|
|
|
type PrismaEnvWithAuth = Prisma.RuntimeEnvironmentGetPayload<{ include: typeof authIncludeBase }>;
|
|
type PrismaEnvWithAuthAndParent = Prisma.RuntimeEnvironmentGetPayload<{
|
|
include: typeof authIncludeWithParent;
|
|
}>;
|
|
|
|
// Coerce a Prisma RuntimeEnvironment payload to the slim
|
|
// AuthenticatedEnvironment shape. Drops the columns handlers don't read
|
|
// and converts `concurrencyLimitBurstFactor` from Prisma's Decimal to a
|
|
// plain number (lossless at this scale). The optional union accepts both
|
|
// query shapes — with parentEnvironment loaded, or without it.
|
|
export function toAuthenticated(
|
|
env: PrismaEnvWithAuth | PrismaEnvWithAuthAndParent
|
|
): AuthenticatedEnvironment {
|
|
return {
|
|
id: env.id,
|
|
slug: env.slug,
|
|
type: env.type,
|
|
apiKey: env.apiKey,
|
|
organizationId: env.organizationId,
|
|
projectId: env.projectId,
|
|
orgMemberId: env.orgMemberId,
|
|
parentEnvironmentId: env.parentEnvironmentId,
|
|
branchName: env.branchName,
|
|
archivedAt: env.archivedAt,
|
|
paused: env.paused,
|
|
shortcode: env.shortcode,
|
|
maximumConcurrencyLimit: env.maximumConcurrencyLimit,
|
|
// Coerce Prisma's Decimal to a plain number — the slim type accepts
|
|
// both, but downstream consumers shouldn't have to narrow before
|
|
// doing arithmetic. Lossless at this scale (Decimal(4,2)).
|
|
concurrencyLimitBurstFactor: env.concurrencyLimitBurstFactor.toNumber(),
|
|
builtInEnvironmentVariableOverrides: env.builtInEnvironmentVariableOverrides,
|
|
createdAt: env.createdAt,
|
|
updatedAt: env.updatedAt,
|
|
project: {
|
|
id: env.project.id,
|
|
slug: env.project.slug,
|
|
name: env.project.name,
|
|
externalRef: env.project.externalRef,
|
|
engine: env.project.engine,
|
|
deletedAt: env.project.deletedAt,
|
|
defaultWorkerGroupId: env.project.defaultWorkerGroupId,
|
|
organizationId: env.project.organizationId,
|
|
builderProjectId: env.project.builderProjectId,
|
|
},
|
|
organization: {
|
|
id: env.organization.id,
|
|
slug: env.organization.slug,
|
|
title: env.organization.title,
|
|
streamBasinName: env.organization.streamBasinName,
|
|
maximumConcurrencyLimit: env.organization.maximumConcurrencyLimit,
|
|
runsEnabled: env.organization.runsEnabled,
|
|
maximumDevQueueSize: env.organization.maximumDevQueueSize,
|
|
maximumDeployedQueueSize: env.organization.maximumDeployedQueueSize,
|
|
featureFlags: env.organization.featureFlags,
|
|
apiRateLimiterConfig: env.organization.apiRateLimiterConfig,
|
|
batchRateLimitConfig: env.organization.batchRateLimitConfig,
|
|
batchQueueConcurrencyConfig: env.organization.batchQueueConcurrencyConfig,
|
|
},
|
|
orgMember: env.orgMember,
|
|
parentEnvironment: "parentEnvironment" in env ? env.parentEnvironment : null,
|
|
};
|
|
}
|
|
|
|
export async function findEnvironmentByApiKey(
|
|
apiKey: string,
|
|
branchName: string | undefined,
|
|
tx: PrismaClientOrTransaction = $replica
|
|
): Promise<AuthenticatedEnvironment | null> {
|
|
const branch = sanitizeBranchName(branchName) ?? undefined;
|
|
|
|
const include = {
|
|
...authIncludeBase,
|
|
childEnvironments: branch
|
|
? {
|
|
where: {
|
|
branchName: branch,
|
|
archivedAt: null,
|
|
},
|
|
}
|
|
: undefined,
|
|
} satisfies Prisma.RuntimeEnvironmentInclude;
|
|
|
|
let environment = await tx.runtimeEnvironment.findFirst({
|
|
where: {
|
|
apiKey,
|
|
},
|
|
include,
|
|
});
|
|
|
|
// Fall back to keys that were revoked within the grace window
|
|
if (!environment) {
|
|
const revokedApiKey = await tx.revokedApiKey.findFirst({
|
|
where: {
|
|
apiKey,
|
|
expiresAt: { gt: new Date() },
|
|
},
|
|
include: {
|
|
runtimeEnvironment: { include },
|
|
},
|
|
});
|
|
|
|
environment = revokedApiKey?.runtimeEnvironment ?? null;
|
|
}
|
|
|
|
if (!environment) {
|
|
return null;
|
|
}
|
|
|
|
//don't return deleted projects
|
|
if (environment.project.deletedAt !== null) {
|
|
return null;
|
|
}
|
|
|
|
if (environment.type === "PREVIEW") {
|
|
if (!branch) {
|
|
logger.warn("findEnvironmentByApiKey(): Preview env with no branch name provided", {
|
|
environmentId: environment.id,
|
|
});
|
|
return null;
|
|
}
|
|
|
|
const childEnvironment = environment.childEnvironments.at(0);
|
|
|
|
if (childEnvironment) {
|
|
return toAuthenticated({
|
|
...childEnvironment,
|
|
apiKey: environment.apiKey,
|
|
orgMember: environment.orgMember,
|
|
organization: environment.organization,
|
|
project: environment.project,
|
|
});
|
|
}
|
|
|
|
//A branch was specified but no child environment was found
|
|
return null;
|
|
}
|
|
|
|
// If there is a named DEV branch (other than default), return it
|
|
if (environment.type === "DEVELOPMENT" && branch !== undefined && !isDefaultDevBranch(branch)) {
|
|
const childEnvironment = environment.childEnvironments.at(0);
|
|
|
|
if (childEnvironment) {
|
|
return toAuthenticated({
|
|
...childEnvironment,
|
|
apiKey: environment.apiKey,
|
|
orgMember: environment.orgMember,
|
|
organization: environment.organization,
|
|
project: environment.project,
|
|
});
|
|
}
|
|
|
|
//A branch was specified but no child environment was found
|
|
return null;
|
|
}
|
|
|
|
return toAuthenticated(environment);
|
|
}
|
|
|
|
/**
|
|
* @deprecated We don't use public API keys (`pk_*` tokens) anymore — public
|
|
* access goes through public JWTs (see `isPublicJWT` / `validatePublicJwtKey`).
|
|
*
|
|
* Still exported because a handful of pre-RBAC routes that haven't been
|
|
* migrated to the apiBuilder still wire this lookup into their
|
|
* `authenticateApiKey` / `authenticateApiKeyWithFailure` flow. The new RBAC
|
|
* fallback (`internal-packages/rbac/src/fallback.ts`) intentionally does NOT
|
|
* call this — any pk_*-authenticated request that hits an apiBuilder route
|
|
* returns 401. That's a deliberate cutover, not an oversight.
|
|
*/
|
|
export async function findEnvironmentByPublicApiKey(
|
|
apiKey: string,
|
|
branchName: string | undefined
|
|
): Promise<AuthenticatedEnvironment | null> {
|
|
const environment = await $replica.runtimeEnvironment.findFirst({
|
|
where: {
|
|
pkApiKey: apiKey,
|
|
},
|
|
include: authIncludeBase,
|
|
});
|
|
|
|
if (!environment || environment.project.deletedAt !== null) {
|
|
return null;
|
|
}
|
|
|
|
return toAuthenticated(environment);
|
|
}
|
|
|
|
export async function findEnvironmentById(id: string): Promise<AuthenticatedEnvironment | null> {
|
|
const environment = await $replica.runtimeEnvironment.findFirst({
|
|
where: {
|
|
id,
|
|
},
|
|
include: authIncludeWithParent,
|
|
});
|
|
|
|
if (!environment || environment.project.deletedAt !== null) {
|
|
return null;
|
|
}
|
|
|
|
return toAuthenticated(environment);
|
|
}
|
|
|
|
export async function findEnvironmentBySlug(
|
|
projectId: string,
|
|
envSlug: string,
|
|
userId: string
|
|
): Promise<AuthenticatedEnvironment | null> {
|
|
const environment = await $replica.runtimeEnvironment.findFirst({
|
|
where: {
|
|
projectId: projectId,
|
|
slug: envSlug,
|
|
OR: [
|
|
{
|
|
type: {
|
|
in: ["PREVIEW", "STAGING", "PRODUCTION"],
|
|
},
|
|
},
|
|
{
|
|
type: "DEVELOPMENT",
|
|
orgMember: {
|
|
userId,
|
|
},
|
|
},
|
|
],
|
|
},
|
|
include: authIncludeBase,
|
|
});
|
|
return environment ? toAuthenticated(environment) : null;
|
|
}
|
|
|
|
// The authenticated environment plus the run scalars the realtime publish needs.
|
|
// Both come from one taskRun read — see findEnvironmentFromRun.
|
|
export type EnvironmentFromRun = {
|
|
environment: AuthenticatedEnvironment;
|
|
runTags: string[];
|
|
batchId: string | null;
|
|
};
|
|
|
|
export async function findEnvironmentFromRun(
|
|
runId: string,
|
|
tx?: PrismaClientOrTransaction
|
|
): Promise<EnvironmentFromRun | null> {
|
|
// Run-ops scalars (runTags/batchId/runtimeEnvironmentId) from the run store; the env half is
|
|
// resolved via the control-plane resolver so the run-ops DB can split without a cross-DB join.
|
|
const taskRun = await runStore.findRun(
|
|
{
|
|
id: runId,
|
|
},
|
|
{
|
|
select: {
|
|
runTags: true,
|
|
batchId: true,
|
|
runtimeEnvironmentId: true,
|
|
},
|
|
},
|
|
tx ?? $replica
|
|
);
|
|
if (!taskRun) {
|
|
return null;
|
|
}
|
|
const environment = await controlPlaneResolver.resolveAuthenticatedEnv(
|
|
taskRun.runtimeEnvironmentId
|
|
);
|
|
if (!environment) {
|
|
return null;
|
|
}
|
|
return {
|
|
environment,
|
|
runTags: taskRun.runTags,
|
|
batchId: taskRun.batchId,
|
|
};
|
|
}
|
|
|
|
export async function createNewSession(
|
|
environment: Pick<RuntimeEnvironment, "id">,
|
|
ipAddress: string
|
|
) {
|
|
const session = await prisma.runtimeEnvironmentSession.create({
|
|
data: {
|
|
environmentId: environment.id,
|
|
ipAddress,
|
|
},
|
|
});
|
|
|
|
await prisma.runtimeEnvironment.update({
|
|
where: {
|
|
id: environment.id,
|
|
},
|
|
data: {
|
|
currentSessionId: session.id,
|
|
},
|
|
});
|
|
|
|
return session;
|
|
}
|
|
|
|
export async function disconnectSession(environmentId: string) {
|
|
const environment = await prisma.runtimeEnvironment.findFirst({
|
|
where: {
|
|
id: environmentId,
|
|
},
|
|
});
|
|
|
|
if (!environment || !environment.currentSessionId) {
|
|
return null;
|
|
}
|
|
|
|
const session = await prisma.runtimeEnvironmentSession.update({
|
|
where: {
|
|
id: environment.currentSessionId,
|
|
},
|
|
data: {
|
|
disconnectedAt: new Date(),
|
|
},
|
|
});
|
|
|
|
await prisma.runtimeEnvironment.update({
|
|
where: {
|
|
id: environment.id,
|
|
},
|
|
data: {
|
|
currentSessionId: null,
|
|
},
|
|
});
|
|
|
|
return session;
|
|
}
|
|
|
|
export async function findLatestSession(
|
|
environmentId: string,
|
|
client: PrismaClientOrTransaction = $replica
|
|
) {
|
|
const session = await client.runtimeEnvironmentSession.findFirst({
|
|
where: {
|
|
environmentId,
|
|
},
|
|
orderBy: {
|
|
createdAt: "desc",
|
|
},
|
|
});
|
|
|
|
return session;
|
|
}
|
|
|
|
export type DisplayableInputEnvironment = Prisma.RuntimeEnvironmentGetPayload<{
|
|
select: {
|
|
id: true;
|
|
type: true;
|
|
slug: true;
|
|
orgMember: {
|
|
select: {
|
|
user: {
|
|
select: {
|
|
id: true;
|
|
name: true;
|
|
displayName: true;
|
|
};
|
|
};
|
|
};
|
|
};
|
|
};
|
|
}>;
|
|
|
|
export function displayableEnvironment(
|
|
environment: DisplayableInputEnvironment,
|
|
userId: string | undefined
|
|
) {
|
|
let userName: string | undefined = undefined;
|
|
|
|
if (environment.type === "DEVELOPMENT") {
|
|
if (!environment.orgMember) {
|
|
userName = "Deleted";
|
|
} else if (environment.orgMember.user.id !== userId) {
|
|
userName = getUsername(environment.orgMember.user);
|
|
}
|
|
}
|
|
|
|
return {
|
|
id: environment.id,
|
|
type: environment.type,
|
|
slug: environment.slug,
|
|
userName,
|
|
};
|
|
}
|
|
|
|
export async function findDisplayableEnvironment(
|
|
environmentId: string,
|
|
userId: string | undefined
|
|
) {
|
|
const environment = await $replica.runtimeEnvironment.findFirst({
|
|
where: {
|
|
id: environmentId,
|
|
},
|
|
select: {
|
|
id: true,
|
|
type: true,
|
|
slug: true,
|
|
orgMember: {
|
|
select: {
|
|
user: {
|
|
select: {
|
|
id: true,
|
|
name: true,
|
|
displayName: true,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
});
|
|
|
|
if (!environment) {
|
|
return;
|
|
}
|
|
|
|
return displayableEnvironment(environment, userId);
|
|
}
|
|
|
|
export async function hasAccessToEnvironment({
|
|
environmentId,
|
|
projectId,
|
|
organizationId,
|
|
userId,
|
|
}: {
|
|
environmentId: string;
|
|
projectId: string;
|
|
organizationId: string;
|
|
userId: string;
|
|
}): Promise<boolean> {
|
|
const environment = await $replica.runtimeEnvironment.findFirst({
|
|
where: {
|
|
id: environmentId,
|
|
projectId: projectId,
|
|
organizationId: organizationId,
|
|
organization: { members: { some: { userId } } },
|
|
},
|
|
});
|
|
|
|
return environment !== null;
|
|
}
|