github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Realtime and task run performance improvements (#2158)

Browse Source 
* Add createdAt filter to realtime subscribing with tags

* Filter realtime colums and expose ability to skip some columns

* Add sharding support for electric

* Use unkey cache for the created at filter caching

* Remove 2 unused indexes on TaskRun

* Run list now filters by a single runtime environment

* Remove project ID indexes

* Use clickhouse in task list aggregation queries instead of pg (keep pg for self-hosters)

* WIP clickhouse powered runs list
stuff

* Improve the query to get the latest tasks for the task list presenter

* Update the usage task list to use clickhouse

* Implement next runs list powered by clickhouse

* Add new index for TaskRun for the runs list, by environment ID

* Add runTags gin index

* Handle possibly malicious inputs

* Ignore claude settings

* Better handling not finding an environment on the schedule page

* Use ms since epoch in test, not seconds

* Remove unused function

* Fix test

* Use an env var for the realtime maximum createdAt filter duration (defaults to 1 day)

* Fixed the query builder to correct the group by / order by order

* Make sure runs.list still works

* Create small-birds-arrive.md
This commit is contained in:
Eric Allam
2025-06-10 12:11:01 +01:00
committed by GitHub
parent e617c14456
commit b38405cb88
62 changed files with 4258 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/small-birds-arrive.md
+16
View File
@@ -0,0 +1,16 @@
---
"@trigger.dev/react-hooks": patch
---
Added the ability to specify a "createdAt" filter when subscribing to tags in our useRealtime hooks:
```tsx
// Only subscribe to runs created in the last 10 hours
useRealtimeRunWithTags("my-tag", { createdAt: "10h" })
```
You can also now choose to skip subscribing to specific columns by specifying the `skipColumns` option:
```tsx
useRealtimeRun(run.id, { skipColumns: ["usageDurationMs"] });
```
.gitignore
+1
View File
@@ -63,3 +63,4 @@ apps/**/public/build
/packages/core/src/package.json
/packages/trigger-sdk/src/package.json
/packages/python/src/package.json
.claude
apps/webapp/app/components/runs/v3/TaskRunsTable.tsx
+1 -1
View File
@@ -539,7 +539,7 @@ function BlankState({ isLoading, filters }: Pick<RunsTableProps, "isLoading" | "
const environment = useEnvironment();
if (isLoading) return <TableBlankRow colSpan={15}></TableBlankRow>;
const { environments, tasks, from, to, ...otherFilters } = filters;
const { tasks, from, to, ...otherFilters } = filters;
if (
filters.tasks.length === 1 &&
apps/webapp/app/env.server.ts
+16
View File
@@ -35,6 +35,9 @@ const EnvironmentSchema = z.object({
API_ORIGIN: z.string().optional(),
STREAM_ORIGIN: z.string().optional(),
ELECTRIC_ORIGIN: z.string().default("http://localhost:3060"),
// A comma separated list of electric origins to shard into different electric instances by environmentId
// example: "http://localhost:3060,http://localhost:3061,http://localhost:3062"
ELECTRIC_ORIGIN_SHARDS: z.string().optional(),
APP_ENV: z.string().default(process.env.NODE_ENV),
SERVICE_NAME: z.string().default("trigger.dev webapp"),
POSTHOG_PROJECT_KEY: z.string().default("phc_LFH7kJiGhdIlnO22hTAKgHpaKhpM8gkzWAFvHmf5vfS"),
@@ -161,6 +164,11 @@ const EnvironmentSchema = z.object({
.default(process.env.REDIS_TLS_DISABLED ?? "false"),
REALTIME_STREAMS_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS: z.coerce
.number()
.int()
.default(24 * 60 * 60 * 1000), // 1 day in milliseconds
PUBSUB_REDIS_HOST: z
.string()
.optional()
@@ -738,6 +746,14 @@ const EnvironmentSchema = z.object({
RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
RUN_REPLICATION_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
// Clickhouse
CLICKHOUSE_URL: z.string().optional(),
CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
CLICKHOUSE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
// Bootstrap
TRIGGER_BOOTSTRAP_ENABLED: z.string().default("0"),
TRIGGER_BOOTSTRAP_WORKER_GROUP_NAME: z.string().optional(),
apps/webapp/app/models/runtimeEnvironment.server.ts
+33
View File
@@ -275,3 +275,36 @@ export function displayableEnvironment(
userName,
};
}
export async function findDisplayableEnvironment(
environmentId: string,
userId: string | undefined
) {
const environment = await prisma.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);
}
apps/webapp/app/models/task.server.ts
+2 -2
View File
@@ -7,7 +7,7 @@ import { PrismaClientOrTransaction, sqlDatabaseSchema } from "~/db.server";
* It has indexes for fast performance.
* It does NOT care about versions, so includes all tasks ever created.
*/
export function getAllTaskIdentifiers(prisma: PrismaClientOrTransaction, projectId: string) {
export function getAllTaskIdentifiers(prisma: PrismaClientOrTransaction, environmentId: string) {
return prisma.$queryRaw<
{
slug: string;
@@ -16,6 +16,6 @@ export function getAllTaskIdentifiers(prisma: PrismaClientOrTransaction, project
>`
SELECT DISTINCT(slug), "triggerSource"
FROM ${sqlDatabaseSchema}."BackgroundWorkerTask"
WHERE "projectId" = ${projectId}
WHERE "runtimeEnvironmentId" = ${environmentId}
ORDER BY slug ASC;`;
}
apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts
+11 -4
View File
@@ -12,6 +12,7 @@ import { CoercedDate } from "~/utils/zod";
import { ApiRetrieveRunPresenter } from "./ApiRetrieveRunPresenter.server";
import { type RunListOptions, RunListPresenter } from "./RunListPresenter.server";
import { BasePresenter } from "./basePresenter.server";
import { ServiceValidationError } from "~/v3/services/baseService.server";
export const ApiRunListSearchParams = z.object({
"page[size]": z.coerce.number().int().positive().min(1).max(100).optional(),
@@ -134,9 +135,11 @@ export class ApiRunListPresenter extends BasePresenter {
options.direction = "backward";
}
let environmentId: string | undefined;
// filters
if (environment) {
options.environments = [environment.id];
environmentId = environment.id;
} else {
if (searchParams["filter[env]"]) {
const environments = await this._prisma.runtimeEnvironment.findMany({
@@ -148,10 +151,14 @@ export class ApiRunListPresenter extends BasePresenter {
},
});
options.environments = environments.map((env) => env.id);
environmentId = environments.at(0)?.id;
}
}
if (!environmentId) {
throw new ServiceValidationError("No environment found");
}
if (searchParams["filter[status]"]) {
options.statuses = searchParams["filter[status]"].flatMap((status) =>
ApiRunListPresenter.apiStatusToRunStatuses(status)
@@ -202,9 +209,9 @@ export class ApiRunListPresenter extends BasePresenter {
logger.debug("Calling RunListPresenter", { options });
const results = await presenter.call(options);
const results = await presenter.call(environmentId, options);
logger.debug("RunListPresenter results", { results });
logger.debug("RunListPresenter results", { runs: results.runs.length });
const data: ListRunResponseItem[] = await Promise.all(
results.runs.map(async (run) => {
apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts
+294
View File
@@ -0,0 +1,294 @@
import { ClickHouse } from "@internal/clickhouse";
import { PrismaClient, PrismaClientOrTransaction, type TaskRunStatus } from "@trigger.dev/database";
import { type Direction } from "~/components/ListPagination";
import { timeFilters } from "~/components/runs/v3/SharedFilters";
import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
import { getAllTaskIdentifiers } from "~/models/task.server";
import { RunsRepository } from "~/services/runsRepository.server";
import { ServiceValidationError } from "~/v3/services/baseService.server";
import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus";
import parseDuration from "parse-duration";
export type RunListOptions = {
userId?: string;
projectId: string;
//filters
tasks?: string[];
versions?: string[];
statuses?: TaskRunStatus[];
tags?: string[];
scheduleId?: string;
period?: string;
bulkId?: string;
from?: number;
to?: number;
isTest?: boolean;
rootOnly?: boolean;
batchId?: string;
runIds?: string[];
//pagination
direction?: Direction;
cursor?: string;
pageSize?: number;
};
const DEFAULT_PAGE_SIZE = 25;
export type NextRunList = Awaited<ReturnType<NextRunListPresenter["call"]>>;
export type NextRunListItem = NextRunList["runs"][0];
export type NextRunListAppliedFilters = NextRunList["filters"];
export class NextRunListPresenter {
constructor(
private readonly replica: PrismaClientOrTransaction,
private readonly clickhouse: ClickHouse
) {}
public async call(
environmentId: string,
{
userId,
projectId,
tasks,
versions,
statuses,
tags,
scheduleId,
period,
bulkId,
isTest,
rootOnly,
batchId,
runIds,
from,
to,
direction = "forward",
cursor,
pageSize = DEFAULT_PAGE_SIZE,
}: RunListOptions
) {
//get the time values from the raw values (including a default period)
const time = timeFilters({
period,
from,
to,
});
const periodMs = time.period ? parseDuration(time.period) : undefined;
const hasStatusFilters = statuses && statuses.length > 0;
const hasFilters =
(tasks !== undefined && tasks.length > 0) ||
(versions !== undefined && versions.length > 0) ||
hasStatusFilters ||
(bulkId !== undefined && bulkId !== "") ||
(scheduleId !== undefined && scheduleId !== "") ||
(tags !== undefined && tags.length > 0) ||
batchId !== undefined ||
(runIds !== undefined && runIds.length > 0) ||
typeof isTest === "boolean" ||
rootOnly === true ||
!time.isDefault;
//get all possible tasks
const possibleTasksAsync = getAllTaskIdentifiers(this.replica, environmentId);
//get possible bulk actions
// TODO: we should replace this with the new bulk stuff and make it environment scoped
const bulkActionsAsync = this.replica.bulkActionGroup.findMany({
select: {
friendlyId: true,
type: true,
createdAt: true,
},
where: {
projectId: projectId,
},
orderBy: {
createdAt: "desc",
},
take: 20,
});
const [possibleTasks, bulkActions, displayableEnvironment] = await Promise.all([
possibleTasksAsync,
bulkActionsAsync,
findDisplayableEnvironment(environmentId, userId),
]);
if (!displayableEnvironment) {
throw new ServiceValidationError("No environment found");
}
//we can restrict to specific runs using bulkId, or batchId
let restrictToRunIds: undefined | string[] = undefined;
//bulk id
if (bulkId) {
const bulkAction = await this.replica.bulkActionGroup.findFirst({
select: {
items: {
select: {
destinationRunId: true,
},
},
},
where: {
friendlyId: bulkId,
},
});
if (bulkAction) {
const runIds = bulkAction.items.map((item) => item.destinationRunId).filter(Boolean);
restrictToRunIds = runIds;
}
}
//batch id is a friendly id
if (batchId) {
const batch = await this.replica.batchTaskRun.findFirst({
select: {
id: true,
},
where: {
friendlyId: batchId,
runtimeEnvironmentId: environmentId,
},
});
if (batch) {
batchId = batch.id;
}
}
//scheduleId can be a friendlyId
if (scheduleId && scheduleId.startsWith("sched_")) {
const schedule = await this.replica.taskSchedule.findFirst({
select: {
id: true,
},
where: {
friendlyId: scheduleId,
projectId: projectId,
},
});
if (schedule) {
scheduleId = schedule?.id;
}
}
//show all runs if we are filtering by batchId or runId
if (batchId || runIds?.length || scheduleId || tasks?.length) {
rootOnly = false;
}
const runsRepository = new RunsRepository({
clickhouse: this.clickhouse,
prisma: this.replica as PrismaClient,
});
const { runs, pagination } = await runsRepository.listRuns({
environmentId,
projectId,
tasks,
versions,
statuses,
tags,
scheduleId,
period: periodMs ?? undefined,
from,
to,
isTest,
rootOnly,
batchId,
runFriendlyIds: runIds,
runIds: restrictToRunIds,
page: {
size: pageSize,
cursor,
direction,
},
});
let hasAnyRuns = runs.length > 0;
if (!hasAnyRuns) {
const firstRun = await this.replica.taskRun.findFirst({
where: {
runtimeEnvironmentId: environmentId,
},
});
if (firstRun) {
hasAnyRuns = true;
}
}
return {
runs: runs.map((run) => {
const hasFinished = isFinalRunStatus(run.status);
const startedAt = run.startedAt ?? run.lockedAt;
return {
id: run.id,
number: 1,
friendlyId: run.friendlyId,
createdAt: run.createdAt.toISOString(),
updatedAt: run.updatedAt.toISOString(),
startedAt: startedAt ? startedAt.toISOString() : undefined,
delayUntil: run.delayUntil ? run.delayUntil.toISOString() : undefined,
hasFinished,
finishedAt: hasFinished
? run.completedAt?.toISOString() ?? run.updatedAt.toISOString()
: undefined,
isTest: run.isTest,
status: run.status,
version: run.taskVersion,
taskIdentifier: run.taskIdentifier,
spanId: run.spanId,
isReplayable: true,
isCancellable: isCancellableRunStatus(run.status),
isPending: isPendingRunStatus(run.status),
environment: displayableEnvironment,
idempotencyKey: run.idempotencyKey ? run.idempotencyKey : undefined,
ttl: run.ttl ? run.ttl : undefined,
expiredAt: run.expiredAt ? run.expiredAt.toISOString() : undefined,
costInCents: run.costInCents,
baseCostInCents: run.baseCostInCents,
usageDurationMs: Number(run.usageDurationMs),
tags: run.runTags ? run.runTags.sort((a, b) => a.localeCompare(b)) : [],
depth: run.depth,
rootTaskRunId: run.rootTaskRunId,
metadata: run.metadata,
metadataType: run.metadataType,
};
}),
pagination: {
next: pagination.nextCursor ?? undefined,
previous: pagination.previousCursor ?? undefined,
},
possibleTasks: possibleTasks
.map((task) => ({ slug: task.slug, triggerSource: task.triggerSource }))
.sort((a, b) => {
return a.slug.localeCompare(b.slug);
}),
bulkActions: bulkActions.map((bulkAction) => ({
id: bulkAction.friendlyId,
type: bulkAction.type,
createdAt: bulkAction.createdAt,
})),
filters: {
tasks: tasks || [],
versions: versions || [],
statuses: statuses || [],
from: time.from,
to: time.to,
},
hasFilters,
hasAnyRuns,
};
}
}
apps/webapp/app/presenters/v3/RunListPresenter.server.ts
+47 -85
View File
@@ -1,12 +1,13 @@
import { Prisma, type TaskRunStatus } from "@trigger.dev/database";
import parse from "parse-duration";
import { sqlDatabaseSchema } from "~/db.server";
import { displayableEnvironment } from "~/models/runtimeEnvironment.server";
import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus";
import { BasePresenter } from "./basePresenter.server";
import { getAllTaskIdentifiers } from "~/models/task.server";
import { type Direction } from "~/components/ListPagination";
import { timeFilters } from "~/components/runs/v3/SharedFilters";
import { sqlDatabaseSchema } from "~/db.server";
import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
import { getAllTaskIdentifiers } from "~/models/task.server";
import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus";
import { BasePresenter } from "./basePresenter.server";
import { ServiceValidationError } from "~/v3/services/baseService.server";
export type RunListOptions = {
userId?: string;
@@ -15,7 +16,6 @@ export type RunListOptions = {
tasks?: string[];
versions?: string[];
statuses?: TaskRunStatus[];
environments?: string[];
tags?: string[];
scheduleId?: string;
period?: string;
@@ -39,27 +39,29 @@ export type RunListItem = RunList["runs"][0];
export type RunListAppliedFilters = RunList["filters"];
export class RunListPresenter extends BasePresenter {
public async call({
userId,
projectId,
tasks,
versions,
statuses,
environments,
tags,
scheduleId,
period,
bulkId,
isTest,
rootOnly,
batchId,
runIds,
from,
to,
direction = "forward",
cursor,
pageSize = DEFAULT_PAGE_SIZE,
}: RunListOptions) {
public async call(
environmentId: string,
{
userId,
projectId,
tasks,
versions,
statuses,
tags,
scheduleId,
period,
bulkId,
isTest,
rootOnly,
batchId,
runIds,
from,
to,
direction = "forward",
cursor,
pageSize = DEFAULT_PAGE_SIZE,
}: RunListOptions
) {
//get the time values from the raw values (including a default period)
const time = timeFilters({
period,
@@ -82,38 +84,11 @@ export class RunListPresenter extends BasePresenter {
rootOnly === true ||
!time.isDefault;
// Find the project scoped to the organization
const project = await this._replica.project.findFirstOrThrow({
select: {
id: true,
environments: {
select: {
id: true,
type: true,
slug: true,
orgMember: {
select: {
user: {
select: {
id: true,
name: true,
displayName: true,
},
},
},
},
},
},
},
where: {
id: projectId,
},
});
//get all possible tasks
const possibleTasksAsync = getAllTaskIdentifiers(this._replica, project.id);
const possibleTasksAsync = getAllTaskIdentifiers(this._replica, environmentId);
//get possible bulk actions
// TODO: we should replace this with the new bulk stuff and make it environment scoped
const bulkActionsAsync = this._replica.bulkActionGroup.findMany({
select: {
friendlyId: true,
@@ -121,7 +96,7 @@ export class RunListPresenter extends BasePresenter {
createdAt: true,
},
where: {
projectId: project.id,
projectId: projectId,
},
orderBy: {
createdAt: "desc",
@@ -129,7 +104,15 @@ export class RunListPresenter extends BasePresenter {
take: 20,
});
const [possibleTasks, bulkActions] = await Promise.all([possibleTasksAsync, bulkActionsAsync]);
const [possibleTasks, bulkActions, displayableEnvironment] = await Promise.all([
possibleTasksAsync,
bulkActionsAsync,
findDisplayableEnvironment(environmentId, userId),
]);
if (!displayableEnvironment) {
throw new ServiceValidationError("No environment found");
}
//we can restrict to specific runs using bulkId, or batchId
let restrictToRunIds: undefined | string[] = undefined;
@@ -163,6 +146,7 @@ export class RunListPresenter extends BasePresenter {
},
where: {
friendlyId: batchId,
runtimeEnvironmentId: environmentId,
},
});
@@ -179,6 +163,7 @@ export class RunListPresenter extends BasePresenter {
},
where: {
friendlyId: scheduleId,
projectId: projectId,
},
});
@@ -202,7 +187,6 @@ export class RunListPresenter extends BasePresenter {
runFriendlyId: string;
taskIdentifier: string;
version: string | null;
runtimeEnvironmentId: string;
status: TaskRunStatus;
createdAt: Date;
startedAt: Date | null;
@@ -231,8 +215,7 @@ export class RunListPresenter extends BasePresenter {
tr.number,
tr."friendlyId" AS "runFriendlyId",
tr."taskIdentifier" AS "taskIdentifier",
bw.version AS version,
tr."runtimeEnvironmentId" AS "runtimeEnvironmentId",
tr."taskVersion" AS version,
tr.status AS status,
tr."createdAt" AS "createdAt",
tr."startedAt" AS "startedAt",
@@ -255,11 +238,9 @@ export class RunListPresenter extends BasePresenter {
tr."metadataType" AS "metadataType"
FROM
${sqlDatabaseSchema}."TaskRun" tr
LEFT JOIN
${sqlDatabaseSchema}."BackgroundWorker" bw ON tr."lockedToVersionId" = bw.id
WHERE
-- project
tr."projectId" = ${project.id}
tr."runtimeEnvironmentId" = ${environmentId}
-- cursor
${
cursor
@@ -288,11 +269,6 @@ WHERE
? Prisma.sql`AND tr.status = ANY(ARRAY[${Prisma.join(statuses)}]::"TaskRunStatus"[])`
: Prisma.empty
}
${
environments && environments.length > 0
? Prisma.sql`AND tr."runtimeEnvironmentId" IN (${Prisma.join(environments)})`
: Prisma.empty
}
${scheduleId ? Prisma.sql`AND tr."scheduleId" = ${scheduleId}` : Prisma.empty}
${typeof isTest === "boolean" ? Prisma.sql`AND tr."isTest" = ${isTest}` : Prisma.empty}
${
@@ -314,8 +290,6 @@ WHERE
: Prisma.empty
}
${rootOnly === true ? Prisma.sql`AND tr."rootTaskRunId" IS NULL` : Prisma.empty}
GROUP BY
tr.id, bw.version
ORDER BY
${direction === "forward" ? Prisma.sql`tr.id DESC` : Prisma.sql`tr.id ASC`}
LIMIT ${pageSize + 1}`;
@@ -350,12 +324,7 @@ WHERE
if (!hasAnyRuns) {
const firstRun = await this._replica.taskRun.findFirst({
where: {
projectId: project.id,
runtimeEnvironmentId: environments
? {
in: environments,
}
: undefined,
runtimeEnvironmentId: environmentId,
},
});
@@ -366,12 +335,6 @@ WHERE
return {
runs: runsToReturn.map((run) => {
const environment = project.environments.find((env) => env.id === run.runtimeEnvironmentId);
if (!environment) {
throw new Error(`Environment not found for TaskRun ${run.id}`);
}
const hasFinished = isFinalRunStatus(run.status);
const startedAt = run.startedAt ?? run.lockedAt;
@@ -396,7 +359,7 @@ WHERE
isReplayable: true,
isCancellable: isCancellableRunStatus(run.status),
isPending: isPendingRunStatus(run.status),
environment: displayableEnvironment(environment, userId),
environment: displayableEnvironment,
idempotencyKey: run.idempotencyKey ? run.idempotencyKey : undefined,
ttl: run.ttl ? run.ttl : undefined,
expiredAt: run.expiredAt ? run.expiredAt.toISOString() : undefined,
@@ -428,7 +391,6 @@ WHERE
tasks: tasks || [],
versions: versions || [],
statuses: statuses || [],
environments: environments || [],
from: time.from,
to: time.to,
},
apps/webapp/app/presenters/v3/TaskListPresenter.server.ts
+91 -221
View File
@@ -1,15 +1,20 @@
import { CURRENT_DEPLOYMENT_LABEL } from "@trigger.dev/core/v3/isomorphic";
import {
Prisma,
type TaskRunStatus as DBTaskRunStatus,
type TaskRunStatus as TaskRunStatusType,
PrismaClientOrTransaction,
RuntimeEnvironmentType,
type TaskTriggerSource,
} from "@trigger.dev/database";
import { QUEUED_STATUSES } from "~/components/runs/v3/TaskRunStatus";
import { TaskRunStatus } from "~/database-types";
import { sqlDatabaseSchema } from "~/db.server";
import { logger } from "~/services/logger.server";
import { BasePresenter } from "./basePresenter.server";
import { $replica } from "~/db.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import {
AverageDurations,
ClickHouseEnvironmentMetricsRepository,
CurrentRunningStats,
DailyTaskActivity,
EnvironmentMetricsRepository,
PostgrestEnvironmentMetricsRepository,
} from "~/services/environmentMetricsRepository.server";
import { singleton } from "~/utils/singleton";
import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server";
export type TaskListItem = {
slug: string;
@@ -18,225 +23,90 @@ export type TaskListItem = {
triggerSource: TaskTriggerSource;
};
type Return = Awaited<ReturnType<TaskListPresenter["call"]>>;
export type TaskActivity = DailyTaskActivity[string];
export type TaskActivity = Awaited<Return["activity"]>[string];
export class TaskListPresenter {
constructor(
private readonly environmentMetricsRepository: EnvironmentMetricsRepository,
private readonly _replica: PrismaClientOrTransaction
) {}
export class TaskListPresenter extends BasePresenter {
public async call({ environmentId, projectId }: { environmentId: string; projectId: string }) {
const tasks = await this._replica.$queryRaw<
public async call({
environmentId,
environmentType,
}: {
environmentId: string;
environmentType: RuntimeEnvironmentType;
}) {
const currentWorker = await findCurrentWorkerFromEnvironment(
{
id: string;
slug: string;
filePath: string;
createdAt: Date;
triggerSource: TaskTriggerSource;
}[]
>`
WITH non_dev_workers AS (
SELECT wd."workerId" AS id
FROM ${sqlDatabaseSchema}."WorkerDeploymentPromotion" wdp
INNER JOIN ${sqlDatabaseSchema}."WorkerDeployment" wd
ON wd.id = wdp."deploymentId"
WHERE wdp."environmentId" = ${environmentId}
AND wdp."label" = ${CURRENT_DEPLOYMENT_LABEL}
),
workers AS (
SELECT DISTINCT ON ("runtimeEnvironmentId") id, "runtimeEnvironmentId", version
FROM ${sqlDatabaseSchema}."BackgroundWorker"
WHERE "runtimeEnvironmentId" = ${environmentId}
OR id IN (SELECT id FROM non_dev_workers)
ORDER BY "runtimeEnvironmentId", "createdAt" DESC
)
SELECT tasks.id, slug, "filePath", "triggerSource", tasks."runtimeEnvironmentId", tasks."createdAt"
FROM workers
JOIN ${sqlDatabaseSchema}."BackgroundWorkerTask" tasks ON tasks."workerId" = workers.id
ORDER BY slug ASC;`;
//then get the activity for each task
const activity = this.#getActivity(
tasks.map((t) => t.slug),
projectId,
environmentId
id: environmentId,
type: environmentType,
},
this._replica
);
const runningStats = this.#getRunningStats(
tasks.map((t) => t.slug),
projectId,
environmentId
);
if (!currentWorker) {
return {
tasks: [],
activity: Promise.resolve({} as DailyTaskActivity),
runningStats: Promise.resolve({} as CurrentRunningStats),
durations: Promise.resolve({} as AverageDurations),
};
}
const durations = this.#getAverageDurations(
tasks.map((t) => t.slug),
projectId,
environmentId
);
const tasks = await this._replica.backgroundWorkerTask.findMany({
where: {
workerId: currentWorker.id,
},
select: {
id: true,
slug: true,
filePath: true,
triggerSource: true,
createdAt: true,
},
orderBy: {
slug: "asc",
},
});
const slugs = tasks.map((t) => t.slug);
// IMPORTANT: Don't await these, we want to return the promises
// so we can defer the loading of the data
const activity = this.environmentMetricsRepository.getDailyTaskActivity({
environmentId,
days: 6, // This actually means 7 days, because we want to show the current day too
tasks: slugs,
});
const runningStats = this.environmentMetricsRepository.getCurrentRunningStats({
environmentId,
days: 6,
tasks: slugs,
});
const durations = this.environmentMetricsRepository.getAverageDurations({
environmentId,
days: 6,
tasks: slugs,
});
return { tasks, activity, runningStats, durations };
}
async #getActivity(tasks: string[], projectId: string, environmentId: string) {
if (tasks.length === 0) {
return {};
}
const activity = await this._replica.$queryRaw<
{
taskIdentifier: string;
status: TaskRunStatusType;
day: Date;
count: BigInt;
}[]
>`
SELECT
tr."taskIdentifier",
tr."status",
DATE(tr."createdAt") as day,
COUNT(*)
FROM
${sqlDatabaseSchema}."TaskRun" as tr
WHERE
tr."taskIdentifier" IN (${Prisma.join(tasks)})
AND tr."projectId" = ${projectId}
AND tr."runtimeEnvironmentId" = ${environmentId}
AND tr."createdAt" >= (current_date - interval '6 days')
GROUP BY
tr."taskIdentifier",
tr."status",
day
ORDER BY
tr."taskIdentifier" ASC,
day ASC,
tr."status" ASC;`;
//today with no time
const today = new Date();
today.setUTCHours(0, 0, 0, 0);
return activity.reduce((acc, a) => {
let existingTask = acc[a.taskIdentifier];
if (!existingTask) {
existingTask = [];
//populate the array with the past 7 days
for (let i = 6; i >= 0; i--) {
const day = new Date(today);
day.setUTCDate(today.getDate() - i);
day.setUTCHours(0, 0, 0, 0);
existingTask.push({
day: day.toISOString(),
[TaskRunStatus.COMPLETED_SUCCESSFULLY]: 0,
} as { day: string } & Record<TaskRunStatusType, number>);
}
acc[a.taskIdentifier] = existingTask;
}
const dayString = a.day.toISOString();
const day = existingTask.find((d) => d.day === dayString);
if (!day) {
logger.warn(`Day not found for TaskRun`, {
day: dayString,
taskIdentifier: a.taskIdentifier,
existingTask,
});
return acc;
}
day[a.status] = Number(a.count);
return acc;
}, {} as Record<string, ({ day: string } & Record<TaskRunStatusType, number>)[]>);
}
async #getRunningStats(tasks: string[], projectId: string, environmentId: string) {
if (tasks.length === 0) {
return {};
}
const stats = await this._replica.$queryRaw<
{
taskIdentifier: string;
status: DBTaskRunStatus;
count: BigInt;
}[]
>`
SELECT
tr."taskIdentifier",
tr.status,
COUNT(*)
FROM
${sqlDatabaseSchema}."TaskRun" as tr
WHERE
tr."taskIdentifier" IN (${Prisma.join(tasks)})
AND tr."projectId" = ${projectId}
AND tr."runtimeEnvironmentId" = ${environmentId}
AND tr."status" = ANY(ARRAY[${Prisma.join([
...QUEUED_STATUSES,
"EXECUTING",
])}]::\"TaskRunStatus\"[])
GROUP BY
tr."taskIdentifier",
tr.status
ORDER BY
tr."taskIdentifier" ASC`;
//create an object combining the queued and concurrency counts
const result: Record<string, { queued: number; running: number }> = {};
for (const task of tasks) {
const queued = stats.filter(
(q) => q.taskIdentifier === task && QUEUED_STATUSES.includes(q.status)
);
const queuedCount =
queued.length === 0
? 0
: queued.reduce((acc, q) => {
return acc + Number(q.count);
}, 0);
const running = stats.filter((r) => r.taskIdentifier === task && r.status === "EXECUTING");
const runningCount =
running.length === 0
? 0
: running.reduce((acc, r) => {
return acc + Number(r.count);
}, 0);
result[task] = {
queued: queuedCount,
running: runningCount,
};
}
return result;
}
async #getAverageDurations(tasks: string[], projectId: string, environmentId: string) {
if (tasks.length === 0) {
return {};
}
const durations = await this._replica.$queryRaw<
{
taskIdentifier: string;
duration: Number;
}[]
>`
SELECT
tr."taskIdentifier",
AVG(EXTRACT(EPOCH FROM (tr."updatedAt" - COALESCE(tr."startedAt", tr."lockedAt")))) as duration
FROM
${sqlDatabaseSchema}."TaskRun" as tr
WHERE
tr."taskIdentifier" IN (${Prisma.join(tasks)})
AND tr."projectId" = ${projectId}
AND tr."runtimeEnvironmentId" = ${environmentId}
AND tr."createdAt" >= (current_date - interval '6 days')
AND tr."status" IN ('COMPLETED_SUCCESSFULLY', 'COMPLETED_WITH_ERRORS')
GROUP BY
tr."taskIdentifier";`;
return Object.fromEntries(durations.map((s) => [s.taskIdentifier, Number(s.duration)]));
}
}
export const taskListPresenter = singleton("taskListPresenter", setupTaskListPresenter);
function setupTaskListPresenter() {
const environmentMetricsRepository = clickhouseClient
? new ClickHouseEnvironmentMetricsRepository({
clickhouse: clickhouseClient,
})
: new PostgrestEnvironmentMetricsRepository({
prisma: $replica,
});
return new TaskListPresenter(environmentMetricsRepository, $replica);
}
apps/webapp/app/presenters/v3/UsagePresenter.server.ts
+64 -26
View File
@@ -1,9 +1,10 @@
import { sqlDatabaseSchema } from "~/db.server";
import { PrismaClientOrTransaction, sqlDatabaseSchema } from "~/db.server";
import { env } from "~/env.server";
import { getUsage, getUsageSeries } from "~/services/platform.v3.server";
import { createTimeSeriesData } from "~/utils/graphs";
import { BasePresenter } from "./basePresenter.server";
import { DataPoint, linear } from "regression";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
type Options = {
organizationId: string;
@@ -103,27 +104,69 @@ export class UsagePresenter extends BasePresenter {
});
//usage by task
const tasks = this._replica.$queryRaw<TaskUsageItem[]>`
const tasks = await getTaskUsageByOrganization(
organizationId,
startOfMonth,
endOfMonth,
this._replica
);
return {
usage,
tasks,
};
}
}
async function getTaskUsageByOrganization(
organizationId: string,
startOfMonth: Date,
endOfMonth: Date,
replica: PrismaClientOrTransaction
) {
if (clickhouseClient) {
const [queryError, tasks] = await clickhouseClient.taskRuns.getTaskUsageByOrganization({
startTime: startOfMonth.getTime(),
endTime: endOfMonth.getTime(),
organizationId,
});
if (queryError) {
throw queryError;
}
return tasks
.map((task) => ({
taskIdentifier: task.task_identifier,
runCount: Number(task.run_count),
averageDuration: Number(task.average_duration),
averageCost: Number(task.average_cost) + env.CENTS_PER_RUN / 100,
totalDuration: Number(task.total_duration),
totalCost: Number(task.total_cost) + Number(task.total_base_cost),
}))
.sort((a, b) => b.totalCost - a.totalCost);
} else {
return replica.$queryRaw<TaskUsageItem[]>`
SELECT
tr."taskIdentifier",
COUNT(*) AS "runCount",
AVG(tr."usageDurationMs") AS "averageDuration",
SUM(tr."usageDurationMs") AS "totalDuration",
AVG(tr."costInCents") / 100.0 AS "averageCost",
SUM(tr."costInCents") / 100.0 AS "totalCost",
SUM(tr."baseCostInCents") / 100.0 AS "totalBaseCost"
FROM
${sqlDatabaseSchema}."TaskRun" tr
JOIN ${sqlDatabaseSchema}."Project" pr ON pr.id = tr."projectId"
JOIN ${sqlDatabaseSchema}."Organization" org ON org.id = pr."organizationId"
JOIN ${sqlDatabaseSchema}."RuntimeEnvironment" env ON env."id" = tr."runtimeEnvironmentId"
WHERE
env.type <> 'DEVELOPMENT'
AND tr."createdAt" > ${startOfMonth}
AND tr."createdAt" < ${endOfMonth}
AND org.id = ${organizationId}
GROUP BY
tr."taskIdentifier";
tr."taskIdentifier",
COUNT(*) AS "runCount",
AVG(tr."usageDurationMs") AS "averageDuration",
SUM(tr."usageDurationMs") AS "totalDuration",
AVG(tr."costInCents") / 100.0 AS "averageCost",
SUM(tr."costInCents") / 100.0 AS "totalCost",
SUM(tr."baseCostInCents") / 100.0 AS "totalBaseCost"
FROM
${sqlDatabaseSchema}."TaskRun" tr
JOIN ${sqlDatabaseSchema}."Project" pr ON pr.id = tr."projectId"
JOIN ${sqlDatabaseSchema}."Organization" org ON org.id = pr."organizationId"
JOIN ${sqlDatabaseSchema}."RuntimeEnvironment" env ON env."id" = tr."runtimeEnvironmentId"
WHERE
env.type <> 'DEVELOPMENT'
AND tr."createdAt" > ${startOfMonth}
AND tr."createdAt" < ${endOfMonth}
AND org.id = ${organizationId}
GROUP BY
tr."taskIdentifier";
`.then((data) => {
return data
.map((item) => ({
@@ -136,10 +179,5 @@ export class UsagePresenter extends BasePresenter {
}))
.sort((a, b) => b.totalCost - a.totalCost);
});
return {
usage,
tasks,
};
}
}
apps/webapp/app/presenters/v3/ViewSchedulePresenter.server.ts
+3 -2
View File
@@ -8,6 +8,7 @@ type ViewScheduleOptions = {
userId?: string;
projectId: string;
friendlyId: string;
environmentId: string;
};
export class ViewSchedulePresenter {
@@ -17,7 +18,7 @@ export class ViewSchedulePresenter {
this.#prismaClient = prismaClient;
}
public async call({ userId, projectId, friendlyId }: ViewScheduleOptions) {
public async call({ userId, projectId, friendlyId, environmentId }: ViewScheduleOptions) {
const schedule = await this.#prismaClient.taskSchedule.findFirst({
select: {
id: true,
@@ -76,7 +77,7 @@ export class ViewSchedulePresenter {
const runPresenter = new RunListPresenter(this.#prismaClient);
const { runs } = await runPresenter.call({
const { runs } = await runPresenter.call(environmentId, {
projectId: schedule.project.id,
scheduleId: schedule.id,
pageSize: 5,
apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts
+1 -2
View File
@@ -78,9 +78,8 @@ export class WaitpointPresenter extends BasePresenter {
if (connectedRunIds.length > 0) {
const runPresenter = new RunListPresenter();
const { runs } = await runPresenter.call({
const { runs } = await runPresenter.call(environmentId, {
projectId: projectId,
environments: [environmentId],
runIds: connectedRunIds,
pageSize: 5,
});
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam._index/route.tsx
+3 -3
View File
@@ -75,6 +75,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import {
type TaskActivity,
type TaskListItem,
taskListPresenter,
TaskListPresenter,
} from "~/presenters/v3/TaskListPresenter.server";
import {
@@ -123,10 +124,9 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
}
try {
const presenter = new TaskListPresenter();
const { tasks, activity, runningStats, durations } = await presenter.call({
const { tasks, activity, runningStats, durations } = await taskListPresenter.call({
environmentId: environment.id,
projectId: project.id,
environmentType: environment.type,
});
const usefulLinksPreference = await getUsefulLinksPreference(request);
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.next.runs._index/route.tsx
+9 -4
View File
@@ -34,12 +34,14 @@ import { TextLink } from "~/components/primitives/TextLink";
import { RunsFilters, TaskRunListSearchFilters } from "~/components/runs/v3/RunFilters";
import { TaskRunsTable } from "~/components/runs/v3/TaskRunsTable";
import { BULK_ACTION_RUN_LIMIT } from "~/consts";
import { $replica } from "~/db.server";
import { useEnvironment } from "~/hooks/useEnvironment";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { RunListPresenter } from "~/presenters/v3/RunListPresenter.server";
import { NextRunListPresenter } from "~/presenters/v3/NextRunListPresenter.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import {
getRootOnlyFilterPreference,
setRootOnlyFilterPreference,
@@ -121,14 +123,17 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
scheduleId,
} = TaskRunListSearchFilters.parse(s);
const presenter = new RunListPresenter();
const list = presenter.call({
if (!clickhouseClient) {
throw new Error("Clickhouse is not supported yet");
}
const presenter = new NextRunListPresenter($replica, clickhouseClient);
const list = presenter.call(environment.id, {
userId,
projectId: project.id,
tasks,
versions,
statuses,
environments,
tags,
period,
bulkId,
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx
+1 -2
View File
@@ -122,13 +122,12 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
} = TaskRunListSearchFilters.parse(s);
const presenter = new RunListPresenter();
const list = presenter.call({
const list = presenter.call(environment.id, {
userId,
projectId: project.id,
tasks,
versions,
statuses,
environments,
tags,
period,
bulkId,
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.schedules.$scheduleParam/route.tsx
+9 -2
View File
@@ -45,6 +45,7 @@ import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { ViewSchedulePresenter } from "~/presenters/v3/ViewSchedulePresenter.server";
import { requireUserId } from "~/services/session.server";
import { cn } from "~/utils/cn";
@@ -59,7 +60,8 @@ import { SetActiveOnTaskScheduleService } from "~/v3/services/setActiveOnTaskSch
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
const { projectParam, organizationSlug, scheduleParam } = v3ScheduleParams.parse(params);
const { projectParam, organizationSlug, envParam, scheduleParam } =
v3ScheduleParams.parse(params);
// Find the project scoped to the organization
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
@@ -68,11 +70,17 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
return redirectWithErrorMessage("/", request, "Project not found");
}
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
if (!environment) {
return redirectWithErrorMessage("/", request, "Environment not found");
}
const presenter = new ViewSchedulePresenter();
const result = await presenter.call({
userId,
projectId: project.id,
friendlyId: scheduleParam,
environmentId: environment.id,
});
if (!result) {
@@ -304,7 +312,6 @@ export default function Page() {
tasks: [],
versions: [],
statuses: [],
environments: [],
from: undefined,
to: undefined,
}}
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.tokens.$waitpointParam/route.tsx
-1
View File
@@ -120,7 +120,6 @@ export default function Page() {
tasks: [],
versions: [],
statuses: [],
environments: [],
from: undefined,
to: undefined,
}}
apps/webapp/app/routes/api.v1.schedules.$scheduleId.activate.ts
+1
View File
@@ -59,6 +59,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
const result = await presenter.call({
projectId: authenticationResult.environment.projectId,
friendlyId: parsedParams.data.scheduleId,
environmentId: authenticationResult.environment.id,
});
if (!result) {
apps/webapp/app/routes/api.v1.schedules.$scheduleId.deactivate.ts
+1
View File
@@ -58,6 +58,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
const result = await presenter.call({
projectId: authenticationResult.environment.projectId,
friendlyId: parsedParams.data.scheduleId,
environmentId: authenticationResult.environment.id,
});
if (!result) {
apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts
+1
View File
@@ -140,6 +140,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
const result = await presenter.call({
projectId: authenticationResult.environment.projectId,
friendlyId: parsedParams.data.scheduleId,
environmentId: authenticationResult.environment.id,
});
if (!result) {
apps/webapp/app/routes/realtime.v1.runs.ts
+1
View File
@@ -9,6 +9,7 @@ const SearchParamsSchema = z.object({
.transform((value) => {
return value ? value.split(",") : undefined;
}),
createdAt: z.string().optional(),
});
export const loader = createLoaderApiRoute(
apps/webapp/app/services/clickhouseInstance.server.ts
+30
View File
@@ -0,0 +1,30 @@
import { ClickHouse } from "@internal/clickhouse";
import { env } from "~/env.server";
import { singleton } from "~/utils/singleton";
export const clickhouseClient = singleton("clickhouseClient", initializeClickhouseClient);
function initializeClickhouseClient() {
if (!env.CLICKHOUSE_URL) {
console.log("🗃️ Clickhouse service not enabled");
return;
}
console.log("🗃️ Clickhouse service enabled");
const clickhouse = new ClickHouse({
url: env.CLICKHOUSE_URL,
name: "clickhouse-instance",
keepAlive: {
enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
},
logLevel: env.CLICKHOUSE_LOG_LEVEL,
compression: {
request: true,
},
maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
return clickhouse;
}
apps/webapp/app/services/environmentMetricsRepository.server.ts
+352
View File
@@ -0,0 +1,352 @@
import { ClickHouse } from "@internal/clickhouse";
import { Logger, LogLevel } from "@trigger.dev/core/logger";
import type { PrismaClientOrTransaction, TaskRunStatus } from "@trigger.dev/database";
import { Prisma } from "@trigger.dev/database";
import { QUEUED_STATUSES } from "~/components/runs/v3/TaskRunStatus";
export type DailyTaskActivity = Record<string, ({ day: string } & Record<TaskRunStatus, number>)[]>;
export type CurrentRunningStats = Record<string, { queued: number; running: number }>;
export type AverageDurations = Record<string, number>;
export interface EnvironmentMetricsRepository {
getDailyTaskActivity(options: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<DailyTaskActivity>;
getCurrentRunningStats(options: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<CurrentRunningStats>;
getAverageDurations(options: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<AverageDurations>;
}
export type PostgrestEnvironmentMetricsRepositoryOptions = {
prisma: PrismaClientOrTransaction;
schema?: string;
logger?: Logger;
logLevel?: LogLevel;
};
export class PostgrestEnvironmentMetricsRepository implements EnvironmentMetricsRepository {
private readonly logger: Logger;
private readonly schema: string;
constructor(private readonly options: PostgrestEnvironmentMetricsRepositoryOptions) {
this.logger =
options.logger ??
new Logger("PostgrestEnvironmentMetricsRepository", options.logLevel ?? "info");
this.schema = options.schema ?? "public";
}
public async getDailyTaskActivity({
environmentId,
days,
tasks,
}: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<DailyTaskActivity> {
if (tasks.length === 0) {
return {};
}
const activity = await this.options.prisma.$queryRaw<
{
taskIdentifier: string;
status: TaskRunStatus;
day: Date;
count: BigInt;
}[]
>`
SELECT
tr."taskIdentifier",
tr."status",
DATE(tr."createdAt") as day,
COUNT(*)
FROM
${Prisma.sql([this.schema])}."TaskRun" as tr
WHERE
tr."taskIdentifier" IN (${Prisma.join(tasks)})
AND tr."runtimeEnvironmentId" = ${environmentId}
AND tr."createdAt" >= (current_date - interval '1 day' * ${days})
GROUP BY
tr."taskIdentifier",
tr."status",
day
ORDER BY
tr."taskIdentifier" ASC,
day ASC,
tr."status" ASC;`;
return fillInDailyTaskActivity(activity, days);
}
public async getCurrentRunningStats({
environmentId,
days,
tasks,
}: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<CurrentRunningStats> {
if (tasks.length === 0) {
return {};
}
const stats = await this.options.prisma.$queryRaw<
{
taskIdentifier: string;
status: TaskRunStatus;
count: BigInt;
}[]
>`
SELECT
tr."taskIdentifier",
tr.status,
COUNT(*)
FROM
${Prisma.sql([this.schema])}."TaskRun" as tr
WHERE
tr."taskIdentifier" IN (${Prisma.join(tasks)})
AND tr."runtimeEnvironmentId" = ${environmentId}
AND tr."createdAt" >= (current_date - interval '1 day' * ${days})
AND tr."status" = ANY(ARRAY[${Prisma.join([
...QUEUED_STATUSES,
"EXECUTING",
])}]::\"TaskRunStatus\"[])
GROUP BY
tr."taskIdentifier",
tr.status
ORDER BY
tr."taskIdentifier" ASC`;
return fillInCurrentRunningStats(stats, tasks);
}
public async getAverageDurations({
environmentId,
days,
tasks,
}: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<AverageDurations> {
if (tasks.length === 0) {
return {};
}
const durations = await this.options.prisma.$queryRaw<
{
taskIdentifier: string;
duration: Number;
}[]
>`
SELECT
tr."taskIdentifier",
AVG(EXTRACT(EPOCH FROM (tr."updatedAt" - COALESCE(tr."startedAt", tr."lockedAt")))) as duration
FROM
${Prisma.sql([this.schema])}."TaskRun" as tr
WHERE
tr."taskIdentifier" IN (${Prisma.join(tasks)})
AND tr."runtimeEnvironmentId" = ${environmentId}
AND tr."createdAt" >= (current_date - interval '1 day' * ${days})
AND tr."status" IN ('COMPLETED_SUCCESSFULLY', 'COMPLETED_WITH_ERRORS')
GROUP BY
tr."taskIdentifier";`;
return Object.fromEntries(durations.map((s) => [s.taskIdentifier, Number(s.duration)]));
}
}
export type ClickHouseEnvironmentMetricsRepositoryOptions = {
clickhouse: ClickHouse;
};
export class ClickHouseEnvironmentMetricsRepository implements EnvironmentMetricsRepository {
constructor(private readonly options: ClickHouseEnvironmentMetricsRepositoryOptions) {}
public async getDailyTaskActivity({
environmentId,
days,
tasks,
}: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<DailyTaskActivity> {
if (tasks.length === 0) {
return {};
}
const [queryError, activity] = await this.options.clickhouse.taskRuns.getTaskActivity({
environmentId,
days,
});
if (queryError) {
throw queryError;
}
return fillInDailyTaskActivity(
activity.map((a) => ({
taskIdentifier: a.task_identifier,
status: a.status as TaskRunStatus,
day: new Date(a.day),
count: BigInt(a.count),
})),
days
);
}
public async getCurrentRunningStats({
environmentId,
days,
tasks,
}: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<CurrentRunningStats> {
if (tasks.length === 0) {
return {};
}
const [queryError, stats] = await this.options.clickhouse.taskRuns.getCurrentRunningStats({
environmentId,
days,
});
if (queryError) {
throw queryError;
}
return fillInCurrentRunningStats(
stats.map((s) => ({
taskIdentifier: s.task_identifier,
status: s.status as TaskRunStatus,
count: BigInt(s.count),
})),
tasks
);
}
public async getAverageDurations({
environmentId,
days,
tasks,
}: {
environmentId: string;
days: number;
tasks: string[];
}): Promise<AverageDurations> {
if (tasks.length === 0) {
return {};
}
const [queryError, durations] = await this.options.clickhouse.taskRuns.getAverageDurations({
environmentId,
days,
});
if (queryError) {
throw queryError;
}
return Object.fromEntries(durations.map((d) => [d.task_identifier, Number(d.duration)]));
}
}
type TaskActivityResults = Array<{
taskIdentifier: string;
status: TaskRunStatus;
day: Date;
count: BigInt;
}>;
function fillInDailyTaskActivity(activity: TaskActivityResults, days: number): DailyTaskActivity {
//today with no time
const today = new Date();
today.setUTCHours(0, 0, 0, 0);
return activity.reduce((acc, a) => {
let existingTask = acc[a.taskIdentifier];
if (!existingTask) {
existingTask = [];
//populate the array with the past 7 days
for (let i = days; i >= 0; i--) {
const day = new Date(today);
day.setUTCDate(today.getDate() - i);
day.setUTCHours(0, 0, 0, 0);
existingTask.push({
day: day.toISOString(),
["COMPLETED_SUCCESSFULLY"]: 0,
} as { day: string } & Record<TaskRunStatus, number>);
}
acc[a.taskIdentifier] = existingTask;
}
const dayString = a.day.toISOString();
const day = existingTask.find((d) => d.day === dayString);
if (!day) {
return acc;
}
day[a.status] = Number(a.count);
return acc;
}, {} as DailyTaskActivity);
}
type CurrentRunningStatsResults = Array<{
taskIdentifier: string;
status: TaskRunStatus;
count: BigInt;
}>;
function fillInCurrentRunningStats(
stats: CurrentRunningStatsResults,
tasks: string[]
): CurrentRunningStats {
//create an object combining the queued and concurrency counts
const result: Record<string, { queued: number; running: number }> = {};
for (const task of tasks) {
const queued = stats.filter(
(q) => q.taskIdentifier === task && QUEUED_STATUSES.includes(q.status)
);
const queuedCount =
queued.length === 0
? 0
: queued.reduce((acc, q) => {
return acc + Number(q.count);
}, 0);
const running = stats.filter((r) => r.taskIdentifier === task && r.status === "EXECUTING");
const runningCount =
running.length === 0
? 0
: running.reduce((acc, r) => {
return acc + Number(r.count);
}, 0);
result[task] = {
queued: queuedCount,
running: runningCount,
};
}
return result;
}
apps/webapp/app/services/realtimeClient.server.ts
+217 -8
View File
@@ -1,16 +1,54 @@
import { json } from "@remix-run/server-runtime";
import Redis, { Callback, Result, type RedisOptions } from "ioredis";
import { tryCatch } from "@trigger.dev/core/utils";
import { safeParseNaturalLanguageDurationAgo } from "@trigger.dev/core/v3/isomorphic";
import { Callback, Result } from "ioredis";
import { randomUUID } from "node:crypto";
import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server";
import { longPollingFetch } from "~/utils/longPollingFetch";
import { logger } from "./logger.server";
import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server";
import { jumpHash } from "@trigger.dev/core/v3/serverOnly";
import { Cache, createCache, DefaultStatefulContext, Namespace } from "@unkey/cache";
import { MemoryStore } from "@unkey/cache/stores";
import { RedisCacheStore } from "./unkey/redisCacheStore.server";
import { env } from "~/env.server";
export interface CachedLimitProvider {
getCachedLimit: (organizationId: string, defaultValue: number) => Promise<number | undefined>;
}
const DEFAULT_ELECTRIC_COLUMNS = [
"id",
"taskIdentifier",
"createdAt",
"updatedAt",
"startedAt",
"delayUntil",
"queuedAt",
"expiredAt",
"completedAt",
"friendlyId",
"number",
"isTest",
"status",
"usageDurationMs",
"costInCents",
"baseCostInCents",
"ttl",
"payload",
"payloadType",
"metadata",
"metadataType",
"output",
"outputType",
"runTags",
"error",
];
const RESERVED_COLUMNS = ["id", "taskIdentifier", "friendlyId", "status", "createdAt"];
const RESERVED_SEARCH_PARAMS = ["createdAt", "tags", "skipColumns"];
export type RealtimeClientOptions = {
electricOrigin: string;
electricOrigin: string | string[];
redis: RedisWithClusterOptions;
cachedLimitProvider: CachedLimitProvider;
keyPrefix: string;
@@ -24,18 +62,45 @@ export type RealtimeEnvironment = {
export type RealtimeRunsParams = {
tags?: string[];
createdAt?: string;
};
export class RealtimeClient {
private redis: RedisClient;
private expiryTimeInSeconds: number;
private cachedLimitProvider: CachedLimitProvider;
private cache: Cache<{ createdAtFilter: string }>;
constructor(private options: RealtimeClientOptions) {
this.redis = createRedisClient("trigger:realtime", options.redis);
this.expiryTimeInSeconds = options.expiryTimeInSeconds ?? 60 * 5; // default to 5 minutes
this.cachedLimitProvider = options.cachedLimitProvider;
this.#registerCommands();
const ctx = new DefaultStatefulContext();
const memory = new MemoryStore({ persistentMap: new Map() });
const redisCacheStore = new RedisCacheStore({
connection: {
keyPrefix: "tr:cache:realtime",
port: options.redis.port,
host: options.redis.host,
username: options.redis.username,
password: options.redis.password,
tlsDisabled: options.redis.tlsDisabled,
clusterMode: options.redis.clusterMode,
},
});
// This cache holds the limits fetched from the platform service
const cache = createCache({
createdAtFilter: new Namespace<string>(ctx, {
stores: [memory, redisCacheStore],
fresh: 60_000 * 60 * 24 * 7, // 1 week
stale: 60_000 * 60 * 24 * 14, // 2 weeks
}),
});
this.cache = cache;
}
async streamChunks(
@@ -92,9 +157,99 @@ export class RealtimeClient {
whereClauses.push(`"runTags" @> ARRAY[${params.tags.map((t) => `'${t}'`).join(",")}]`);
}
const createdAtFilter = await this.#calculateCreatedAtFilter(url, params.createdAt);
if (createdAtFilter) {
whereClauses.push(`"createdAt" > '${createdAtFilter.toISOString()}'`);
}
const whereClause = whereClauses.join(" AND ");
return this.#streamRunsWhere(url, environment, whereClause, clientVersion);
const response = await this.#streamRunsWhere(url, environment, whereClause, clientVersion);
if (createdAtFilter) {
const [setCreatedAtFilterError] = await tryCatch(
this.#setCreatedAtFilterFromResponse(response, createdAtFilter)
);
if (setCreatedAtFilterError) {
logger.error("[realtimeClient] Failed to set createdAt filter", {
error: setCreatedAtFilterError,
createdAtFilter,
responseHeaders: Object.fromEntries(response.headers.entries()),
responseStatus: response.status,
});
}
}
return response;
}
async #calculateCreatedAtFilter(url: URL | string, createdAt?: string) {
const duration = createdAt ?? "24h";
const $url = new URL(url.toString());
const shapeId = extractShapeId($url);
if (!shapeId) {
// This means we need to calculate the createdAt filter and store it in redis after we get back the response
const createdAtFilter = safeParseNaturalLanguageDurationAgo(duration);
// Validate that the createdAt filter is in the past, and not more than the maximum age in the past.
// if it's more than the maximum age in the past, just return the maximum age in the past Date
if (
createdAtFilter &&
createdAtFilter < new Date(Date.now() - env.REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS)
) {
return new Date(Date.now() - env.REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS);
}
return createdAtFilter;
} else {
// We need to get the createdAt filter value from redis, if there is none we need to return undefined
const [createdAtFilterError, createdAtFilter] = await tryCatch(
this.#getCreatedAtFilter(shapeId)
);
if (createdAtFilterError) {
logger.error("[realtimeClient] Failed to get createdAt filter", {
shapeId,
error: createdAtFilterError,
});
return;
}
return createdAtFilter;
}
}
async #getCreatedAtFilter(shapeId: string) {
const createdAtFilterCacheResult = await this.cache.createdAtFilter.get(shapeId);
if (createdAtFilterCacheResult.err) {
logger.error("[realtimeClient] Failed to get createdAt filter", {
shapeId,
error: createdAtFilterCacheResult.err,
});
return;
}
if (!createdAtFilterCacheResult.val) {
return;
}
return new Date(createdAtFilterCacheResult.val);
}
async #setCreatedAtFilterFromResponse(response: Response, createdAtFilter: Date) {
const shapeId = extractShapeIdFromResponse(response);
if (!shapeId) {
return;
}
await this.cache.createdAtFilter.set(shapeId, createdAtFilter.toISOString());
}
async #streamRunsWhere(
@@ -103,18 +258,33 @@ export class RealtimeClient {
whereClause: string,
clientVersion?: string
) {
const electricUrl = this.#constructRunsElectricUrl(url, whereClause, clientVersion);
const electricUrl = this.#constructRunsElectricUrl(
url,
environment,
whereClause,
clientVersion
);
return this.#performElectricRequest(electricUrl, environment, undefined, clientVersion);
}
#constructRunsElectricUrl(url: URL | string, whereClause: string, clientVersion?: string): URL {
#constructRunsElectricUrl(
url: URL | string,
environment: RealtimeEnvironment,
whereClause: string,
clientVersion?: string
): URL {
const $url = new URL(url.toString());
const electricUrl = new URL(`${this.options.electricOrigin}/v1/shape`);
const electricOrigin = this.#resolveElectricOrigin(environment.id);
const electricUrl = new URL(`${electricOrigin}/v1/shape`);
// Copy over all the url search params to the electric url
$url.searchParams.forEach((value, key) => {
if (RESERVED_SEARCH_PARAMS.includes(key)) {
return;
}
electricUrl.searchParams.set(key, value);
});
@@ -127,6 +297,27 @@ export class RealtimeClient {
electricUrl.searchParams.set("handle", electricUrl.searchParams.get("shape_id") ?? "");
}
const skipColumnsRaw = $url.searchParams.get("skipColumns");
if (skipColumnsRaw) {
const skipColumns = skipColumnsRaw
.split(",")
.map((c) => c.trim())
.filter((c) => c !== "" && !RESERVED_COLUMNS.includes(c));
electricUrl.searchParams.set(
"columns",
DEFAULT_ELECTRIC_COLUMNS.filter((c) => !skipColumns.includes(c))
.map((c) => `"${c}"`)
.join(",")
);
} else {
electricUrl.searchParams.set(
"columns",
DEFAULT_ELECTRIC_COLUMNS.map((c) => `"${c}"`).join(",")
);
}
return electricUrl;
}
@@ -232,6 +423,10 @@ export class RealtimeClient {
// ... (rest of your existing code for the long polling request)
const response = await longPollingFetch(url.toString(), { signal }, rewriteResponseHeaders);
// If this is the initial request, the response.headers['electric-handle'] will be the shapeId
// And we may need to set the "createdAt" filter timestamp keyed by the shapeId
// Then in the next request, we will get the createdAt timestamp value via the shapeId and use it to filter the results
// Decrement the counter after the long polling request is complete
await this.#decrementConcurrency(environment.id, requestId);
@@ -275,6 +470,16 @@ export class RealtimeClient {
return `${this.options.keyPrefix}:${environmentId}`;
}
#resolveElectricOrigin(environmentId: string) {
if (typeof this.options.electricOrigin === "string") {
return this.options.electricOrigin;
}
const index = jumpHash(environmentId, this.options.electricOrigin.length);
return this.options.electricOrigin[index] ?? this.options.electricOrigin[0];
}
#registerCommands() {
this.redis.defineCommand("incrementAndCheckConcurrency", {
numberOfKeys: 1,
@@ -314,7 +519,11 @@ export class RealtimeClient {
}
function extractShapeId(url: URL) {
return url.searchParams.get("handle");
return url.searchParams.get("handle") ?? url.searchParams.get("shape_id");
}
function extractShapeIdFromResponse(response: Response) {
return response.headers.get("electric-handle");
}
function isLiveRequestUrl(url: URL) {
apps/webapp/app/services/realtimeClientGlobal.server.ts
+3 -1
View File
@@ -4,8 +4,10 @@ import { RealtimeClient } from "./realtimeClient.server";
import { getCachedLimit } from "./platform.v3.server";
function initializeRealtimeClient() {
const electricOrigin = env.ELECTRIC_ORIGIN_SHARDS?.split(",") ?? env.ELECTRIC_ORIGIN;
return new RealtimeClient({
electricOrigin: env.ELECTRIC_ORIGIN,
electricOrigin: electricOrigin,
keyPrefix: "tr:realtime:concurrency",
redis: {
port: env.RATE_LIMIT_REDIS_PORT,
apps/webapp/app/services/runsRepository.server.ts
+226
View File
@@ -0,0 +1,226 @@
import { ClickHouse } from "@internal/clickhouse";
import { Tracer } from "@internal/tracing";
import { Logger, LogLevel } from "@trigger.dev/core/logger";
import { TaskRunStatus } from "@trigger.dev/database";
import { PrismaClient } from "~/db.server";
export type RunsRepositoryOptions = {
clickhouse: ClickHouse;
prisma: PrismaClient;
logger?: Logger;
logLevel?: LogLevel;
tracer?: Tracer;
};
export type ListRunsOptions = {
projectId: string;
environmentId: string;
//filters
tasks?: string[];
versions?: string[];
statuses?: TaskRunStatus[];
tags?: string[];
scheduleId?: string;
period?: number;
from?: number;
to?: number;
isTest?: boolean;
rootOnly?: boolean;
batchId?: string;
runFriendlyIds?: string[];
runIds?: string[];
//pagination
page: {
size: number;
cursor?: string;
direction?: "forward" | "backward";
};
};
export class RunsRepository {
constructor(private readonly options: RunsRepositoryOptions) {}
async listRuns(options: ListRunsOptions) {
const queryBuilder = this.options.clickhouse.taskRuns.queryBuilder();
queryBuilder
.where("environment_id = {environmentId: String}", {
environmentId: options.environmentId,
})
.where("project_id = {projectId: String}", {
projectId: options.projectId,
});
if (options.tasks && options.tasks.length > 0) {
queryBuilder.where("task_identifier IN {tasks: Array(String)}", { tasks: options.tasks });
}
if (options.versions && options.versions.length > 0) {
queryBuilder.where("task_version IN {versions: Array(String)}", {
versions: options.versions,
});
}
if (options.statuses && options.statuses.length > 0) {
queryBuilder.where("status IN {statuses: Array(String)}", { statuses: options.statuses });
}
if (options.tags && options.tags.length > 0) {
queryBuilder.where("hasAny(tags, {tags: Array(String)})", { tags: options.tags });
}
if (options.scheduleId) {
queryBuilder.where("schedule_id = {scheduleId: String}", { scheduleId: options.scheduleId });
}
// Period is a number of milliseconds duration
if (options.period) {
queryBuilder.where("created_at >= fromUnixTimestamp64Milli({period: Int64})", {
period: new Date(Date.now() - options.period).getTime(),
});
}
if (options.from) {
queryBuilder.where("created_at >= fromUnixTimestamp64Milli({from: Int64})", {
from: options.from,
});
}
if (options.to) {
queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", { to: options.to });
}
if (typeof options.isTest === "boolean") {
queryBuilder.where("is_test = {isTest: Boolean}", { isTest: options.isTest });
}
if (options.rootOnly) {
queryBuilder.where("root_run_id = ''");
}
if (options.batchId) {
queryBuilder.where("batch_id = {batchId: String}", { batchId: options.batchId });
}
if (options.runFriendlyIds && options.runFriendlyIds.length > 0) {
queryBuilder.where("friendly_id IN {runFriendlyIds: Array(String)}", {
runFriendlyIds: options.runFriendlyIds,
});
}
if (options.runIds && options.runIds.length > 0) {
queryBuilder.where("run_id IN {runIds: Array(String)}", { runIds: options.runIds });
}
if (options.page.cursor) {
if (options.page.direction === "forward") {
queryBuilder
.where("run_id < {runId: String}", { runId: options.page.cursor })
.orderBy("run_id DESC")
.limit(options.page.size + 1);
} else {
queryBuilder
.where("run_id > {runId: String}", { runId: options.page.cursor })
.orderBy("run_id DESC")
.limit(options.page.size + 1);
}
} else {
// Initial page - no cursor provided
queryBuilder.orderBy("run_id DESC").limit(options.page.size + 1);
}
const [queryError, result] = await queryBuilder.execute();
if (queryError) {
throw queryError;
}
const runIds = result.map((row) => row.run_id);
// If there are more runs than the page size, we need to fetch the next page
const hasMore = runIds.length > options.page.size;
let nextCursor: string | null = null;
let previousCursor: string | null = null;
//get cursors for next and previous pages
if (options.page.cursor) {
switch (options.page.direction) {
case "forward":
previousCursor = runIds.at(0) ?? null;
if (hasMore) {
// The next cursor should be the last run ID from this page
nextCursor = runIds[options.page.size - 1];
}
break;
case "backward":
// No need to reverse since we're using DESC ordering consistently
if (hasMore) {
previousCursor = runIds[options.page.size - 1];
}
nextCursor = runIds.at(0) ?? null;
break;
default:
// This shouldn't happen if cursor is provided, but handle it
if (hasMore) {
nextCursor = runIds[options.page.size - 1];
}
break;
}
} else {
// Initial page - no cursor
if (hasMore) {
// The next cursor should be the last run ID from this page
nextCursor = runIds[options.page.size - 1];
}
}
const runIdsToReturn = hasMore ? runIds.slice(0, -1) : runIds;
const runs = await this.options.prisma.taskRun.findMany({
where: {
id: {
in: runIdsToReturn,
},
},
orderBy: {
id: "desc",
},
select: {
id: true,
friendlyId: true,
taskIdentifier: true,
taskVersion: true,
runtimeEnvironmentId: true,
status: true,
createdAt: true,
startedAt: true,
lockedAt: true,
delayUntil: true,
updatedAt: true,
completedAt: true,
isTest: true,
spanId: true,
idempotencyKey: true,
ttl: true,
expiredAt: true,
costInCents: true,
baseCostInCents: true,
usageDurationMs: true,
runTags: true,
depth: true,
rootTaskRunId: true,
batchId: true,
metadata: true,
metadataType: true,
},
});
return {
runs,
pagination: {
nextCursor,
previousCursor,
},
};
}
}
apps/webapp/test/runsRepository.test.ts
+1501
View File
File diff suppressed because it is too large Load Diff
apps/webapp/test/utils/replicationUtils.ts
+54
View File
@@ -0,0 +1,54 @@
import { ClickHouse } from "@internal/clickhouse";
import { RedisOptions } from "@internal/redis";
import { PrismaClient } from "~/db.server";
import { RunsReplicationService } from "~/services/runsReplicationService.server";
import { afterEach } from "vitest";
export async function setupClickhouseReplication({
prisma,
databaseUrl,
clickhouseUrl,
redisOptions,
}: {
prisma: PrismaClient;
databaseUrl: string;
clickhouseUrl: string;
redisOptions: RedisOptions;
}) {
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
const clickhouse = new ClickHouse({
url: clickhouseUrl,
name: "runs-replication",
compression: {
request: true,
},
});
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: databaseUrl,
serviceName: "runs-replication",
slotName: "task_runs_to_clickhouse_v1",
publicationName: "task_runs_to_clickhouse_v1_publication",
redisOptions,
maxFlushConcurrency: 1,
flushIntervalMs: 100,
flushBatchSize: 1,
leaderLockTimeoutMs: 5000,
leaderLockExtendIntervalMs: 1000,
ackIntervalSeconds: 5,
});
await runsReplicationService.start();
// Runs after each test in the current context
afterEach(async () => {
// Clean up resources here
await runsReplicationService.stop();
});
return {
clickhouse,
};
}
internal-packages/clickhouse/src/__snapshots__/taskRuns.test.ts.snap
+20
View File
@@ -0,0 +1,20 @@
// Vitest Snapshot v1, https://vitest.dev/guide/snapshot.html
exports[`Task Runs V2 > should be able to query task runs using the query builder 1`] = `
{
"params": {
"environmentId": "cm9kddfcs01zqdy88ld9mmrli",
},
"query": "SELECT run_id FROM trigger_dev.task_runs_v2 FINAL WHERE environment_id = {environmentId: String}",
}
`;
exports[`Task Runs V2 > should be able to query task runs using the query builder 2`] = `
{
"params": {
"environmentId": "cm9kddfcs01zqdy88ld9mmrli",
"status": "COMPLETED_SUCCESSFULLY",
},
"query": "SELECT run_id FROM trigger_dev.task_runs_v2 FINAL WHERE environment_id = {environmentId: String} AND status = {status: String}",
}
`;
internal-packages/clickhouse/src/client/client.ts
+23
View File
@@ -11,6 +11,7 @@ import { z } from "zod";
import { InsertError, QueryError } from "./errors.js";
import type {
ClickhouseInsertFunction,
ClickhouseQueryBuilderFunction,
ClickhouseQueryFunction,
ClickhouseReader,
ClickhouseWriter,
@@ -19,6 +20,7 @@ import { generateErrorMessage } from "zod-error";
import { Logger, type LogLevel } from "@trigger.dev/core/logger";
import type { Agent as HttpAgent } from "http";
import type { Agent as HttpsAgent } from "https";
import { ClickhouseQueryBuilder } from "./queryBuilder.js";
export type ClickhouseConfig = {
name: string;
@@ -102,6 +104,14 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
}): ClickhouseQueryFunction<z.input<TIn>, z.output<TOut>> {
return async (params, options) => {
return await startSpan(this.tracer, "query", async (span) => {
this.logger.debug("Querying clickhouse", {
name: req.name,
query: req.query.replace(/\s+/g, " "),
params,
settings: req.settings,
attributes: options?.attributes,
});
span.setAttributes({
"clickhouse.clientName": this.name,
"clickhouse.operationName": req.name,
@@ -205,6 +215,19 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
};
}
public queryBuilder<TOut extends z.ZodSchema<any>>(req: {
name: string;
baseQuery: string;
schema: TOut;
settings?: ClickHouseSettings;
}): ClickhouseQueryBuilderFunction<z.input<TOut>> {
return (chSettings) =>
new ClickhouseQueryBuilder(req.name, req.baseQuery, this, req.schema, {
...req.settings,
...chSettings?.settings,
});
}
public insert<TSchema extends z.ZodSchema<any>>(req: {
name: string;
table: string;
internal-packages/clickhouse/src/client/noop.ts
+12 -1
View File
@@ -1,15 +1,26 @@
import { Result } from "@trigger.dev/core/v3";
import { InsertError, QueryError } from "./errors.js";
import { ClickhouseWriter } from "./types.js";
import { ClickhouseQueryBuilderFunction, ClickhouseWriter } from "./types.js";
import { ClickhouseReader } from "./types.js";
import { z } from "zod";
import { ClickHouseSettings, InsertResult } from "@clickhouse/client";
import { ClickhouseQueryBuilder } from "./queryBuilder.js";
export class NoopClient implements ClickhouseReader, ClickhouseWriter {
public async close() {
return;
}
public queryBuilder<TOut extends z.ZodSchema<any>>(req: {
name: string;
baseQuery: string;
schema: TOut;
settings?: ClickHouseSettings;
}): ClickhouseQueryBuilderFunction<z.input<TOut>> {
return () =>
new ClickhouseQueryBuilder(req.name, req.baseQuery, this, req.schema, req.settings);
}
public query<TIn extends z.ZodSchema<any>, TOut extends z.ZodSchema<any>>(req: {
query: string;
params?: TIn;
internal-packages/clickhouse/src/client/queryBuilder.ts
+93
View File
@@ -0,0 +1,93 @@
import { z } from "zod";
import { ClickhouseQueryFunction, ClickhouseReader } from "./types.js";
import { ClickHouseSettings } from "@clickhouse/client";
export type QueryParamValue = string | number | boolean | Array<string | number | boolean> | null;
export type QueryParams = Record<string, QueryParamValue>;
export class ClickhouseQueryBuilder<TOutput> {
private name: string;
private baseQuery: string;
private whereClauses: string[] = [];
private params: QueryParams = {};
private orderByClause: string | null = null;
private limitClause: string | null = null;
private reader: ClickhouseReader;
private schema: z.ZodSchema<TOutput>;
private settings: ClickHouseSettings | undefined;
private groupByClause: string | null = null;
constructor(
name: string,
baseQuery: string,
reader: ClickhouseReader,
schema: z.ZodSchema<TOutput>,
settings?: ClickHouseSettings
) {
this.name = name;
this.baseQuery = baseQuery;
this.reader = reader;
this.schema = schema;
this.settings = settings;
}
where(clause: string, params?: QueryParams): this {
this.whereClauses.push(clause);
if (params) {
Object.assign(this.params, params);
}
return this;
}
whereIf(condition: any, clause: string, params?: QueryParams): this {
if (condition) {
this.where(clause, params);
}
return this;
}
groupBy(clause: string): this {
this.groupByClause = clause;
return this;
}
orderBy(clause: string): this {
this.orderByClause = clause;
return this;
}
limit(limit: number): this {
this.limitClause = `LIMIT ${limit}`;
return this;
}
execute(): ReturnType<ClickhouseQueryFunction<void, TOutput>> {
const { query, params } = this.build();
const queryFunction = this.reader.query({
name: this.name,
query,
params: z.any(),
schema: this.schema,
settings: this.settings,
});
return queryFunction(params);
}
build(): { query: string; params: QueryParams } {
let query = this.baseQuery;
if (this.whereClauses.length > 0) {
query += " WHERE " + this.whereClauses.join(" AND ");
}
if (this.groupByClause) {
query += ` GROUP BY ${this.groupByClause}`;
}
if (this.orderByClause) {
query += ` ORDER BY ${this.orderByClause}`;
}
if (this.limitClause) {
query += ` ${this.limitClause}`;
}
return { query, params: this.params };
}
}
internal-packages/clickhouse/src/client/types.ts
+29
View File
@@ -3,6 +3,7 @@ import type { z } from "zod";
import type { InsertError, QueryError } from "./errors.js";
import { ClickHouseSettings } from "@clickhouse/client";
import type { BaseQueryParams, InsertResult } from "@clickhouse/client";
import { ClickhouseQueryBuilder } from "./queryBuilder.js";
export type ClickhouseQueryFunction<TInput, TOutput> = (
params: TInput,
@@ -12,6 +13,10 @@ export type ClickhouseQueryFunction<TInput, TOutput> = (
}
) => Promise<Result<TOutput[], QueryError>>;
export type ClickhouseQueryBuilderFunction<TOutput> = (options?: {
settings?: ClickHouseSettings;
}) => ClickhouseQueryBuilder<TOutput>;
export interface ClickhouseReader {
query<TIn extends z.ZodSchema<any>, TOut extends z.ZodSchema<any>>(req: {
/**
@@ -42,6 +47,30 @@ export interface ClickhouseReader {
settings?: ClickHouseSettings;
}): ClickhouseQueryFunction<z.input<TIn>, z.output<TOut>>;
queryBuilder<TOut extends z.ZodSchema<any>>(req: {
/**
* The name of the operation.
* This will be used to identify the operation in the span.
*/
name: string;
/**
* The initial select clause
*
* @example SELECT run_id from trigger_dev.task_runs_v1
*/
baseQuery: string;
/**
* The schema of the output of each row
* Example: z.object({ id: z.string() })
*/
schema: TOut;
/**
* The settings to use for the query.
* These will be merged with the default settings.
*/
settings?: ClickHouseSettings;
}): ClickhouseQueryBuilderFunction<z.input<TOut>>;
close(): Promise<void>;
}
internal-packages/clickhouse/src/index.ts
+14 -1
View File
@@ -2,7 +2,15 @@ import { ClickHouseSettings } from "@clickhouse/client";
import { ClickhouseClient } from "./client/client.js";
import { ClickhouseReader, ClickhouseWriter } from "./client/types.js";
import { NoopClient } from "./client/noop.js";
import { insertTaskRuns, insertRawTaskRunPayloads } from "./taskRuns.js";
import {
insertTaskRuns,
insertRawTaskRunPayloads,
getTaskRunsQueryBuilder,
getTaskActivityQueryBuilder,
getCurrentRunningStats,
getAverageDurations,
getTaskUsageByOrganization,
} from "./taskRuns.js";
import { Logger, type LogLevel } from "@trigger.dev/core/logger";
import type { Agent as HttpAgent } from "http";
import type { Agent as HttpsAgent } from "https";
@@ -135,6 +143,11 @@ export class ClickHouse {
return {
insert: insertTaskRuns(this.writer),
insertPayloads: insertRawTaskRunPayloads(this.writer),
queryBuilder: getTaskRunsQueryBuilder(this.reader),
getTaskActivity: getTaskActivityQueryBuilder(this.reader),
getCurrentRunningStats: getCurrentRunningStats(this.reader),
getAverageDurations: getAverageDurations(this.reader),
getTaskUsageByOrganization: getTaskUsageByOrganization(this.reader),
};
}
}
internal-packages/clickhouse/src/taskRuns.test.ts
+95 -1
View File
@@ -1,7 +1,7 @@
import { clickhouseTest } from "@internal/testcontainers";
import { z } from "zod";
import { ClickhouseClient } from "./client/client.js";
import { insertRawTaskRunPayloads, insertTaskRuns } from "./taskRuns.js";
import { getTaskRunsQueryBuilder, insertRawTaskRunPayloads, insertTaskRuns } from "./taskRuns.js";
describe("Task Runs V2", () => {
clickhouseTest("should be able to insert task runs", async ({ clickhouseContainer }) => {
@@ -250,4 +250,98 @@ describe("Task Runs V2", () => {
])
);
});
clickhouseTest(
"should be able to query task runs using the query builder",
async ({ clickhouseContainer }) => {
const client = new ClickhouseClient({
name: "test",
url: clickhouseContainer.getConnectionUrl(),
});
const insert = insertTaskRuns(client, {
async_insert: 0, // turn off async insert for this test
});
const [insertError, insertResult] = await insert([
{
environment_id: "cm9kddfcs01zqdy88ld9mmrli",
organization_id: "cm8zs78wb0002dy616dg75tv3",
project_id: "cm9kddfbz01zpdy88t9dstecu",
run_id: "cma45oli70002qrdy47w0j4n7",
environment_type: "PRODUCTION",
friendly_id: "run_cma45oli70002qrdy47w0j4n7",
attempt: 1,
engine: "V2",
status: "PENDING",
task_identifier: "retry-task",
queue: "task/retry-task",
schedule_id: "",
batch_id: "",
root_run_id: "",
parent_run_id: "",
depth: 0,
span_id: "538677637f937f54",
trace_id: "20a28486b0b9f50c647b35e8863e36a5",
idempotency_key: "",
created_at: new Date("2025-04-30 16:34:04.312").getTime(),
updated_at: new Date("2025-04-30 16:34:04.312").getTime(),
started_at: null,
executed_at: null,
completed_at: null,
delay_until: null,
queued_at: new Date("2025-04-30 16:34:04.311").getTime(),
expired_at: null,
expiration_ttl: "",
usage_duration_ms: 0,
cost_in_cents: 0,
base_cost_in_cents: 0,
output: null,
error: null,
tags: [],
task_version: "",
sdk_version: "",
cli_version: "",
machine_preset: "",
is_test: true,
_version: "1",
},
]);
const queryBuilder = getTaskRunsQueryBuilder(client)();
queryBuilder.where("environment_id = {environmentId: String}", {
environmentId: "cm9kddfcs01zqdy88ld9mmrli",
});
expect(queryBuilder.build()).toMatchSnapshot();
const [queryError, result] = await queryBuilder.execute();
expect(queryError).toBeNull();
expect(result).toEqual(
expect.arrayContaining([
expect.objectContaining({
run_id: "cma45oli70002qrdy47w0j4n7",
}),
])
);
const queryBuilder2 = getTaskRunsQueryBuilder(client)();
queryBuilder2
.where("environment_id = {environmentId: String}", {
environmentId: "cm9kddfcs01zqdy88ld9mmrli",
})
.whereIf(true, "status = {status: String}", {
status: "COMPLETED_SUCCESSFULLY",
});
expect(queryBuilder2.build()).toMatchSnapshot();
const [queryError2, result2] = await queryBuilder2.execute();
expect(queryError2).toBeNull();
expect(result2).toEqual([]);
}
);
});
internal-packages/clickhouse/src/taskRuns.ts
+180 -1
View File
@@ -1,6 +1,6 @@
import { ClickHouseSettings } from "@clickhouse/client";
import { z } from "zod";
import { ClickhouseWriter } from "./client/types.js";
import { ClickhouseReader, ClickhouseWriter } from "./client/types.js";
export const TaskRunV2 = z.object({
environment_id: z.string(),
@@ -87,3 +87,182 @@ export function insertRawTaskRunPayloads(ch: ClickhouseWriter, settings?: ClickH
},
});
}
export const TaskRunV2QueryResult = z.object({
run_id: z.string(),
});
export type TaskRunV2QueryResult = z.infer<typeof TaskRunV2QueryResult>;
export function getTaskRunsQueryBuilder(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.queryBuilder({
name: "getTaskRuns",
baseQuery: "SELECT run_id FROM trigger_dev.task_runs_v2 FINAL",
schema: TaskRunV2QueryResult,
settings,
});
}
export const TaskActivityQueryResult = z.object({
task_identifier: z.string(),
status: z.string(),
day: z.string(),
count: z.number().int(),
});
export type TaskActivityQueryResult = z.infer<typeof TaskActivityQueryResult>;
export const TaskActivityQueryParams = z.object({
environmentId: z.string(),
days: z.number().int(),
});
export function getTaskActivityQueryBuilder(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.query({
name: "getTaskActivity",
query: `
SELECT
task_identifier,
status,
toDate(created_at) as day,
count() as count
FROM trigger_dev.task_runs_v2 FINAL
WHERE
environment_id = {environmentId: String}
AND created_at >= today() - {days: Int64}
AND _is_deleted = 0
GROUP BY
task_identifier,
status,
day
ORDER BY
task_identifier ASC,
day ASC,
status ASC
`,
schema: TaskActivityQueryResult,
params: TaskActivityQueryParams,
settings,
});
}
export const CurrentRunningStatsQueryResult = z.object({
task_identifier: z.string(),
status: z.string(),
count: z.number().int(),
});
export type CurrentRunningStatsQueryResult = z.infer<typeof CurrentRunningStatsQueryResult>;
export const CurrentRunningStatsQueryParams = z.object({
environmentId: z.string(),
days: z.number().int(),
});
export function getCurrentRunningStats(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.query({
name: "getCurrentRunningStats",
query: `
SELECT
task_identifier,
status,
count() as count
FROM trigger_dev.task_runs_v2 FINAL
WHERE
environment_id = {environmentId: String}
AND status IN ('PENDING', 'WAITING_FOR_DEPLOY', 'WAITING_TO_RESUME', 'QUEUED', 'EXECUTING')
AND _is_deleted = 0
AND created_at >= now() - INTERVAL {days: Int64} DAY
GROUP BY
task_identifier,
status
ORDER BY
task_identifier ASC
`,
schema: CurrentRunningStatsQueryResult,
params: CurrentRunningStatsQueryParams,
settings,
});
}
export const AverageDurationsQueryResult = z.object({
task_identifier: z.string(),
duration: z.number(),
});
export type AverageDurationsQueryResult = z.infer<typeof AverageDurationsQueryResult>;
export const AverageDurationsQueryParams = z.object({
environmentId: z.string(),
days: z.number().int(),
});
export function getAverageDurations(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.query({
name: "getAverageDurations",
query: `
SELECT
task_identifier,
avg(toUnixTimestamp(completed_at) - toUnixTimestamp(started_at)) as duration
FROM trigger_dev.task_runs_v2 FINAL
WHERE
environment_id = {environmentId: String}
AND created_at >= today() - {days: Int64}
AND status IN ('COMPLETED_SUCCESSFULLY', 'COMPLETED_WITH_ERRORS')
AND started_at IS NOT NULL
AND completed_at IS NOT NULL
AND _is_deleted = 0
GROUP BY
task_identifier
`,
schema: AverageDurationsQueryResult,
params: AverageDurationsQueryParams,
settings,
});
}
export const TaskUsageByOrganizationQueryResult = z.object({
task_identifier: z.string(),
run_count: z.number(),
average_duration: z.number(),
total_duration: z.number(),
average_cost: z.number(),
total_cost: z.number(),
total_base_cost: z.number(),
});
export const TaskUsageByOrganizationQueryParams = z.object({
startTime: z.number().int(),
endTime: z.number().int(),
organizationId: z.string(),
});
export function getTaskUsageByOrganization(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.query({
name: "getTaskUsageByOrganization",
query: `
SELECT
task_identifier,
count() AS run_count,
avg(usage_duration_ms) AS average_duration,
sum(usage_duration_ms) AS total_duration,
avg(cost_in_cents) / 100.0 AS average_cost,
sum(cost_in_cents) / 100.0 AS total_cost,
sum(base_cost_in_cents) / 100.0 AS total_base_cost
FROM trigger_dev.task_runs_v2 FINAL
WHERE
environment_type != 'DEVELOPMENT'
AND created_at >= fromUnixTimestamp64Milli({startTime: Int64})
AND created_at < fromUnixTimestamp64Milli({endTime: Int64})
AND organization_id = {organizationId: String}
AND _is_deleted = 0
GROUP BY
task_identifier
ORDER BY
total_cost DESC
`,
schema: TaskUsageByOrganizationQueryResult,
params: TaskUsageByOrganizationQueryParams,
settings,
});
}
internal-packages/database/prisma/migrations/20250605220110_remove_unused_completed_at_idnex/migration.sql
+1
View File
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS "TaskRun_completedAt_idx";
internal-packages/database/prisma/migrations/20250605220552_remove_unused_schedule_instance_id_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX CONCURRENTLY IF EXISTS "TaskRun_scheduleInstanceId_idx";
internal-packages/database/prisma/migrations/20250606131749_remove_task_run_project_id_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX CONCURRENTLY IF EXISTS "TaskRun_projectId_idx";
internal-packages/database/prisma/migrations/20250606132144_remove_task_run_project_id_task_identifier_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX CONCURRENTLY IF EXISTS "TaskRun_projectId_taskIdentifier_idx";
internal-packages/database/prisma/migrations/20250606132316_remove_task_run_project_id_status_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX CONCURRENTLY IF EXISTS "TaskRun_projectId_status_idx";
internal-packages/database/prisma/migrations/20250606132630_remove_task_run_project_id_task_identifier_status_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX CONCURRENTLY IF EXISTS "TaskRun_projectId_taskIdentifier_status_idx";
internal-packages/database/prisma/migrations/20250606132928_remove_project_id_created_at_task_identifier_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX CONCURRENTLY IF EXISTS "TaskRun_projectId_createdAt_taskIdentifier_idx";
internal-packages/database/prisma/migrations/20250606133133_remove_task_run_project_id_id_desc_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX CONCURRENTLY IF EXISTS "TaskRun_projectId_id_idx";
internal-packages/database/prisma/migrations/20250609163214_add_task_run_environment_created_at_id_index/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- CreateIndex
CREATE INDEX CONCURRENTLY IF NOT EXISTS "TaskRun_runtimeEnvironmentId_createdAt_id_idx" ON "TaskRun"("runtimeEnvironmentId", "createdAt" DESC, "id" DESC);
internal-packages/database/prisma/migrations/20250609164201_add_task_run_run_tags_gin_index/migration.sql
+1
View File
@@ -0,0 +1 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS "TaskRun_runTags_idx" ON "TaskRun" USING GIN ("runTags" array_ops);
internal-packages/database/prisma/schema.prisma
+2 -11
View File
@@ -672,26 +672,17 @@ model TaskRun {
@@index([parentTaskRunId])
// Finding ancestor runs
@@index([rootTaskRunId])
// Task activity graph
@@index([projectId, createdAt, taskIdentifier])
//Runs list
@@index([projectId])
@@index([projectId, id(sort: Desc)])
@@index([projectId, taskIdentifier])
@@index([projectId, status])
@@index([projectId, taskIdentifier, status])
//Schedules
@@index([scheduleId])
@@index([scheduleInstanceId])
// Run page inspector
@@index([spanId])
@@index([parentSpanId])
// Finding completed runs
@@index([completedAt])
// Schedule list page
@@index([scheduleId, createdAt(sort: Desc)])
// Finding runs in a batch
@@index([runtimeEnvironmentId, batchId])
@@index([runtimeEnvironmentId, createdAt(sort: Desc), id(sort: Desc)])
@@index([runTags(ops: ArrayOps)], type: Gin)
}
enum TaskRunStatus {
packages/core/src/v3/apiClient/index.ts
+52 -16
View File
@@ -66,6 +66,7 @@ import {
SSEStreamSubscriptionFactory,
TaskRunShape,
runShapeStream,
RealtimeRunSkipColumns,
} from "./runStream.js";
import {
CreateEnvironmentVariableParams,
@@ -88,6 +89,7 @@ export type {
ImportEnvironmentVariablesParams,
SubscribeToRunsQueryParams,
UpdateEnvironmentVariableParams,
RealtimeRunSkipColumns,
};
export type ClientTriggerOptions = {
@@ -890,24 +892,37 @@ export class ApiClient {
signal?: AbortSignal;
closeOnComplete?: boolean;
onFetchError?: (error: Error) => void;
skipColumns?: string[];
}
) {
return runShapeStream<TRunTypes>(`${this.baseUrl}/realtime/v1/runs/${runId}`, {
closeOnComplete:
typeof options?.closeOnComplete === "boolean" ? options.closeOnComplete : true,
headers: this.#getRealtimeHeaders(),
client: this,
signal: options?.signal,
onFetchError: options?.onFetchError,
});
const queryParams = new URLSearchParams();
if (options?.skipColumns) {
queryParams.append("skipColumns", options.skipColumns.join(","));
}
return runShapeStream<TRunTypes>(
`${this.baseUrl}/realtime/v1/runs/${runId}${queryParams ? `?${queryParams}` : ""}`,
{
closeOnComplete:
typeof options?.closeOnComplete === "boolean" ? options.closeOnComplete : true,
headers: this.#getRealtimeHeaders(),
client: this,
signal: options?.signal,
onFetchError: options?.onFetchError,
}
);
}
subscribeToRunsWithTag<TRunTypes extends AnyRunTypes>(
tag: string | string[],
filters?: { createdAt?: string; skipColumns?: string[] },
options?: { signal?: AbortSignal; onFetchError?: (error: Error) => void }
) {
const searchParams = createSearchQueryForSubscribeToRuns({
tags: tag,
...(filters ? { createdAt: filters.createdAt } : {}),
...(filters?.skipColumns ? { skipColumns: filters.skipColumns } : {}),
});
return runShapeStream<TRunTypes>(
@@ -924,15 +939,28 @@ export class ApiClient {
subscribeToBatch<TRunTypes extends AnyRunTypes>(
batchId: string,
options?: { signal?: AbortSignal; onFetchError?: (error: Error) => void }
options?: {
signal?: AbortSignal;
onFetchError?: (error: Error) => void;
skipColumns?: string[];
}
) {
return runShapeStream<TRunTypes>(`${this.baseUrl}/realtime/v1/batches/${batchId}`, {
closeOnComplete: false,
headers: this.#getRealtimeHeaders(),
client: this,
signal: options?.signal,
onFetchError: options?.onFetchError,
});
const queryParams = new URLSearchParams();
if (options?.skipColumns) {
queryParams.append("skipColumns", options.skipColumns.join(","));
}
return runShapeStream<TRunTypes>(
`${this.baseUrl}/realtime/v1/batches/${batchId}${queryParams ? `?${queryParams}` : ""}`,
{
closeOnComplete: false,
headers: this.#getRealtimeHeaders(),
client: this,
signal: options?.signal,
onFetchError: options?.onFetchError,
}
);
}
async fetchStream<T>(
@@ -1043,6 +1071,14 @@ function createSearchQueryForSubscribeToRuns(query?: SubscribeToRunsQueryParams)
if (query.tags) {
searchParams.append("tags", Array.isArray(query.tags) ? query.tags.join(",") : query.tags);
}
if (query.createdAt) {
searchParams.append("createdAt", query.createdAt);
}
if (query.skipColumns) {
searchParams.append("skipColumns", query.skipColumns.join(","));
}
}
return searchParams;
packages/core/src/v3/apiClient/runStream.ts
+28 -7
View File
@@ -55,6 +55,27 @@ export type TaskRunShape<TTask extends AnyTask> = RunShape<InferRunTypes<TTask>>
export type RealtimeRun<TTask extends AnyTask> = TaskRunShape<TTask>;
export type AnyRealtimeRun = RealtimeRun<AnyTask>;
export type RealtimeRunSkipColumns = Array<
| "startedAt"
| "delayUntil"
| "queuedAt"
| "expiredAt"
| "completedAt"
| "number"
| "isTest"
| "usageDurationMs"
| "costInCents"
| "baseCostInCents"
| "ttl"
| "payload"
| "payloadType"
| "metadata"
| "output"
| "outputType"
| "runTags"
| "error"
>;
export type RunStreamCallback<TRunTypes extends AnyRunTypes> = (
run: RunShape<TRunTypes>
) => void | Promise<void>;
@@ -395,16 +416,16 @@ export class RunSubscription<TRunTypes extends AnyRunTypes> {
return {
id: row.friendlyId,
payload,
output,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
taskIdentifier: row.taskIdentifier,
number: row.number,
status: apiStatusFromRunStatus(row.status),
durationMs: row.usageDurationMs,
costInCents: row.costInCents,
baseCostInCents: row.baseCostInCents,
payload,
output,
number: row.number ?? 0,
durationMs: row.usageDurationMs ?? 0,
costInCents: row.costInCents ?? 0,
baseCostInCents: row.baseCostInCents ?? 0,
tags: row.runTags ?? [],
idempotencyKey: row.idempotencyKey ?? undefined,
expiredAt: row.expiredAt ?? undefined,
@@ -413,7 +434,7 @@ export class RunSubscription<TRunTypes extends AnyRunTypes> {
delayedUntil: row.delayUntil ?? undefined,
queuedAt: row.queuedAt ?? undefined,
error: row.error ? createJsonErrorObject(row.error) : undefined,
isTest: row.isTest,
isTest: row.isTest ?? false,
metadata,
} as RunShape<TRunTypes>;
}
packages/core/src/v3/apiClient/stream.ts
+6 -9
View File
@@ -102,19 +102,16 @@ class ReadableShapeStream<T extends Row<unknown> = Row> {
// Create the source stream that will receive messages
const source = new ReadableStream<Message<T>[]>({
start: (controller) => {
this.#unsubscribe = this.#stream.subscribe(
(messages) => {
if (!this.#isStreamClosed) {
controller.enqueue(messages);
}
},
this.#handleError.bind(this)
);
this.#unsubscribe = this.#stream.subscribe((messages) => {
if (!this.#isStreamClosed) {
controller.enqueue(messages);
}
}, this.#handleError.bind(this));
},
cancel: () => {
this.#isStreamClosed = true;
this.#unsubscribe?.();
}
},
});
// Create the transformed stream that processes messages and emits complete rows
packages/core/src/v3/apiClient/types.ts
+2
View File
@@ -41,6 +41,8 @@ export interface ListProjectRunsQueryParams extends CursorPageParams, ListRunsQu
export interface SubscribeToRunsQueryParams {
tasks?: Array<string> | string;
tags?: Array<string> | string;
createdAt?: string;
skipColumns?: string[];
}
export interface ListWaitpointTokensQueryParams extends CursorPageParams {
packages/core/src/v3/isomorphic/duration.ts
+139 -35
View File
@@ -1,55 +1,159 @@
export function parseNaturalLanguageDuration(duration: string): Date | undefined {
const regexPattern = /^(\d+w)?(\d+d)?(\d+h)?(\d+m)?(\d+s)?$/;
// Handle Code scanning alert #44 (https://github.com/triggerdotdev/trigger.dev/security/code-scanning/44) by limiting the length of the input string
if (duration.length > 100) {
return undefined;
}
const result: Date = new Date();
// More flexible regex that captures all units individually regardless of order
const weekMatch = duration.match(/(\d+)w/);
const dayMatch = duration.match(/(\d+)d/);
const hourMatch = duration.match(/(\d+)(?:hr|h)/);
const minuteMatch = duration.match(/(\d+)m/);
const secondMatch = duration.match(/(\d+)s/);
// Check if the entire string consists only of valid duration units
const validPattern = /^(\d+(?:w|d|hr|h|m|s))+$/;
if (!validPattern.test(duration)) {
return undefined;
}
let totalMilliseconds = 0;
let hasMatch = false;
const elements = duration.match(regexPattern);
if (elements) {
if (elements[1]) {
const weeks = Number(elements[1].slice(0, -1));
if (weeks >= 0) {
result.setDate(result.getDate() + 7 * weeks);
hasMatch = true;
}
if (weekMatch) {
const weeks = Number(weekMatch[1]);
if (weeks >= 0) {
totalMilliseconds += weeks * 7 * 24 * 60 * 60 * 1000;
hasMatch = true;
}
if (elements[2]) {
const days = Number(elements[2].slice(0, -1));
if (days >= 0) {
result.setDate(result.getDate() + days);
hasMatch = true;
}
}
if (dayMatch) {
const days = Number(dayMatch[1]);
if (days >= 0) {
totalMilliseconds += days * 24 * 60 * 60 * 1000;
hasMatch = true;
}
if (elements[3]) {
const hours = Number(elements[3].slice(0, -1));
if (hours >= 0) {
result.setHours(result.getHours() + hours);
hasMatch = true;
}
}
if (hourMatch) {
const hours = Number(hourMatch[1]);
if (hours >= 0) {
totalMilliseconds += hours * 60 * 60 * 1000;
hasMatch = true;
}
if (elements[4]) {
const minutes = Number(elements[4].slice(0, -1));
if (minutes >= 0) {
result.setMinutes(result.getMinutes() + minutes);
hasMatch = true;
}
}
if (minuteMatch) {
const minutes = Number(minuteMatch[1]);
if (minutes >= 0) {
totalMilliseconds += minutes * 60 * 1000;
hasMatch = true;
}
if (elements[5]) {
const seconds = Number(elements[5].slice(0, -1));
if (seconds >= 0) {
result.setSeconds(result.getSeconds() + seconds);
hasMatch = true;
}
}
if (secondMatch) {
const seconds = Number(secondMatch[1]);
if (seconds >= 0) {
totalMilliseconds += seconds * 1000;
hasMatch = true;
}
}
if (hasMatch) {
return result;
return new Date(Date.now() + totalMilliseconds);
}
return undefined;
}
export function safeParseNaturalLanguageDuration(duration: string): Date | undefined {
try {
return parseNaturalLanguageDuration(duration);
} catch (error) {
return undefined;
}
}
// ... existing code ...
export function parseNaturalLanguageDurationAgo(duration: string): Date | undefined {
// Handle Code scanning alert #44 (https://github.com/triggerdotdev/trigger.dev/security/code-scanning/44) by limiting the length of the input string
if (duration.length > 100) {
return undefined;
}
// More flexible regex that captures all units individually regardless of order
const weekMatch = duration.match(/(\d+)w/);
const dayMatch = duration.match(/(\d+)d/);
const hourMatch = duration.match(/(\d+)(?:hr|h)/);
const minuteMatch = duration.match(/(\d+)m/);
const secondMatch = duration.match(/(\d+)s/);
// Check if the entire string consists only of valid duration units
const validPattern = /^(\d+(?:w|d|hr|h|m|s))+$/;
if (!validPattern.test(duration)) {
return undefined;
}
let totalMilliseconds = 0;
let hasMatch = false;
if (weekMatch) {
const weeks = Number(weekMatch[1]);
if (weeks >= 0) {
totalMilliseconds += weeks * 7 * 24 * 60 * 60 * 1000;
hasMatch = true;
}
}
if (dayMatch) {
const days = Number(dayMatch[1]);
if (days >= 0) {
totalMilliseconds += days * 24 * 60 * 60 * 1000;
hasMatch = true;
}
}
if (hourMatch) {
const hours = Number(hourMatch[1]);
if (hours >= 0) {
totalMilliseconds += hours * 60 * 60 * 1000;
hasMatch = true;
}
}
if (minuteMatch) {
const minutes = Number(minuteMatch[1]);
if (minutes >= 0) {
totalMilliseconds += minutes * 60 * 1000;
hasMatch = true;
}
}
if (secondMatch) {
const seconds = Number(secondMatch[1]);
if (seconds >= 0) {
totalMilliseconds += seconds * 1000;
hasMatch = true;
}
}
if (hasMatch) {
return new Date(Date.now() - totalMilliseconds);
}
return undefined;
}
export function safeParseNaturalLanguageDurationAgo(duration: string): Date | undefined {
try {
return parseNaturalLanguageDurationAgo(duration);
} catch (error) {
return undefined;
}
}
export function stringifyDuration(seconds: number): string | undefined {
if (seconds <= 0) {
return;
packages/core/src/v3/schemas/api.ts
+9 -9
View File
@@ -870,7 +870,9 @@ const RawOptionalShapeDate = z
export const SubscribeRunRawShape = z.object({
id: z.string(),
idempotencyKey: z.string().nullish(),
taskIdentifier: z.string(),
friendlyId: z.string(),
status: z.string(),
createdAt: RawShapeDate,
updatedAt: RawShapeDate,
startedAt: RawOptionalShapeDate,
@@ -878,14 +880,12 @@ export const SubscribeRunRawShape = z.object({
queuedAt: RawOptionalShapeDate,
expiredAt: RawOptionalShapeDate,
completedAt: RawOptionalShapeDate,
taskIdentifier: z.string(),
friendlyId: z.string(),
number: z.number(),
isTest: z.boolean(),
status: z.string(),
usageDurationMs: z.number(),
costInCents: z.number(),
baseCostInCents: z.number(),
idempotencyKey: z.string().nullish(),
number: z.number().default(0),
isTest: z.boolean().default(false),
usageDurationMs: z.number().default(0),
costInCents: z.number().default(0),
baseCostInCents: z.number().default(0),
ttl: z.string().nullish(),
payload: z.string().nullish(),
payloadType: z.string().nullish(),
packages/core/test/duration.test.ts
+374
View File
@@ -0,0 +1,374 @@
import {
parseNaturalLanguageDuration,
safeParseNaturalLanguageDuration,
parseNaturalLanguageDurationAgo,
safeParseNaturalLanguageDurationAgo,
stringifyDuration,
} from "../src/v3/isomorphic/duration.js";
describe("parseNaturalLanguageDuration", () => {
let baseTime: Date;
beforeEach(() => {
// Set a fixed base time for consistent testing
baseTime = new Date("2024-01-01T12:00:00.000Z");
vi.setSystemTime(baseTime);
});
afterEach(() => {
vi.useRealTimers();
});
describe("valid duration strings", () => {
it("parses seconds correctly", () => {
const result = parseNaturalLanguageDuration("30s");
expect(result).toEqual(new Date("2024-01-01T12:00:30.000Z"));
});
it("parses minutes correctly", () => {
const result = parseNaturalLanguageDuration("15m");
expect(result).toEqual(new Date("2024-01-01T12:15:00.000Z"));
});
it("parses hours correctly", () => {
const result = parseNaturalLanguageDuration("2h");
expect(result).toEqual(new Date("2024-01-01T14:00:00.000Z"));
});
it("parses hours with 'hr' format correctly", () => {
const result = parseNaturalLanguageDuration("2hr");
expect(result).toEqual(new Date("2024-01-01T14:00:00.000Z"));
});
it("parses days correctly", () => {
const result = parseNaturalLanguageDuration("3d");
expect(result).toEqual(new Date("2024-01-04T12:00:00.000Z"));
});
it("parses weeks correctly", () => {
const result = parseNaturalLanguageDuration("1w");
expect(result).toEqual(new Date("2024-01-08T12:00:00.000Z"));
});
it("parses combined durations correctly", () => {
const result = parseNaturalLanguageDuration("1w2d3h4m5s");
expect(result).toEqual(new Date("2024-01-10T15:04:05.000Z"));
});
it("parses combined durations with 'hr' format correctly", () => {
const result = parseNaturalLanguageDuration("1w2d3hr4m5s");
expect(result).toEqual(new Date("2024-01-10T15:04:05.000Z"));
});
it("parses partial combined durations correctly", () => {
const result = parseNaturalLanguageDuration("2d30m");
expect(result).toEqual(new Date("2024-01-03T12:30:00.000Z"));
});
it("parses durations with units in any order", () => {
const result = parseNaturalLanguageDuration("30s2h1d");
expect(result).toEqual(new Date("2024-01-02T14:00:30.000Z"));
});
it("parses durations with 'hr' in any order", () => {
const result = parseNaturalLanguageDuration("30s2hr1d");
expect(result).toEqual(new Date("2024-01-02T14:00:30.000Z"));
});
it("handles zero values", () => {
const result = parseNaturalLanguageDuration("0s");
expect(result).toEqual(new Date("2024-01-01T12:00:00.000Z"));
});
it("handles large numbers", () => {
const result = parseNaturalLanguageDuration("100d");
expect(result).toEqual(new Date("2024-04-10T12:00:00.000Z"));
});
});
describe("invalid duration strings", () => {
it("returns undefined for empty string", () => {
const result = parseNaturalLanguageDuration("");
expect(result).toBeUndefined();
});
it("returns undefined for invalid format", () => {
const result = parseNaturalLanguageDuration("invalid");
expect(result).toBeUndefined();
});
it("returns undefined for negative numbers", () => {
const result = parseNaturalLanguageDuration("-1h");
expect(result).toBeUndefined();
});
it("returns undefined for invalid units", () => {
const result = parseNaturalLanguageDuration("1x");
expect(result).toBeUndefined();
});
it("returns undefined for mixed valid/invalid format", () => {
const result = parseNaturalLanguageDuration("1h2x");
expect(result).toBeUndefined();
});
it("returns undefined for decimal numbers", () => {
const result = parseNaturalLanguageDuration("1.5h");
expect(result).toBeUndefined();
});
it("returns undefined for units without numbers", () => {
const result = parseNaturalLanguageDuration("h");
expect(result).toBeUndefined();
});
});
});
describe("safeParseNaturalLanguageDuration", () => {
let baseTime: Date;
beforeEach(() => {
baseTime = new Date("2024-01-01T12:00:00.000Z");
vi.setSystemTime(baseTime);
});
afterEach(() => {
vi.useRealTimers();
});
it("returns the same result as parseNaturalLanguageDuration for valid input", () => {
const duration = "1h30m";
const result1 = parseNaturalLanguageDuration(duration);
const result2 = safeParseNaturalLanguageDuration(duration);
expect(result1).toEqual(result2);
});
it("returns undefined for invalid input without throwing", () => {
const result = safeParseNaturalLanguageDuration("invalid");
expect(result).toBeUndefined();
});
it("handles exceptions gracefully", () => {
// Mock parseNaturalLanguageDuration to throw an error
const originalParse = parseNaturalLanguageDuration;
const mockParse = vi.fn().mockImplementation(() => {
throw new Error("Test error");
});
// This test demonstrates the safe wrapper behavior
expect(() => safeParseNaturalLanguageDuration("1h")).not.toThrow();
});
});
describe("parseNaturalLanguageDurationAgo", () => {
let baseTime: Date;
beforeEach(() => {
baseTime = new Date("2024-01-01T12:00:00.000Z");
vi.setSystemTime(baseTime);
});
afterEach(() => {
vi.useRealTimers();
});
describe("valid duration strings", () => {
it("parses seconds ago correctly", () => {
const result = parseNaturalLanguageDurationAgo("30s");
expect(result).toEqual(new Date("2024-01-01T11:59:30.000Z"));
});
it("parses minutes ago correctly", () => {
const result = parseNaturalLanguageDurationAgo("15m");
expect(result).toEqual(new Date("2024-01-01T11:45:00.000Z"));
});
it("parses hours ago correctly", () => {
const result = parseNaturalLanguageDurationAgo("2h");
expect(result).toEqual(new Date("2024-01-01T10:00:00.000Z"));
});
it("parses hours ago with 'hr' format correctly", () => {
const result = parseNaturalLanguageDurationAgo("2hr");
expect(result).toEqual(new Date("2024-01-01T10:00:00.000Z"));
});
it("parses days ago correctly", () => {
const result = parseNaturalLanguageDurationAgo("3d");
expect(result).toEqual(new Date("2023-12-29T12:00:00.000Z"));
});
it("parses weeks ago correctly", () => {
const result = parseNaturalLanguageDurationAgo("1w");
expect(result).toEqual(new Date("2023-12-25T12:00:00.000Z"));
});
it("parses combined durations ago correctly", () => {
const result = parseNaturalLanguageDurationAgo("1w2d3h4m5s");
expect(result).toEqual(new Date("2023-12-23T08:55:55.000Z"));
});
it("parses combined durations ago with 'hr' format correctly", () => {
const result = parseNaturalLanguageDurationAgo("1w2d3hr4m5s");
expect(result).toEqual(new Date("2023-12-23T08:55:55.000Z"));
});
it("parses partial combined durations ago correctly", () => {
const result = parseNaturalLanguageDurationAgo("2d30m");
expect(result).toEqual(new Date("2023-12-30T11:30:00.000Z"));
});
it("handles zero values", () => {
const result = parseNaturalLanguageDurationAgo("0s");
expect(result).toEqual(new Date("2024-01-01T12:00:00.000Z"));
});
it("handles large numbers in the past", () => {
const result = parseNaturalLanguageDurationAgo("100d");
expect(result).toEqual(new Date("2023-09-23T12:00:00.000Z"));
});
});
describe("invalid duration strings", () => {
it("returns undefined for empty string", () => {
const result = parseNaturalLanguageDurationAgo("");
expect(result).toBeUndefined();
});
it("returns undefined for invalid format", () => {
const result = parseNaturalLanguageDurationAgo("invalid");
expect(result).toBeUndefined();
});
it("returns undefined for negative numbers", () => {
const result = parseNaturalLanguageDurationAgo("-1h");
expect(result).toBeUndefined();
});
it("returns undefined for invalid units", () => {
const result = parseNaturalLanguageDurationAgo("1x");
expect(result).toBeUndefined();
});
});
});
describe("safeParseNaturalLanguageDurationAgo", () => {
let baseTime: Date;
beforeEach(() => {
baseTime = new Date("2024-01-01T12:00:00.000Z");
vi.setSystemTime(baseTime);
});
afterEach(() => {
vi.useRealTimers();
});
it("returns the same result as parseNaturalLanguageDurationAgo for valid input", () => {
const duration = "1h30m";
const result1 = parseNaturalLanguageDurationAgo(duration);
const result2 = safeParseNaturalLanguageDurationAgo(duration);
expect(result1).toEqual(result2);
});
it("returns undefined for invalid input without throwing", () => {
const result = safeParseNaturalLanguageDurationAgo("invalid");
expect(result).toBeUndefined();
});
it("handles exceptions gracefully", () => {
expect(() => safeParseNaturalLanguageDurationAgo("1h")).not.toThrow();
});
});
describe("stringifyDuration", () => {
it("returns undefined for zero or negative seconds", () => {
expect(stringifyDuration(0)).toBeUndefined();
expect(stringifyDuration(-1)).toBeUndefined();
});
it("formats seconds correctly", () => {
expect(stringifyDuration(30)).toBe("30s");
});
it("formats minutes correctly", () => {
expect(stringifyDuration(90)).toBe("1m30s");
});
it("formats hours correctly", () => {
expect(stringifyDuration(3661)).toBe("1h1m1s");
});
it("formats days correctly", () => {
expect(stringifyDuration(90061)).toBe("1d1h1m1s");
});
it("formats weeks correctly", () => {
expect(stringifyDuration(694861)).toBe("1w1d1h1m1s");
});
it("omits zero units", () => {
expect(stringifyDuration(3600)).toBe("1h");
expect(stringifyDuration(3660)).toBe("1h1m");
expect(stringifyDuration(86400)).toBe("1d");
});
it("handles large durations", () => {
expect(stringifyDuration(1209600)).toBe("2w"); // 2 weeks
});
it("handles complex durations", () => {
expect(stringifyDuration(694861)).toBe("1w1d1h1m1s");
});
});
describe("duration function consistency", () => {
let baseTime: Date;
beforeEach(() => {
baseTime = new Date("2024-01-01T12:00:00.000Z");
vi.setSystemTime(baseTime);
});
afterEach(() => {
vi.useRealTimers();
});
it("parseNaturalLanguageDuration and parseNaturalLanguageDurationAgo should be symmetric", () => {
const duration = "1h30m15s";
const futureDate = parseNaturalLanguageDuration(duration);
const pastDate = parseNaturalLanguageDurationAgo(duration);
expect(futureDate).toBeDefined();
expect(pastDate).toBeDefined();
if (futureDate && pastDate) {
const now = new Date();
const futureOffset = futureDate.getTime() - now.getTime();
const pastOffset = now.getTime() - pastDate.getTime();
expect(futureOffset).toBe(pastOffset);
}
});
it("safe versions should match unsafe versions for valid input", () => {
const duration = "2d4h";
const future1 = parseNaturalLanguageDuration(duration);
const future2 = safeParseNaturalLanguageDuration(duration);
const past1 = parseNaturalLanguageDurationAgo(duration);
const past2 = safeParseNaturalLanguageDurationAgo(duration);
expect(future1).toEqual(future2);
expect(past1).toEqual(past2);
});
it("safe versions should return undefined for invalid input", () => {
const invalidDuration = "invalid-duration";
expect(parseNaturalLanguageDuration(invalidDuration)).toBeUndefined();
expect(safeParseNaturalLanguageDuration(invalidDuration)).toBeUndefined();
expect(parseNaturalLanguageDurationAgo(invalidDuration)).toBeUndefined();
expect(safeParseNaturalLanguageDurationAgo(invalidDuration)).toBeUndefined();
});
});
packages/react-hooks/src/hooks/useRealtime.ts
+52 -3
View File
@@ -1,6 +1,12 @@
"use client";
import { AnyTask, ApiClient, InferRunTypes, RealtimeRun } from "@trigger.dev/core/v3";
import {
AnyTask,
ApiClient,
InferRunTypes,
RealtimeRun,
RealtimeRunSkipColumns,
} from "@trigger.dev/core/v3";
import { useCallback, useEffect, useId, useRef, useState } from "react";
import { KeyedMutator, useSWR } from "../utils/trigger-swr.js";
import { useApiClient, UseApiClientOptions } from "./useApiClient.js";
@@ -29,6 +35,13 @@ export type UseRealtimeSingleRunOptions<TTask extends AnyTask = AnyTask> = UseRe
* Set this to false if you are making updates to the run metadata after completion through child runs
*/
stopOnCompletion?: boolean;
/**
* Skip columns from the subscription.
*
* @default []
*/
skipColumns?: RealtimeRunSkipColumns;
};
export type UseRealtimeRunInstance<TTask extends AnyTask = AnyTask> = {
@@ -101,6 +114,7 @@ export function useRealtimeRun<TTask extends AnyTask>(
await processRealtimeRun(
runId,
{ skipColumns: options?.skipColumns },
apiClient,
mutateRun,
setError,
@@ -261,6 +275,7 @@ export function useRealtimeRunWithStreams<
await processRealtimeRunWithStreams(
runId,
{ skipColumns: options?.skipColumns },
apiClient,
mutateRun,
mutateStreams,
@@ -334,6 +349,32 @@ export type UseRealtimeRunsInstance<TTask extends AnyTask = AnyTask> = {
stop: () => void;
};
export type UseRealtimeRunsWithTagOptions = UseRealtimeRunOptions & {
/**
* Filter runs by the time they were created. You must specify the duration string like "1h", "10s", "30m", etc.
*
* @example
* "1h" - 1 hour ago
* "10s" - 10 seconds ago
* "30m" - 30 minutes ago
* "1d" - 1 day ago
* "1w" - 1 week ago
*
* The maximum duration is 1 week
*
* @note The timestamp will be calculated on the server side when you first subscribe to the runs.
*
*/
createdAt?: string;
/**
* Skip columns from the subscription.
*
* @default []
*/
skipColumns?: RealtimeRunSkipColumns;
};
/**
* Hook to subscribe to realtime updates of task runs filtered by tag(s).
*
@@ -348,11 +389,13 @@ export type UseRealtimeRunsInstance<TTask extends AnyTask = AnyTask> = {
* const { runs, error } = useRealtimeRunsWithTag<typeof myTask>('my-tag');
* // Or with multiple tags
* const { runs, error } = useRealtimeRunsWithTag<typeof myTask>(['tag1', 'tag2']);
* // Or with a createdAt filter
* const { runs, error } = useRealtimeRunsWithTag<typeof myTask>('my-tag', { createdAt: '1h' });
* ```
*/
export function useRealtimeRunsWithTag<TTask extends AnyTask>(
tag: string | string[],
options?: UseRealtimeRunOptions
options?: UseRealtimeRunsWithTagOptions
): UseRealtimeRunsInstance<TTask> {
const hookId = useId();
const idKey = options?.id ?? hookId;
@@ -396,6 +439,7 @@ export function useRealtimeRunsWithTag<TTask extends AnyTask>(
await processRealtimeRunsWithTag(
tag,
{ createdAt: options?.createdAt, skipColumns: options?.skipColumns },
apiClient,
mutateRuns,
runsRef,
@@ -568,13 +612,14 @@ function insertRunShapeInOrder<TTask extends AnyTask>(
async function processRealtimeRunsWithTag<TTask extends AnyTask = AnyTask>(
tag: string | string[],
filters: { createdAt?: string; skipColumns?: RealtimeRunSkipColumns },
apiClient: ApiClient,
mutateRunsData: KeyedMutator<RealtimeRun<TTask>[]>,
existingRunsRef: React.MutableRefObject<RealtimeRun<TTask>[]>,
onError: (e: Error) => void,
abortControllerRef: React.MutableRefObject<AbortController | null>
) {
const subscription = apiClient.subscribeToRunsWithTag<InferRunTypes<TTask>>(tag, {
const subscription = apiClient.subscribeToRunsWithTag<InferRunTypes<TTask>>(tag, filters, {
signal: abortControllerRef.current?.signal,
onFetchError: onError,
});
@@ -610,6 +655,7 @@ async function processRealtimeRunWithStreams<
TStreams extends Record<string, any> = Record<string, any>,
>(
runId: string,
filters: { skipColumns?: RealtimeRunSkipColumns },
apiClient: ApiClient,
mutateRunData: KeyedMutator<RealtimeRun<TTask>>,
mutateStreamData: KeyedMutator<StreamResults<TStreams>>,
@@ -623,6 +669,7 @@ async function processRealtimeRunWithStreams<
signal: abortControllerRef.current?.signal,
closeOnComplete: stopOnCompletion,
onFetchError: onError,
skipColumns: filters.skipColumns,
});
type StreamUpdate = {
@@ -669,6 +716,7 @@ async function processRealtimeRunWithStreams<
async function processRealtimeRun<TTask extends AnyTask = AnyTask>(
runId: string,
filters: { skipColumns?: RealtimeRunSkipColumns },
apiClient: ApiClient,
mutateRunData: KeyedMutator<RealtimeRun<TTask>>,
onError: (e: Error) => void,
@@ -679,6 +727,7 @@ async function processRealtimeRun<TTask extends AnyTask = AnyTask>(
signal: abortControllerRef.current?.signal,
closeOnComplete: stopOnCompletion,
onFetchError: onError,
skipColumns: filters.skipColumns,
});
for await (const part of subscription) {
packages/react-hooks/src/hooks/useTaskTrigger.ts
+8
View File
@@ -9,6 +9,7 @@ import {
RunHandleFromTypes,
stringifyIO,
type TriggerOptions,
type RealtimeRunSkipColumns,
} from "@trigger.dev/core/v3";
import useSWRMutation from "swr/mutation";
import { useApiClient, UseApiClientOptions } from "./useApiClient.js";
@@ -113,6 +114,13 @@ export type UseRealtimeTaskTriggerOptions = UseTaskTriggerOptions & {
enabled?: boolean;
/** Optional throttle time in milliseconds for stream updates */
experimental_throttleInMs?: number;
/**
* Skip columns from the subscription.
*
* @default []
*/
skipColumns?: RealtimeRunSkipColumns;
};
export type RealtimeTriggerInstanceWithStreams<
packages/trigger-sdk/src/v3/runs.ts
+46 -2
View File
@@ -15,6 +15,7 @@ import type {
AnyBatchedRunHandle,
AsyncIterableStream,
ApiPromise,
RealtimeRunSkipColumns,
} from "@trigger.dev/core/v3";
import {
CanceledRunResponse,
@@ -345,6 +346,18 @@ export type SubscribeToRunOptions = {
* Set this to false if you are making updates to the run metadata after completion through child runs
*/
stopOnCompletion?: boolean;
/**
* Skip columns from the subscription.
*
* @default []
*
* @example
* ```ts
* runs.subscribeToRun("123", { skipColumns: ["payload", "output"] });
* ```
*/
skipColumns?: RealtimeRunSkipColumns;
};
/**
@@ -389,9 +402,36 @@ function subscribeToRun<TRunId extends AnyRunHandle | AnyTask | string>(
return apiClient.subscribeToRun($runId, {
closeOnComplete:
typeof options?.stopOnCompletion === "boolean" ? options.stopOnCompletion : true,
skipColumns: options?.skipColumns,
});
}
export type SubscribeToRunsFilterOptions = {
/**
* Filter runs by the time they were created. You must specify the duration string like "1h", "10s", "30m", etc.
*
* @example
* "1h" - 1 hour ago
* "10s" - 10 seconds ago
* "30m" - 30 minutes ago
* "1d" - 1 day ago
* "1w" - 1 week ago
*
* The maximum duration is 1 week
*
* @note The timestamp will be calculated on the server side when you first subscribe to the runs.
*
*/
createdAt?: string;
/**
* Skip columns from the subscription.
*
* @default []
*/
skipColumns?: RealtimeRunSkipColumns;
};
/**
* Subscribes to real-time updates for all runs that have specific tags.
*
@@ -423,11 +463,15 @@ function subscribeToRun<TRunId extends AnyRunHandle | AnyTask | string>(
* ```
*/
function subscribeToRunsWithTag<TTasks extends AnyTask>(
tag: string | string[]
tag: string | string[],
filters?: SubscribeToRunsFilterOptions,
options?: { signal?: AbortSignal }
): RunSubscription<InferRunTypes<TTasks>> {
const apiClient = apiClientManager.clientOrThrow();
return apiClient.subscribeToRunsWithTag<InferRunTypes<TTasks>>(tag);
return apiClient.subscribeToRunsWithTag<InferRunTypes<TTasks>>(tag, filters, {
...(options ? { signal: options.signal } : {}),
});
}
/**
references/d3-chat/src/components/chat-container.tsx
+1
View File
@@ -97,6 +97,7 @@ export function useTodoChat({ accessToken }: { accessToken: string }) {
const triggerInstance = useRealtimeTaskTriggerWithStreams<typeof todoChat, STREAMS>("todo-chat", {
accessToken,
baseURL: process.env.NEXT_PUBLIC_TRIGGER_API_URL,
skipColumns: ["isTest"],
});
const messages = triggerInstance.run
references/d3-chat/src/trigger/chat.ts
+7
View File
@@ -111,6 +111,13 @@ const getUserId = tool({
export const todoChat = schemaTask({
id: "todo-chat",
description: "Chat with the todo app",
retry: {
maxAttempts: 3,
maxTimeoutInMs: 10000,
},
queue: {
concurrencyLimit: 1,
},
schema: z.object({
input: z
.string()
references/hello-world/src/trigger/realtime.ts
+34
View File
@@ -0,0 +1,34 @@
import { logger, runs, task } from "@trigger.dev/sdk";
import { helloWorldTask } from "./example.js";
export const realtimeByTagsTask = task({
id: "realtime-by-tags",
run: async (payload: any, { ctx, signal }) => {
await helloWorldTask.trigger(
{ hello: "world" },
{
tags: ["hello-world", "realtime"],
}
);
const timeoutSignal = AbortSignal.timeout(10000);
const $signal = AbortSignal.any([signal, timeoutSignal]);
$signal.addEventListener("abort", () => {
logger.info("signal aborted");
});
for await (const run of runs.subscribeToRunsWithTag(
"hello-world",
{ createdAt: "2m", skipColumns: ["payload", "output", "number"] },
{ signal: $signal }
)) {
logger.info("run", { run });
}
return {
message: "Hello, world!",
};
},
});
references/hello-world/src/trigger/sdk.ts
+25
View File
@@ -0,0 +1,25 @@
import { logger, runs, task } from "@trigger.dev/sdk";
export const sdkMethods = task({
id: "sdk-methods",
run: async (payload: any, { ctx }) => {
for await (const run of runs.list({
status: ["COMPLETED"],
from: new Date(Date.now() - 1000 * 60 * 60), // 1 hour ago
to: new Date(), // now
})) {
logger.info("completed run", { run });
}
for await (const run of runs.list({
status: ["FAILED"],
from: new Date(Date.now() - 1000 * 60 * 60), // 1 hour ago
to: new Date(), // now
limit: 50,
})) {
logger.info("failed run", { run });
}
return runs;
},
});