Files
triggerdotdev--trigger.dev/apps/webapp/app/models/runtimeEnvironment.server.ts
Eric Allam f9d57d3bd5 feat(webapp): add a new backend for the realtime runs feed (#3864)
## 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.
2026-06-11 07:56:10 +01:00

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;
}