f9d57d3bd5
## Summary Adds a second backend for the realtime runs feed (`useRealtimeRun`, `subscribeToRunsWithTag`, `subscribeToBatch`), built to stay healthy when a single busy environment has many subscribers watching many runs at once. It is gated behind a feature flag with the existing backend as the default, so nothing changes for users until it is enabled per environment. ## Design A run change is published once, as a small self-describing record, to a single per-environment channel. Every feed is then a predicate over that one stream rather than owning a channel: - A per-instance router indexes the currently-held feeds by run, tag, and batch. When a run changes it hydrates the affected rows once and serializes them once, then fans the result to every matching feed. One hot shared tag watched by many subscribers costs a single database query and serialize, not one per subscriber. - Feeds that don't match a change are never woken, wake delivery per environment is coalesced on a leading edge (250ms default) so a burst of changes costs one wake, and cold reads coalesce onto a single short-TTL-cached resolve. - An admission gate bounds how many cold ClickHouse resolves run concurrently, so a mass reconnect across many distinct filters queues instead of stampeding the database. - Changes that land while a client is between long-polls are delivered on its next poll instead of waiting for the periodic backstop: each environment buffers its recent change records, subscriptions linger briefly after the last feed closes, and a newly-armed poll replays exactly the connection's gap. - The per-connection replay cursors behind that are shared across instances via Redis (a single timestamp each), so a poll landing on a different instance behind the load balancer still reads the connection's true gap instead of falling back to a cold resolve. Cursor reads have a bounded deadline and degrade to the cold-read path on any Redis trouble. - Tag subscriptions with multiple tags match runs carrying all of the tags, mirroring the existing backend's filter semantics, and live long-polls hold for about 20 seconds to match its cadence. - The per-environment channel supports Redis Cluster sharded pub/sub, so the wake path scales horizontally across shards by environment. - The backend reports its health through OpenTelemetry metrics (delivery lag, poll resolution paths, backstop outcomes, replay and cursor-store activity), with a provisioned Grafana dashboard for local development. Everything is behind the feature flag and tunable via env vars; the existing backend remains the default.
436 lines
12 KiB
TypeScript
436 lines
12 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 { logger } from "~/services/logger.server";
|
|
import { getUsername } from "~/utils/username";
|
|
import { 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
|
|
): Promise<AuthenticatedEnvironment | null> {
|
|
const include = {
|
|
...authIncludeBase,
|
|
childEnvironments: branchName
|
|
? {
|
|
where: {
|
|
branchName: sanitizeBranchName(branchName),
|
|
archivedAt: null,
|
|
},
|
|
}
|
|
: undefined,
|
|
} satisfies Prisma.RuntimeEnvironmentInclude;
|
|
|
|
let environment = await $replica.runtimeEnvironment.findFirst({
|
|
where: {
|
|
apiKey,
|
|
},
|
|
include,
|
|
});
|
|
|
|
// Fall back to keys that were revoked within the grace window
|
|
if (!environment) {
|
|
const revokedApiKey = await $replica.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 (!branchName) {
|
|
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;
|
|
}
|
|
|
|
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> {
|
|
// The include (no select) already pulls every taskRun scalar, so runTags/batchId
|
|
// ride along for free — no extra query for the realtime publish to send a full record.
|
|
const taskRun = await (tx ?? $replica).taskRun.findFirst({
|
|
where: {
|
|
id: runId,
|
|
},
|
|
include: {
|
|
runtimeEnvironment: { include: authIncludeBase },
|
|
},
|
|
});
|
|
if (!taskRun?.runtimeEnvironment) {
|
|
return null;
|
|
}
|
|
return {
|
|
environment: toAuthenticated(taskRun.runtimeEnvironment),
|
|
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) {
|
|
const session = await $replica.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;
|
|
}
|