Compare commits

...

87 Commits

Author SHA1 Message Date
github-actions[bot] 185f4ecaf9 chore: release v4.2.0 (#2713)
🚀 Publish Trigger.dev Docker / units (push) Failing after 0s
🚀 Publish Trigger.dev Docker / typecheck (push) Failing after 19s
🚀 Publish Trigger.dev Docker / publish-webapp (push) Has been skipped
🚀 Publish Trigger.dev Docker / publish-worker (push) Has been skipped
🚀 Publish Trigger.dev Docker / publish-worker-v4 (push) Has been skipped
This PR was opened by the [Changesets
release](https://github.com/changesets/action) GitHub action. When
you're ready to do a release, you can merge this and publish to npm
yourself or [setup this action to publish
automatically](https://github.com/changesets/action#with-publishing). If
you're not ready to do a release yet, that's fine, whenever you add more
changesets to main, this PR will be updated.


# Releases
## trigger.dev@4.2.0

### Minor Changes

- feat(cli): upgrade bun deployments to v1.3.3
([#2756](https://github.com/triggerdotdev/trigger.dev/pull/2756))

### Patch Changes

- fix(otel): exported logs and spans will now have matching trace IDs
([#2724](https://github.com/triggerdotdev/trigger.dev/pull/2724))
- The `--force-local-build` flag is now renamed to just `--local-build`
([#2702](https://github.com/triggerdotdev/trigger.dev/pull/2702))
- fix(cli): header will always print the correct profile
([#2728](https://github.com/triggerdotdev/trigger.dev/pull/2728))
- feat: add ability to set custom resource properties through
trigger.config.ts or via the OTEL_RESOURCE_ATTRIBUTES env var
([#2704](https://github.com/triggerdotdev/trigger.dev/pull/2704))
- feat(cli): implements content-addressable store for the dev CLI build
outputs, reducing disk usage
([#2725](https://github.com/triggerdotdev/trigger.dev/pull/2725))
- Added support for native build server builds in the deploy command
(`--native-build-server`)
([#2702](https://github.com/triggerdotdev/trigger.dev/pull/2702))
-   Updated dependencies:
    -   `@trigger.dev/build@4.2.0`
    -   `@trigger.dev/core@4.2.0`
    -   `@trigger.dev/schema-to-json@4.2.0`

## @trigger.dev/build@4.2.0

### Patch Changes

- syncVercelEnvVars to skip API and read env vars directly from
env.process for Vercel build environments. New syncNeonEnvVars build
extension for syncing environment variablesfrom Neon database projects
to Trigger.dev. The extension automatically detects branches and builds
appropriate PostgreSQL connection strings for non-production, non-dev
environments (staging, preview).
([#2729](https://github.com/triggerdotdev/trigger.dev/pull/2729))
-   Updated dependencies:
    -   `@trigger.dev/core@4.2.0`

## @trigger.dev/core@4.2.0

### Patch Changes

- fix: prevent ERR_IPC_CHANNEL_CLOSED errors from causing an unhandled
exception on TaskRunProcess
([#2743](https://github.com/triggerdotdev/trigger.dev/pull/2743))
- Added support for native build server builds in the deploy command
(`--native-build-server`)
([#2702](https://github.com/triggerdotdev/trigger.dev/pull/2702))

## @trigger.dev/python@4.2.0

### Patch Changes

-   Updated dependencies:
    -   `@trigger.dev/build@4.2.0`
    -   `@trigger.dev/sdk@4.2.0`
    -   `@trigger.dev/core@4.2.0`

## @trigger.dev/react-hooks@4.2.0

### Patch Changes

- fix: prevent infinite useEffect when passing an array of tags to
useRealtimeRunsWithTag
([#2705](https://github.com/triggerdotdev/trigger.dev/pull/2705))
-   Updated dependencies:
    -   `@trigger.dev/core@4.2.0`

## @trigger.dev/redis-worker@4.2.0

### Patch Changes

-   Updated dependencies:
    -   `@trigger.dev/core@4.2.0`

## @trigger.dev/rsc@4.2.0

### Patch Changes

-   Updated dependencies:
    -   `@trigger.dev/core@4.2.0`

## @trigger.dev/schema-to-json@4.2.0

### Patch Changes

-   Updated dependencies:
    -   `@trigger.dev/core@4.2.0`

## @trigger.dev/sdk@4.2.0

### Patch Changes

- fix(sdk): Re-export schemaTask types to prevent the TypeScript error
TS2742: The inferred type of 'task' cannot be named without a reference
to '@trigger.dev/core/v3'. This is likely not portable.
([#2735](https://github.com/triggerdotdev/trigger.dev/pull/2735))
- feat: add ability to set custom resource properties through
trigger.config.ts or via the OTEL_RESOURCE_ATTRIBUTES env var
([#2704](https://github.com/triggerdotdev/trigger.dev/pull/2704))
-   Updated dependencies:
    -   `@trigger.dev/core@4.2.0`

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
2025-12-05 16:27:21 +01:00
nicktrn 0d764d4c46 feat(cli): upgrade bun deployments to v1.3.3 (#2756)
New deployments with `runtime: "bun"` will now use Bun v1.3.3

Link to Bun release: https://bun.com/blog/bun-v1.3.3
2025-12-05 13:31:27 +00:00
Saadi Myftija beb52b9800 chore(releases): adjust gh workflow title (#2755)
Just a tiny workflow label change
2025-12-05 13:33:35 +01:00
Saadi Myftija 3401a1d0a9 feat(releases): require approval for package releases (#2753)
**Background**
Currently the changeset PR creation and the publishing is handled by the
same workflow. This is not ideal:
- The build steps are executed on every run of the pipeline, even though
they're only needed for the publish case.
- The PR creation workflow does not need permissions to publish to npm,
only the release path needs them.
- Adding an approval step is painful as we'd need to also approve each
changeset PR creation workflow run.

**Changes in this PR**
- Separated the changeset PR creation into its own workflow and minimum
permission set.
- Added a GH environment with an approval step for the package
publishing workflow (also for prereleases).
- New publish workflow runs will not cancel in-progress runs; helps
avoid partial failures in publishing.

These changes also enable hardening the npm OIDC setup by tying it to a
GH environment that requires approval.
2025-12-05 13:25:25 +01:00
nicktrn 3f982ed366 docs: update email regex examples (#2749)
The previous examples lacked start and end anchors
2025-12-04 17:58:25 +00:00
Saadi Myftija 4dc956470d fix(deployments): misc fixes for the native build server deployment flow (#2748)
Improved a couple of error messages.

Also fixed an issue with the s2 token caching.
2025-12-04 17:01:47 +01:00
Oskar Otwinowski 7fddadcce9 chore: Improve Vercel/Neon syncEnvVars build extensions docs (#2747)
chore: Improve Vercel/Neon `syncEnvVars` build extensions docs
2025-12-04 16:34:57 +01:00
Oskar 6e038d4d1e chore: Improve Vercel/Neon syncEnvVars extensions docs 2025-12-04 16:32:15 +01:00
Oskar Otwinowski 05b53967ea feat(build): Add NeonDB branch resolution for Vercel preview environments (#2729)
Vercel's NeonDB integration renders database connection environment
variables at runtime, which means Trigger.dev cannot directly sync these
values during the build process. This change adds support for fetching
branch-specific NeonDB connection strings via the Neon API.

feat(build): Add syncNeonEnvVars extension and improve Vercel env var
syncing

Add a new `syncNeonEnvVars` build extension for syncing environment
variables
from Neon database projects to Trigger.dev. The extension automatically
detects
branches and builds appropriate PostgreSQL connection strings for
non-production
environments (staging, dev, preview).

Features of `syncNeonEnvVars`:
- Fetches branch-specific database credentials from Neon API
- Generates all standard Postgres connection strings (DATABASE_URL,
POSTGRES_URL,
  POSTGRES_PRISMA_URL, etc.) with both pooled and unpooled variants
- Supports custom database name, role name, and env var prefix options
- Skips automatically in Vercel environments (Neon's Vercel integration
handles this)
- Skips for production environments (designed for preview/staging/dev
branches)

Improvements to `syncVercelEnvVars`:
- When running in a Vercel build environment (detected via VERCEL env
var),
values are now read from process.env instead of the Vercel API response
- This ensures the build uses the actual runtime values Vercel provides
- Removed embedded Neon-specific logic (now handled by separate
extension)
- Simplified and cleaned up the extension code

Documentation updates for both extensions with usage examples and
configuration
options.

Closes #2714

##  Checklist

- [x] I have followed every step in the [contributing
guide](https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md)
- [x] The PR title follows the convention.
- [x] I ran and tested the code works

---

## Testing

Set up Vercel + Trigger.dev envs, used Vercel's chat-bot-ai template.
2025-12-04 16:26:35 +01:00
Oskar 702f3b4bca chore: Improve changeset message 2025-12-04 16:26:24 +01:00
nicktrn 8fcd93001d chore: insert missing spaces in changeset 2025-12-04 13:42:32 +00:00
Eric Allam b01b8740cc fix(engine): prevent overriding a queue concurrency limit resuming a paused queue (#2745) 2025-12-04 13:33:12 +00:00
Oskar 357aa99309 Add changeset 2025-12-04 14:12:42 +01:00
Oskar 7cbf82a4ae fix: PR feedback, doc improvements for neon/vercel syncEnvVars 2025-12-04 14:07:31 +01:00
Eric Allam e7fec4097f fix(dev): CLI now properly cleans up the store dir on dev CLI exit (#2744) 2025-12-04 13:13:01 +01:00
Eric Allam d279988e38 fix(workers): prevent ERR_IPC_CHANNEL_CLOSED errors from causing an unhandled exception on TaskRunProcess (#2743) 2025-12-04 12:02:30 +00:00
Saadi Myftija 748ae658f7 fix(releases): use npm 11.x for OIDC support in the release workflow (#2742)
Support for OIDC requires npm v11.5.1 or newer.
2025-12-04 12:21:21 +01:00
Oskar Otwinowski 249878ed92 feat(webapp): link to task-specific test page from filtered runs table (#2741)
When viewing runs filtered to a single task, the "Create a test run" and
"Run a test" buttons now navigate directly to the task-specific test
page instead of the generic test page.

This improves UX by pre-populating the test form with the filtered task,
saving users from having to manually select it again.


##  Checklist

- [x] I have followed every step in the [contributing
guide](https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md)
- [x] The PR title follows the convention.
- [x] I ran and tested the code works

## Screenshots

<img width="1706" height="1392" alt="image"
src="https://github.com/user-attachments/assets/d8b5d445-73b5-426c-83a4-90ac2a95b955"
/>
2025-12-04 10:56:24 +00:00
Eric Allam 117b1d5a53 chore(dependabot): upgrade next.js for CVE-2025-66478 in d3-chat example project (#2740) 2025-12-04 10:30:37 +00:00
Eric Allam 04173a93b9 fix(replication): detect misconfigered run replication publication and output helpful error messages (#2736)
Add validation for logical replication publication configuration. Helps
diagnose an issue where runs are no longer replicated to clickhouse
because of a configuration issue with the replication publication.

## Problem

The `LogicalReplicationClient` only checked if a publication existed,
not if it was correctly configured. This caused a silent failure where:

- Replication would start successfully
- Transaction boundaries (begin/commit) were received
- **But no actual data changes were replicated**

This happened when a publication existed but:
1. Had no tables associated with it
2. Was missing required actions (e.g., `delete`)

## Solution

Added `#validatePublicationConfiguration()` method that validates:
-  Publication includes the expected table
-  Publication has all required actions configured

When validation fails, error messages include the exact SQL command to
fix the issue:

**Missing table:**
```
Publication 'task_runs_to_clickhouse_v1_publication' exists but has NO TABLES configured. 
Expected table: "public.TaskRun". 
Run: ALTER PUBLICATION task_runs_to_clickhouse_v1_publication ADD TABLE "TaskRun";
```

**Missing actions:**
```
Publication 'task_runs_to_clickhouse_v1_publication' is missing required actions. 
Expected: [insert, update, delete], Current: [insert, update], Missing: [delete]. 
Run: ALTER PUBLICATION task_runs_to_clickhouse_v1_publication SET (publish = 'insert, update, delete');
```

This prevents silent data loss and makes debugging configuration issues
much easier.
2025-12-04 10:29:13 +00:00
Saadi Myftija da4c753b68 fix(releases): add missing GH token in prerelease workflow (#2739)
The `changeset version` command needs the GH token too.
2025-12-04 11:27:11 +01:00
Eric Allam 6ae3b69745 chore(cursor): add deslop command (#2722) 2025-12-04 10:24:09 +00:00
Saadi Myftija 652d95c7eb feat(releases): add a prerelease workflow (#2737)
Adds a manual trigger to the `release.yml` workflow for publishing
prerelease versions. Needs to be in the same workflow file due to a NPM
limitation on how OIDC claims are checked.

Currently there is a validation step on the ref for the prerelease: it
must be merged to the main branch. We can revisit this in the future in
case we find it too limiting.
2025-12-04 10:17:05 +01:00
Saadi Myftija 341e27d213 chore(deployments): lazily update ECR repo cache settings (#2734)
To avoid doing a migration for the ECR repo cache settings, we lazily do
it on the next deployment for that project. Failures to update the repo
settings are just logged and will not cause the deployment to fail.
2025-12-03 17:15:59 +01:00
Eric Allam 9821d02af7 fix(sdk): Re-export schemaTask types to prevent the TypeScript error TS2742 (#2735)
Fixes this type of error when exporting a `schemaTask` in a monorepo:

```
error TS2742: The inferred type of 'helloWorldSchema' cannot be named without a reference to '@trigger.dev/core/v3'. This is likely not portable.
```
2025-12-03 16:12:23 +00:00
Saadi Myftija 255a73a2fe feat(deployments): --native-build-server support for the deploy command (#2702)
This PR adds support for CLI deployments using the native build server.

**Background**

The deployment command currently does the following:
- bundles the code
- submits the build context to our external build provider and waits for
the build
- triggers deployment state transitions using the platform API

Upstream build provider outages cause issue with deployments,
potentially blocking deployments entirely. We recently introduced the
`--force-local-build` flag as a fallback to enable deployment without a
dependency on the upstream build provider, though it requires users to
have docker in their systems. This PR continues that work by providing a
remote build path which uses our own build server and does not rely on
the external provider.

**Changes in this PR**

Introduced the new `--native-build-server` flag, which does the
following:
- scans all files relevant for the Trigger deployment and evaluates
ignore rules
- packages it up in an archive and uploads it as a deployment artifact
- queues the deployment and triggers the build
- streams logs from the build server

This no longer relies on external build services. Also deployment state
transitions happen on the server-side, giving us more flexibility to
evolve the flow and schemas of related deployment API endpoints. In
general it gives us better control of the whole build and deployment
process. This path will eventually become the default.

The `--detach` flag is also new, allowing to trigger deployments without
waiting for the result.

The deployment artifacts are uploaded via pre-signed URLs to avoid
unnecessary load on the platform. The new `/artifacts` endpoint
generates the pre-signed URLs; size limits are enforced on s3. This
endpoint is deliberately generic, we could extend it in the future to
upload other artifacts client-side in a similar way, e.g., large payload
packets.
2025-12-03 16:40:21 +01:00
Saadi Myftija c5f7a8daf7 fix(releases): explicitly use no frozen lockfile in the install step (#2733) 2025-12-03 16:28:06 +01:00
Saadi Myftija 8b0f51b317 chore(releases): automatically update the lockfile in changeset PRs (#2732)
This is a step which we currently need to do manually and it's rather
painful. The lockfile update is necessary due to cross references in our
packages.

Added it as a separate job instead of a step to start from fresh
workspace, as the state that the `changeset` step leaves the workdir is
not explicitly clear to the reader.
2025-12-03 16:04:30 +01:00
Oskar 53f21e1330 feat(build): Add syncNeonEnvVars extension and improve Vercel env var syncing
Add a new `syncNeonEnvVars` build extension for syncing environment variables
from Neon database projects to Trigger.dev. The extension automatically detects
branches and builds appropriate PostgreSQL connection strings for non-production
environments (staging, dev, preview).

Features of `syncNeonEnvVars`:
- Fetches branch-specific database credentials from Neon API
- Generates all standard Postgres connection strings (DATABASE_URL, POSTGRES_URL,
  POSTGRES_PRISMA_URL, etc.) with both pooled and unpooled variants
- Supports custom database name, role name, and env var prefix options
- Skips automatically in Vercel environments (Neon's Vercel integration handles this)
- Skips for production environments (designed for preview/staging/dev branches)

Improvements to `syncVercelEnvVars`:
- When running in a Vercel build environment (detected via VERCEL env var),
  values are now read from process.env instead of the Vercel API response
- This ensures the build uses the actual runtime values Vercel provides
- Removed embedded Neon-specific logic (now handled by separate extension)
- Simplified and cleaned up the extension code

Documentation updates for both extensions with usage examples and configuration
options.
2025-12-03 16:03:41 +01:00
Saadi Myftija 331882f59c chore(releases): add package version in changeset PRs (#2730)
* Add the release version to changeset PRs

* Add missing id-token permission, needed for oidc

* Remove a couple of unnecesary steps

* Reference the `changeset-release/main` branch explicitly
2025-12-03 14:42:40 +01:00
nicktrn 1276491a83 docs: clarify slack channel on pro (#2731) 2025-12-03 13:23:38 +00:00
Eric Allam 5b7dfe23b5 feat(cli): implements content-addressable store for the dev CLI build outputs, reducing disk usage (#2725)
* feat(cli): implements content-addressable store for the dev CLI build outputs, reducing disk usage

* fix a few things
2025-12-03 10:17:35 +00:00
Oskar 49b2f683f4 feat(build): Add NeonDB branch resolution for Vercel preview environments
Vercel's NeonDB integration renders database connection environment
variables at runtime, which means Trigger.dev cannot directly sync
these values during the build process. This change adds support for
fetching branch-specific NeonDB connection strings via the Neon API.

Changes:
- Discover NEON_PROJECT_ID from incoming Vercel environment variables
- Call NeonDB API to search for branches matching the git branch name
- Filter branches to find exact matches with Vercel environment prefix
  (e.g., "preview/branch-name") to avoid false positives from partial
  string matches
- Retrieve branch endpoints and select the write endpoint (or first
  available)
- Build connection strings (DATABASE_URL, POSTGRES_URL, etc.) using
  the branch endpoint host while preserving user/password credentials

Safety measures for non-production environments:
- Filter out all Neon-related env vars (DATABASE_URL, PGHOST, etc.)
  before calling the Neon API to prevent accidental use of production
  database credentials
- Only add branch-specific database env vars if a matching Neon branch
  is found and the API call succeeds
- If neonDbAccessToken is not provided or the API fails, non-production
  environments will not receive any database connection env vars

Usage:
Users must provide a NEON_ACCESS_TOKEN (via options or env var) to
enable automatic branch resolution for preview deployments. Production
environments continue to use Vercel's standard env var sync without
modification.
2025-12-02 19:52:45 +01:00
nicktrn af9b3e1c99 fix(cli): header will always print the correct profile (#2728) 2025-12-02 17:45:35 +00:00
Eric Allam df4ab97d59 fix(otel): fix broken schedule run spans (#2727)
schedule spans can sometimes show as generic spans when using the 
task_events_v2 table because of the inserted_at filter. Increasing the 
buffer for the start time does the trick and doesn’t cause any perf 
Issues (and is in general just more robust)
2025-12-02 17:16:08 +00:00
Eric Allam 9f27422472 fix(otel): exported logs and spans will now have matching trace IDs (#2724)
When using custom OTLP exporters via `telemetry.exporters` and 

This occurred when tasks were triggered **without** a parent trace 
context (e.g., via API or dashboard). In this scenario: - Spans were 
correctly rewritten to use the generated `externalTraceId` - Logs kept 
their original internal trace ID due to a bug in the early return logic

### Root Cause

In `ExternalLogRecordExporterWrapper.transformLogRecord()`, the early 
return condition incorrectly included `!this.externalTraceContext`:

```typescript
if (!logRecord.spanContext || !this.externalTraceId || 
!this.externalTraceContext) {   return logRecord;  // Bug: Returns early
when externalTraceContext is undefined }

// This fallback logic was never reached:
const externalTraceId = this.externalTraceContext
  ? this.externalTraceContext.traceId
  : this.externalTraceId;
```

### Fix

1. **Reordered logic in `transformLogRecord()`**: Move the 
1. `externalTraceId` calculation before the early return, and check the 
1. culated value instead of `this.externalTraceContext`:

```typescript
const externalTraceId = this.externalTraceContext
  ? this.externalTraceContext.traceId
  : this.externalTraceId;

if (!logRecord.spanContext || !externalTraceId) {
  return logRecord;
}
```

2. **Clarified `_isExternallySampled` logic**: Updated both 
2. `ExternalSpanExporterWrapper` and `ExternalLogRecordExporterWrapper` 
2. explicitly handle the case where there's no external trace context 
2. a generated `externalTraceId` exists:

```typescript
this._isExternallySampled = externalTraceContext
  ? isTraceFlagSampled(externalTraceContext.traceFlags)
  : !!externalTraceId;
```

### Impact

Logs and spans from the same task run will now have matching trace IDs 
when exported to external observability tools, enabling proper trace correlation regardless of whether the task was triggered with or without a parent trace context. 
`telemetry.logExporters` in `trigger.config.ts`, logs and spans were 
exported with **different trace IDs**, breaking trace correlation in
external observability tools like Datadog.
2025-12-02 14:16:06 +00:00
nicktrn 2f1a72b109 security: remedy dependabot alerts (#2723)
* security: override js-yaml

* security: upgrade vite

* security: update nodemailer
2025-12-02 12:02:48 +00:00
Eric Allam 3c326a4b4a fix(clickhouse): ensure start_time is never older than X ms to prevent old partition merge issues (#2721) 2025-12-01 15:43:10 +00:00
Saadi Myftija 6ae1317b69 chore(clickhouse): enable dropping run debug events (#2720)
Added a new env var (`EVENT_REPOSITORY_DEBUG_LOGS_DISABLED`) that allows disabling writing run debug logs in the event repository.
2025-12-01 15:42:35 +01:00
Eric Allam 2e1c4f6df6 fix(clickhouse): partition by insertion date to prevent "Too many parts" errors when partitioning by start time (#2719) 2025-12-01 12:01:06 +00:00
nicktrn 2bf86dc20e fix(supervisor): image builds with pnpm v10 (#2718) 2025-12-01 11:28:05 +00:00
nicktrn 485782cae1 feat(ch): optionally disable migrations (#2715) 2025-11-28 19:39:29 +00:00
Eric Allam 61b338bea7 chore(repo): upgrade repo to pnpm@10 to prevent executing dep scripts on install (#2712)
* chore: migrate pnpm lockfile to v9 format via pnpm@9

* Upgrade to pnpm 10.23.0

* update the dockerfile and added a few deps to bundle in remix app
2025-11-27 16:26:19 +00:00
Kim Hallberg 83ddf721a4 docs: update Firecrawl example (#2652)
Update the example to use Firecrawl v2 API
2025-11-24 14:54:51 +00:00
Kim Hallberg a4dd2562d2 docs: format embedding example (#2656) 2025-11-24 14:51:53 +00:00
Felipe Martinez Albeche 5ff21a758c fix(docs): tooltip color contrast accessibility (#2662)
* fix(docs): color contrast accessibility on tooltip

* add changeset

* Delete .changeset/tough-feet-accept.md

---------

Co-authored-by: Eric Allam <eric@trigger.dev>
2025-11-24 14:39:14 +00:00
Lindsey 47a64c0335 docs: improve env var documentation with .env upload + ctx environment details (#2680) 2025-11-24 14:37:04 +00:00
Eric Allam fc351cb6c4 chore(repo): remove GITHUB_TOKEN requirement for publishing prerelease packages (#2679) 2025-11-24 14:33:40 +00:00
Eric Allam c4f2a9d065 chore(references): added prisma-generator-ts-enums to prisma ref project as an example (#2701) 2025-11-24 14:33:22 +00:00
Eric Allam 2762c542c2 chore(repo): remove format script to AI stops calling it (#2703) 2025-11-24 14:33:14 +00:00
Eric Allam 72e286af2f feat(otel): support for custom resource attributes via config#telemetry.resource and OTEL_RESOURCE_ATTRIBUTES env var (#2704) 2025-11-24 14:33:07 +00:00
Eric Allam f7240a99e7 fix(react): prevent infinite useEffect when passing an array of tags to useRealtimeRunsWithTag (#2705) 2025-11-24 14:32:57 +00:00
github-actions[bot] 6e47377766 chore: Update version for release (#2700)
🚀 Publish Trigger.dev Docker / typecheck (push) Failing after 0s
🚀 Publish Trigger.dev Docker / publish-webapp (push) Has been skipped
🚀 Publish Trigger.dev Docker / publish-worker (push) Has been skipped
🚀 Publish Trigger.dev Docker / publish-worker-v4 (push) Has been skipped
🚀 Publish Trigger.dev Docker / units (push) Failing after 21s
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
2025-11-24 10:30:01 +00:00
nicktrn a8563ca534 fix(docker): support the latest docker version (#2686) 2025-11-21 15:24:31 +00:00
Eric Allam 5e5c97ea4c fix(cli): stop dev runs stuck in dequeued status (fix #2639) (#2699)
* stop deleting the first dev version files on the first change, prevents system failures

* prevent dev runs getting stuck in dequeued status by deleting workers

* add changeset
2025-11-21 15:24:07 +00:00
Matt Aitken abd99aa27f Don't show (or count) archived branches/projects against allocated concurrency (#2698)
Archived preview branches were being counted. Also archived projects.
2025-11-20 11:41:12 +00:00
Eric Allam 4347499799 fix(api): Fix preview branch targeting in environment variable API routes (#2697)
This PR fixes the `x-trigger-branch` header support for targeting specific preview branches when managing environment variables. The header was documented but not actually being extracted or used in the environment variable API routes. Additionally, the query logic in `authenticatedEnvironmentForAuthentication` was fundamentally broken—it searched for environments with both `slug: "preview"` (parent environment property) AND a specific `branchName` (child environment property), which no environment could satisfy simultaneously. The fix extracts the branch name using `branchNameFromRequest()` and correctly queries for child branch environments using `type: "PREVIEW"` and the specific `branchName`. This ensures that environment variable operations (create, update, get, list) properly target individual preview branches instead of affecting all preview environments.
2025-11-19 18:01:12 +00:00
Eric Allam abee783d3f chore(logs): remove unnecessary debug logs (#2696) 2025-11-19 13:56:15 +00:00
Eric Allam 6464eeed53 fix(webapp): correctly generate JWT tokens for preview branches after triggering a run (fix #2678) (#2695) 2025-11-19 13:42:12 +00:00
Matt Aitken bee59de3a0 Concurrency self serve (#2681)
* Don't use the organization max concurrency anymore

* Early draft of the concurrency page

* WIP adding a new stepper input component

* Move stepper to be alphabetical

* When max value is reached, disabled the + button

* Show placeholder if you delete all numbers

* Make all the html input values available to the component

* Adds size variants

* Move stepper into its own component

* Work on showing the extra concurrency

* The purchase form styling and functionality (minus actually purchasing)

* New style for outline input fields

* Concurrency purchasing working

* Purchasing concurrency and quota emails working

* Improvements to the modal

* Show cost breakdown in the modal

* Fix for allocated concurrency including DEV

* Improved types

* Allocating concurrency is working

* Live updates total env concurrency

* Implemented reset

* Fix for concurrency allocation editing across multiple projects

* Tabular numbers

* Added an error from allocating concurrency

* Fixes for allocating concurrency where it didn't calculate correctly

* "Increase limit" link to concurrency page

* Indent environments

* Added Preview limit when updating concurrency for an org

* Show error when changing plan fails

* Added maximumProjectCount column to Org

* Limit project count and display a rich error toast (with title and button now)

* Added title and button to toasts. Use it for new project error

* @trigger.dev/platform 1.0.20

* Allow submitting zero concurrency so you can downgrade back to nothing

* Use the server as the truth for omitted environments

* Updated the pricing panels

---------

Co-authored-by: James Ritchie <james@trigger.dev>
2025-11-19 11:22:23 +00:00
Eric Allam 3af7303156 feat(docs): update prismaExtension for new mode functionality in 4.1.1 (#2690) 2025-11-19 11:16:35 +00:00
github-actions[bot] 6231ddc67a Release v4.1.1
* chore: Update version for release

* chore: Update pnpm-lock.yaml (#2694)

* Initial plan

* Update pnpm-lock.yaml

Co-authored-by: ericallam <534+ericallam@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: ericallam <534+ericallam@users.noreply.github.com>

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com>
Co-authored-by: ericallam <534+ericallam@users.noreply.github.com>
2025-11-19 11:09:26 +00:00
Eric Allam 15fef916f6 feat(build): update prisma extension to work with generated clients and rust-free clients (#2689)
* prisma extension fixes WIP

* More prisma stuff

* more prisma stuff

* remove changelog

* upgrade github workflows to use node 20.19 because installing prisma@7 breaks with lower versions

* Don't use generate for the prisma reference projects

* make sure it works if no mode is passed in
2025-11-19 10:48:15 +00:00
Eric Allam e2a703bfa8 fix(dashboard): continuously apply log filter during live run (#2692) 2025-11-18 17:00:29 +00:00
Saadi Myftija 1a7ee24b9e fix(ui): refresh deployments logs stream on status change (#2688)
The logs stream is created after the deployment moves from `PENDING` status to `INSTALLING`.
2025-11-17 14:50:05 +01:00
Saadi Myftija bb99af52cb feat(deployments): expose native build server option in build settings (#2685)
Adds a build setting about using our build server for remote builds.
2025-11-17 12:46:57 +01:00
Saadi Myftija 01797a1668 feat(deployments): ECR repo adaptations to enable external build cache (#2684)
* Update aws sdk ecr client to the latest version

* Exlude the cache tag from the immutability enforcement

* Attach a policy to ECR repos to expire untagged images

* Fix filterType
2025-11-17 12:34:54 +01:00
Eric Allam 19fa669318 Release v4.1.0 (#2683) 2025-11-14 09:55:01 +00:00
github-actions[bot] 78fcba518a chore: Update version for release (#2666)
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
2025-11-14 09:52:22 +00:00
Saadi Myftija 53047ab648 feat(deployments): external cache support for local builds (#2682)
* Add registry cache support for local builds

* Add changeset
2025-11-14 10:46:31 +01:00
Eric Allam bb5cefa92c docs: realtime streams v2 (#2673) 2025-11-14 09:42:20 +00:00
Eric Allam b8b198579c chore(rules): upgrade rules for 4.1.0 (#2676) 2025-11-14 09:42:15 +00:00
Eric Allam 892bed8c4c Upgrade to electricsql 1.2.4 (#2668) 2025-11-13 15:19:59 +00:00
Eric Allam a94a11f44d feat(sdk): replace onStart lifecycle hook with onStartAttempt (#2515)
* fix(sdk): prevent uncaught errors thrown onSuccess, onComplete, and onFailure hooks to fail attempts & in some cases runs

* Add onStartAttempt hook and deprecate onSuccess

* Add onStartAttempt hook and deprecate onStart hook

* Fix onStartAttempt overload types

* Update lifecycle functions diagram
2025-11-13 14:51:13 +00:00
Matt Aitken f116e93e01 Docs: tags can be up to 128 chars (#2678) 2025-11-13 14:43:31 +00:00
Eric Allam 6137338da9 feat(streams): make v2 streams the default when using 4.1.0+ if they are supported (#2677) 2025-11-13 13:53:42 +00:00
Lindsey 8cec3b763b docs: add section on using .env.production and dotenvx for environmen… (#2674)
* docs: add section on using .env.production and dotenvx for environment variables

* fix: correct dotenvx API usage in documentation example
2025-11-13 11:28:26 +00:00
Eric Allam 343ba54c69 fix(streams): restore realtime stream writing for v3 tasks (#2675) 2025-11-13 11:21:48 +00:00
Eric Allam a70ab10809 fix(streams): fixed broken wrapping in streams inspector (#2672) 2025-11-12 16:00:54 +00:00
Eric Allam f7cb637b32 fix(streams): scope s2 access token to environment and fix streams v1 appends (#2670)
* fix(streams): scope s2 access token to environment and fix streams v1 appends

* Less stale time
2025-11-12 12:34:15 +00:00
Eric Allam 668559ec1a fix(streams): buffer v1 streams on read to prevent split chunks (#2669) 2025-11-11 21:08:49 +00:00
Eric Allam d0ad38d684 chore(docker): remove unused seed copy from dockerfile (#2667) 2025-11-11 15:13:10 +00:00
Eric Allam 536d9fa217 feat(realtime): Realtime streams v2 (#2632) 2025-11-11 14:54:00 +00:00
github-actions[bot] d75c3aeadd chore: Update version for release (#2665)
* chore: Update version for release

* Release 4.0.7

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Matt Aitken <matt@mattaitken.com>
2025-11-11 14:29:24 +00:00
Matt Aitken a342332146 Fix for the MCP tool that gets logs for debugging runs (#2653)
* Fix for the MCP tool that gets logs for debugging runs

This was broken when we changed the data on the backend that returns 
log/span data from runs. We changed the data structured and the internal
API that the MCP client uses was failing to parse with the Zod schema

* add changeset

* Revert "add changeset"

This reverts commit 86eca836d5907fa0d0f8ac595d4d5ebade140514.

---------

Co-authored-by: nicktrn <55853254+nicktrn@users.noreply.github.com>
2025-11-11 13:26:07 +00:00
Saadi Myftija 9624465ee2 fix: error handling issue with s2 streams (#2664)
* Fix logger import

*old man yells at auto-import*

* Fix s2 error handling for non-existing streams
2025-11-11 09:44:54 +01:00
Saadi Myftija 42f53b12b7 feat(deployments): stream build server logs (#2663)
* Use read-only project-scoped s2 tokens for streaming deployment logs

* Add http2 to remix polyfills

Needed for using s2 client-side.

* Stream build-server logs in the deployment details page

* Disable 12-hour format in the DateTime component

* Enable collapsing the logs panel

* Auto-collapse logs for succesful/timedout/queued deployments

* Make S2 env vars optional

* Show the logs section only for gh-triggered deployments

* Cache s2 access tokens in redis

* Reset streaming state

* Expose 12h format as a param for the Datetime components
2025-11-10 12:42:40 +01:00
416 changed files with 51812 additions and 23008 deletions
+11
View File
@@ -0,0 +1,11 @@
# Remove AI code slop
Check the diff against main, and remove all AI generated slop introduced in this branch.
This includes:
- Extra comments that a human wouldn't add or is inconsistent with the rest of the file
- Extra defensive checks or try/catch blocks that are abnormal for that area of the codebase (especially if called by trusted / validated codepaths)
- Casts to any to get around type issues
- Any other style that is inconsistent with the file
Report at the end with only a 1-3 sentence summary of what you changed
+2 -2
View File
@@ -34,9 +34,9 @@ DEPLOY_REGISTRY_HOST=localhost:5000
# OPTIONAL VARIABLES
# This is used for validating emails that are allowed to log in. Every email that do not match this regex will be rejected.
# WHITELISTED_EMAILS="authorized@yahoo\.com|authorized@gmail\.com"
# WHITELISTED_EMAILS="^(authorized@yahoo\.com|authorized@gmail\.com)$"
# Accounts with these emails will get global admin rights. This grants access to the admin UI.
# ADMIN_EMAILS="admin@example\.com|another-admin@example\.com"
# ADMIN_EMAILS="^(admin@example\.com|another-admin@example\.com)$"
# This is used for logging in via GitHub. You can leave these commented out if you don't want to use GitHub for authentication.
# AUTH_GITHUB_CLIENT_ID=
# AUTH_GITHUB_CLIENT_SECRET=
+102
View File
@@ -0,0 +1,102 @@
name: 🦋 Changeset PR
on:
push:
branches:
- main
paths:
- "packages/**"
- ".changeset/**"
- "package.json"
- "pnpm-lock.yaml"
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true
jobs:
release-pr:
name: Create Release PR
runs-on: ubuntu-latest
permissions:
contents: write
pull-requests: write
if: github.repository == 'triggerdotdev/trigger.dev'
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup pnpm
uses: pnpm/action-setup@v4
- name: Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.19.0
cache: "pnpm"
- name: Install dependencies
run: pnpm install --frozen-lockfile
- name: Create release PR
id: changesets
uses: changesets/action@v1
with:
version: pnpm run changeset:version
commit: "chore: release"
title: "chore: release"
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: Update PR title with version
if: steps.changesets.outputs.published != 'true'
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
PR_NUMBER=$(gh pr list --head changeset-release/main --json number --jq '.[0].number')
if [ -n "$PR_NUMBER" ]; then
git fetch origin changeset-release/main
# we arbitrarily reference the version of the cli package here; it is the same for all package releases
VERSION=$(git show origin/changeset-release/main:packages/cli-v3/package.json | jq -r '.version')
gh pr edit "$PR_NUMBER" --title "chore: release v$VERSION"
fi
update-lockfile:
name: Update lockfile on release PR
runs-on: ubuntu-latest
needs: release-pr
permissions:
contents: write
steps:
- name: Checkout release branch
uses: actions/checkout@v4
with:
ref: changeset-release/main
- name: Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 10.23.0
- name: Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.19.0
- name: Install and update lockfile
run: pnpm install --no-frozen-lockfile
- name: Commit and push lockfile
run: |
set -e
if git diff --quiet pnpm-lock.yaml; then
echo "No lockfile changes"
else
git config user.name "github-actions[bot]"
git config user.email "github-actions[bot]@users.noreply.github.com"
git add pnpm-lock.yaml
git commit -m "chore: update lockfile for release"
git push origin changeset-release/main
fi
+2 -2
View File
@@ -31,12 +31,12 @@ jobs:
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
- name: 📥 Download deps
run: pnpm install --frozen-lockfile --filter trigger.dev...
+104 -48
View File
@@ -1,98 +1,154 @@
name: 🦋 Changesets Release
permissions:
contents: write
on:
push:
pull_request:
types: [closed]
branches:
- main
paths-ignore:
- "docs/**"
- "**.md"
- ".github/CODEOWNERS"
- ".github/ISSUE_TEMPLATE/**"
workflow_dispatch:
inputs:
ref:
description: "The ref (branch, tag, or SHA) to checkout and release from"
required: true
type: string
tag:
description: "The npm dist-tag for the prerelease (e.g., 'v4-prerelease')"
required: true
type: string
default: "prerelease"
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true
group: ${{ github.workflow }}
cancel-in-progress: false
jobs:
release:
name: 🦋 Changesets Release
name: 🚀 Release npm packages
runs-on: ubuntu-latest
environment: npm-publish
permissions:
contents: write
contents: read
packages: write
pull-requests: write
if: github.repository == 'triggerdotdev/trigger.dev'
id-token: write
if: |
github.repository == 'triggerdotdev/trigger.dev' &&
github.event_name != 'workflow_dispatch' &&
github.event.pull_request.merged == true &&
startsWith(github.event.pull_request.head.ref, 'changeset-release/')
outputs:
published: ${{ steps.changesets.outputs.published }}
published_packages: ${{ steps.changesets.outputs.publishedPackages }}
published_package_version: ${{ steps.get_version.outputs.package_version }}
steps:
- name: ⬇️ Checkout repo
- name: Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup pnpm
- name: Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: Setup node
- name: Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
cache: "pnpm"
- name: 📥 Download deps
- name: Install dependencies
run: pnpm install --frozen-lockfile
- name: 📀 Generate Prisma Client
- name: Generate Prisma client
run: pnpm run generate
- name: 🏗️ Build
- name: Build
run: pnpm run build --filter "@trigger.dev/*" --filter "trigger.dev"
- name: 🔎 Type check
- name: Type check
run: pnpm run typecheck --filter "@trigger.dev/*" --filter "trigger.dev"
- name: 🔐 Setup npm auth
run: |
echo "registry=https://registry.npmjs.org" >> ~/.npmrc
echo "//registry.npmjs.org/:_authToken=${{ secrets.NPM_TOKEN }}" >> ~/.npmrc
# This action has two responsibilities. The first time the workflow runs
# (initial push to the `main` branch) it will create a new branch and
# then open a PR with the related changes for the new version. After the
# PR is merged, the workflow will run again and this action will build +
# publish to npm.
- name: 🚀 PR / Publish
if: ${{ !env.ACT }}
- name: Publish
id: changesets
uses: changesets/action@v1
with:
version: pnpm run changeset:version
commit: "chore: Update version for release"
title: "chore: Update version for release"
publish: pnpm run changeset:release
createGithubReleases: true
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
NPM_TOKEN: ${{ secrets.NPM_TOKEN }}
# - name: 🚀 PR / Publish (mock)
# if: ${{ env.ACT }}
# id: changesets
# run: |
# echo "published=true" >> "$GITHUB_OUTPUT"
# echo "publishedPackages=[{\"name\": \"@xx/xx\", \"version\": \"1.2.0\"}, {\"name\": \"@xx/xy\", \"version\": \"0.8.9\"}]" >> "$GITHUB_OUTPUT"
- name: 📦 Get package version
- name: Show package version
if: steps.changesets.outputs.published == 'true'
id: get_version
run: |
package_version=$(echo '${{ steps.changesets.outputs.publishedPackages }}' | jq -r '.[0].version')
echo "package_version=${package_version}" >> "$GITHUB_OUTPUT"
# this triggers the publish workflow for the docker images
- name: Create and push Docker tag
if: steps.changesets.outputs.published == 'true'
run: |
set -e
git tag "v.docker.${{ steps.get_version.outputs.package_version }}"
git push origin "v.docker.${{ steps.get_version.outputs.package_version }}"
# The prerelease job needs to be on the same workflow file due to a limitation related to how npm verifies OIDC claims.
prerelease:
name: 🧪 Prerelease
runs-on: ubuntu-latest
environment: npm-publish
permissions:
contents: read
id-token: write
if: github.repository == 'triggerdotdev/trigger.dev' && github.event_name == 'workflow_dispatch'
steps:
- name: Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
ref: ${{ github.event.inputs.ref }}
- name: Validate ref is on main
run: |
if ! git merge-base --is-ancestor ${{ github.event.inputs.ref }} origin/main; then
echo "Error: ref must be an ancestor of main (i.e., already merged)"
exit 1
fi
- name: Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 10.23.0
- name: Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.19.0
cache: "pnpm"
# npm v11.5.1 or newer is required for OIDC support
# https://github.blog/changelog/2025-07-31-npm-trusted-publishing-with-oidc-is-generally-available/#whats-new
- name: Setup npm 11.x for OIDC
run: npm install -g npm@11.6.4
- name: Download deps
run: pnpm install --frozen-lockfile
- name: Generate Prisma Client
run: pnpm run generate
- name: Snapshot version
run: pnpm exec changeset version --snapshot ${{ github.event.inputs.tag }}
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: Clean
run: pnpm run clean --filter "@trigger.dev/*" --filter "trigger.dev"
- name: Build
run: pnpm run build --filter "@trigger.dev/*" --filter "trigger.dev"
- name: Publish prerelease
run: pnpm exec changeset publish --no-git-tag --snapshot --tag ${{ github.event.inputs.tag }}
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+2 -2
View File
@@ -19,12 +19,12 @@ jobs:
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
cache: "pnpm"
- name: 📥 Download deps
+4 -4
View File
@@ -53,12 +53,12 @@ jobs:
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
cache: "pnpm"
# ..to avoid rate limits when pulling images
@@ -111,12 +111,12 @@ jobs:
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
# no cache enabled, we're not installing deps
- name: Download blob reports from GitHub Actions Artifacts
+4 -4
View File
@@ -53,12 +53,12 @@ jobs:
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
cache: "pnpm"
# ..to avoid rate limits when pulling images
@@ -111,12 +111,12 @@ jobs:
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
# no cache enabled, we're not installing deps
- name: Download blob reports from GitHub Actions Artifacts
+4 -4
View File
@@ -53,12 +53,12 @@ jobs:
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
cache: "pnpm"
# ..to avoid rate limits when pulling images
@@ -119,12 +119,12 @@ jobs:
- name: ⎔ Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 8.15.5
version: 10.23.0
- name: ⎔ Setup node
uses: buildjet/setup-node@v4
with:
node-version: 20.11.1
node-version: 20.19.0
# no cache enabled, we're not installing deps
- name: Download blob reports from GitHub Actions Artifacts
+3 -5
View File
@@ -29,12 +29,10 @@ yarn-debug.log*
yarn-error.log*
# local env files
.env.docker
.env
.env.*
.docker/*.env
.env.local
.env.development.local
.env.test.local
.env.production.local
!.env.example
# turbo
.turbo
-5
View File
@@ -1,5 +0,0 @@
link-workspace-packages=false
public-hoist-pattern[]=*prisma*
prefer-workspace-packages=true
update-notifier=false
side-effects-cache=false
+1 -1
View File
@@ -1 +1 @@
v20.11.1
v20.19.0
+1 -1
View File
@@ -13,7 +13,7 @@ This repository is a pnpm monorepo managed with Turbo. It contains multiple apps
See `ai/references/repo.md` for a more complete explanation of the workspaces.
## Development setup
1. Install dependencies with `pnpm i` (pnpm `8.15.5` and Node.js `20.11.1` are required).
1. Install dependencies with `pnpm i` (pnpm `10.23.0` and Node.js `20.11.1` are required).
2. Copy `.env.example` to `.env` and generate a random 16 byte hex string for `ENCRYPTION_KEY` (`openssl rand -hex 16`). Update other secrets if needed.
3. Start the local services with Docker:
```bash
+2 -2
View File
@@ -15,7 +15,7 @@ branch are tagged into a release periodically.
### Prerequisites
- [Node.js](https://nodejs.org/en) version 20.11.1
- [pnpm package manager](https://pnpm.io/installation) version 8.15.5
- [pnpm package manager](https://pnpm.io/installation) version 10.23.0
- [Docker](https://www.docker.com/get-started/)
- [protobuf](https://github.com/protocolbuffers/protobuf)
@@ -36,7 +36,7 @@ branch are tagged into a release periodically.
```
3. Ensure you are on the correct version of Node.js (20.11.1). If you are using `nvm`, there is an `.nvmrc` file that will automatically select the correct version of Node.js when you navigate to the repository.
4. Run `corepack enable` to use the correct version of pnpm (`8.15.5`) as specified in the root `package.json` file.
4. Run `corepack enable` to use the correct version of pnpm (`10.23.0`) as specified in the root `package.json` file.
5. Install the required packages using pnpm.
```
+1 -1
View File
@@ -1,6 +1,6 @@
## Repo Overview
This is a pnpm 8.15.5 monorepo that uses turborepo @turbo.json. The following workspaces are relevant
This is a pnpm 10.23.0 monorepo that uses turborepo @turbo.json. The following workspaces are relevant
## Apps
+3 -3
View File
@@ -5,7 +5,7 @@ WORKDIR /app
FROM node-22-alpine AS pruner
COPY --chown=node:node . .
RUN npx -q turbo@1.10.9 prune --scope=supervisor --docker
RUN npx -q turbo@2.5.4 prune --scope=supervisor --docker
FROM node-22-alpine AS base
@@ -16,7 +16,7 @@ COPY --from=pruner --chown=node:node /app/out/json/ .
COPY --from=pruner --chown=node:node /app/out/pnpm-lock.yaml ./pnpm-lock.yaml
COPY --from=pruner --chown=node:node /app/out/pnpm-workspace.yaml ./pnpm-workspace.yaml
RUN corepack enable && corepack prepare --activate
RUN corepack enable && corepack prepare pnpm@10.23.0 --activate
FROM base AS deps-fetcher
RUN apk add --no-cache python3-dev py3-setuptools make g++ gcc linux-headers
@@ -37,7 +37,7 @@ COPY --chown=node:node scripts/updateVersion.ts scripts/updateVersion.ts
RUN pnpm run generate && \
pnpm run --filter supervisor... build&& \
pnpm deploy --filter=supervisor --prod /prod/supervisor
pnpm deploy --legacy --filter=supervisor --prod /prod/supervisor
FROM base AS runner
+1 -1
View File
@@ -57,7 +57,7 @@ const Env = z.object({
RESOURCE_MONITOR_OVERRIDE_MEMORY_TOTAL_GB: z.coerce.number().optional(),
// Docker settings
DOCKER_API_VERSION: z.string().default("v1.41"),
DOCKER_API_VERSION: z.string().optional(),
DOCKER_PLATFORM: z.string().optional(), // e.g. linux/amd64, linux/arm64
DOCKER_STRIP_IMAGE_DIGEST: BoolEnv.default(true),
DOCKER_REGISTRY_USERNAME: z.string().optional(),
@@ -0,0 +1,13 @@
export function ConcurrencyIcon({ className }: { className?: string }) {
return (
<svg className={className} viewBox="0 0 18 18" fill="none" xmlns="http://www.w3.org/2000/svg">
<circle cx="3.75" cy="3.75" r="2.25" fill="currentColor" />
<circle cx="9" cy="3.75" r="2.25" fill="currentColor" />
<circle cx="14.25" cy="3.75" r="2.25" fill="currentColor" />
<circle cx="3.75" cy="9" r="2.25" fill="currentColor" />
<circle cx="9" cy="9" r="2.25" fill="currentColor" />
<circle cx="9" cy="14.25" r="1.75" stroke="currentColor" />
<circle cx="14.25" cy="9" r="2.25" fill="currentColor" />
</svg>
);
}
@@ -0,0 +1,30 @@
export function ListBulletIcon({ className }: { className?: string }) {
return (
<svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path
d="M9 5H20"
stroke="currentColor"
strokeWidth="2"
strokeLinecap="round"
strokeLinejoin="round"
/>
<path
d="M9 12H20"
stroke="currentColor"
strokeWidth="2"
strokeLinecap="round"
strokeLinejoin="round"
/>
<path
d="M9 19H20"
stroke="currentColor"
strokeWidth="2"
strokeLinecap="round"
strokeLinejoin="round"
/>
<circle cx="4" cy="5" r="1" fill="currentColor" />
<circle cx="4" cy="12" r="1" fill="currentColor" />
<circle cx="4" cy="19" r="1" fill="currentColor" />
</svg>
);
}
@@ -0,0 +1,27 @@
export function MoveToBottomIcon({ className }: { className?: string }) {
return (
<svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path
d="M12 15L12 3"
stroke="currentColor"
strokeWidth="2"
strokeLinecap="round"
strokeLinejoin="round"
/>
<path
d="M3 21L21 21"
stroke="currentColor"
strokeWidth="2"
strokeLinecap="round"
strokeLinejoin="round"
/>
<path
d="M7.5 12.5L12 17L16.5 12.5"
stroke="currentColor"
strokeWidth="2"
strokeLinecap="round"
strokeLinejoin="round"
/>
</svg>
);
}
@@ -0,0 +1,20 @@
export function SnakedArrowIcon({ className }: { className?: string }) {
return (
<svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path
d="M5 5H16C17.6569 5 19 6.34315 19 8L19 8.5C19 10.1569 17.6569 11.5 16 11.5H8C6.34314 11.5 5 12.8431 5 14.5L5 15C4.99999 16.6569 6.34314 18 8 18H18.634"
stroke="currentColor"
strokeWidth="2"
strokeLinecap="round"
strokeLinejoin="round"
/>
<path
d="M16 21L19 18L16 15"
stroke="currentColor"
strokeWidth="2"
strokeLinecap="round"
strokeLinejoin="round"
/>
</svg>
);
}
@@ -0,0 +1,10 @@
export function StreamsIcon({ className }: { className?: string }) {
return (
<svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M3 19C3 19 5.01155 17 8 17C10.9885 17 13 18.9973 16 18.9973C19 18.9973 21 17 21 17" stroke="currentColor" strokeWidth="2" strokeLinecap="round"/>
<path d="M3 13.0001C3 13.0001 5.01155 11 8 11C10.9885 11 13 13 16 13C19 13 21 11.0001 21 11.0001" stroke="currentColor" strokeWidth="2" strokeLinecap="round"/>
<path d="M3 7C3 7 5.01155 5 8 5C10.9885 5 13 6.9973 16 6.9973C19 6.9973 21 5 21 5" stroke="currentColor" strokeWidth="2" strokeLinecap="round"/>
</svg>
);
}
+23 -3
View File
@@ -2,7 +2,7 @@ import { conform, useForm } from "@conform-to/react";
import { parse } from "@conform-to/zod";
import { InformationCircleIcon, ArrowUpCircleIcon } from "@heroicons/react/20/solid";
import { EnvelopeIcon } from "@heroicons/react/24/solid";
import { Form, useActionData, useLocation, useNavigation } from "@remix-run/react";
import { Form, useActionData, useLocation, useNavigation, useSearchParams } from "@remix-run/react";
import { type ReactNode, useEffect, useState } from "react";
import { type FeedbackType, feedbackTypeLabel, schema } from "~/routes/resources.feedback";
import { Button } from "./primitives/Buttons";
@@ -23,10 +23,12 @@ import { DialogClose } from "@radix-ui/react-dialog";
type FeedbackProps = {
button: ReactNode;
defaultValue?: FeedbackType;
onOpenChange?: (open: boolean) => void;
};
export function Feedback({ button, defaultValue = "bug" }: FeedbackProps) {
export function Feedback({ button, defaultValue = "bug", onOpenChange }: FeedbackProps) {
const [open, setOpen] = useState(false);
const [searchParams, setSearchParams] = useSearchParams();
const location = useLocation();
const lastSubmission = useActionData();
const navigation = useNavigation();
@@ -52,8 +54,26 @@ export function Feedback({ button, defaultValue = "bug" }: FeedbackProps) {
}
}, [navigation, form]);
// Handle URL param functionality
useEffect(() => {
const open = searchParams.get("feedbackPanel");
if (open) {
setType(open as FeedbackType);
setOpen(true);
// Clone instead of mutating in place
const next = new URLSearchParams(searchParams);
next.delete("feedbackPanel");
setSearchParams(next);
}
}, [searchParams]);
const handleOpenChange = (value: boolean) => {
setOpen(value);
onOpenChange?.(value);
};
return (
<Dialog open={open} onOpenChange={setOpen}>
<Dialog open={open} onOpenChange={handleOpenChange}>
<DialogTrigger asChild>{button}</DialogTrigger>
<DialogContent>
<DialogHeader>Contact us</DialogHeader>
@@ -24,6 +24,7 @@ import { Link, useNavigation } from "@remix-run/react";
import { useEffect, useRef, useState, type ReactNode } from "react";
import simplur from "simplur";
import { BranchEnvironmentIconSmall } from "~/assets/icons/EnvironmentIcons";
import { ConcurrencyIcon } from "~/assets/icons/ConcurrencyIcon";
import { ListCheckedIcon } from "~/assets/icons/ListCheckedIcon";
import { RunsIconExtraSmall } from "~/assets/icons/RunsIcon";
import { TaskIconSmall } from "~/assets/icons/TaskIcon";
@@ -43,6 +44,7 @@ import {
accountPath,
adminPath,
branchesPath,
concurrencyPath,
logoutPath,
newOrganizationPath,
newProjectPath,
@@ -122,6 +124,7 @@ export function SideMenu({
const { isConnected } = useDevPresence();
const isFreeUser = currentPlan?.v3Subscription?.isPaying === false;
const isAdmin = useHasAdminAccess();
const { isManagedCloud } = useFeatures();
useEffect(() => {
const handleScroll = () => {
@@ -313,6 +316,15 @@ export function SideMenu({
data-action="preview-branches"
badge={<V4Badge />}
/>
{isManagedCloud && (
<SideMenuItem
name="Concurrency"
icon={ConcurrencyIcon}
activeIconColor="text-amber-500"
to={concurrencyPath(organization, project, environment)}
data-action="concurrency"
/>
)}
<SideMenuItem
name="Regions"
icon={GlobeAmericasIcon}
@@ -13,7 +13,9 @@ type DateTimeProps = {
includeTime?: boolean;
showTimezone?: boolean;
showTooltip?: boolean;
hideDate?: boolean;
previousDate?: Date | string | null; // Add optional previous date for comparison
hour12?: boolean;
};
export const DateTime = ({
@@ -23,6 +25,7 @@ export const DateTime = ({
includeTime = true,
showTimezone = false,
showTooltip = true,
hour12 = true,
}: DateTimeProps) => {
const locales = useLocales();
const [localTimeZone, setLocalTimeZone] = useState<string>("UTC");
@@ -50,7 +53,8 @@ export const DateTime = ({
timeZone ?? localTimeZone,
locales,
includeSeconds,
includeTime
includeTime,
hour12
).replace(/\s/g, String.fromCharCode(32))}
{showTimezone ? ` (${timeZone ?? "UTC"})` : null}
</Fragment>
@@ -66,7 +70,8 @@ export function formatDateTime(
timeZone: string,
locales: string[],
includeSeconds: boolean,
includeTime: boolean
includeTime: boolean,
hour12: boolean = true
): string {
return new Intl.DateTimeFormat(locales, {
year: "numeric",
@@ -76,6 +81,7 @@ export function formatDateTime(
minute: includeTime ? "numeric" : undefined,
second: includeTime && includeSeconds ? "numeric" : undefined,
timeZone,
hour12,
}).format(date);
}
@@ -122,7 +128,7 @@ export function formatDateTimeISO(date: Date, timeZone: string): string {
}
// New component that only shows date when it changes
export const SmartDateTime = ({ date, previousDate = null, timeZone = "UTC" }: DateTimeProps) => {
export const SmartDateTime = ({ date, previousDate = null, timeZone = "UTC", hour12 = true }: DateTimeProps) => {
const locales = useLocales();
const realDate = typeof date === "string" ? new Date(date) : date;
const realPrevDate = previousDate
@@ -132,8 +138,8 @@ export const SmartDateTime = ({ date, previousDate = null, timeZone = "UTC" }: D
: null;
// Initial formatted values
const initialTimeOnly = formatTimeOnly(realDate, timeZone, locales);
const initialWithDate = formatSmartDateTime(realDate, timeZone, locales);
const initialTimeOnly = formatTimeOnly(realDate, timeZone, locales, hour12);
const initialWithDate = formatSmartDateTime(realDate, timeZone, locales, hour12);
// State for the formatted time
const [formattedDateTime, setFormattedDateTime] = useState<string>(
@@ -150,10 +156,10 @@ export const SmartDateTime = ({ date, previousDate = null, timeZone = "UTC" }: D
// Format with appropriate function
setFormattedDateTime(
showDatePart
? formatSmartDateTime(realDate, userTimeZone, locales)
: formatTimeOnly(realDate, userTimeZone, locales)
? formatSmartDateTime(realDate, userTimeZone, locales, hour12)
: formatTimeOnly(realDate, userTimeZone, locales, hour12)
);
}, [locales, realDate, realPrevDate]);
}, [locales, realDate, realPrevDate, hour12]);
return <Fragment>{formattedDateTime.replace(/\s/g, String.fromCharCode(32))}</Fragment>;
};
@@ -168,7 +174,7 @@ function isSameDay(date1: Date, date2: Date): boolean {
}
// Format with date and time
function formatSmartDateTime(date: Date, timeZone: string, locales: string[]): string {
function formatSmartDateTime(date: Date, timeZone: string, locales: string[], hour12: boolean = true): string {
return new Intl.DateTimeFormat(locales, {
month: "short",
day: "numeric",
@@ -178,18 +184,20 @@ function formatSmartDateTime(date: Date, timeZone: string, locales: string[]): s
timeZone,
// @ts-ignore fractionalSecondDigits works in most modern browsers
fractionalSecondDigits: 3,
hour12,
}).format(date);
}
// Format time only
function formatTimeOnly(date: Date, timeZone: string, locales: string[]): string {
function formatTimeOnly(date: Date, timeZone: string, locales: string[], hour12: boolean = true): string {
return new Intl.DateTimeFormat(locales, {
hour: "numeric",
hour: "2-digit",
minute: "numeric",
second: "numeric",
timeZone,
// @ts-ignore fractionalSecondDigits works in most modern browsers
fractionalSecondDigits: 3,
hour12,
}).format(date);
}
@@ -198,6 +206,8 @@ export const DateTimeAccurate = ({
timeZone = "UTC",
previousDate = null,
showTooltip = true,
hideDate = false,
hour12 = true,
}: DateTimeProps) => {
const locales = useLocales();
const [localTimeZone, setLocalTimeZone] = useState<string>("UTC");
@@ -214,11 +224,13 @@ export const DateTimeAccurate = ({
}, []);
// Smart formatting based on whether date changed
const formattedDateTime = realPrevDate
const formattedDateTime = hideDate
? formatTimeOnly(realDate, localTimeZone, locales, hour12)
: realPrevDate
? isSameDay(realDate, realPrevDate)
? formatTimeOnly(realDate, localTimeZone, locales)
: formatDateTimeAccurate(realDate, localTimeZone, locales)
: formatDateTimeAccurate(realDate, localTimeZone, locales);
? formatTimeOnly(realDate, localTimeZone, locales, hour12)
: formatDateTimeAccurate(realDate, localTimeZone, locales, hour12)
: formatDateTimeAccurate(realDate, localTimeZone, locales, hour12);
if (!showTooltip)
return <Fragment>{formattedDateTime.replace(/\s/g, String.fromCharCode(32))}</Fragment>;
@@ -241,7 +253,7 @@ export const DateTimeAccurate = ({
);
};
function formatDateTimeAccurate(date: Date, timeZone: string, locales: string[]): string {
function formatDateTimeAccurate(date: Date, timeZone: string, locales: string[], hour12: boolean = true): string {
const formattedDateTime = new Intl.DateTimeFormat(locales, {
month: "short",
day: "numeric",
@@ -251,26 +263,27 @@ function formatDateTimeAccurate(date: Date, timeZone: string, locales: string[])
timeZone,
// @ts-ignore fractionalSecondDigits works in most modern browsers
fractionalSecondDigits: 3,
hour12,
}).format(date);
return formattedDateTime;
}
export const DateTimeShort = ({ date, timeZone = "UTC" }: DateTimeProps) => {
export const DateTimeShort = ({ date, timeZone = "UTC", hour12 = true }: DateTimeProps) => {
const locales = useLocales();
const realDate = typeof date === "string" ? new Date(date) : date;
const initialFormattedDateTime = formatDateTimeShort(realDate, timeZone, locales);
const initialFormattedDateTime = formatDateTimeShort(realDate, timeZone, locales, hour12);
const [formattedDateTime, setFormattedDateTime] = useState<string>(initialFormattedDateTime);
useEffect(() => {
const resolvedOptions = Intl.DateTimeFormat().resolvedOptions();
setFormattedDateTime(formatDateTimeShort(realDate, resolvedOptions.timeZone, locales));
}, [locales, realDate]);
setFormattedDateTime(formatDateTimeShort(realDate, resolvedOptions.timeZone, locales, hour12));
}, [locales, realDate, hour12]);
return <Fragment>{formattedDateTime.replace(/\s/g, String.fromCharCode(32))}</Fragment>;
};
function formatDateTimeShort(date: Date, timeZone: string, locales: string[]): string {
function formatDateTimeShort(date: Date, timeZone: string, locales: string[], hour12: boolean = true): string {
const formattedDateTime = new Intl.DateTimeFormat(locales, {
hour: "numeric",
minute: "numeric",
@@ -278,6 +291,7 @@ function formatDateTimeShort(date: Date, timeZone: string, locales: string[]): s
timeZone,
// @ts-ignore fractionalSecondDigits works in most modern browsers
fractionalSecondDigits: 3,
hour12,
}).format(date);
return formattedDateTime;
@@ -44,6 +44,24 @@ const variants = {
iconSize: "size-3 ml-0.5",
accessory: "pr-0.5",
},
"outline/large": {
container: "px-1 h-10 w-full rounded border border-grid-bright hover:border-charcoal-550",
input: "px-2 rounded text-sm",
iconSize: "size-4 ml-1",
accessory: "pr-1",
},
"outline/medium": {
container: "px-1 h-8 w-full rounded border border-grid-bright hover:border-charcoal-550",
input: "px-1 rounded text-sm",
iconSize: "size-4 ml-0.5",
accessory: "pr-1",
},
"outline/small": {
container: "px-1 h-6 w-full rounded border border-grid-bright hover:border-charcoal-550",
input: "px-1 rounded text-xs",
iconSize: "size-3 ml-0.5",
accessory: "pr-0.5",
},
};
export type InputProps = React.InputHTMLAttributes<HTMLInputElement> & {
@@ -0,0 +1,220 @@
import { MinusIcon, PlusIcon } from "@heroicons/react/20/solid";
import { type ChangeEvent, useRef } from "react";
import { cn } from "~/utils/cn";
type InputNumberStepperProps = Omit<JSX.IntrinsicElements["input"], "min" | "max" | "step"> & {
step?: number;
min?: number;
max?: number;
round?: boolean;
controlSize?: "base" | "large";
};
export function InputNumberStepper({
value,
onChange,
step = 50,
min,
max,
round = true,
controlSize = "base",
name,
id,
disabled = false,
readOnly = false,
className,
placeholder = "Type a number",
...props
}: InputNumberStepperProps) {
const inputRef = useRef<HTMLInputElement>(null);
const handleStepUp = () => {
if (!inputRef.current || disabled) return;
// If rounding is enabled, ensure we start from a rounded base before stepping
if (round) {
// If field is empty, treat as 0 (or min if provided) before stepping up
if (inputRef.current.value === "") {
inputRef.current.value = String(min ?? 0);
} else {
commitRoundedFromInput();
}
}
inputRef.current.stepUp();
const event = new Event("change", { bubbles: true });
inputRef.current.dispatchEvent(event);
};
const handleStepDown = () => {
if (!inputRef.current || disabled) return;
// If rounding is enabled, ensure we start from a rounded base before stepping
if (round) {
// If field is empty, treat as 0 (or min if provided) before stepping down
if (inputRef.current.value === "") {
inputRef.current.value = String(min ?? 0);
} else {
commitRoundedFromInput();
}
}
inputRef.current.stepDown();
const event = new Event("change", { bubbles: true });
inputRef.current.dispatchEvent(event);
};
const numericValue = value === "" ? NaN : (value as number);
const isMinDisabled = min !== undefined && !Number.isNaN(numericValue) && numericValue <= min;
const isMaxDisabled = max !== undefined && !Number.isNaN(numericValue) && numericValue >= max;
function clamp(val: number): number {
if (Number.isNaN(val)) return typeof value === "number" ? value : min ?? 0;
let next = val;
if (min !== undefined) next = Math.max(min, next);
if (max !== undefined) next = Math.min(max, next);
return next;
}
function roundToStep(val: number): number {
if (step <= 0) return val;
const base = min ?? 0;
const shifted = val - base;
const quotient = shifted / step;
const floored = Math.floor(quotient);
const ceiled = Math.ceil(quotient);
const down = base + floored * step;
const up = base + ceiled * step;
const distDown = Math.abs(val - down);
const distUp = Math.abs(up - val);
return distUp < distDown ? up : down;
}
function commitRoundedFromInput() {
if (!inputRef.current || disabled || readOnly) return;
const el = inputRef.current;
const raw = el.value;
if (raw === "") return; // do not coerce empty to 0; keep placeholder visible
const numeric = Number(raw);
if (Number.isNaN(numeric)) return; // ignore non-numeric
const rounded = clamp(roundToStep(numeric));
if (String(rounded) === String(value)) return;
// Update the real input's value for immediate UI feedback
el.value = String(rounded);
// Invoke consumer onChange with the real element as target/currentTarget
onChange?.({
target: el,
currentTarget: el,
} as unknown as ChangeEvent<HTMLInputElement>);
}
const sizeStyles = {
base: {
container: "h-9",
input: "text-sm px-3",
button: "size-6",
icon: "size-3.5",
gap: "gap-1 pr-1.5",
},
large: {
container: "h-11 rounded-md",
input: "text-base px-3.5",
button: "size-8",
icon: "size-5",
gap: "gap-[0.3125rem] pr-[0.3125rem]",
},
} as const;
const size = sizeStyles[controlSize];
return (
<div
className={cn(
"flex items-center rounded border border-charcoal-600 bg-tertiary transition hover:border-charcoal-550/80 hover:bg-charcoal-600/80",
size.container,
"has-[:focus-visible]:outline has-[:focus-visible]:outline-1 has-[:focus-visible]:outline-offset-0 has-[:focus-visible]:outline-text-link",
disabled && "cursor-not-allowed opacity-50",
className
)}
>
<input
ref={inputRef}
type="number"
id={id}
name={name}
value={value}
placeholder={placeholder}
onChange={(e) => {
// Allow empty string to pass through so user can clear the field
if (e.currentTarget.value === "") {
// reflect emptiness in the input and notify consumer as empty
if (inputRef.current) inputRef.current.value = "";
onChange?.({
target: e.currentTarget,
currentTarget: e.currentTarget,
} as ChangeEvent<HTMLInputElement>);
return;
}
onChange?.(e);
}}
onBlur={(e) => {
// If blur is caused by clicking our step buttons, we prevent pointerdown
// so blur shouldn't fire. This is for safety in case of keyboard focus move.
if (round) commitRoundedFromInput();
}}
onKeyDown={(e) => {
if (e.key === "Enter" && round) {
e.preventDefault();
commitRoundedFromInput();
}
}}
step={step}
min={min}
max={max}
disabled={disabled}
readOnly={readOnly}
className={cn(
"placeholder:text-muted-foreground h-full grow border-0 bg-transparent text-left text-text-bright outline-none ring-0 focus:border-0 focus:outline-none focus:ring-0 disabled:cursor-not-allowed",
size.input,
// Hide number input arrows
"[type=number]:border-0 [appearance:textfield] [&::-webkit-inner-spin-button]:appearance-none [&::-webkit-outer-spin-button]:appearance-none"
)}
{...props}
/>
<div className={cn("flex items-center", size.gap)}>
<button
type="button"
onClick={handleStepDown}
onPointerDown={(e) => e.preventDefault()}
disabled={disabled || isMinDisabled}
aria-label={`Decrease by ${step}`}
className={cn(
"flex items-center justify-center rounded border border-error/30 bg-error/20 transition",
size.button,
"hover:border-error/50 hover:bg-error/30",
"disabled:cursor-not-allowed disabled:opacity-40",
"focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-text-link"
)}
>
<MinusIcon className={cn("text-error", size.icon)} />
</button>
<button
type="button"
onClick={handleStepUp}
onPointerDown={(e) => e.preventDefault()}
disabled={disabled || isMaxDisabled}
aria-label={`Increase by ${step}`}
className={cn(
"flex items-center justify-center rounded border border-success/30 bg-success/10 transition",
size.button,
"hover:border-success/40 hover:bg-success/20",
"disabled:cursor-not-allowed disabled:opacity-40 disabled:hover:bg-transparent",
"focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-text-link"
)}
>
<PlusIcon className={cn("text-success", size.icon)} />
</button>
</div>
</div>
);
}
@@ -17,6 +17,10 @@ const paragraphVariants = {
text: "font-sans text-sm font-normal text-text-bright",
spacing: "mb-2",
},
"small/dimmed": {
text: "font-sans text-sm font-normal text-text-dimmed",
spacing: "mb-2",
},
"extra-small": {
text: "font-sans text-xs font-normal text-text-dimmed",
spacing: "mb-1.5",
@@ -25,6 +29,14 @@ const paragraphVariants = {
text: "font-sans text-xs font-normal text-text-bright",
spacing: "mb-1.5",
},
"extra-small/dimmed": {
text: "font-sans text-xs font-normal text-text-dimmed",
spacing: "mb-1.5",
},
"extra-small/dimmed/mono": {
text: "font-mono text-xs font-normal text-text-dimmed",
spacing: "mb-1.5",
},
"extra-small/mono": {
text: "font-mono text-xs font-normal text-text-dimmed",
spacing: "mb-1.5",
+85 -13
View File
@@ -1,12 +1,15 @@
import { ExclamationCircleIcon, XMarkIcon } from "@heroicons/react/20/solid";
import { EnvelopeIcon, ExclamationCircleIcon, XMarkIcon } from "@heroicons/react/20/solid";
import { CheckCircleIcon } from "@heroicons/react/24/solid";
import { Toaster, toast } from "sonner";
import { useTypedLoaderData } from "remix-typedjson";
import { loader } from "~/root";
import { useSearchParams } from "@remix-run/react";
import { useEffect } from "react";
import { Paragraph } from "./Paragraph";
import { useTypedLoaderData } from "remix-typedjson";
import { Toaster, toast } from "sonner";
import { type ToastMessageAction } from "~/models/message.server";
import { type loader } from "~/root";
import { cn } from "~/utils/cn";
import { Button, LinkButton } from "./Buttons";
import { Header2 } from "./Headers";
import { Paragraph } from "./Paragraph";
const defaultToastDuration = 5000;
const permanentToastDuration = 60 * 60 * 24 * 1000;
@@ -19,9 +22,22 @@ export function Toast() {
}
const { message, type, options } = toastMessage;
toast.custom((t) => <ToastUI variant={type} message={message} t={t as string} />, {
duration: options.ephemeral ? defaultToastDuration : permanentToastDuration,
});
const ephemeral = options.action ? false : options.ephemeral;
toast.custom(
(t) => (
<ToastUI
variant={type}
message={message}
t={t as string}
title={options.title}
action={options.action}
/>
),
{
duration: ephemeral ? defaultToastDuration : permanentToastDuration,
}
);
}, [toastMessage]);
return <Toaster />;
@@ -32,11 +48,15 @@ export function ToastUI({
message,
t,
toastWidth = 356, // Default width, matches what sonner provides by default
title,
action,
}: {
variant: "error" | "success";
message: string;
t: string;
toastWidth?: string | number;
title?: string;
action?: ToastMessageAction;
}) {
return (
<div
@@ -51,13 +71,19 @@ export function ToastUI({
>
<div className="flex w-full items-start gap-2 rounded-lg p-3">
{variant === "success" ? (
<CheckCircleIcon className="mt-1 size-6 min-w-6 text-success" />
<CheckCircleIcon className="mt-1 size-4 min-w-4 text-success" />
) : (
<ExclamationCircleIcon className="mt-1 size-6 min-w-6 text-error" />
<ExclamationCircleIcon className="mt-1 size-4 min-w-4 text-error" />
)}
<Paragraph className="py-1 text-text-bright">{message}</Paragraph>
<div className="flex flex-col">
{title && <Header2 className="pt-0">{title}</Header2>}
<Paragraph variant="small/dimmed" className="pb-1 pt-0.5">
{message}
</Paragraph>
<Action action={action} toastId={t} className="my-2" />
</div>
<button
className="hover:bg-midnight-800 ms-auto rounded p-2 text-text-dimmed transition hover:text-text-bright"
className="hover:bg-midnight-800 -mr-1 -mt-1 ms-auto rounded p-2 text-text-dimmed transition hover:text-text-bright"
onClick={() => toast.dismiss(t)}
>
<XMarkIcon className="size-4" />
@@ -66,3 +92,49 @@ export function ToastUI({
</div>
);
}
function Action({
action,
toastId,
className,
}: {
action?: ToastMessageAction;
toastId: string;
className?: string;
}) {
const [_, setSearchParams] = useSearchParams();
if (!action) return null;
switch (action.action.type) {
case "link": {
return (
<LinkButton
className={className}
variant={action.variant ?? "secondary/small"}
to={action.action.path}
>
{action.label}
</LinkButton>
);
}
case "help": {
const feedbackType = action.action.feedbackType;
return (
<Button
className={className}
variant={action.variant ?? "secondary/small"}
LeadingIcon={EnvelopeIcon}
onClick={() => {
setSearchParams({
feedbackPanel: feedbackType,
});
toast.dismiss(toastId);
}}
>
{action.label}
</Button>
);
}
}
}
@@ -39,14 +39,14 @@ export function concreteStateFromInput({
}
const nodes = concreteStateFromPartialState(tree, state);
return {
return applyFilterToState({
tree,
nodes,
changes: { selectedId },
filter,
filteredNodes: nodes,
visibleNodeIds: visibleNodes(tree, nodes).map((node) => node.id),
};
visibleNodeIds: [],
});
}
export function concreteStateFromPartialState<TData>(
@@ -20,6 +20,7 @@ import { TriggerIcon } from "~/assets/icons/TriggerIcon";
import { PythonLogoIcon } from "~/assets/icons/PythonLogoIcon";
import { TraceIcon } from "~/assets/icons/TraceIcon";
import { WaitpointTokenIcon } from "~/assets/icons/WaitpointTokenIcon";
import { StreamsIcon } from "~/assets/icons/StreamsIcon";
type TaskIconProps = {
name: string | undefined;
@@ -97,6 +98,7 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
return <RunFunctionIcon className={cn(className, "text-text-dimmed")} />;
case "task-hook-init":
case "task-hook-onStart":
case "task-hook-onStartAttempt":
case "task-hook-onSuccess":
case "task-hook-onWait":
case "task-hook-onResume":
@@ -107,6 +109,8 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
case "task-hook-onFailure":
case "task-hook-catchError":
return <FunctionIcon className={cn(className, "text-error")} />;
case "streams":
return <StreamsIcon className={cn(className, "text-text-dimmed")} />;
}
return <InformationCircleIcon className={cn(className, "text-text-dimmed")} />;
@@ -31,7 +31,7 @@ import {
type NextRunListItem,
} from "~/presenters/v3/NextRunListPresenter.server";
import { formatCurrencyAccurate } from "~/utils/numberFormatter";
import { docsPath, v3RunSpanPath, v3TestPath } from "~/utils/pathBuilder";
import { docsPath, v3RunSpanPath, v3TestPath,v3TestTaskPath } from "~/utils/pathBuilder";
import { DateTime } from "../../primitives/DateTime";
import { Paragraph } from "../../primitives/Paragraph";
import { Spinner } from "../../primitives/Spinner";
@@ -565,6 +565,8 @@ function BlankState({ isLoading, filters }: Pick<RunsTableProps, "isLoading" | "
if (isLoading) return <TableBlankRow colSpan={15}></TableBlankRow>;
const { tasks, from, to, ...otherFilters } = filters;
const singleTaskFromFilters = filters.tasks.length === 1 ? filters.tasks[0] : null;
const testPath = singleTaskFromFilters ? v3TestTaskPath(organization, project, environment, {taskIdentifier: singleTaskFromFilters}) : v3TestPath(organization, project, environment);
if (
filters.tasks.length === 1 &&
@@ -579,7 +581,7 @@ function BlankState({ isLoading, filters }: Pick<RunsTableProps, "isLoading" | "
</Paragraph>
<div className="mt-6 flex items-center justify-center gap-2">
<LinkButton
to={v3TestPath(organization, project, environment)}
to={testPath}
variant="tertiary/medium"
LeadingIcon={BeakerIcon}
className="inline-flex"
@@ -620,7 +622,7 @@ function BlankState({ isLoading, filters }: Pick<RunsTableProps, "isLoading" | "
<LinkButton
LeadingIcon={BeakerIcon}
variant="tertiary/medium"
to={v3TestPath(organization, project, environment)}
to={testPath}
>
Run a test
</LinkButton>
+53 -2
View File
@@ -25,6 +25,27 @@ const GithubAppEnvSchema = z.preprocess(
])
);
// eventually we can make all S2 env vars required once the S2 OSS version is out
const S2EnvSchema = z.preprocess(
(val) => {
const obj = val as any;
if (!obj || !obj.S2_ENABLED) {
return { ...obj, S2_ENABLED: "0" };
}
return obj;
},
z.discriminatedUnion("S2_ENABLED", [
z.object({
S2_ENABLED: z.literal("1"),
S2_ACCESS_TOKEN: z.string(),
S2_DEPLOYMENT_LOGS_BASIN_NAME: z.string(),
}),
z.object({
S2_ENABLED: z.literal("0"),
}),
])
);
const EnvironmentSchema = z
.object({
NODE_ENV: z.union([z.literal("development"), z.literal("production"), z.literal("test")]),
@@ -198,6 +219,7 @@ const EnvironmentSchema = z
.string()
.default(process.env.REDIS_TLS_DISABLED ?? "false"),
REALTIME_STREAMS_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
REALTIME_STREAMS_INACTIVITY_TIMEOUT_MS: z.coerce.number().int().default(60000), // 1 minute
REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS: z.coerce
.number()
@@ -323,6 +345,12 @@ const EnvironmentSchema = z
OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
OBJECT_STORE_REGION: z.string().optional(),
OBJECT_STORE_SERVICE: z.string().default("s3"),
ARTIFACTS_OBJECT_STORE_BUCKET: z.string().optional(),
ARTIFACTS_OBJECT_STORE_BASE_URL: z.string().optional(),
ARTIFACTS_OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
ARTIFACTS_OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
ARTIFACTS_OBJECT_STORE_REGION: z.string().optional(),
EVENTS_BATCH_SIZE: z.coerce.number().int().default(100),
EVENTS_BATCH_INTERVAL: z.coerce.number().int().default(1000),
EVENTS_DEFAULT_LOG_RETENTION: z.coerce.number().int().default(7),
@@ -1127,8 +1155,15 @@ const EnvironmentSchema = z
EVENTS_CLICKHOUSE_WAIT_FOR_ASYNC_INSERT: z.string().default("1"),
EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE: z.coerce.number().int().default(10485760),
EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS: z.coerce.number().int().default(5000),
EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS: z.coerce
.number()
.int()
.default(60_000 * 5), // 5 minutes
EVENT_REPOSITORY_CLICKHOUSE_ROLLOUT_PERCENT: z.coerce.number().optional(),
EVENT_REPOSITORY_DEFAULT_STORE: z.enum(["postgres", "clickhouse"]).default("postgres"),
EVENT_REPOSITORY_DEFAULT_STORE: z
.enum(["postgres", "clickhouse", "clickhouse_v2"])
.default("postgres"),
EVENT_REPOSITORY_DEBUG_LOGS_DISABLED: BoolEnv.default(false),
EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),
EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(5_000),
EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING: z.coerce.number().int().default(2000),
@@ -1201,8 +1236,24 @@ const EnvironmentSchema = z
EVENT_LOOP_MONITOR_UTILIZATION_SAMPLE_RATE: z.coerce.number().default(0.05),
VERY_SLOW_QUERY_THRESHOLD_MS: z.coerce.number().int().optional(),
REALTIME_STREAMS_S2_BASIN: z.string().optional(),
REALTIME_STREAMS_S2_ACCESS_TOKEN: z.string().optional(),
REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS: z.coerce
.number()
.int()
.default(60_000 * 60 * 24), // 1 day
REALTIME_STREAMS_S2_LOG_LEVEL: z
.enum(["log", "error", "warn", "info", "debug"])
.default("info"),
REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS: z.coerce.number().int().default(100),
REALTIME_STREAMS_S2_MAX_RETRIES: z.coerce.number().int().default(10),
REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(60),
REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),
WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),
})
.and(GithubAppEnvSchema);
.and(GithubAppEnvSchema)
.and(S2EnvSchema);
export type Environment = z.infer<typeof EnvironmentSchema>;
export const env = EnvironmentSchema.parse(process.env);
+22 -2
View File
@@ -1,7 +1,8 @@
import { json, Session } from "@remix-run/node";
import { createCookieSessionStorage } from "@remix-run/node";
import { json, createCookieSessionStorage, type Session } from "@remix-run/node";
import { redirect, typedjson } from "remix-typedjson";
import { ButtonVariant } from "~/components/primitives/Buttons";
import { env } from "~/env.server";
import { type FeedbackType } from "~/routes/resources.feedback";
export type ToastMessage = {
message: string;
@@ -9,9 +10,26 @@ export type ToastMessage = {
options: Required<ToastMessageOptions>;
};
export type ToastMessageAction = {
label: string;
variant?: ButtonVariant;
action:
| {
type: "link";
path: string;
}
| {
type: "help";
feedbackType: FeedbackType;
};
};
export type ToastMessageOptions = {
title?: string;
/** Ephemeral means it disappears after a delay, defaults to true */
ephemeral?: boolean;
/** This display a button and make it not ephemeral, unless ephemeral is explicitlyset to false */
action?: ToastMessageAction;
};
const ONE_YEAR = 1000 * 60 * 60 * 24 * 365;
@@ -36,6 +54,7 @@ export function setSuccessMessage(
message,
type: "success",
options: {
...options,
ephemeral: options?.ephemeral ?? true,
},
} as ToastMessage);
@@ -46,6 +65,7 @@ export function setErrorMessage(session: Session, message: string, options?: Toa
message,
type: "error",
options: {
...options,
ephemeral: options?.ephemeral ?? true,
},
} as ToastMessage);
@@ -12,7 +12,7 @@ import { prisma, type PrismaClientOrTransaction } from "~/db.server";
import { env } from "~/env.server";
import { featuresForUrl } from "~/features.server";
import { createApiKeyForEnv, createPkApiKeyForEnv, envSlug } from "./api-key.server";
import { getDefaultEnvironmentConcurrencyLimit } from "~/services/platform.v3.server";
export type { Organization };
const nanoid = customAlphabet("1234567890abcdef", 4);
@@ -66,7 +66,7 @@ export async function createOrganization(
role: "ADMIN",
},
},
v3Enabled: !features.isManagedCloud,
v3Enabled: true,
},
include: {
members: true,
@@ -96,6 +96,8 @@ export async function createEnvironment({
const pkApiKey = createPkApiKeyForEnv(type);
const shortcode = createShortcode().join("-");
const limit = await getDefaultEnvironmentConcurrencyLimit(organization.id, type);
return await prismaClient.runtimeEnvironment.create({
data: {
slug,
@@ -103,7 +105,7 @@ export async function createEnvironment({
pkApiKey,
shortcode,
autoEnableInternalSources: type !== "DEVELOPMENT",
maximumConcurrencyLimit: organization.maximumConcurrencyLimit / 3,
maximumConcurrencyLimit: limit,
organization: {
connect: {
id: organization.id,
+27
View File
@@ -16,12 +16,26 @@ type Options = {
version: "v2" | "v3";
};
export class ExceededProjectLimitError extends Error {
constructor(message: string) {
super(message);
this.name = "ExceededProjectLimitError";
}
}
export async function createProject(
{ organizationSlug, name, userId, version }: Options,
attemptCount = 0
): Promise<Project & { organization: Organization }> {
//check the user has permissions to do this
const organization = await prisma.organization.findFirst({
select: {
id: true,
slug: true,
v3Enabled: true,
maximumConcurrencyLimit: true,
maximumProjectCount: true,
},
where: {
slug: organizationSlug,
members: { some: { userId } },
@@ -40,6 +54,19 @@ export async function createProject(
}
}
const projectCount = await prisma.project.count({
where: {
organizationId: organization.id,
deletedAt: null,
},
});
if (projectCount >= organization.maximumProjectCount) {
throw new ExceededProjectLimitError(
`This organization has reached the maximum number of projects (${organization.maximumProjectCount}).`
);
}
//ensure the slug is globally unique
const uniqueProjectSlug = `${slug(name)}-${nanoid(4)}`;
const projectWithSameSlug = await prisma.project.findFirst({
@@ -3,7 +3,7 @@ import {
ExternalBuildData,
prepareDeploymentError,
} from "@trigger.dev/core/v3";
import { RuntimeEnvironment, type WorkerDeployment } from "@trigger.dev/database";
import { type RuntimeEnvironment, type WorkerDeployment } from "@trigger.dev/database";
import { type PrismaClient, prisma } from "~/db.server";
import { type Organization } from "~/models/organization.server";
import { type Project } from "~/models/project.server";
@@ -11,6 +11,24 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { type User } from "~/models/user.server";
import { getUsername } from "~/utils/username";
import { processGitMetadata } from "./BranchesPresenter.server";
import { S2 } from "@s2-dev/streamstore";
import { env } from "~/env.server";
import { createRedisClient } from "~/redis.server";
import { tryCatch } from "@trigger.dev/core";
import { logger } from "~/services/logger.server";
const S2_TOKEN_KEY_PREFIX = "s2-token:project:";
const s2TokenRedis = createRedisClient("s2-token-cache", {
host: env.CACHE_REDIS_HOST,
port: env.CACHE_REDIS_PORT,
username: env.CACHE_REDIS_USERNAME,
password: env.CACHE_REDIS_PASSWORD,
tlsDisabled: env.CACHE_REDIS_TLS_DISABLED === "true",
clusterMode: env.CACHE_REDIS_CLUSTER_MODE_ENABLED === "1",
});
const s2 = env.S2_ENABLED === "1" ? new S2({ accessToken: env.S2_ACCESS_TOKEN }) : undefined;
export type ErrorData = {
name: string;
@@ -43,6 +61,7 @@ export class DeploymentPresenter {
select: {
id: true,
organizationId: true,
externalRef: true,
},
where: {
slug: projectSlug,
@@ -138,11 +157,31 @@ export class DeploymentPresenter {
},
});
const gitMetadata = processGitMetadata(deployment.git);
const externalBuildData = deployment.externalBuildData
? ExternalBuildData.safeParse(deployment.externalBuildData)
: undefined;
let eventStream = undefined;
if (env.S2_ENABLED === "1" && gitMetadata?.source === "trigger_github_app") {
const [error, accessToken] = await tryCatch(this.getS2AccessToken(project.externalRef));
if (error) {
logger.error("Failed getting S2 access token", { error });
} else {
eventStream = {
s2: {
basin: env.S2_DEPLOYMENT_LOGS_BASIN_NAME,
stream: `projects/${project.externalRef}/deployments/${deployment.shortCode}`,
accessToken,
},
};
}
}
return {
eventStream,
deployment: {
id: deployment.id,
shortCode: deployment.shortCode,
@@ -178,11 +217,46 @@ export class DeploymentPresenter {
errorData: DeploymentPresenter.prepareErrorData(deployment.errorData),
isBuilt: !!deployment.builtAt,
type: deployment.type,
git: processGitMetadata(deployment.git),
git: gitMetadata,
},
};
}
private async getS2AccessToken(projectRef: string): Promise<string> {
if (env.S2_ENABLED !== "1" || !s2) {
throw new Error("Failed getting S2 access token: S2 is not enabled");
}
const redisKey = `${S2_TOKEN_KEY_PREFIX}${projectRef}`;
const cachedToken = await s2TokenRedis.get(redisKey);
if (cachedToken) {
return cachedToken;
}
const { access_token: accessToken } = await s2.accessTokens.issue({
id: `${projectRef}-${new Date().getTime()}`,
expires_at: new Date(Date.now() + 60 * 60 * 1000).toISOString(), // 1 hour
scope: {
ops: ["read"],
basins: {
exact: env.S2_DEPLOYMENT_LOGS_BASIN_NAME,
},
streams: {
prefix: `projects/${projectRef}/deployments/`,
},
},
});
await s2TokenRedis.setex(
redisKey,
59 * 60, // slightly shorter than the token validity period
accessToken
);
return accessToken;
}
public static prepareErrorData(errorData: WorkerDeployment["errorData"]): ErrorData | undefined {
if (!errorData) {
return;
@@ -0,0 +1,141 @@
import { type RuntimeEnvironmentType } from "@trigger.dev/database";
import {
getCurrentPlan,
getDefaultEnvironmentLimitFromPlan,
getPlans,
} from "~/services/platform.v3.server";
import { BasePresenter } from "./basePresenter.server";
import { sortEnvironments } from "~/utils/environmentSort";
export type ConcurrencyResult = {
canAddConcurrency: boolean;
environments: EnvironmentWithConcurrency[];
extraConcurrency: number;
extraAllocatedConcurrency: number;
extraUnallocatedConcurrency: number;
maxQuota: number;
concurrencyPricing: {
stepSize: number;
centsPerStep: number;
};
};
export type EnvironmentWithConcurrency = {
id: string;
type: RuntimeEnvironmentType;
isBranchableEnvironment: boolean;
branchName: string | null;
parentEnvironmentId: string | null;
maximumConcurrencyLimit: number;
planConcurrencyLimit: number;
};
export class ManageConcurrencyPresenter extends BasePresenter {
public async call({
userId,
projectId,
organizationId,
}: {
userId: string;
projectId: string;
organizationId: string;
}): Promise<ConcurrencyResult> {
// Get plan
const currentPlan = await getCurrentPlan(organizationId);
if (!currentPlan) {
throw new Error("No plan found");
}
const canAddConcurrency =
currentPlan.v3Subscription.plan?.limits.concurrentRuns.canExceed === true;
const environments = await this._replica.runtimeEnvironment.findMany({
select: {
id: true,
projectId: true,
type: true,
branchName: true,
parentEnvironmentId: true,
isBranchableEnvironment: true,
maximumConcurrencyLimit: true,
orgMember: {
select: {
userId: true,
},
},
project: {
select: {
deletedAt: true,
},
},
},
where: {
organizationId,
archivedAt: null,
},
});
const extraConcurrency = currentPlan?.v3Subscription.addOns?.concurrentRuns?.purchased ?? 0;
// Go through all environments and add up extra concurrency above their allowed allocation
let extraAllocatedConcurrency = 0;
const projectEnvironments: EnvironmentWithConcurrency[] = [];
for (const environment of environments) {
// Don't count parent environments
if (environment.isBranchableEnvironment) continue;
// Don't count deleted projects
if (environment.project.deletedAt) continue;
const limit = currentPlan
? getDefaultEnvironmentLimitFromPlan(environment.type, currentPlan)
: 0;
if (!limit) continue;
// If it's not DEV and they've increased, track that
// You can't spend money to increase DEV concurrency
if (environment.type !== "DEVELOPMENT" && environment.maximumConcurrencyLimit > limit) {
extraAllocatedConcurrency += environment.maximumConcurrencyLimit - limit;
}
// We only want to show this project's environments
if (environment.projectId === projectId) {
if (environment.type === "DEVELOPMENT" && environment.orgMember?.userId !== userId) {
continue;
}
projectEnvironments.push({
id: environment.id,
type: environment.type,
isBranchableEnvironment: environment.isBranchableEnvironment,
branchName: environment.branchName,
parentEnvironmentId: environment.parentEnvironmentId,
maximumConcurrencyLimit: environment.maximumConcurrencyLimit,
planConcurrencyLimit: limit,
});
}
}
const extraAllocated = Math.min(extraConcurrency, extraAllocatedConcurrency);
const plans = await getPlans();
if (!plans) {
throw new Error("Couldn't retrieve add on pricing");
}
return {
canAddConcurrency,
extraConcurrency,
extraAllocatedConcurrency: extraAllocated,
extraUnallocatedConcurrency: extraConcurrency - extraAllocated,
maxQuota: currentPlan.v3Subscription.addOns?.concurrentRuns?.quota ?? 0,
environments: sortEnvironments(projectEnvironments, [
"PRODUCTION",
"STAGING",
"PREVIEW",
"DEVELOPMENT",
]),
concurrencyPricing: plans.addOnPricing.concurrency,
};
}
}
@@ -19,6 +19,7 @@ import { WaitpointPresenter } from "./WaitpointPresenter.server";
import { engine } from "~/v3/runEngine.server";
import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
import { IEventRepository, SpanDetail } from "~/v3/eventRepository/eventRepository.types";
import { safeJsonParse } from "~/utils/json";
type Result = Awaited<ReturnType<SpanPresenter["call"]>>;
export type Span = NonNullable<NonNullable<Result>["span"]>;
@@ -272,6 +273,7 @@ export class SpanPresenter extends BasePresenter {
spanId: run.spanId,
isCached: !!originalRunId,
machinePreset: machine?.name,
taskEventStore: run.taskEventStore,
externalTraceId,
};
}
@@ -496,7 +498,18 @@ export class SpanPresenter extends BasePresenter {
duration: span.duration,
events: span.events,
style: span.style,
properties: span.properties ? JSON.stringify(span.properties, null, 2) : undefined,
properties:
span.properties &&
typeof span.properties === "object" &&
Object.keys(span.properties).length > 0
? JSON.stringify(span.properties, null, 2)
: undefined,
resourceProperties:
span.resourceProperties &&
typeof span.resourceProperties === "object" &&
Object.keys(span.resourceProperties).length > 0
? JSON.stringify(span.resourceProperties, null, 2)
: undefined,
entity: span.entity,
metadata: span.metadata,
triggeredRuns,
@@ -551,6 +564,41 @@ export class SpanPresenter extends BasePresenter {
},
};
}
case "realtime-stream": {
if (!span.entity.id) {
logger.error(`SpanPresenter: No realtime stream id`, {
spanId,
realtimeStreamId: span.entity.id,
});
return { ...data, entity: null };
}
const [runId, streamKey] = span.entity.id.split(":");
if (!runId || !streamKey) {
logger.error(`SpanPresenter: Invalid realtime stream id`, {
spanId,
realtimeStreamId: span.entity.id,
});
return { ...data, entity: null };
}
const metadata = span.entity.metadata
? (safeJsonParse(span.entity.metadata) as Record<string, unknown> | undefined)
: undefined;
return {
...data,
entity: {
type: "realtime-stream" as const,
object: {
runId,
streamKey,
metadata,
},
},
};
}
default:
return { ...data, entity: null };
}
@@ -0,0 +1,808 @@
import { conform, useFieldList, useForm } from "@conform-to/react";
import { parse } from "@conform-to/zod";
import {
EnvelopeIcon,
ExclamationTriangleIcon,
InformationCircleIcon,
PlusIcon,
} from "@heroicons/react/20/solid";
import { DialogClose } from "@radix-ui/react-dialog";
import {
Form,
useActionData,
useNavigate,
useNavigation,
useSearchParams,
type MetaFunction,
} from "@remix-run/react";
import { json, type ActionFunctionArgs, type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { tryCatch } from "@trigger.dev/core";
import { useEffect, useState } from "react";
import { typedjson, useTypedLoaderData } from "remix-typedjson";
import { z } from "zod";
import { AdminDebugTooltip } from "~/components/admin/debugTooltip";
import { EnvironmentCombo } from "~/components/environments/EnvironmentLabel";
import {
MainHorizontallyCenteredContainer,
PageBody,
PageContainer,
} from "~/components/layout/AppLayout";
import { Button, LinkButton } from "~/components/primitives/Buttons";
import { Dialog, DialogContent, DialogHeader, DialogTrigger } from "~/components/primitives/Dialog";
import { Fieldset } from "~/components/primitives/Fieldset";
import { FormButtons } from "~/components/primitives/FormButtons";
import { FormError } from "~/components/primitives/FormError";
import { Header2, Header3 } from "~/components/primitives/Headers";
import { Input } from "~/components/primitives/Input";
import { InputGroup } from "~/components/primitives/InputGroup";
import { InputNumberStepper } from "~/components/primitives/InputNumberStepper";
import { Label } from "~/components/primitives/Label";
import { NavBar, PageAccessories, PageTitle } from "~/components/primitives/PageHeader";
import { Paragraph } from "~/components/primitives/Paragraph";
import * as Property from "~/components/primitives/PropertyTable";
import {
Table,
TableBody,
TableCell,
TableHeader,
TableHeaderCell,
TableRow,
} from "~/components/primitives/Table";
import { InfoIconTooltip } from "~/components/primitives/Tooltip";
import { useFeatures } from "~/hooks/useFeatures";
import { useOrganization } from "~/hooks/useOrganizations";
import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
import { findProjectBySlug } from "~/models/project.server";
import {
ManageConcurrencyPresenter,
type ConcurrencyResult,
type EnvironmentWithConcurrency,
} from "~/presenters/v3/ManageConcurrencyPresenter.server";
import { getPlans } from "~/services/platform.v3.server";
import { requireUserId } from "~/services/session.server";
import { formatCurrency, formatNumber } from "~/utils/numberFormatter";
import { concurrencyPath, EnvironmentParamSchema, v3BillingPath } from "~/utils/pathBuilder";
import { SetConcurrencyAddOnService } from "~/v3/services/setConcurrencyAddOn.server";
import { useCurrentPlan } from "../_app.orgs.$organizationSlug/route";
import { SpinnerWhite } from "~/components/primitives/Spinner";
import { cn } from "~/utils/cn";
import { logger } from "~/services/logger.server";
import { AllocateConcurrencyService } from "~/v3/services/allocateConcurrency.server";
export const meta: MetaFunction = () => {
return [
{
title: `Manage concurrency | Trigger.dev`,
},
];
};
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
if (!project) {
throw new Response(undefined, {
status: 404,
statusText: "Project not found",
});
}
const presenter = new ManageConcurrencyPresenter();
const [error, result] = await tryCatch(
presenter.call({
userId: userId,
projectId: project.id,
organizationId: project.organizationId,
})
);
if (error) {
throw new Response(undefined, {
status: 400,
statusText: error.message,
});
}
const plans = await tryCatch(getPlans());
if (!plans) {
throw new Response(null, { status: 404, statusText: "Plans not found" });
}
return typedjson(result);
};
const FormSchema = z.discriminatedUnion("action", [
z.object({
action: z.enum(["purchase"]),
amount: z.coerce.number().min(0, "Amount must be 0 or more"),
}),
z.object({
action: z.enum(["quota-increase"]),
amount: z.coerce.number().min(1, "Amount must be greater than 0"),
}),
z.object({
action: z.enum(["allocate"]),
// It will only update environments that are passed in
environments: z.array(
z.object({
id: z.string(),
amount: z.coerce.number().min(0, "Amount must be 0 or more"),
})
),
}),
]);
export const action = async ({ request, params }: ActionFunctionArgs) => {
const userId = await requireUserId(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
const redirectPath = concurrencyPath(
{ slug: organizationSlug },
{ slug: projectParam },
{ slug: envParam }
);
if (!project) {
throw redirectWithErrorMessage(redirectPath, request, "Project not found");
}
const formData = await request.formData();
const submission = parse(formData, { schema: FormSchema });
if (!submission.value || submission.intent !== "submit") {
return json(submission);
}
if (submission.value.action === "allocate") {
const allocate = new AllocateConcurrencyService();
const [error, result] = await tryCatch(
allocate.call({
userId,
projectId: project.id,
organizationId: project.organizationId,
environments: submission.value.environments,
})
);
if (error) {
submission.error.environments = [error instanceof Error ? error.message : "Unknown error"];
return json(submission);
}
if (!result.success) {
submission.error.environments = [result.error];
return json(submission);
}
return redirectWithSuccessMessage(
`${redirectPath}?success=true`,
request,
"Concurrency allocated successfully"
);
}
const service = new SetConcurrencyAddOnService();
const [error, result] = await tryCatch(
service.call({
userId,
projectId: project.id,
organizationId: project.organizationId,
action: submission.value.action,
amount: submission.value.amount,
})
);
if (error) {
submission.error.amount = [error instanceof Error ? error.message : "Unknown error"];
return json(submission);
}
if (!result.success) {
submission.error.amount = [result.error];
return json(submission);
}
return redirectWithSuccessMessage(
`${redirectPath}?success=true`,
request,
submission.value.action === "purchase"
? "Concurrency updated successfully"
: "Requested extra concurrency, we'll get back to you soon."
);
};
export default function Page() {
const {
canAddConcurrency,
extraConcurrency,
extraAllocatedConcurrency,
extraUnallocatedConcurrency,
environments,
concurrencyPricing,
maxQuota,
} = useTypedLoaderData<typeof loader>();
return (
<PageContainer>
<NavBar>
<PageTitle title="Concurrency" />
<PageAccessories>
<AdminDebugTooltip>
<Property.Table>
{environments.map((environment) => (
<Property.Item key={environment.id}>
<Property.Label>
{environment.type}{" "}
{environment.branchName ? ` (${environment.branchName})` : ""}
</Property.Label>
<Property.Value>{environment.id}</Property.Value>
</Property.Item>
))}
</Property.Table>
</AdminDebugTooltip>
</PageAccessories>
</NavBar>
<PageBody scrollable={true}>
<MainHorizontallyCenteredContainer>
{canAddConcurrency ? (
<Upgradable
canAddConcurrency={canAddConcurrency}
extraConcurrency={extraConcurrency}
extraAllocatedConcurrency={extraAllocatedConcurrency}
extraUnallocatedConcurrency={extraUnallocatedConcurrency}
environments={environments}
concurrencyPricing={concurrencyPricing}
maxQuota={maxQuota}
/>
) : (
<NotUpgradable environments={environments} />
)}
</MainHorizontallyCenteredContainer>
</PageBody>
</PageContainer>
);
}
function initialAllocation(environments: ConcurrencyResult["environments"]) {
return new Map<string, number>(
environments
.filter((e) => e.type !== "DEVELOPMENT")
.map((e) => [e.id, Math.max(0, e.maximumConcurrencyLimit - e.planConcurrencyLimit)])
);
}
function allocationTotal(environments: ConcurrencyResult["environments"]) {
const allocation = initialAllocation(environments);
return Array.from(allocation.values()).reduce((e, acc) => e + acc, 0);
}
function Upgradable({
extraConcurrency,
extraAllocatedConcurrency,
extraUnallocatedConcurrency,
environments,
concurrencyPricing,
maxQuota,
}: ConcurrencyResult) {
const lastSubmission = useActionData();
const [form, { environments: formEnvironments }] = useForm({
id: "purchase-concurrency",
// TODO: type this
lastSubmission: lastSubmission as any,
onValidate({ formData }) {
return parse(formData, { schema: FormSchema });
},
shouldRevalidate: "onSubmit",
});
const navigation = useNavigation();
const isLoading = navigation.state !== "idle" && navigation.formMethod === "POST";
const [allocation, setAllocation] = useState(initialAllocation(environments));
const allocatedInProject = Array.from(allocation.values()).reduce((e, acc) => e + acc, 0);
const initialAllocationInProject = allocationTotal(environments);
const changeInAllocation = allocatedInProject - initialAllocationInProject;
const unallocated = extraUnallocatedConcurrency - changeInAllocation;
const allocationModified = changeInAllocation !== 0;
return (
<div className="flex flex-col gap-3">
<div className="border-b border-grid-dimmed pb-1">
<Header2>Manage your concurrency</Header2>
</div>
<Paragraph variant="small">
Concurrency limits determine how many runs you can execute at the same time. You can add
extra concurrency to your organization which you can allocate to environments in your
projects.
</Paragraph>
<div className="mt-3 flex flex-col gap-6">
<div className="flex flex-col gap-2">
<div className="flex items-center first-letter:pb-1">
<Header3 className="grow">Extra concurrency</Header3>
<PurchaseConcurrencyModal
concurrencyPricing={concurrencyPricing}
extraConcurrency={extraConcurrency}
extraUnallocatedConcurrency={extraUnallocatedConcurrency}
maxQuota={maxQuota}
disabled={unallocated < 0 ? false : allocationModified}
/>
</div>
<Table>
<TableBody>
<TableRow>
<TableCell className="pl-0 text-text-bright">Extra concurrency purchased</TableCell>
<TableCell alignment="right" className="tabular-nums text-text-bright">
{extraConcurrency}
</TableCell>
</TableRow>
<TableRow>
<TableCell>Allocated concurrency</TableCell>
<TableCell alignment="right" className={"tabular-nums text-text-bright"}>
{allocationModified ? (
<>
<span className="text-text-dimmed line-through">
{extraAllocatedConcurrency}
</span>{" "}
{extraAllocatedConcurrency + changeInAllocation}
</>
) : (
extraAllocatedConcurrency
)}
</TableCell>
</TableRow>
<TableRow>
<TableCell>Unallocated concurrency</TableCell>
<TableCell
alignment="right"
className={cn(
"tabular-nums",
unallocated > 0
? "text-success"
: unallocated < 0
? "text-error"
: "text-text-bright"
)}
>
{allocationModified ? (
<>
<span className="text-text-dimmed line-through">
{extraUnallocatedConcurrency}
</span>{" "}
{extraUnallocatedConcurrency - changeInAllocation}
</>
) : (
extraUnallocatedConcurrency
)}
</TableCell>
</TableRow>
<TableRow className={allocationModified ? undefined : "after:bg-transparent"}>
<TableCell colSpan={2} className="py-0">
<div className="flex h-10 items-center">
{allocationModified ? (
unallocated < 0 ? (
<div className="flex items-center gap-1">
<ExclamationTriangleIcon className="size-4 text-error" />
<span className="text-error">
You're trying to allocate more concurrency than your total purchased
amount.
</span>
</div>
) : (
<div className="flex w-full items-center justify-between gap-3">
<div className="flex items-center gap-1">
<InformationCircleIcon className="size-4 text-text-dimmed" />
<span>
Save your changes or{" "}
<button
className="inline text-indigo-500 hover:text-indigo-300"
onClick={() => {
setAllocation(initialAllocation(environments));
}}
>
reset
</button>
.
</span>
</div>
<Button
variant="primary/small"
type="submit"
form="allocate"
disabled={unallocated < 0 || isLoading}
LeadingIcon={isLoading ? SpinnerWhite : undefined}
>
Save
</Button>
</div>
)
) : (
<></>
)}
</div>
</TableCell>
</TableRow>
</TableBody>
</Table>
<FormError id={formEnvironments.id}>{formEnvironments.error}</FormError>
</div>
<Form className="flex flex-col gap-2" method="post" {...form.props} id="allocate">
<input type="hidden" name="action" value="allocate" />
<div className="flex items-center pb-1">
<Header3 className="grow">Concurrency allocation</Header3>
</div>
<Table>
<TableHeader>
<TableRow>
<TableHeaderCell className="pl-0">Environment</TableHeaderCell>
<TableHeaderCell alignment="right">
<span className="flex items-center justify-end gap-x-1">
Included{" "}
<InfoIconTooltip content="This is the included concurrency based on your plan." />
</span>
</TableHeaderCell>
<TableHeaderCell alignment="right">Extra concurrency</TableHeaderCell>
<TableHeaderCell alignment="right">Total</TableHeaderCell>
</TableRow>
</TableHeader>
<TableBody>
{environments.map((environment, index) => (
<TableRow key={environment.id}>
<TableCell>
<EnvironmentCombo environment={environment} />
</TableCell>
<TableCell alignment="right">{environment.planConcurrencyLimit}</TableCell>
<TableCell alignment="right">
<div className="flex items-center justify-end">
{environment.type === "DEVELOPMENT" ? (
Math.max(
0,
environment.maximumConcurrencyLimit - environment.planConcurrencyLimit
)
) : (
<>
<input
type="hidden"
name={`environments[${index}].id`}
value={environment.id}
/>
<Input
name={`environments[${index}].amount`}
type="number"
variant="outline/small"
className="text-right"
containerClassName="w-16"
fullWidth={false}
value={allocation.get(environment.id)}
onChange={(e) => {
const value = e.target.value === "" ? 0 : Number(e.target.value);
setAllocation(new Map(allocation).set(environment.id, value));
}}
min={0}
/>
</>
)}
</div>
</TableCell>
<TableCell alignment="right">
{environment.planConcurrencyLimit + (allocation.get(environment.id) ?? 0)}
</TableCell>
</TableRow>
))}
</TableBody>
</Table>
</Form>
</div>
</div>
);
}
function NotUpgradable({ environments }: { environments: EnvironmentWithConcurrency[] }) {
const { isManagedCloud } = useFeatures();
const plan = useCurrentPlan();
const organization = useOrganization();
return (
<div className="flex flex-col gap-3">
<div className="border-b border-grid-dimmed pb-1">
<Header2>Your concurrency</Header2>
</div>
{isManagedCloud ? (
<>
<Paragraph variant="small">
Concurrency limits determine how many runs you can execute at the same time. You can
upgrade your plan to get more concurrency. You are currently on the{" "}
{plan?.v3Subscription?.plan?.title ?? "Free"} plan.
</Paragraph>
<LinkButton variant="primary/small" to={v3BillingPath(organization)}>
Upgrade for more concurrency
</LinkButton>
</>
) : null}
<div className="mt-3 flex flex-col gap-3">
<Table>
<TableHeader>
<TableRow>
<TableHeaderCell className="pl-0">Environment</TableHeaderCell>
<TableHeaderCell alignment="right">Concurrency limit</TableHeaderCell>
</TableRow>
</TableHeader>
<TableBody>
{environments.map((environment) => (
<TableRow key={environment.id}>
<TableCell className="pl-0">
<EnvironmentCombo environment={environment} />
</TableCell>
<TableCell alignment="right">{environment.maximumConcurrencyLimit}</TableCell>
</TableRow>
))}
</TableBody>
</Table>
</div>
</div>
);
}
function PurchaseConcurrencyModal({
concurrencyPricing,
extraConcurrency,
extraUnallocatedConcurrency,
maxQuota,
disabled,
}: {
concurrencyPricing: {
stepSize: number;
centsPerStep: number;
};
extraConcurrency: number;
extraUnallocatedConcurrency: number;
maxQuota: number;
disabled: boolean;
}) {
const lastSubmission = useActionData();
const [form, { amount }] = useForm({
id: "purchase-concurrency",
// TODO: type this
lastSubmission: lastSubmission as any,
onValidate({ formData }) {
return parse(formData, { schema: FormSchema });
},
shouldRevalidate: "onSubmit",
});
const [amountValue, setAmountValue] = useState(extraConcurrency);
const navigation = useNavigation();
const isLoading = navigation.state !== "idle" && navigation.formMethod === "POST";
// Close the panel, when we've succeeded
// This is required because a redirect to the same path doesn't clear state
const [searchParams, setSearchParams] = useSearchParams();
const [open, setOpen] = useState(false);
useEffect(() => {
const success = searchParams.get("success");
if (success) {
setOpen(false);
setSearchParams((s) => {
s.delete("success");
return s;
});
}
}, [searchParams.get("success")]);
const state = updateState({
value: amountValue,
existingValue: extraConcurrency,
quota: maxQuota,
extraUnallocatedConcurrency,
});
const changeClassName =
state === "decrease" ? "text-error" : state === "increase" ? "text-success" : undefined;
const title = extraConcurrency === 0 ? "Purchase extra concurrency" : "Add/remove concurrency";
return (
<Dialog open={open} onOpenChange={setOpen}>
<DialogTrigger asChild>
<Button
variant="primary/small"
disabled={disabled}
onClick={() => {
setOpen(true);
}}
>
{title}
</Button>
</DialogTrigger>
<DialogContent>
<DialogHeader>{title}</DialogHeader>
<Form method="post" {...form.props}>
<div className="flex flex-col gap-4 pt-2">
<Paragraph variant="base/bright" spacing>
You can purchase bundles of {concurrencyPricing.stepSize} concurrency for{" "}
{formatCurrency(concurrencyPricing.centsPerStep / 100, false)}/month. Or you can
remove any extra concurrency after you have unallocated it from your environments
first.
</Paragraph>
<Fieldset>
<InputGroup fullWidth>
<Label htmlFor="amount" className="text-text-dimmed">
Total extra concurrency
</Label>
<InputNumberStepper
{...conform.input(amount, { type: "number" })}
step={concurrencyPricing.stepSize}
min={0}
max={undefined}
value={amountValue}
onChange={(e) => setAmountValue(Number(e.target.value))}
disabled={isLoading}
/>
<FormError id={amount.errorId}>{amount.error}</FormError>
<FormError>{form.error}</FormError>
</InputGroup>
</Fieldset>
{state === "need_to_increase_unallocated" ? (
<div className="flex flex-col pb-3">
<Paragraph variant="small" className="text-warning" spacing>
You need to unallocate{" "}
{formatNumber(extraConcurrency - amountValue - extraUnallocatedConcurrency)} more
concurrency from your environments in order to remove{" "}
{formatNumber(extraConcurrency - amountValue)} concurrency from your account.
</Paragraph>
</div>
) : state === "above_quota" ? (
<div className="flex flex-col pb-3">
<Paragraph variant="small" className="text-warning" spacing>
Currently you can only have up to {maxQuota} extra concurrency. Send a request
below to lift your current limit. We'll get back to you soon.
</Paragraph>
</div>
) : (
<div className="flex flex-col pb-3 tabular-nums">
<div className="grid grid-cols-2 border-b border-grid-dimmed pb-1">
<Header3 className="font-normal text-text-dimmed">Summary</Header3>
<Header3 className="justify-self-end font-normal text-text-dimmed">Total</Header3>
</div>
<div className="grid grid-cols-2 pt-2">
<Header3 className="pb-0 font-normal text-text-dimmed">
<span className="text-text-bright">{formatNumber(extraConcurrency)}</span>{" "}
current total
</Header3>
<Header3 className="justify-self-end font-normal text-text-bright">
{formatCurrency(
(extraConcurrency * concurrencyPricing.centsPerStep) /
concurrencyPricing.stepSize /
100,
true
)}
</Header3>
</div>
<div className="grid grid-cols-2 text-xs">
<span className="text-text-dimmed">
({extraConcurrency / concurrencyPricing.stepSize} bundles)
</span>
<span className="justify-self-end text-text-dimmed">/mth</span>
</div>
<div className="grid grid-cols-2 pt-2">
<Header3 className={cn("pb-0 font-normal", changeClassName)}>
{state === "increase" ? "+" : null}
{formatNumber(amountValue - extraConcurrency)}
</Header3>
<Header3 className={cn("justify-self-end font-normal", changeClassName)}>
{state === "increase" ? "+" : null}
{formatCurrency(
((amountValue - extraConcurrency) * concurrencyPricing.centsPerStep) /
concurrencyPricing.stepSize /
100,
true
)}
</Header3>
</div>
<div className="grid grid-cols-2 text-xs">
<span className="text-text-dimmed">
({(amountValue - extraConcurrency) / concurrencyPricing.stepSize} bundles @{" "}
{formatCurrency(concurrencyPricing.centsPerStep / 100, true)}/mth)
</span>
<span className="justify-self-end text-text-dimmed">/mth</span>
</div>
<div className="grid grid-cols-2 pt-2">
<Header3 className="pb-0 font-normal text-text-dimmed">
<span className="text-text-bright">{formatNumber(amountValue)}</span> new total
</Header3>
<Header3 className="justify-self-end font-normal text-text-bright">
{formatCurrency(
(amountValue * concurrencyPricing.centsPerStep) /
concurrencyPricing.stepSize /
100,
true
)}
</Header3>
</div>
<div className="grid grid-cols-2 text-xs">
<span className="text-text-dimmed">
({amountValue / concurrencyPricing.stepSize} bundles)
</span>
<span className="justify-self-end text-text-dimmed">/mth</span>
</div>
</div>
)}
</div>
<FormButtons
confirmButton={
state === "above_quota" ? (
<>
<input type="hidden" name="action" value="quota-increase" />
<Button
LeadingIcon={isLoading ? SpinnerWhite : EnvelopeIcon}
variant="primary/medium"
type="submit"
disabled={isLoading}
>
{`Send request for ${formatNumber(amountValue)}`}
</Button>
</>
) : state === "decrease" || state === "need_to_increase_unallocated" ? (
<>
<input type="hidden" name="action" value="purchase" />
<Button
variant="danger/medium"
type="submit"
disabled={isLoading || state === "need_to_increase_unallocated"}
LeadingIcon={isLoading ? SpinnerWhite : undefined}
>
{`Remove ${formatNumber(extraConcurrency - amountValue)} concurrency`}
</Button>
</>
) : (
<>
<input type="hidden" name="action" value="purchase" />
<Button
variant="primary/medium"
type="submit"
disabled={isLoading || state === "no_change"}
LeadingIcon={isLoading ? SpinnerWhite : undefined}
>
{`Purchase ${formatNumber(amountValue - extraConcurrency)} concurrency`}
</Button>
</>
)
}
cancelButton={
<DialogClose asChild>
<Button variant="tertiary/medium" disabled={isLoading}>
Cancel
</Button>
</DialogClose>
}
/>
</Form>
</DialogContent>
</Dialog>
);
}
function updateState({
value,
existingValue,
quota,
extraUnallocatedConcurrency,
}: {
value: number;
existingValue: number;
quota: number;
extraUnallocatedConcurrency: number;
}): "no_change" | "increase" | "decrease" | "above_quota" | "need_to_increase_unallocated" {
if (value === existingValue) return "no_change";
if (value < existingValue) {
const difference = existingValue - value;
if (difference > extraUnallocatedConcurrency) {
return "need_to_increase_unallocated";
}
return "decrease";
}
if (value > quota) return "above_quota";
return "increase";
}
@@ -1,6 +1,9 @@
import { Link, useLocation } from "@remix-run/react";
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { typedjson, useTypedLoaderData } from "remix-typedjson";
import { useEffect, useState, useRef, useCallback } from "react";
import { S2, S2Error } from "@s2-dev/streamstore";
import { Clipboard, ClipboardCheck, ChevronDown, ChevronUp } from "lucide-react";
import { ExitIcon } from "~/assets/icons/ExitIcon";
import { GitMetadata } from "~/components/GitMetadata";
import { RuntimeIcon } from "~/components/RuntimeIcon";
@@ -22,16 +25,22 @@ import {
} from "~/components/primitives/Table";
import { DeploymentError } from "~/components/runs/v3/DeploymentError";
import { DeploymentStatus } from "~/components/runs/v3/DeploymentStatus";
import {
Tooltip,
TooltipContent,
TooltipProvider,
TooltipTrigger,
} from "~/components/primitives/Tooltip";
import { useEnvironment } from "~/hooks/useEnvironment";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { useUser } from "~/hooks/useUser";
import { DeploymentPresenter } from "~/presenters/v3/DeploymentPresenter.server";
import { requireUserId } from "~/services/session.server";
import { cn } from "~/utils/cn";
import { v3DeploymentParams, v3DeploymentsPath, v3RunsPath } from "~/utils/pathBuilder";
import { capitalizeWord } from "~/utils/string";
import { UserTag } from "../_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments/route";
import { DeploymentEventFromString } from "@trigger.dev/core/v3/schemas";
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
@@ -40,7 +49,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
try {
const presenter = new DeploymentPresenter();
const { deployment } = await presenter.call({
const { deployment, eventStream } = await presenter.call({
userId,
organizationSlug,
projectSlug: projectParam,
@@ -48,7 +57,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
deploymentShortCode: deploymentParam,
});
return typedjson({ deployment });
return typedjson({ deployment, eventStream });
} catch (error) {
console.error(error);
throw new Response(undefined, {
@@ -58,15 +67,122 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
}
};
type LogEntry = {
message: string;
timestamp: Date;
level: "info" | "error" | "warn" | "debug";
};
export default function Page() {
const { deployment } = useTypedLoaderData<typeof loader>();
const { deployment, eventStream } = useTypedLoaderData<typeof loader>();
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const location = useLocation();
const user = useUser();
const page = new URLSearchParams(location.search).get("page");
const logsDisabled = eventStream === undefined;
const [logs, setLogs] = useState<LogEntry[]>([]);
const [isStreaming, setIsStreaming] = useState(true);
const [streamError, setStreamError] = useState<string | null>(null);
const isPending = deployment.status === "PENDING";
useEffect(() => {
if (logsDisabled) return;
const abortController = new AbortController();
setLogs([]);
setStreamError(null);
setIsStreaming(true);
const streamLogs = async () => {
try {
const s2 = new S2({ accessToken: eventStream.s2.accessToken });
const basin = s2.basin(eventStream.s2.basin);
const stream = basin.stream(eventStream.s2.stream);
const readSession = await stream.readSession(
{
seq_num: 0,
wait: 60,
as: "bytes",
},
{ signal: abortController.signal }
);
const decoder = new TextDecoder();
for await (const record of readSession) {
const decoded = decoder.decode(record.body);
const result = DeploymentEventFromString.safeParse(decoded);
if (!result.success) {
// fallback to the previous format in s2 logs for compatibility
try {
const headers: Record<string, string> = {};
if (record.headers) {
for (const [nameBytes, valueBytes] of record.headers) {
headers[decoder.decode(nameBytes)] = decoder.decode(valueBytes);
}
}
const level = (headers["level"]?.toLowerCase() as LogEntry["level"]) ?? "info";
setLogs((prevLogs) => [
...prevLogs,
{
timestamp: new Date(record.timestamp),
message: decoded,
level,
},
]);
} catch (err) {
console.error("Failed to parse log record:", err);
}
continue;
}
const event = result.data;
if (event.type !== "log") {
continue;
}
setLogs((prevLogs) => [
...prevLogs,
{
timestamp: new Date(record.timestamp),
message: event.data.message,
level: event.data.level,
},
]);
}
} catch (error) {
if (abortController.signal.aborted) return;
const isNotFoundError =
error instanceof S2Error &&
error.code &&
["permission_denied", "stream_not_found"].includes(error.code);
if (isNotFoundError) return;
console.error("Failed to stream logs:", error);
setStreamError("Failed to stream logs");
} finally {
if (!abortController.signal.aborted) {
setIsStreaming(false);
}
}
};
streamLogs();
return () => {
abortController.abort();
};
}, [eventStream?.s2?.basin, eventStream?.s2?.stream, eventStream?.s2?.accessToken, isPending]);
return (
<div className="grid h-full max-h-full grid-rows-[2.5rem_1fr] overflow-hidden bg-background-bright">
<div className="mx-3 flex items-center justify-between gap-2 border-b border-grid-dimmed">
@@ -158,6 +274,19 @@ export default function Page() {
/>
</Property.Value>
</Property.Item>
{!logsDisabled && (
<Property.Item>
<Property.Label>Logs</Property.Label>
<LogsDisplay
logs={logs}
isStreaming={isStreaming}
streamError={streamError}
initialCollapsed={(
["PENDING", "DEPLOYED", "TIMED_OUT"] satisfies (typeof deployment.status)[]
).includes(deployment.status)}
/>
</Property.Item>
)}
{deployment.canceledAt && (
<Property.Item>
<Property.Label>Canceled at</Property.Label>
@@ -320,3 +449,189 @@ export default function Page() {
</div>
);
}
function LogsDisplay({
logs,
isStreaming,
streamError,
initialCollapsed = false,
}: {
logs: LogEntry[];
isStreaming: boolean;
streamError: string | null;
initialCollapsed?: boolean;
}) {
const [copied, setCopied] = useState(false);
const [mouseOver, setMouseOver] = useState(false);
const [collapsed, setCollapsed] = useState(initialCollapsed);
const logsContainerRef = useRef<HTMLDivElement>(null);
useEffect(() => {
setCollapsed(initialCollapsed);
}, [initialCollapsed]);
// auto-scroll log container to bottom when new logs arrive
useEffect(() => {
if (logsContainerRef.current) {
logsContainerRef.current.scrollTop = logsContainerRef.current.scrollHeight;
}
}, [logs]);
const onCopyLogs = useCallback(
(event: React.MouseEvent<HTMLButtonElement>) => {
event.preventDefault();
event.stopPropagation();
const logsText = logs.map((log) => log.message).join("\n");
navigator.clipboard.writeText(logsText);
setCopied(true);
setTimeout(() => {
setCopied(false);
}, 1500);
},
[logs]
);
const errorCount = logs.filter((log) => log.level === "error").length;
const warningCount = logs.filter((log) => log.level === "warn").length;
return (
<div className="mt-1.5 overflow-hidden rounded-md border border-grid-bright">
<div className="flex items-center justify-between border-b border-grid-dimmed px-3 py-2">
<div className="flex items-center gap-4">
<div className="flex items-center gap-1.5">
<div
className={cn(
"h-2 w-2 rounded-full",
errorCount > 0 ? "bg-error/80" : "bg-charcoal-600"
)}
/>
<Paragraph variant="extra-small/dimmed/mono" className="w-[ch-10]">
{`${errorCount} ${errorCount === 1 ? "error" : "errors"}`}
</Paragraph>
</div>
<div className="flex items-center gap-1.5">
<div
className={cn(
"h-2 w-2 rounded-full",
warningCount > 0 ? "bg-warning/80" : "bg-charcoal-600"
)}
/>
<Paragraph variant="extra-small/dimmed/mono">
{`${warningCount} ${warningCount === 1 ? "warning" : "warnings"}`}
</Paragraph>
</div>
</div>
{logs.length > 0 && (
<div className="flex items-center gap-3">
<TooltipProvider>
<Tooltip open={copied || mouseOver} disableHoverableContent>
<TooltipTrigger
onClick={onCopyLogs}
onMouseEnter={() => setMouseOver(true)}
onMouseLeave={() => setMouseOver(false)}
className={cn(
"transition-colors duration-100 focus-custom hover:cursor-pointer",
copied ? "text-success" : "text-text-dimmed hover:text-text-bright"
)}
>
<div className="size-4 shrink-0">
{copied ? (
<ClipboardCheck className="size-full" />
) : (
<Clipboard className="size-full" />
)}
</div>
</TooltipTrigger>
<TooltipContent side="left" className="text-xs">
{copied ? "Copied" : "Copy"}
</TooltipContent>
</Tooltip>
</TooltipProvider>
<TooltipProvider>
<Tooltip disableHoverableContent>
<TooltipTrigger
onClick={() => setCollapsed(!collapsed)}
className={cn(
"transition-colors duration-100 focus-custom hover:cursor-pointer",
"text-text-dimmed hover:text-text-bright"
)}
>
{collapsed ? (
<ChevronDown className="size-4" />
) : (
<ChevronUp className="size-4" />
)}
</TooltipTrigger>
<TooltipContent side="left" className="text-xs">
{collapsed ? "Expand" : "Collapse"}
</TooltipContent>
</Tooltip>
</TooltipProvider>
</div>
)}
</div>
<div className="relative">
<div
ref={logsContainerRef}
className={cn(
"grow overflow-x-auto overflow-y-scroll font-mono text-xs transition-all duration-200 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600",
collapsed ? "h-16" : "h-64"
)}
>
<div className="flex w-fit min-w-full flex-col">
{logs.length === 0 && (
<div className="flex gap-x-2.5 border-l-2 border-transparent px-2.5 py-1">
{streamError ? (
<span className="text-error">Failed fetching logs</span>
) : (
<span className="text-text-dimmed">
{isStreaming ? "Waiting for logs..." : "No logs yet"}
</span>
)}
</div>
)}
{logs.map((log, index) => {
return (
<div
key={index}
className={cn(
"flex w-full gap-x-2.5 border-l-2 px-2.5 py-1",
log.level === "error" && "border-error/60 bg-error/15 hover:bg-error/25",
log.level === "warn" && "border-warning/60 bg-warning/20 hover:bg-warning/30",
log.level === "info" && "border-transparent hover:bg-charcoal-750"
)}
>
<span
className={cn(
"select-none whitespace-nowrap py-px",
log.level === "error" && "text-error/80",
log.level === "warn" && "text-warning/70",
log.level === "info" && "text-text-dimmed"
)}
>
<DateTimeAccurate date={log.timestamp} hideDate hour12={false} />
</span>
<span
className={cn(
"whitespace-nowrap",
log.level === "error" && "text-error",
log.level === "warn" && "text-warning",
log.level === "info" && "text-text-bright"
)}
>
{log.message}
</span>
</div>
);
})}
</div>
</div>
{collapsed && (
<div className="pointer-events-none absolute bottom-0 left-0 right-0 h-8 bg-gradient-to-t from-charcoal-800/90 to-transparent" />
)}
</div>
</div>
);
}
@@ -372,7 +372,7 @@ export default function Page() {
{deploymentParam && (
<>
<ResizableHandle id="deployments-handle" />
<ResizablePanel id="deployments-inspector" min="400px" max="700px">
<ResizablePanel id="deployments-inspector" min="500px" max="800px">
<Outlet />
</ResizablePanel>
</>
@@ -68,11 +68,18 @@ import { EnvironmentQueuePresenter } from "~/presenters/v3/EnvironmentQueuePrese
import { QueueListPresenter } from "~/presenters/v3/QueueListPresenter.server";
import { requireUserId } from "~/services/session.server";
import { cn } from "~/utils/cn";
import { docsPath, EnvironmentParamSchema, v3BillingPath, v3RunsPath } from "~/utils/pathBuilder";
import {
concurrencyPath,
docsPath,
EnvironmentParamSchema,
v3BillingPath,
v3RunsPath,
} from "~/utils/pathBuilder";
import { concurrencySystem } from "~/v3/services/concurrencySystemInstance.server";
import { PauseEnvironmentService } from "~/v3/services/pauseEnvironment.server";
import { PauseQueueService } from "~/v3/services/pauseQueue.server";
import { useCurrentPlan } from "../_app.orgs.$organizationSlug/route";
import { ConcurrencyIcon } from "~/assets/icons/ConcurrencyIcon";
const SearchParamsSchema = z.object({
query: z.string().optional(),
@@ -406,18 +413,14 @@ export default function Page() {
accessory={
plan ? (
plan?.v3Subscription?.plan?.limits.concurrentRuns.canExceed ? (
<Feedback
button={
<Button
variant="tertiary/small"
LeadingIcon={ChatBubbleLeftEllipsisIcon}
leadingIconClassName="text-indigo-500"
>
Increase limit
</Button>
}
defaultValue="concurrency"
/>
<LinkButton
to={concurrencyPath(organization, project, env)}
variant="tertiary/small"
LeadingIcon={ConcurrencyIcon}
leadingIconClassName="text-amber-500"
>
Increase limit
</LinkButton>
) : (
<LinkButton
to={v3BillingPath(organization, "Upgrade your plan for more concurrency")}
@@ -30,6 +30,7 @@ import {
PageContainer,
} from "~/components/layout/AppLayout";
import { Button, LinkButton } from "~/components/primitives/Buttons";
import { CheckboxWithLabel } from "~/components/primitives/Checkbox";
import { ClipboardField } from "~/components/primitives/ClipboardField";
import { Fieldset } from "~/components/primitives/Fieldset";
import { FormButtons } from "~/components/primitives/FormButtons";
@@ -180,6 +181,10 @@ const UpdateBuildSettingsFormSchema = z.object({
.refine((val) => !val || val.length <= 500, {
message: "Pre-build command must not exceed 500 characters",
}),
useNativeBuildServer: z
.string()
.optional()
.transform((val) => val === "on"),
});
type UpdateBuildSettingsFormSchema = z.infer<typeof UpdateBuildSettingsFormSchema>;
@@ -407,12 +412,14 @@ export const action: ActionFunction = async ({ request, params }) => {
});
}
case "update-build-settings": {
const { installCommand, preBuildCommand, triggerConfigFilePath } = submission.value;
const { installCommand, preBuildCommand, triggerConfigFilePath, useNativeBuildServer } =
submission.value;
const resultOrFail = await projectSettingsService.updateBuildSettings(projectId, {
installCommand: installCommand || undefined,
preBuildCommand: preBuildCommand || undefined,
triggerConfigFilePath: triggerConfigFilePath || undefined,
useNativeBuildServer: useNativeBuildServer,
});
if (resultOrFail.isErr()) {
@@ -1135,13 +1142,15 @@ function BuildSettingsForm({ buildSettings }: { buildSettings: BuildSettings })
preBuildCommand: buildSettings?.preBuildCommand || "",
installCommand: buildSettings?.installCommand || "",
triggerConfigFilePath: buildSettings?.triggerConfigFilePath || "",
useNativeBuildServer: buildSettings?.useNativeBuildServer || false,
});
useEffect(() => {
const hasChanges =
buildSettingsValues.preBuildCommand !== (buildSettings?.preBuildCommand || "") ||
buildSettingsValues.installCommand !== (buildSettings?.installCommand || "") ||
buildSettingsValues.triggerConfigFilePath !== (buildSettings?.triggerConfigFilePath || "");
buildSettingsValues.triggerConfigFilePath !== (buildSettings?.triggerConfigFilePath || "") ||
buildSettingsValues.useNativeBuildServer !== (buildSettings?.useNativeBuildServer || false);
setHasBuildSettingsChanges(hasChanges);
}, [buildSettingsValues, buildSettings]);
@@ -1222,6 +1231,30 @@ function BuildSettingsForm({ buildSettings }: { buildSettings: BuildSettings })
</Hint>
<FormError id={fields.preBuildCommand.errorId}>{fields.preBuildCommand.error}</FormError>
</InputGroup>
<div className="border-t border-grid-dimmed pt-4">
<InputGroup>
<CheckboxWithLabel
id={fields.useNativeBuildServer.id}
{...conform.input(fields.useNativeBuildServer, { type: "checkbox" })}
label="Use native build server"
variant="simple/small"
defaultChecked={buildSettings?.useNativeBuildServer || false}
onChange={(isChecked) => {
setBuildSettingsValues((prev) => ({
...prev,
useNativeBuildServer: isChecked,
}));
}}
/>
<Hint>
Native build server builds do not rely on external build providers and will become the
default in the future. Version 4.1.3 or newer is required.
</Hint>
<FormError id={fields.useNativeBuildServer.errorId}>
{fields.useNativeBuildServer.error}
</FormError>
</InputGroup>
</div>
<FormError>{buildSettingsForm.error}</FormError>
<FormButtons
confirmButton={
@@ -83,6 +83,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
export default function ChoosePlanPage() {
const {
plans,
addOnPricing,
v3Subscription,
organizationSlug,
periodStart,
@@ -141,6 +142,7 @@ export default function ChoosePlanPage() {
<div>
<PricingPlans
plans={plans}
concurrencyAddOnPricing={addOnPricing.concurrency}
subscription={v3Subscription}
organizationSlug={organizationSlug}
hasPromotedPlan={false}
@@ -8,6 +8,7 @@ import { redirect, typedjson, useTypedLoaderData } from "remix-typedjson";
import invariant from "tiny-invariant";
import { z } from "zod";
import { BackgroundWrapper } from "~/components/BackgroundWrapper";
import { Feedback } from "~/components/Feedback";
import { AppContainer, MainCenteredContainer } from "~/components/layout/AppLayout";
import { Button, LinkButton } from "~/components/primitives/Buttons";
import { Callout } from "~/components/primitives/Callout";
@@ -21,10 +22,11 @@ import { Label } from "~/components/primitives/Label";
import { ButtonSpinner } from "~/components/primitives/Spinner";
import { prisma } from "~/db.server";
import { featuresForRequest } from "~/features.server";
import { redirectWithSuccessMessage } from "~/models/message.server";
import { createProject } from "~/models/project.server";
import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
import { createProject, ExceededProjectLimitError } from "~/models/project.server";
import { requireUserId } from "~/services/session.server";
import {
newProjectPath,
OrganizationParamsSchema,
organizationPath,
selectPlanPath,
@@ -114,8 +116,29 @@ export const action: ActionFunction = async ({ request, params }) => {
request,
`${submission.value.projectName} created`
);
} catch (error: any) {
return json({ errors: { body: error.message } }, { status: 400 });
} catch (error) {
if (error instanceof ExceededProjectLimitError) {
return redirectWithErrorMessage(
newProjectPath({ slug: organizationSlug }),
request,
error.message,
{
title: "Failed to create project",
action: {
label: "Request more projects",
variant: "secondary/small",
action: { type: "help", feedbackType: "help" },
},
}
);
}
return redirectWithErrorMessage(
newProjectPath({ slug: organizationSlug }),
request,
error instanceof Error ? error.message : "Something went wrong",
{ ephemeral: false }
);
}
};
@@ -191,6 +214,7 @@ export default function Page() {
</Fieldset>
</Form>
</div>
<Feedback button={<></>} />
</MainCenteredContainer>
</BackgroundWrapper>
</AppContainer>
@@ -45,7 +45,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
}
export default function ChoosePlanPage() {
const { plans, v3Subscription, organizationSlug, periodEnd } =
const { plans, v3Subscription, organizationSlug, periodEnd, addOnPricing } =
useTypedLoaderData<typeof loader>();
return (
@@ -57,6 +57,7 @@ export default function ChoosePlanPage() {
<div className="w-full rounded-lg border border-grid-bright bg-background-dimmed p-5 shadow-lg">
<PricingPlans
plans={plans}
concurrencyAddOnPricing={addOnPricing.concurrency}
subscription={v3Subscription}
organizationSlug={organizationSlug}
hasPromotedPlan
@@ -74,6 +74,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
limit = body.development;
break;
}
case "PREVIEW":
case "STAGING": {
limit = body.staging;
break;
@@ -0,0 +1,91 @@
import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
import {
type CreateArtifactResponseBody,
CreateArtifactRequestBody,
tryCatch,
} from "@trigger.dev/core/v3";
import { authenticateRequest } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
import { ArtifactsService } from "~/v3/services/artifacts.server";
export async function action({ request }: ActionFunctionArgs) {
if (request.method.toUpperCase() !== "POST") {
return json({ error: "Method Not Allowed" }, { status: 405 });
}
const authenticationResult = await authenticateRequest(request, {
apiKey: true,
organizationAccessToken: false,
personalAccessToken: false,
});
if (!authenticationResult || !authenticationResult.result.ok) {
logger.info("Invalid or missing api key", { url: request.url });
return json({ error: "Invalid or Missing API key" }, { status: 401 });
}
const [, rawBody] = await tryCatch(request.json());
const body = CreateArtifactRequestBody.safeParse(rawBody ?? {});
if (!body.success) {
return json({ error: "Invalid request body", issues: body.error.issues }, { status: 400 });
}
const { environment: authenticatedEnv } = authenticationResult.result;
const service = new ArtifactsService();
return await service
.createArtifact(body.data.type, authenticatedEnv, body.data.contentLength)
.match(
(result) => {
return json(
{
artifactKey: result.artifactKey,
uploadUrl: result.uploadUrl,
uploadFields: result.uploadFields,
expiresAt: result.expiresAt.toISOString(),
} satisfies CreateArtifactResponseBody,
{ status: 201 }
);
},
(error) => {
switch (error.type) {
case "artifact_size_exceeds_limit": {
logger.warn("Artifact size exceeds limit", { error });
const sizeMB = parseFloat((error.contentLength / (1024 * 1024)).toFixed(1));
const limitMB = parseFloat((error.sizeLimit / (1024 * 1024)).toFixed(1));
let errorMessage;
switch (body.data.type) {
case "deployment_context":
errorMessage = `Artifact size (${sizeMB} MB) exceeds the allowed limit of ${limitMB} MB. Make sure you are in the correct directory of your Trigger.dev project. Reach out to us if you are seeing this error consistently.`;
break;
default:
body.data.type satisfies never;
errorMessage = `Artifact size (${sizeMB} MB) exceeds the allowed limit of ${limitMB} MB`;
}
return json(
{
error: errorMessage,
},
{ status: 400 }
);
}
case "failed_to_create_presigned_post": {
logger.error("Failed to create presigned POST", { error });
return json({ error: "Failed to generate artifact upload URL" }, { status: 500 });
}
case "artifacts_bucket_not_configured": {
logger.error("Artifacts bucket not configured", { error });
return json({ error: "Internal server error" }, { status: 500 });
}
default: {
error satisfies never;
logger.error("Failed creating artifact", { error });
return json({ error: "Internal server error" }, { status: 500 });
}
}
}
);
}
+2 -1
View File
@@ -37,7 +37,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
const service = new InitializeDeploymentService();
try {
const { deployment, imageRef } = await service.call(authenticatedEnv, body.data);
const { deployment, imageRef, eventStream } = await service.call(authenticatedEnv, body.data);
const responseBody: InitializeDeploymentResponseBody = {
id: deployment.friendlyId,
@@ -48,6 +48,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
deployment.externalBuildData as InitializeDeploymentResponseBody["externalBuildData"],
imageTag: imageRef,
imagePlatform: deployment.imagePlatform,
eventStream,
};
return json(responseBody, { status: 200 });
@@ -4,6 +4,7 @@ import {
CreateProjectRequestBody,
GetProjectResponseBody,
GetProjectsResponseBody,
tryCatch,
} from "@trigger.dev/core/v3";
import { z } from "zod";
import { prisma } from "~/db.server";
@@ -99,12 +100,18 @@ export async function action({ request, params }: ActionFunctionArgs) {
return json({ error: "Invalid request body" }, { status: 400 });
}
const project = await createProject({
organizationSlug: organization.slug,
name: parsedBody.data.name,
userId: authenticationResult.userId,
version: "v3",
});
const [error, project] = await tryCatch(
createProject({
organizationSlug: organization.slug,
name: parsedBody.data.name,
userId: authenticationResult.userId,
version: "v3",
})
);
if (error) {
return json({ error: error.message }, { status: 400 });
}
const result: GetProjectResponseBody = {
id: project.id,
@@ -5,6 +5,7 @@ import { env as processEnv } from "~/env.server";
import {
authenticatedEnvironmentForAuthentication,
authenticateRequest,
branchNameFromRequest,
} from "~/services/apiAuth.server";
const ParamsSchema = z.object({
@@ -32,7 +33,8 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
const environment = await authenticatedEnvironmentForAuthentication(
authenticationResult,
projectRef,
env
env,
branchNameFromRequest(request)
);
const result: GetProjectEnvResponse = {
@@ -4,6 +4,7 @@ import { prisma } from "~/db.server";
import {
authenticateRequest,
authenticatedEnvironmentForAuthentication,
branchNameFromRequest,
} from "~/services/apiAuth.server";
import zlib from "node:zlib";
@@ -29,7 +30,8 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
const environment = await authenticatedEnvironmentForAuthentication(
authenticationResult,
parsedParams.data.projectRef,
parsedParams.data.envSlug
parsedParams.data.envSlug,
branchNameFromRequest(request)
);
// Find the background worker and tasks and files
@@ -5,6 +5,7 @@ import { prisma } from "~/db.server";
import {
authenticateRequest,
authenticatedEnvironmentForAuthentication,
branchNameFromRequest,
} from "~/services/apiAuth.server";
import { EnvironmentVariablesRepository } from "~/v3/environmentVariables/environmentVariablesRepository.server";
@@ -30,7 +31,8 @@ export async function action({ params, request }: ActionFunctionArgs) {
const environment = await authenticatedEnvironmentForAuthentication(
authenticationResult,
parsedParams.data.projectRef,
parsedParams.data.slug
parsedParams.data.slug,
branchNameFromRequest(request)
);
// Find the environment variable
@@ -106,7 +108,8 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
const environment = await authenticatedEnvironmentForAuthentication(
authenticationResult,
parsedParams.data.projectRef,
parsedParams.data.slug
parsedParams.data.slug,
branchNameFromRequest(request)
);
// Find the environment variable
@@ -4,6 +4,7 @@ import { z } from "zod";
import {
authenticateRequest,
authenticatedEnvironmentForAuthentication,
branchNameFromRequest,
} from "~/services/apiAuth.server";
import { EnvironmentVariablesRepository } from "~/v3/environmentVariables/environmentVariablesRepository.server";
@@ -28,7 +29,8 @@ export async function action({ params, request }: ActionFunctionArgs) {
const environment = await authenticatedEnvironmentForAuthentication(
authenticationResult,
parsedParams.data.projectRef,
parsedParams.data.slug
parsedParams.data.slug,
branchNameFromRequest(request)
);
const jsonBody = await request.json();
@@ -75,7 +77,8 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
const environment = await authenticatedEnvironmentForAuthentication(
authenticationResult,
parsedParams.data.projectRef,
parsedParams.data.slug
parsedParams.data.slug,
branchNameFromRequest(request)
);
const repository = new EnvironmentVariablesRepository();
@@ -9,6 +9,7 @@ import { authenticateApiRequest } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
import { BatchTriggerV3Service } from "~/v3/services/batchTriggerV3.server";
import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
const ParamsSchema = z.object({
taskId: z.string(),
@@ -40,6 +41,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
"trigger-version": triggerVersion,
"x-trigger-span-parent-as-link": spanParentAsLink,
"x-trigger-worker": isFromWorker,
"x-trigger-realtime-streams-version": realtimeStreamsVersion,
traceparent,
tracestate,
} = headers.data;
@@ -100,6 +102,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
triggerVersion: triggerVersion ?? undefined,
traceContext,
spanParentAsLink: spanParentAsLink === 1,
realtimeStreamsVersion: determineRealtimeStreamsVersion(realtimeStreamsVersion ?? undefined),
});
if (!result) {
@@ -11,6 +11,8 @@ import { prisma } from "~/db.server";
import { env } from "~/env.server";
import { ApiAuthenticationResultSuccess, getOneTimeUseToken } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
import {
@@ -33,6 +35,7 @@ export const HeadersSchema = z.object({
"x-trigger-client": z.string().nullish(),
"x-trigger-engine-version": RunEngineVersionSchema.nullish(),
"x-trigger-request-idempotency-key": z.string().nullish(),
"x-trigger-realtime-streams-version": z.string().nullish(),
traceparent: z.string().optional(),
tracestate: z.string().optional(),
});
@@ -63,6 +66,7 @@ const { action, loader } = createActionApiRoute(
"x-trigger-client": triggerClient,
"x-trigger-engine-version": engineVersion,
"x-trigger-request-idempotency-key": requestIdempotencyKey,
"x-trigger-realtime-streams-version": realtimeStreamsVersion,
} = headers;
const cachedResponse = await handleRequestIdempotency(requestIdempotencyKey, {
@@ -82,7 +86,7 @@ const { action, loader } = createActionApiRoute(
isCached: false,
}),
buildResponseHeaders: async (responseBody, cachedEntity) => {
return await responseHeaders(cachedEntity, authentication, triggerClient);
return await responseHeaders(cachedEntity, authentication);
},
});
@@ -99,25 +103,6 @@ const { action, loader } = createActionApiRoute(
const oneTimeUseToken = await getOneTimeUseToken(authentication);
logger.debug("Triggering task", {
taskId: params.taskId,
idempotencyKey,
idempotencyKeyTTL,
triggerVersion,
headers,
options: body.options,
isFromWorker,
traceContext,
});
logger.debug("[otelContext]", {
taskId: params.taskId,
headers,
options: body.options,
isFromWorker,
traceContext,
});
const idempotencyKeyExpiresAt = resolveIdempotencyKeyTTL(idempotencyKeyTTL);
const result = await service.call(
@@ -131,6 +116,9 @@ const { action, loader } = createActionApiRoute(
traceContext,
spanParentAsLink: spanParentAsLink === 1,
oneTimeUseToken,
realtimeStreamsVersion: determineRealtimeStreamsVersion(
realtimeStreamsVersion ?? undefined
),
},
engineVersion ?? undefined
);
@@ -141,7 +129,7 @@ const { action, loader } = createActionApiRoute(
await saveRequestIdempotency(requestIdempotencyKey, "trigger", result.run.id);
const $responseHeaders = await responseHeaders(result.run, authentication, triggerClient);
const $responseHeaders = await responseHeaders(result.run, authentication);
return json(
{
@@ -171,39 +159,26 @@ const { action, loader } = createActionApiRoute(
async function responseHeaders(
run: Pick<TaskRun, "friendlyId">,
authentication: ApiAuthenticationResultSuccess,
triggerClient?: string | null
authentication: ApiAuthenticationResultSuccess
): Promise<Record<string, string>> {
const { environment, realtime } = authentication;
const claimsHeader = JSON.stringify({
const claims = {
sub: environment.id,
pub: true,
scopes: [`read:runs:${run.friendlyId}`],
realtime,
};
const jwt = await internal_generateJWT({
secretKey: extractJwtSigningSecretKey(environment),
payload: claims,
expirationTime: "1h",
});
if (triggerClient === "browser") {
const claims = {
sub: environment.id,
pub: true,
scopes: [`read:runs:${run.friendlyId}`],
realtime,
};
const jwt = await internal_generateJWT({
secretKey: environment.apiKey,
payload: claims,
expirationTime: "1h",
});
return {
"x-trigger-jwt-claims": claimsHeader,
"x-trigger-jwt": jwt,
};
}
return {
"x-trigger-jwt-claims": claimsHeader,
"x-trigger-jwt-claims": JSON.stringify(claims),
"x-trigger-jwt": jwt,
};
}
+7 -1
View File
@@ -16,6 +16,8 @@ import {
} from "~/v3/services/batchTriggerV3.server";
import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
const { action, loader } = createActionApiRoute(
{
@@ -69,6 +71,7 @@ const { action, loader } = createActionApiRoute(
"x-trigger-client": triggerClient,
"x-trigger-engine-version": engineVersion,
"batch-processing-strategy": batchProcessingStrategy,
"x-trigger-realtime-streams-version": realtimeStreamsVersion,
traceparent,
tracestate,
} = headers;
@@ -107,6 +110,9 @@ const { action, loader } = createActionApiRoute(
traceContext,
spanParentAsLink: spanParentAsLink === 1,
oneTimeUseToken,
realtimeStreamsVersion: determineRealtimeStreamsVersion(
realtimeStreamsVersion ?? undefined
),
});
const $responseHeaders = await responseHeaders(
@@ -158,7 +164,7 @@ async function responseHeaders(
};
const jwt = await generateJWT({
secretKey: environment.apiKey,
secretKey: extractJwtSigningSecretKey(environment),
payload: claims,
expirationTime: "1h",
});
+7 -1
View File
@@ -18,6 +18,8 @@ import { ServiceValidationError } from "~/v3/services/baseService.server";
import { BatchProcessingStrategy } from "~/v3/services/batchTriggerV3.server";
import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
const { action, loader } = createActionApiRoute(
{
@@ -59,6 +61,7 @@ const { action, loader } = createActionApiRoute(
"x-trigger-engine-version": engineVersion,
"batch-processing-strategy": batchProcessingStrategy,
"x-trigger-request-idempotency-key": requestIdempotencyKey,
"x-trigger-realtime-streams-version": realtimeStreamsVersion,
traceparent,
tracestate,
} = headers;
@@ -119,6 +122,9 @@ const { action, loader } = createActionApiRoute(
traceContext,
spanParentAsLink: spanParentAsLink === 1,
oneTimeUseToken,
realtimeStreamsVersion: determineRealtimeStreamsVersion(
realtimeStreamsVersion ?? undefined
),
});
const $responseHeaders = await responseHeaders(
@@ -173,7 +179,7 @@ async function responseHeaders(
};
const jwt = await generateJWT({
secretKey: environment.apiKey,
secretKey: extractJwtSigningSecretKey(environment),
payload: claims,
expirationTime: "1h",
});
@@ -1,9 +0,0 @@
import { redirect, type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { EnvironmentParamSchema, v3QueuesPath } from "~/utils/pathBuilder";
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
return redirect(
v3QueuesPath({ slug: organizationSlug }, { slug: projectParam }, { slug: envParam })
);
};
@@ -1,22 +1,77 @@
import { ActionFunctionArgs } from "@remix-run/server-runtime";
import { type ActionFunctionArgs } from "@remix-run/server-runtime";
import { z } from "zod";
import { $replica } from "~/db.server";
import { relayRealtimeStreams } from "~/services/realtime/relayRealtimeStreams.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
const ParamsSchema = z.object({
runId: z.string(),
streamId: z.string(),
});
// Plain action for backwards compatibility with older clients that don't send auth headers
export async function action({ request, params }: ActionFunctionArgs) {
const $params = ParamsSchema.parse(params);
const parsedParams = ParamsSchema.safeParse(params);
if (!parsedParams.success) {
return new Response("Invalid parameters", { status: 400 });
}
const { runId, streamId } = parsedParams.data;
// Look up the run without environment scoping for backwards compatibility
const run = await $replica.taskRun.findFirst({
where: {
friendlyId: runId,
},
select: {
id: true,
friendlyId: true,
runtimeEnvironment: {
include: {
project: true,
organization: true,
orgMember: true,
},
},
},
});
if (!run) {
return new Response("Run not found", { status: 404 });
}
// Extract client ID from header, default to "default" if not provided
const clientId = request.headers.get("X-Client-Id") || "default";
const streamVersion = request.headers.get("X-Stream-Version") || "v1";
if (!request.body) {
return new Response("No body provided", { status: 400 });
}
return relayRealtimeStreams.ingestData(request.body, $params.runId, $params.streamId);
const resumeFromChunk = request.headers.get("X-Resume-From-Chunk");
let resumeFromChunkNumber: number | undefined = undefined;
if (resumeFromChunk) {
const parsed = parseInt(resumeFromChunk, 10);
if (isNaN(parsed) || parsed < 0) {
return new Response(`Invalid X-Resume-From-Chunk header value: ${resumeFromChunk}`, {
status: 400,
});
}
resumeFromChunkNumber = parsed;
}
// The runtimeEnvironment from the run is already in the correct shape for AuthenticatedEnvironment
const realtimeStream = getRealtimeStreamInstance(run.runtimeEnvironment, streamVersion);
return realtimeStream.ingestData(
request.body,
run.friendlyId,
streamId,
clientId,
resumeFromChunkNumber
);
}
export const loader = createLoaderApiRoute(
@@ -51,12 +106,32 @@ export const loader = createLoaderApiRoute(
},
},
async ({ params, request, resource: run, authentication }) => {
return relayRealtimeStreams.streamResponse(
request,
run.friendlyId,
params.streamId,
// Get Last-Event-ID header for resuming from a specific position
const lastEventId = request.headers.get("Last-Event-ID") || undefined;
const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined;
if (timeoutInSeconds && isNaN(timeoutInSeconds)) {
return new Response("Invalid timeout seconds", { status: 400 });
}
if (timeoutInSeconds && timeoutInSeconds < 1) {
return new Response("Timeout seconds must be greater than 0", { status: 400 });
}
if (timeoutInSeconds && timeoutInSeconds > 600) {
return new Response("Timeout seconds must be less than 600", { status: 400 });
}
const realtimeStream = getRealtimeStreamInstance(
authentication.environment,
request.signal
run.realtimeStreamsVersion
);
return realtimeStream.streamResponse(request, run.friendlyId, params.streamId, request.signal, {
lastEventId,
timeoutInSeconds,
});
}
);
@@ -0,0 +1,135 @@
import { json } from "@remix-run/server-runtime";
import { tryCatch } from "@trigger.dev/core/utils";
import { nanoid } from "nanoid";
import { z } from "zod";
import { $replica, prisma } from "~/db.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
import { ServiceValidationError } from "~/v3/services/common.server";
const ParamsSchema = z.object({
runId: z.string(),
target: z.enum(["self", "parent", "root"]),
streamId: z.string(),
});
const { action } = createActionApiRoute(
{
params: ParamsSchema,
},
async ({ request, params, authentication }) => {
const run = await $replica.taskRun.findFirst({
where: {
friendlyId: params.runId,
runtimeEnvironmentId: authentication.environment.id,
},
select: {
id: true,
friendlyId: true,
parentTaskRun: {
select: {
friendlyId: true,
},
},
rootTaskRun: {
select: {
friendlyId: true,
},
},
},
});
if (!run) {
return new Response("Run not found", { status: 404 });
}
const targetId =
params.target === "self"
? run.friendlyId
: params.target === "parent"
? run.parentTaskRun?.friendlyId
: run.rootTaskRun?.friendlyId;
if (!targetId) {
return new Response("Target not found", { status: 404 });
}
const targetRun = await prisma.taskRun.findFirst({
where: {
friendlyId: targetId,
runtimeEnvironmentId: authentication.environment.id,
},
select: {
realtimeStreams: true,
realtimeStreamsVersion: true,
completedAt: true,
id: true,
},
});
if (!targetRun) {
return new Response("Run not found", { status: 404 });
}
if (targetRun.completedAt) {
return new Response("Cannot append to a realtime stream on a completed run", {
status: 400,
});
}
if (!targetRun.realtimeStreams.includes(params.streamId)) {
await prisma.taskRun.update({
where: {
id: targetRun.id,
},
data: {
realtimeStreams: {
push: params.streamId,
},
},
});
}
const part = await request.text();
const realtimeStream = getRealtimeStreamInstance(
authentication.environment,
targetRun.realtimeStreamsVersion
);
const partId = request.headers.get("X-Part-Id") ?? nanoid(7);
const [appendError] = await tryCatch(
realtimeStream.appendPart(part, partId, targetId, params.streamId)
);
if (appendError) {
if (appendError instanceof ServiceValidationError) {
return json(
{
ok: false,
error: appendError.message,
},
{ status: appendError.status ?? 422 }
);
} else {
return json(
{
ok: false,
error: appendError.message,
},
{ status: 500 }
);
}
}
return json(
{
ok: true,
},
{ status: 200 }
);
}
);
export { action };
@@ -1,7 +1,11 @@
import { json } from "@remix-run/server-runtime";
import { z } from "zod";
import { $replica } from "~/db.server";
import { relayRealtimeStreams } from "~/services/realtime/relayRealtimeStreams.server";
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
import { $replica, prisma } from "~/db.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import {
createActionApiRoute,
createLoaderApiRoute,
} from "~/services/routeBuilders/apiBuilder.server";
const ParamsSchema = z.object({
runId: z.string(),
@@ -14,10 +18,6 @@ const { action } = createActionApiRoute(
params: ParamsSchema,
},
async ({ request, params, authentication }) => {
if (!request.body) {
return new Response("No body provided", { status: 400 });
}
const run = await $replica.taskRun.findFirst({
where: {
friendlyId: params.runId,
@@ -54,8 +54,145 @@ const { action } = createActionApiRoute(
return new Response("Target not found", { status: 404 });
}
return relayRealtimeStreams.ingestData(request.body, targetId, params.streamId);
if (request.method === "PUT") {
// This is the "create" endpoint
const updatedRun = await prisma.taskRun.update({
where: {
friendlyId: targetId,
runtimeEnvironmentId: authentication.environment.id,
},
data: {
realtimeStreams: {
push: params.streamId,
},
},
select: {
realtimeStreamsVersion: true,
completedAt: true,
},
});
if (updatedRun.completedAt) {
return new Response("Cannot initialize a realtime stream on a completed run", {
status: 400,
});
}
const realtimeStream = getRealtimeStreamInstance(
authentication.environment,
updatedRun.realtimeStreamsVersion
);
const { responseHeaders } = await realtimeStream.initializeStream(targetId, params.streamId);
return json(
{
version: updatedRun.realtimeStreamsVersion,
},
{ status: 202, headers: responseHeaders }
);
} else {
// Extract client ID from header, default to "default" if not provided
const clientId = request.headers.get("X-Client-Id") || "default";
const streamVersion = request.headers.get("X-Stream-Version") || "v1";
if (!request.body) {
return new Response("No body provided", { status: 400 });
}
const resumeFromChunk = request.headers.get("X-Resume-From-Chunk");
let resumeFromChunkNumber: number | undefined = undefined;
if (resumeFromChunk) {
const parsed = parseInt(resumeFromChunk, 10);
if (isNaN(parsed) || parsed < 0) {
return new Response(`Invalid X-Resume-From-Chunk header value: ${resumeFromChunk}`, {
status: 400,
});
}
resumeFromChunkNumber = parsed;
}
const realtimeStream = getRealtimeStreamInstance(authentication.environment, streamVersion);
return realtimeStream.ingestData(
request.body,
targetId,
params.streamId,
clientId,
resumeFromChunkNumber
);
}
}
);
export { action };
const loader = createLoaderApiRoute(
{
params: ParamsSchema,
allowJWT: false,
corsStrategy: "none",
findResource: async (params, authentication) => {
return $replica.taskRun.findFirst({
where: {
friendlyId: params.runId,
runtimeEnvironmentId: authentication.environment.id,
},
select: {
id: true,
friendlyId: true,
parentTaskRun: {
select: {
friendlyId: true,
},
},
rootTaskRun: {
select: {
friendlyId: true,
},
},
},
});
},
},
async ({ request, params, resource: run, authentication }) => {
if (!run) {
return new Response("Run not found", { status: 404 });
}
const targetId =
params.target === "self"
? run.friendlyId
: params.target === "parent"
? run.parentTaskRun?.friendlyId
: run.rootTaskRun?.friendlyId;
if (!targetId) {
return new Response("Target not found", { status: 404 });
}
// Handle HEAD request to get last chunk index
if (request.method !== "HEAD") {
return new Response("Only HEAD requests are allowed for this endpoint", { status: 405 });
}
// Extract client ID from header, default to "default" if not provided
const clientId = request.headers.get("X-Client-Id") || "default";
const streamVersion = request.headers.get("X-Stream-Version") || "v1";
const realtimeStream = getRealtimeStreamInstance(authentication.environment, streamVersion);
const lastChunkIndex = await realtimeStream.getLastChunkIndex(
targetId,
params.streamId,
clientId
);
return new Response(null, {
status: 200,
headers: {
"X-Last-Chunk-Index": lastChunkIndex.toString(),
},
});
}
);
export { action, loader };
@@ -80,6 +80,7 @@ import { createTimelineSpanEventsFromSpanEvents } from "~/utils/timelineSpanEven
import { CompleteWaitpointForm } from "../resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route";
import { requireUserId } from "~/services/session.server";
import type { SpanOverride } from "~/v3/eventRepository/eventRepository.types";
import { RealtimeStreamViewer } from "../resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route";
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
@@ -213,8 +214,8 @@ function SpanBody({
span = applySpanOverrides(span, spanOverrides);
return (
<div className="grid h-full max-h-full grid-rows-[2.5rem_2rem_1fr] overflow-hidden bg-background-bright">
<div className="flex items-center justify-between gap-2 overflow-x-hidden px-3">
<div className="grid h-full max-h-full grid-rows-[2.5rem_1fr] overflow-hidden bg-background-bright">
<div className="flex items-center justify-between gap-2 overflow-x-hidden border-b border-grid-bright px-3 pr-2">
<div className="flex items-center gap-1 overflow-x-hidden">
<RunIcon
name={span.style?.icon}
@@ -228,26 +229,14 @@ function SpanBody({
{runParam && closePanel && (
<Button
onClick={closePanel}
variant="minimal/medium"
LeadingIcon={ExitIcon}
variant="minimal/small"
TrailingIcon={ExitIcon}
shortcut={{ key: "esc" }}
shortcutPosition="before-trailing-icon"
className="pl-1"
/>
)}
</div>
<div className="h-fit overflow-x-auto px-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
<TabContainer>
<TabButton
isActive={!tab || tab === "overview"}
layoutId="span-span"
onClick={() => {
replace({ tab: "overview" });
}}
shortcut={{ key: "o" }}
>
Overview
</TabButton>
</TabContainer>
</div>
<div className="overflow-y-auto scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
<SpanEntity span={span} />
</div>
@@ -307,7 +296,7 @@ function RunBody({
return (
<div className="grid h-full max-h-full grid-rows-[2.5rem_2rem_1fr_3.25rem] overflow-hidden bg-background-bright">
<div className="flex items-center justify-between gap-2 overflow-x-hidden px-3">
<div className="flex items-center justify-between gap-2 overflow-x-hidden px-3 pr-2">
<div className="flex items-center gap-1 overflow-x-hidden">
<RunIcon
name={run.isCached ? "task-cached" : "task"}
@@ -324,9 +313,11 @@ function RunBody({
{runParam && closePanel && (
<Button
onClick={closePanel}
variant="minimal/medium"
LeadingIcon={ExitIcon}
variant="minimal/small"
TrailingIcon={ExitIcon}
shortcut={{ key: "esc" }}
shortcutPosition="before-trailing-icon"
className="pl-1"
/>
)}
</div>
@@ -824,6 +815,10 @@ function RunBody({
<Property.Label>Span ID</Property.Label>
<Property.Value>{run.spanId}</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Task event store</Property.Label>
<Property.Value>{run.taskEventStore}</Property.Value>
</Property.Item>
</div>
)}
</Property.Table>
@@ -1075,6 +1070,20 @@ function SpanEntity({ span }: { span: Span }) {
code={span.properties}
maxLines={20}
showLineNumbers={false}
showCopyButton
showTextWrapping
showOpenInModal
/>
) : null}
{span.resourceProperties !== undefined ? (
<CodeBlock
rowTitle="Resource properties"
code={span.resourceProperties}
maxLines={20}
showLineNumbers={false}
showCopyButton
showTextWrapping
showOpenInModal
/>
) : null}
</div>
@@ -1120,6 +1129,9 @@ function SpanEntity({ span }: { span: Span }) {
code={span.properties}
maxLines={20}
showLineNumbers={false}
showCopyButton
showTextWrapping
showOpenInModal
/>
) : null}
</div>
@@ -1146,6 +1158,15 @@ function SpanEntity({ span }: { span: Span }) {
</div>
);
}
case "realtime-stream": {
return (
<RealtimeStreamViewer
runId={span.entity.object.runId}
streamKey={span.entity.object.streamKey}
metadata={span.entity.object.metadata}
/>
);
}
default: {
assertNever(span.entity);
}
@@ -0,0 +1,544 @@
import { BoltIcon, BoltSlashIcon } from "@heroicons/react/20/solid";
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { type SSEStreamPart, SSEStreamSubscription } from "@trigger.dev/core/v3";
import { useVirtualizer } from "@tanstack/react-virtual";
import { Clipboard, ClipboardCheck } from "lucide-react";
import { useCallback, useEffect, useRef, useState } from "react";
import simplur from "simplur";
import { ListBulletIcon } from "~/assets/icons/ListBulletIcon";
import { MoveToBottomIcon } from "~/assets/icons/MoveToBottomIcon";
import { MoveToTopIcon } from "~/assets/icons/MoveToTopIcon";
import { SnakedArrowIcon } from "~/assets/icons/SnakedArrowIcon";
import { Paragraph } from "~/components/primitives/Paragraph";
import { Spinner } from "~/components/primitives/Spinner";
import {
Tooltip,
TooltipContent,
TooltipProvider,
TooltipTrigger,
} from "~/components/primitives/Tooltip";
import { $replica } from "~/db.server";
import { useEnvironment } from "~/hooks/useEnvironment";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import { requireUserId } from "~/services/session.server";
import { cn } from "~/utils/cn";
import { v3RunStreamParamsSchema } from "~/utils/pathBuilder";
type ViewMode = "list" | "compact";
type StreamChunk = {
id: string;
data: unknown;
timestamp: number;
};
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
const { projectParam, organizationSlug, envParam, runParam, streamKey } =
v3RunStreamParamsSchema.parse(params);
const project = await $replica.project.findFirst({
where: {
slug: projectParam,
organization: {
slug: organizationSlug,
members: {
some: {
userId,
},
},
},
},
});
if (!project) {
throw new Response("Not Found", { status: 404 });
}
const run = await $replica.taskRun.findFirst({
where: {
friendlyId: runParam,
projectId: project.id,
},
include: {
runtimeEnvironment: {
include: {
project: true,
organization: true,
orgMember: true,
},
},
},
});
if (!run) {
throw new Response("Not Found", { status: 404 });
}
if (run.runtimeEnvironment.slug !== envParam) {
throw new Response("Not Found", { status: 404 });
}
// Get Last-Event-ID header for resuming from a specific position
const lastEventId = request.headers.get("Last-Event-ID") || undefined;
const realtimeStream = getRealtimeStreamInstance(
run.runtimeEnvironment,
run.realtimeStreamsVersion
);
return realtimeStream.streamResponse(request, run.friendlyId, streamKey, request.signal, {
lastEventId,
});
};
export function RealtimeStreamViewer({
runId,
streamKey,
metadata,
}: {
runId: string;
streamKey: string;
metadata: Record<string, unknown> | undefined;
}) {
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const resourcePath = `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/runs/${runId}/streams/${streamKey}`;
const startIndex = typeof metadata?.startIndex === "number" ? metadata.startIndex : undefined;
const { chunks, error, isConnected } = useRealtimeStream(resourcePath, startIndex);
const scrollRef = useRef<HTMLDivElement>(null);
const bottomRef = useRef<HTMLDivElement>(null);
const [isAtBottom, setIsAtBottom] = useState(true);
const [viewMode, setViewMode] = useState<ViewMode>("list");
const [mouseOver, setMouseOver] = useState(false);
const [copied, setCopied] = useState(false);
const getCompactText = useCallback(() => {
return chunks
.map((chunk) => {
if (typeof chunk.data === "string") {
return chunk.data;
}
return JSON.stringify(chunk.data);
})
.join("");
}, [chunks]);
const onCopied = useCallback(
(event: React.MouseEvent<HTMLButtonElement>) => {
event.preventDefault();
event.stopPropagation();
navigator.clipboard.writeText(getCompactText());
setCopied(true);
setTimeout(() => {
setCopied(false);
}, 1500);
},
[getCompactText]
);
// Use IntersectionObserver to detect when the bottom element is visible
useEffect(() => {
const bottomElement = bottomRef.current;
const scrollElement = scrollRef.current;
if (!bottomElement || !scrollElement) return;
const observer = new IntersectionObserver(
(entries) => {
const entry = entries[0];
if (entry) {
setIsAtBottom(entry.isIntersecting);
}
},
{
root: scrollElement,
threshold: 0.1,
rootMargin: "0px",
}
);
observer.observe(bottomElement);
// Also add a scroll listener as a backup to ensure state updates
let scrollTimeout: ReturnType<typeof setTimeout> | null = null;
const handleScroll = () => {
if (!scrollElement || !bottomElement) return;
// Clear any existing timeout
if (scrollTimeout) {
clearTimeout(scrollTimeout);
}
// Debounce the state update to avoid interrupting smooth scroll
scrollTimeout = setTimeout(() => {
const scrollBottom = scrollElement.scrollTop + scrollElement.clientHeight;
const isNearBottom = scrollElement.scrollHeight - scrollBottom < 50;
setIsAtBottom(isNearBottom);
}, 100);
};
scrollElement.addEventListener("scroll", handleScroll);
// Check initial state
const scrollBottom = scrollElement.scrollTop + scrollElement.clientHeight;
const isNearBottom = scrollElement.scrollHeight - scrollBottom < 50;
setIsAtBottom(isNearBottom);
return () => {
observer.disconnect();
scrollElement.removeEventListener("scroll", handleScroll);
if (scrollTimeout) {
clearTimeout(scrollTimeout);
}
};
}, [chunks.length, viewMode]);
// Auto-scroll to bottom when new chunks arrive, if we're at the bottom
useEffect(() => {
if (isAtBottom && scrollRef.current) {
// Preserve horizontal scroll position while scrolling to bottom vertically
const currentScrollLeft = scrollRef.current.scrollLeft;
scrollRef.current.scrollTop = scrollRef.current.scrollHeight;
scrollRef.current.scrollLeft = currentScrollLeft;
}
}, [chunks, isAtBottom]);
const firstLineNumber = startIndex ?? 0;
const lastLineNumber = firstLineNumber + chunks.length - 1;
const maxLineNumberWidth = (chunks.length > 0 ? lastLineNumber : firstLineNumber).toString()
.length;
// Virtual rendering for list view
const rowVirtualizer = useVirtualizer({
count: chunks.length,
getScrollElement: () => scrollRef.current,
estimateSize: () => 28,
overscan: 5,
});
return (
<div className="flex h-full flex-col overflow-hidden">
{/* Header */}
<div className="border-b border-grid-bright bg-background-bright @container">
<div className="flex flex-wrap items-center justify-between gap-2 px-3 py-2.5 @[300px]:flex-nowrap">
<div className="flex min-w-0 items-center gap-1.5">
<TooltipProvider>
<Tooltip>
<TooltipTrigger>
{isConnected ? (
<BoltIcon className={cn("size-3.5 animate-pulse text-success")} />
) : (
<BoltSlashIcon className={cn("size-3.5 text-text-dimmed")} />
)}
</TooltipTrigger>
<TooltipContent side="top" className="text-xs">
{isConnected ? "Connected" : "Disconnected"}
</TooltipContent>
</Tooltip>
</TooltipProvider>
<Paragraph
variant="small/bright"
className="mb-0 flex min-w-0 items-center gap-1 truncate whitespace-nowrap"
>
<span>Stream:</span>
<span className="truncate font-mono text-text-dimmed">{streamKey}</span>
</Paragraph>
</div>
<div className="flex w-full flex-wrap items-center justify-between gap-3 @[300px]:w-auto @[300px]:flex-nowrap">
<Paragraph variant="small" className="mb-0 whitespace-nowrap">
{simplur`${chunks.length} chunk[|s]`}
</Paragraph>
<div className="flex items-center gap-3">
<TooltipProvider>
<Tooltip open={chunks.length === 0 ? false : undefined} disableHoverableContent>
<TooltipTrigger
disabled={chunks.length === 0}
onClick={() => setViewMode(viewMode === "list" ? "compact" : "list")}
className={cn(
"text-text-dimmed transition-colors focus-custom",
chunks.length === 0
? "cursor-not-allowed opacity-50"
: "hover:cursor-pointer hover:text-text-bright"
)}
>
{viewMode === "list" ? (
<SnakedArrowIcon className="size-4" />
) : (
<ListBulletIcon className="size-4" />
)}
</TooltipTrigger>
<TooltipContent side="left" className="text-xs">
{viewMode === "list" ? "Flow as text" : "View as list"}
</TooltipContent>
</Tooltip>
</TooltipProvider>
<TooltipProvider>
<Tooltip
open={chunks.length === 0 ? false : copied || mouseOver || undefined}
disableHoverableContent
>
<TooltipTrigger
disabled={chunks.length === 0}
onClick={onCopied}
onMouseEnter={() => setMouseOver(true)}
onMouseLeave={() => setMouseOver(false)}
className={cn(
"transition-colors duration-100 focus-custom",
chunks.length === 0
? "cursor-not-allowed opacity-50"
: copied
? "text-success hover:cursor-pointer"
: "text-text-dimmed hover:cursor-pointer hover:text-text-bright"
)}
>
{copied ? (
<ClipboardCheck className="size-4" />
) : (
<Clipboard className="size-4" />
)}
</TooltipTrigger>
<TooltipContent side="left" className="text-xs">
{copied ? "Copied" : "Copy"}
</TooltipContent>
</Tooltip>
</TooltipProvider>
<TooltipProvider>
<Tooltip open={chunks.length === 0 ? false : undefined} disableHoverableContent>
<TooltipTrigger
disabled={chunks.length === 0}
onClick={() => {
if (isAtBottom) {
scrollRef.current?.scrollTo({ top: 0, behavior: "smooth" });
} else {
bottomRef.current?.scrollIntoView({ behavior: "smooth", block: "end" });
}
}}
className={cn(
"text-text-dimmed transition-colors focus-custom",
chunks.length === 0
? "cursor-not-allowed opacity-50"
: "hover:cursor-pointer hover:text-text-bright"
)}
>
{isAtBottom ? (
<MoveToTopIcon className="size-4" />
) : (
<MoveToBottomIcon className="size-4" />
)}
</TooltipTrigger>
<TooltipContent side="left" className="text-xs">
{isAtBottom ? "Scroll to top" : "Scroll to bottom"}
</TooltipContent>
</Tooltip>
</TooltipProvider>
</div>
</div>
</div>
</div>
{/* Content */}
<div
ref={scrollRef}
className="flex-1 overflow-x-auto overflow-y-auto bg-charcoal-900 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600"
>
{error && (
<div className="border-b border-error/20 bg-error/10 p-3">
<Paragraph variant="small" className="mb-0 text-error">
Error: {error.message}
</Paragraph>
</div>
)}
{chunks.length === 0 && !error && (
<div className="flex h-full items-center justify-center">
{isConnected ? (
<div className="flex items-center gap-2">
<Spinner />
<Paragraph variant="small" className="mb-0 text-text-dimmed">
Waiting for data
</Paragraph>
</div>
) : (
<Paragraph variant="small" className="mb-0 text-text-dimmed">
No data received
</Paragraph>
)}
</div>
)}
{chunks.length > 0 && viewMode === "list" && (
<div className="font-mono text-xs leading-tight">
<div
style={{
height: `${rowVirtualizer.getTotalSize()}px`,
position: "relative",
minWidth: "100%",
}}
>
{rowVirtualizer.getVirtualItems().map((virtualItem) => (
<StreamChunkLine
key={virtualItem.key}
chunk={chunks[virtualItem.index]}
lineNumber={firstLineNumber + virtualItem.index}
maxLineNumberWidth={maxLineNumberWidth}
size={virtualItem.size}
start={virtualItem.start}
/>
))}
{/* Sentinel element for IntersectionObserver */}
<div
ref={bottomRef}
className="h-px"
style={{
position: "absolute",
top: `${rowVirtualizer.getTotalSize()}px`,
}}
/>
</div>
</div>
)}
{chunks.length > 0 && viewMode === "compact" && (
<div className="p-3 font-mono text-xs leading-relaxed">
<CompactStreamView chunks={chunks} />
{/* Sentinel element for IntersectionObserver */}
<div ref={bottomRef} className="h-px" />
</div>
)}
</div>
</div>
);
}
function CompactStreamView({ chunks }: { chunks: StreamChunk[] }) {
const compactText = chunks
.map((chunk) => {
if (typeof chunk.data === "string") {
return chunk.data;
}
return JSON.stringify(chunk.data);
})
.join("");
return <div className="whitespace-pre-wrap break-all text-text-bright">{compactText}</div>;
}
function StreamChunkLine({
chunk,
lineNumber,
maxLineNumberWidth,
size,
start,
}: {
chunk: StreamChunk;
lineNumber: number;
maxLineNumberWidth: number;
size: number;
start: number;
}) {
const formattedData =
typeof chunk.data === "string" ? chunk.data : JSON.stringify(chunk.data, null, 2);
const date = new Date(chunk.timestamp);
const timeString = date.toLocaleTimeString("en-US", {
hour12: false,
hour: "2-digit",
minute: "2-digit",
second: "2-digit",
});
const milliseconds = date.getMilliseconds().toString().padStart(3, "0");
const timestamp = `${timeString}.${milliseconds}`;
return (
<div
className="group flex gap-3 py-1 hover:bg-charcoal-800"
style={{
position: "absolute",
top: 0,
left: 0,
height: `${size}px`,
transform: `translateY(${start}px)`,
}}
>
{/* Line number */}
<div
className="flex-none select-none pl-2 text-right text-charcoal-500"
style={{ width: `${Math.max(maxLineNumberWidth, 3)}ch` }}
>
{lineNumber}
</div>
{/* Timestamp */}
<div className="flex-none select-none pl-1 text-charcoal-500">{timestamp}</div>
{/* Content */}
<div className="whitespace-nowrap text-text-bright">{formattedData}</div>
</div>
);
}
function useRealtimeStream(resourcePath: string, startIndex?: number) {
const [chunks, setChunks] = useState<StreamChunk[]>([]);
const [error, setError] = useState<Error | null>(null);
const [isConnected, setIsConnected] = useState(false);
useEffect(() => {
const abortController = new AbortController();
let reader: ReadableStreamDefaultReader<SSEStreamPart<unknown>> | null = null;
async function connectAndConsume() {
try {
const sseSubscription = new SSEStreamSubscription(resourcePath, {
signal: abortController.signal,
lastEventId: startIndex ? (startIndex - 1).toString() : undefined,
timeoutInSeconds: 30,
});
const stream = await sseSubscription.subscribe();
setIsConnected(true);
reader = stream.getReader();
// Read from the stream
while (true) {
const { done, value } = await reader.read();
if (done) {
break;
}
if (value !== undefined) {
setChunks((prev) => [
...prev,
{
id: value.id,
data: value.chunk,
timestamp: value.timestamp,
},
]);
}
}
} catch (err) {
// Only set error if not aborted
if (!abortController.signal.aborted) {
setError(err instanceof Error ? err : new Error(String(err)));
}
} finally {
setIsConnected(false);
}
}
connectAndConsume();
return () => {
abortController.abort();
reader?.cancel();
};
}, [resourcePath, startIndex]);
return { chunks, error, isConnected };
}
@@ -11,6 +11,7 @@ import { type ActionFunctionArgs } from "@remix-run/server-runtime";
import { uiComponent } from "@team-plain/typescript-sdk";
import { GitHubLightIcon } from "@trigger.dev/companyicons";
import {
AddOnPricing,
type FreePlanDefinition,
type Limits,
type PaidPlanDefinition,
@@ -45,6 +46,8 @@ import { requireUser } from "~/services/session.server";
import { engine } from "~/v3/runEngine.server";
import { cn } from "~/utils/cn";
import { sendToPlain } from "~/utils/plain.server";
import { formatCurrency } from "~/utils/numberFormatter";
import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
const Params = z.object({
organizationSlug: z.string(),
@@ -153,7 +156,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
}
}
return setPlan(organization, request, form.callerPath, payload, {
return await setPlan(organization, request, form.callerPath, payload, {
invalidateBillingCache: engine.invalidateBillingCache.bind(engine),
});
}
@@ -173,7 +176,6 @@ const pricingDefinitions = {
},
additionalConcurrency: {
title: "Additional concurrency",
content: "Then $50/month per 50",
},
taskRun: {
title: "Task runs",
@@ -227,6 +229,7 @@ const pricingDefinitions = {
type PricingPlansProps = {
plans: Plans;
concurrencyAddOnPricing: AddOnPricing;
subscription?: SubscriptionResult;
organizationSlug: string;
hasPromotedPlan: boolean;
@@ -236,6 +239,7 @@ type PricingPlansProps = {
export function PricingPlans({
plans,
concurrencyAddOnPricing,
subscription,
organizationSlug,
hasPromotedPlan,
@@ -258,7 +262,12 @@ export function PricingPlans({
subscription={subscription}
isHighlighted={hasPromotedPlan}
/>
<TierPro plan={plans.pro} organizationSlug={organizationSlug} subscription={subscription} />
<TierPro
plan={plans.pro}
organizationSlug={organizationSlug}
subscription={subscription}
concurrencyAddOnPricing={concurrencyAddOnPricing}
/>
</div>
<div className="mt-3">
<TierEnterprise />
@@ -654,10 +663,12 @@ export function TierHobby({
export function TierPro({
plan,
concurrencyAddOnPricing,
organizationSlug,
subscription,
}: {
plan: PaidPlanDefinition;
concurrencyAddOnPricing: AddOnPricing;
organizationSlug: string;
subscription?: SubscriptionResult;
}) {
@@ -747,7 +758,9 @@ export function TierPro({
</Form>
<ul className="flex flex-col gap-2.5">
<ConcurrentRuns limits={plan.limits}>
{pricingDefinitions.additionalConcurrency.content}
{`Then ${formatCurrency(concurrencyAddOnPricing.centsPerStep / 100, true)}/month per ${
concurrencyAddOnPricing.stepSize
}`}
</ConcurrentRuns>
<FeatureItem checked>
Unlimited{" "}
@@ -963,10 +976,45 @@ function ConcurrentRuns({ limits, children }: { limits: Limits; children?: React
</>
) : (
<>{limits.concurrentRuns.number} </>
)}{" "}
)}
<DefinitionTip
title={pricingDefinitions.concurrentRuns.title}
content={pricingDefinitions.concurrentRuns.content}
content={
<div className="flex flex-col">
<Paragraph variant="small/dimmed" spacing>
{pricingDefinitions.concurrentRuns.content}
</Paragraph>
<div className="flex items-center gap-x-1">
<EnvironmentLabel environment={{ type: "PRODUCTION" }} />
<span>
{limits.concurrentRuns.production}
{limits.concurrentRuns.canExceed ? "+" : ""}
</span>
</div>
<div className="flex items-center gap-x-1">
<EnvironmentLabel environment={{ type: "STAGING" }} />
<span>
{limits.concurrentRuns.staging}
{limits.concurrentRuns.canExceed ? "+" : ""}
</span>
</div>
<div className="flex items-center gap-x-1">
<EnvironmentLabel environment={{ type: "PREVIEW" }} />
<span>
{limits.concurrentRuns.preview}
{limits.concurrentRuns.canExceed ? "+" : ""}
</span>
</div>
<div className="flex items-center gap-x-1">
<EnvironmentLabel environment={{ type: "DEVELOPMENT" }} />
<span>
{limits.concurrentRuns.development}
{limits.concurrentRuns.canExceed ? "+" : ""}
</span>
</div>
</div>
}
>
concurrent runs
</DefinitionTip>
@@ -20,6 +20,9 @@ function InputFieldSet({ disabled }: { disabled?: boolean }) {
<Input disabled={disabled} variant="medium" placeholder="Name" type="text" />
<Input disabled={disabled} variant="small" placeholder="Name" type="text" />
<Input disabled={disabled} variant="tertiary" placeholder="Name" type="text" />
<Input disabled={disabled} variant="outline/large" placeholder="Name" type="text" />
<Input disabled={disabled} variant="outline/medium" placeholder="Name" type="text" />
<Input disabled={disabled} variant="outline/small" placeholder="Name" type="text" />
</div>
<div className="m-8 flex w-64 flex-col gap-4">
<Input
@@ -51,50 +54,6 @@ function InputFieldSet({ disabled }: { disabled?: boolean }) {
accessory={<ShortcutKey shortcut={{ key: "k", modifiers: ["meta"] }} variant="small" />}
/>
</div>
<div className="m-8 flex w-64 flex-col gap-4">
<Input
disabled={disabled}
variant="large"
placeholder="Search"
icon={<EnvironmentLabel environment={{ type: "DEVELOPMENT" }} />}
accessory={<ShortcutKey shortcut={{ key: "k", modifiers: ["meta"] }} variant="medium" />}
/>
<Input
disabled={disabled}
variant="medium"
placeholder="Search"
icon={<EnvironmentLabel environment={{ type: "DEVELOPMENT" }} />}
accessory={<ShortcutKey shortcut={{ key: "k", modifiers: ["meta"] }} variant="medium" />}
/>
<Input
disabled={disabled}
variant="small"
placeholder="Search"
icon={<EnvironmentLabel environment={{ type: "DEVELOPMENT" }} />}
accessory={<ShortcutKey shortcut={{ key: "k", modifiers: ["meta"] }} variant="small" />}
/>
<Input
disabled={disabled}
variant="tertiary"
placeholder="Search"
icon={<EnvironmentLabel environment={{ type: "DEVELOPMENT" }} />}
accessory={<ShortcutKey shortcut={{ key: "k", modifiers: ["meta"] }} variant="small" />}
/>
<Input
disabled={disabled}
variant="tertiary"
placeholder="Search"
icon={<EnvironmentLabel environment={{ type: "STAGING" }} />}
accessory={<ShortcutKey shortcut={{ key: "k", modifiers: ["meta"] }} variant="small" />}
/>
<Input
disabled={disabled}
variant="tertiary"
placeholder="Search"
icon={<EnvironmentLabel environment={{ type: "PRODUCTION" }} />}
accessory={<ShortcutKey shortcut={{ key: "k", modifiers: ["meta"] }} variant="small" />}
/>
</div>
</div>
);
}
@@ -0,0 +1,75 @@
import { useState } from "react";
import { Header2, Header3 } from "~/components/primitives/Headers";
import { InputNumberStepper } from "~/components/primitives/InputNumberStepper";
export default function Story() {
const [value1, setValue1] = useState<number | "">(0);
const [value2, setValue2] = useState<number | "">(100);
const [value3, setValue3] = useState<number | "">(0);
const [value4, setValue4] = useState<number | "">(250);
const [value5, setValue5] = useState<number | "">(250);
return (
<div className="grid h-full w-full place-items-center">
<div className="flex flex-col gap-8">
<div className="flex flex-col gap-2">
<Header2>InputNumberStepper</Header2>
<Header3>Size: base (default)</Header3>
<div className="flex flex-col gap-2">
<label className="text-sm text-text-dimmed">Step: 75</label>
<InputNumberStepper
value={value1}
onChange={(e) => setValue1(e.target.value === "" ? "" : Number(e.target.value))}
step={75}
/>
</div>
<div className="flex flex-col gap-2">
<label className="text-sm text-text-dimmed">Step: 50, Min: 0, Max: 1000</label>
<InputNumberStepper
value={value2}
onChange={(e) => setValue2(e.target.value === "" ? "" : Number(e.target.value))}
step={50}
min={0}
max={1000}
/>
</div>
<div className="flex flex-col gap-2">
<label className="text-sm text-text-dimmed">Disabled state</label>
<InputNumberStepper
value={value3}
onChange={(e) => setValue3(e.target.value === "" ? "" : Number(e.target.value))}
step={50}
disabled
/>
</div>
</div>
<div className="flex flex-col gap-2">
<Header3>Size: large</Header3>
<div className="flex flex-col gap-2">
<label className="text-sm text-text-dimmed">Step: 50</label>
<InputNumberStepper
value={value4}
onChange={(e) => setValue4(e.target.value === "" ? "" : Number(e.target.value))}
step={50}
controlSize="large"
/>
</div>
<div className="flex flex-col gap-2">
<label className="text-sm text-text-dimmed">Step: 50, Disabled</label>
<InputNumberStepper
value={value5}
onChange={(e) => setValue5(e.target.value === "" ? "" : Number(e.target.value))}
step={50}
controlSize="large"
disabled={true}
/>
</div>
</div>
</div>
</div>
);
}
@@ -153,6 +153,10 @@ const stories: Story[] = [
name: "Simple form",
slug: "simple-form",
},
{
name: "Stepper",
slug: "stepper",
},
{
name: "Textarea",
slug: "textarea",
@@ -47,6 +47,7 @@ export type BatchTriggerTaskServiceOptions = {
traceContext?: Record<string, string | undefined | Record<string, string | undefined>>;
spanParentAsLink?: boolean;
oneTimeUseToken?: string;
realtimeStreamsVersion?: "v1" | "v2";
};
/**
@@ -708,6 +709,7 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
batchIndex: currentIndex,
skipChecks: true, // Skip entitlement and queue checks since we already validated at batch/chunk level
planType, // Pass planType from batch-level entitlement check
realtimeStreamsVersion: options?.realtimeStreamsVersion,
},
"V2"
);
@@ -347,6 +347,7 @@ export class RunEngineTriggerTaskService {
createdAt: options.overrideCreatedAt,
bulkActionId: body.options?.bulkActionId,
planType,
realtimeStreamsVersion: options.realtimeStreamsVersion,
},
this.prisma
);
+10 -6
View File
@@ -495,7 +495,9 @@ export async function authenticatedEnvironmentForAuthentication(
throw json({ error: "Project not found" }, { status: 404 });
}
if (!branch) {
const sanitizedBranch = sanitizeBranchName(branch);
if (!sanitizedBranch) {
const environment = await prisma.runtimeEnvironment.findFirst({
where: {
projectId: project.id,
@@ -524,8 +526,8 @@ export async function authenticatedEnvironmentForAuthentication(
const environment = await prisma.runtimeEnvironment.findFirst({
where: {
projectId: project.id,
slug: slug,
branchName: sanitizeBranchName(branch),
type: "PREVIEW",
branchName: sanitizedBranch,
archivedAt: null,
},
include: {
@@ -572,7 +574,9 @@ export async function authenticatedEnvironmentForAuthentication(
throw json({ error: "Project not found" }, { status: 404 });
}
if (!branch) {
const sanitizedBranch = sanitizeBranchName(branch);
if (!sanitizedBranch) {
const environment = await prisma.runtimeEnvironment.findFirst({
where: {
projectId: project.id,
@@ -594,8 +598,8 @@ export async function authenticatedEnvironmentForAuthentication(
const environment = await prisma.runtimeEnvironment.findFirst({
where: {
projectId: project.id,
slug: slug,
branchName: sanitizeBranchName(branch),
type: "PREVIEW",
branchName: sanitizedBranch,
archivedAt: null,
},
include: {
+144 -57
View File
@@ -1,21 +1,24 @@
import type { Organization, Project } from "@trigger.dev/database";
import { MachinePresetName, tryCatch } from "@trigger.dev/core/v3";
import type { Organization, Project, RuntimeEnvironmentType } from "@trigger.dev/database";
import {
BillingClient,
type Limits,
type SetPlanBody,
type UsageSeriesParams,
type UsageResult,
defaultMachine as defaultMachineFromPlatform,
machines as machinesFromPlatform,
type MachineCode,
type UpdateBillingAlertsRequest,
type BillingAlertsResult,
type Limits,
type MachineCode,
type ReportUsageResult,
type ReportUsagePlan,
type SetPlanBody,
type UpdateBillingAlertsRequest,
type UsageResult,
type UsageSeriesParams,
type CurrentPlan,
} from "@trigger.dev/platform";
import { createCache, DefaultStatefulContext, Namespace } from "@unkey/cache";
import { MemoryStore } from "@unkey/cache/stores";
import { existsSync, readFileSync } from "node:fs";
import { redirect } from "remix-typedjson";
import { z } from "zod";
import { env } from "~/env.server";
import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
import { createEnvironment } from "~/models/organization.server";
@@ -23,9 +26,7 @@ import { logger } from "~/services/logger.server";
import { newProjectPath, organizationBillingPath } from "~/utils/pathBuilder";
import { singleton } from "~/utils/singleton";
import { RedisCacheStore } from "./unkey/redisCacheStore.server";
import { existsSync, readFileSync } from "node:fs";
import { z } from "zod";
import { MachinePresetName } from "@trigger.dev/core/v3";
import { $replica } from "~/db.server";
function initializeClient() {
if (isCloud() && process.env.BILLING_API_URL && process.env.BILLING_API_KEY) {
@@ -254,6 +255,52 @@ export async function getLimit(orgId: string, limit: keyof Limits, fallback: num
return fallback;
}
export async function getDefaultEnvironmentConcurrencyLimit(
organizationId: string,
environmentType: RuntimeEnvironmentType
): Promise<number> {
if (!client) {
const org = await $replica.organization.findFirst({
where: {
id: organizationId,
},
select: {
maximumConcurrencyLimit: true,
},
});
if (!org) throw new Error("Organization not found");
return org.maximumConcurrencyLimit;
}
const result = await client.currentPlan(organizationId);
if (!result.success) throw new Error("Error getting current plan");
const limit = getDefaultEnvironmentLimitFromPlan(environmentType, result);
if (!limit) throw new Error("No plan found");
return limit;
}
export function getDefaultEnvironmentLimitFromPlan(
environmentType: RuntimeEnvironmentType,
plan: CurrentPlan
): number | undefined {
if (!plan.v3Subscription?.plan) return undefined;
switch (environmentType) {
case "DEVELOPMENT":
return plan.v3Subscription.plan.limits.concurrentRuns.development;
case "STAGING":
return plan.v3Subscription.plan.limits.concurrentRuns.staging;
case "PREVIEW":
return plan.v3Subscription.plan.limits.concurrentRuns.preview;
case "PRODUCTION":
return plan.v3Subscription.plan.limits.concurrentRuns.production;
default:
return plan.v3Subscription.plan.limits.concurrentRuns.number;
}
}
export async function getCachedLimit(orgId: string, limit: keyof Limits, fallback: number) {
return platformCache.limits.swr(`${orgId}:${limit}`, async () => {
return getLimit(orgId, limit, fallback);
@@ -297,62 +344,74 @@ export async function setPlan(
opts?: { invalidateBillingCache?: (orgId: string) => void }
) {
if (!client) {
throw redirectWithErrorMessage(callerPath, request, "Error setting plan");
return redirectWithErrorMessage(callerPath, request, "Error setting plan", {
ephemeral: false,
});
}
try {
const result = await client.setPlan(organization.id, plan);
const [error, result] = await tryCatch(client.setPlan(organization.id, plan));
if (!result) {
throw redirectWithErrorMessage(callerPath, request, "Error setting plan");
if (error) {
return redirectWithErrorMessage(callerPath, request, error.message, { ephemeral: false });
}
if (!result) {
return redirectWithErrorMessage(callerPath, request, "Error setting plan", {
ephemeral: false,
});
}
if (!result.success) {
return redirectWithErrorMessage(callerPath, request, result.error, { ephemeral: false });
}
switch (result.action) {
case "free_connect_required": {
return redirect(result.connectUrl);
}
if (!result.success) {
throw redirectWithErrorMessage(callerPath, request, result.error);
}
switch (result.action) {
case "free_connect_required": {
return redirect(result.connectUrl);
}
case "free_connected": {
if (result.accepted) {
// Invalidate billing cache since plan changed
opts?.invalidateBillingCache?.(organization.id);
return redirect(newProjectPath(organization, "You're on the Free plan."));
} else {
return redirectWithErrorMessage(
callerPath,
request,
"Free tier unlock failed, your GitHub account is too new."
);
}
}
case "create_subscription_flow_start": {
return redirect(result.checkoutUrl);
}
case "updated_subscription": {
// Invalidate billing cache since subscription changed
case "free_connected": {
if (result.accepted) {
// Invalidate billing cache since plan changed
opts?.invalidateBillingCache?.(organization.id);
return redirectWithSuccessMessage(
return redirect(newProjectPath(organization, "You're on the Free plan."));
} else {
return redirectWithErrorMessage(
callerPath,
request,
"Subscription updated successfully."
"Free tier unlock failed, your GitHub account is too new.",
{ ephemeral: false }
);
}
case "canceled_subscription": {
// Invalidate billing cache since subscription was canceled
opts?.invalidateBillingCache?.(organization.id);
return redirectWithSuccessMessage(callerPath, request, "Subscription canceled.");
}
}
case "create_subscription_flow_start": {
return redirect(result.checkoutUrl);
}
case "updated_subscription": {
// Invalidate billing cache since subscription changed
opts?.invalidateBillingCache?.(organization.id);
return redirectWithSuccessMessage(callerPath, request, "Subscription updated successfully.");
}
case "canceled_subscription": {
// Invalidate billing cache since subscription was canceled
opts?.invalidateBillingCache?.(organization.id);
return redirectWithSuccessMessage(callerPath, request, "Subscription canceled.");
}
}
}
export async function setConcurrencyAddOn(organizationId: string, amount: number) {
if (!client) return undefined;
try {
const result = await client.setAddOn(organizationId, { type: "concurrency", amount });
if (!result.success) {
logger.error("Error setting concurrency add on - no success", { error: result.error });
return undefined;
}
return result;
} catch (e) {
logger.error("Error setting plan", { organizationId: organization.id, error: e });
throw redirectWithErrorMessage(
callerPath,
request,
e instanceof Error ? e.message : "Error setting plan"
);
logger.error("Error setting concurrency add on - caught error", { error: e });
return undefined;
}
}
@@ -462,7 +521,10 @@ export async function getEntitlement(
}
}
export async function projectCreated(organization: Organization, project: Project) {
export async function projectCreated(
organization: Pick<Organization, "id" | "maximumConcurrencyLimit">,
project: Project
) {
if (!isCloud()) {
await createEnvironment({ organization, project, type: "STAGING" });
await createEnvironment({
@@ -529,6 +591,31 @@ export async function generateRegistryCredentials(
return result;
}
export async function enqueueBuild(
projectId: string,
deploymentId: string,
artifactKey: string,
options: {
skipPromotion?: boolean;
configFilePath?: string;
}
) {
if (!client) return undefined;
const result = await client.enqueueBuild(projectId, { deploymentId, artifactKey, options });
if (!result.success) {
logger.error("Error enqueuing build", {
error: result.error,
projectId,
deploymentId,
artifactKey,
options,
});
throw new Error("Failed to enqueue build");
}
return result;
}
function isCloud(): boolean {
const acceptableHosts = [
"https://cloud.trigger.dev",
@@ -2,6 +2,7 @@ import { json } from "@remix-run/server-runtime";
import { validateJWT } from "@trigger.dev/core/v3/jwt";
import { findEnvironmentById } from "~/models/runtimeEnvironment.server";
import { AuthenticatedEnvironment } from "../apiAuth.server";
import { logger } from "../logger.server";
export type ValidatePublicJwtKeySuccess = {
ok: true;
@@ -89,6 +90,12 @@ export function isPublicJWT(token: string): boolean {
}
}
export function extractJwtSigningSecretKey(
environment: AuthenticatedEnvironment & { parentEnvironment?: { apiKey: string } }
) {
return environment.parentEnvironment?.apiKey ?? environment.apiKey;
}
function extractJWTSub(token: string): string | undefined {
// Split the token
const parts = token.split(".");
@@ -1,45 +1,90 @@
import { Logger, LogLevel } from "@trigger.dev/core/logger";
import Redis, { RedisOptions } from "ioredis";
import { AuthenticatedEnvironment } from "../apiAuth.server";
import { logger } from "../logger.server";
import { StreamIngestor, StreamResponder } from "./types";
import { LineTransformStream } from "./utils.server";
import { env } from "~/env.server";
import { StreamIngestor, StreamResponder, StreamResponseOptions } from "./types";
export type RealtimeStreamsOptions = {
redis: RedisOptions | undefined;
logger?: Logger;
logLevel?: LogLevel;
inactivityTimeoutMs?: number; // Close stream after this many ms of no new data (default: 60000)
};
// Legacy constant for backward compatibility (no longer written, but still recognized when reading)
const END_SENTINEL = "<<CLOSE_STREAM>>";
// Internal types for stream pipeline
type StreamChunk =
| { type: "ping" }
| { type: "data"; redisId: string; data: string }
| { type: "legacy-data"; redisId: string; data: string };
// Class implementing both interfaces
export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
constructor(private options: RealtimeStreamsOptions) {}
private logger: Logger;
private inactivityTimeoutMs: number;
constructor(private options: RealtimeStreamsOptions) {
this.logger = options.logger ?? new Logger("RedisRealtimeStreams", options.logLevel ?? "info");
this.inactivityTimeoutMs = options.inactivityTimeoutMs ?? 60000; // Default: 60 seconds
}
async initializeStream(
runId: string,
streamId: string
): Promise<{ responseHeaders?: Record<string, string> }> {
return {};
}
async streamResponse(
request: Request,
runId: string,
streamId: string,
environment: AuthenticatedEnvironment,
signal: AbortSignal
signal: AbortSignal,
options?: StreamResponseOptions
): Promise<Response> {
const redis = new Redis(this.options.redis ?? {});
const streamKey = `stream:${runId}:${streamId}`;
let isCleanedUp = false;
const stream = new ReadableStream({
const stream = new ReadableStream<StreamChunk>({
start: async (controller) => {
let lastId = "0";
// Start from lastEventId if provided, otherwise from beginning
let lastId = options?.lastEventId ?? "0";
let retryCount = 0;
const maxRetries = 3;
let lastDataTime = Date.now();
let lastEnqueueTime = Date.now();
const blockTimeMs = 5000;
const pingIntervalMs = 10000; // 10 seconds
if (options?.lastEventId) {
this.logger.debug("[RealtimeStreams][streamResponse] Resuming from lastEventId", {
streamKey,
lastEventId: options?.lastEventId,
});
}
try {
while (!signal.aborted) {
// Check if we need to send a ping
const timeSinceLastEnqueue = Date.now() - lastEnqueueTime;
if (timeSinceLastEnqueue >= pingIntervalMs) {
controller.enqueue({ type: "ping" });
lastEnqueueTime = Date.now();
}
// Compute inactivity threshold once to use consistently in both branches
const inactivityThresholdMs = options?.timeoutInSeconds
? options.timeoutInSeconds * 1000
: this.inactivityTimeoutMs;
try {
const messages = await redis.xread(
"COUNT",
100,
"BLOCK",
5000,
blockTimeMs,
"STREAMS",
streamKey,
lastId
@@ -49,41 +94,104 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
if (messages && messages.length > 0) {
const [_key, entries] = messages[0];
let foundData = false;
for (let i = 0; i < entries.length; i++) {
const [id, fields] = entries[i];
lastId = id;
if (fields && fields.length >= 2) {
if (fields[1] === END_SENTINEL && i === entries.length - 1) {
controller.close();
return;
// Extract the data field from the Redis entry
// Fields format: ["field1", "value1", "field2", "value2", ...]
let data: string | null = null;
for (let j = 0; j < fields.length; j += 2) {
if (fields[j] === "data") {
data = fields[j + 1];
break;
}
}
if (fields[1] !== END_SENTINEL) {
controller.enqueue(fields[1]);
// Handle legacy entries that don't have field names (just data at index 1)
if (data === null && fields.length >= 2) {
data = fields[1];
}
if (signal.aborted) {
controller.close();
return;
if (data) {
// Skip legacy END_SENTINEL entries (backward compatibility)
if (data === END_SENTINEL) {
continue;
}
// Enqueue structured chunk with Redis stream ID
controller.enqueue({
type: "data",
redisId: id,
data,
});
foundData = true;
lastDataTime = Date.now();
lastEnqueueTime = Date.now();
if (signal.aborted) {
controller.close();
return;
}
}
}
}
// If we didn't find any data in this batch, might have only seen sentinels
if (!foundData) {
// Check for inactivity timeout
const inactiveMs = Date.now() - lastDataTime;
if (inactiveMs >= inactivityThresholdMs) {
this.logger.debug(
"[RealtimeStreams][streamResponse] Closing stream due to inactivity",
{
streamKey,
inactiveMs,
threshold: inactivityThresholdMs,
}
);
controller.close();
return;
}
}
} else {
// No messages received (timed out on BLOCK)
// Check for inactivity timeout
const inactiveMs = Date.now() - lastDataTime;
if (inactiveMs >= inactivityThresholdMs) {
this.logger.debug(
"[RealtimeStreams][streamResponse] Closing stream due to inactivity",
{
streamKey,
inactiveMs,
threshold: inactivityThresholdMs,
}
);
controller.close();
return;
}
}
} catch (error) {
if (signal.aborted) break;
logger.error("[RealtimeStreams][streamResponse] Error reading from Redis stream:", {
error,
});
this.logger.error(
"[RealtimeStreams][streamResponse] Error reading from Redis stream:",
{
error,
}
);
retryCount++;
if (retryCount >= maxRetries) throw error;
await new Promise((resolve) => setTimeout(resolve, 1000 * retryCount));
}
}
} catch (error) {
logger.error("[RealtimeStreams][streamResponse] Fatal error in stream processing:", {
this.logger.error("[RealtimeStreams][streamResponse] Fatal error in stream processing:", {
error,
});
controller.error(error);
@@ -95,12 +203,63 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
await cleanup();
},
})
.pipeThrough(new LineTransformStream())
.pipeThrough(
new TransformStream({
// Transform 1: Buffer partial lines across Redis entries
(() => {
let buffer = "";
let lastRedisId = "0";
return new TransformStream<StreamChunk, StreamChunk & { line: string }>({
transform(chunk, controller) {
if (chunk.type === "ping") {
controller.enqueue(chunk as any);
} else if (chunk.type === "data" || chunk.type === "legacy-data") {
// Buffer partial lines: accumulate until we see newlines
buffer += chunk.data;
// Split on newlines
const lines = buffer.split("\n");
// The last element might be incomplete, hold it back in buffer
buffer = lines.pop() || "";
// Emit complete lines with the Redis ID of the chunk that completed them
for (const line of lines) {
if (line.trim().length > 0) {
controller.enqueue({
...chunk,
line,
});
}
}
// Update last Redis ID for next iteration
lastRedisId = chunk.redisId;
}
},
flush(controller) {
// On stream end, emit any leftover buffered text
if (buffer.trim().length > 0) {
controller.enqueue({
type: "data",
redisId: lastRedisId,
data: "",
line: buffer.trim(),
});
}
},
});
})()
)
.pipeThrough(
// Transform 2: Format as SSE
new TransformStream<StreamChunk & { line?: string }, string>({
transform(chunk, controller) {
for (const line of chunk) {
controller.enqueue(`data: ${line}\n\n`);
if (chunk.type === "ping") {
controller.enqueue(`: ping\n\n`);
} else if ((chunk.type === "data" || chunk.type === "legacy-data") && chunk.line) {
// Use Redis stream ID as SSE event ID
controller.enqueue(`id: ${chunk.redisId}\ndata: ${chunk.line}\n\n`);
}
},
})
@@ -127,16 +286,23 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
async ingestData(
stream: ReadableStream<Uint8Array>,
runId: string,
streamId: string
streamId: string,
clientId: string,
resumeFromChunk?: number
): Promise<Response> {
const redis = new Redis(this.options.redis ?? {});
const streamKey = `stream:${runId}:${streamId}`;
const startChunk = resumeFromChunk ?? 0;
// Start counting from the resume point, not from 0
let currentChunkIndex = startChunk;
const self = this;
async function cleanup() {
try {
await redis.quit();
} catch (error) {
logger.error("[RedisRealtimeStreams][ingestData] Error in cleanup:", { error });
self.logger.error("[RedisRealtimeStreams][ingestData] Error in cleanup:", { error });
}
}
@@ -151,9 +317,13 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
break;
}
logger.debug("[RedisRealtimeStreams][ingestData] Reading data", {
// Write each chunk with its index and clientId
this.logger.debug("[RedisRealtimeStreams][ingestData] Writing chunk", {
streamKey,
runId,
clientId,
chunkIndex: currentChunkIndex,
resumeFromChunk: startChunk,
value,
});
@@ -163,41 +333,137 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
"~",
String(env.REALTIME_STREAM_MAX_LENGTH),
"*",
"clientId",
clientId,
"chunkIndex",
currentChunkIndex.toString(),
"data",
value
);
currentChunkIndex++;
}
// Send the END_SENTINEL and set TTL with a pipeline.
const pipeline = redis.pipeline();
pipeline.xadd(
streamKey,
"MAXLEN",
"~",
String(env.REALTIME_STREAM_MAX_LENGTH),
"*",
"data",
END_SENTINEL
);
pipeline.expire(streamKey, env.REALTIME_STREAM_TTL);
await pipeline.exec();
// Set TTL for cleanup when stream is done
await redis.expire(streamKey, env.REALTIME_STREAM_TTL);
return new Response(null, { status: 200 });
} catch (error) {
if (error instanceof Error) {
if ("code" in error && error.code === "ECONNRESET") {
logger.info("[RealtimeStreams][ingestData] Connection reset during ingestData:", {
this.logger.info("[RealtimeStreams][ingestData] Connection reset during ingestData:", {
error,
});
return new Response(null, { status: 500 });
}
}
logger.error("[RealtimeStreams][ingestData] Error in ingestData:", { error });
this.logger.error("[RealtimeStreams][ingestData] Error in ingestData:", { error });
return new Response(null, { status: 500 });
} finally {
await cleanup();
}
}
async appendPart(part: string, partId: string, runId: string, streamId: string): Promise<void> {
const redis = new Redis(this.options.redis ?? {});
const streamKey = `stream:${runId}:${streamId}`;
await redis.xadd(
streamKey,
"MAXLEN",
"~",
String(env.REALTIME_STREAM_MAX_LENGTH),
"*",
"clientId",
"",
"chunkIndex",
"0",
"data",
JSON.stringify(part) + "\n"
);
// Set TTL for cleanup when stream is done
await redis.expire(streamKey, env.REALTIME_STREAM_TTL);
await redis.quit();
}
async getLastChunkIndex(runId: string, streamId: string, clientId: string): Promise<number> {
const redis = new Redis(this.options.redis ?? {});
const streamKey = `stream:${runId}:${streamId}`;
try {
// Paginate through the stream from newest to oldest until we find this client's last chunk
const batchSize = 100;
let lastId = "+"; // Start from newest
while (true) {
const entries = await redis.xrevrange(streamKey, lastId, "-", "COUNT", batchSize);
if (!entries || entries.length === 0) {
// Reached the beginning of the stream, no chunks from this client
this.logger.debug(
"[RedisRealtimeStreams][getLastChunkIndex] No chunks found for client",
{
streamKey,
clientId,
}
);
return -1;
}
// Search through this batch for the client's last chunk
for (const [id, fields] of entries) {
let entryClientId: string | null = null;
let chunkIndex: number | null = null;
let data: string | null = null;
for (let i = 0; i < fields.length; i += 2) {
if (fields[i] === "clientId") {
entryClientId = fields[i + 1];
}
if (fields[i] === "chunkIndex") {
chunkIndex = parseInt(fields[i + 1], 10);
}
if (fields[i] === "data") {
data = fields[i + 1];
}
}
// Skip legacy END_SENTINEL entries (backward compatibility)
if (data === END_SENTINEL) {
continue;
}
// Check if this entry is from our client and has a chunkIndex
if (entryClientId === clientId && chunkIndex !== null) {
this.logger.debug("[RedisRealtimeStreams][getLastChunkIndex] Found last chunk", {
streamKey,
clientId,
chunkIndex,
});
return chunkIndex;
}
}
// Move to next batch (older entries)
// Use the ID of the last entry in this batch as the new cursor
lastId = `(${entries[entries.length - 1][0]}`; // Exclusive range with (
}
} catch (error) {
this.logger.error("[RedisRealtimeStreams][getLastChunkIndex] Error getting last chunk:", {
error,
streamKey,
clientId,
});
// Return -1 to indicate we don't know what the server has
return -1;
} finally {
await redis.quit().catch((err) => {
this.logger.error("[RedisRealtimeStreams][getLastChunkIndex] Error in cleanup:", { err });
});
}
}
}
@@ -1,263 +0,0 @@
import { AuthenticatedEnvironment } from "../apiAuth.server";
import { logger } from "../logger.server";
import { signalsEmitter } from "../signals.server";
import { StreamIngestor, StreamResponder } from "./types";
import { LineTransformStream } from "./utils.server";
import { v1RealtimeStreams } from "./v1StreamsGlobal.server";
import { singleton } from "~/utils/singleton";
export type RelayRealtimeStreamsOptions = {
ttl: number;
cleanupInterval: number;
fallbackIngestor: StreamIngestor;
fallbackResponder: StreamResponder;
waitForBufferTimeout?: number; // Time to wait for buffer in ms (default: 500ms)
waitForBufferInterval?: number; // Polling interval in ms (default: 50ms)
};
interface RelayedStreamRecord {
stream: ReadableStream<Uint8Array>;
createdAt: number;
lastAccessed: number;
locked: boolean;
finalized: boolean;
}
export class RelayRealtimeStreams implements StreamIngestor, StreamResponder {
private _buffers: Map<string, RelayedStreamRecord> = new Map();
private cleanupInterval: NodeJS.Timeout;
private waitForBufferTimeout: number;
private waitForBufferInterval: number;
constructor(private options: RelayRealtimeStreamsOptions) {
this.waitForBufferTimeout = options.waitForBufferTimeout ?? 1200;
this.waitForBufferInterval = options.waitForBufferInterval ?? 50;
// Periodic cleanup
this.cleanupInterval = setInterval(() => {
this.cleanup();
}, this.options.cleanupInterval).unref();
}
async streamResponse(
request: Request,
runId: string,
streamId: string,
environment: AuthenticatedEnvironment,
signal: AbortSignal
): Promise<Response> {
let record = this._buffers.get(`${runId}:${streamId}`);
if (!record) {
logger.debug(
"[RelayRealtimeStreams][streamResponse] No ephemeral record found, waiting to see if one becomes available",
{
streamId,
runId,
}
);
record = await this.waitForBuffer(`${runId}:${streamId}`);
if (!record) {
logger.debug(
"[RelayRealtimeStreams][streamResponse] No ephemeral record found, using fallback",
{
streamId,
runId,
}
);
// No ephemeral record, use fallback
return this.options.fallbackResponder.streamResponse(
request,
runId,
streamId,
environment,
signal
);
}
}
// Only 1 reader of the stream can use the relayed stream, the rest should use the fallback
if (record.locked) {
logger.debug("[RelayRealtimeStreams][streamResponse] Stream already locked, using fallback", {
streamId,
runId,
});
return this.options.fallbackResponder.streamResponse(
request,
runId,
streamId,
environment,
signal
);
}
record.locked = true;
record.lastAccessed = Date.now();
logger.debug("[RelayRealtimeStreams][streamResponse] Streaming from ephemeral record", {
streamId,
runId,
});
// Create a streaming response from the buffered data
const stream = record.stream
.pipeThrough(new TextDecoderStream())
.pipeThrough(new LineTransformStream())
.pipeThrough(
new TransformStream({
transform(chunk, controller) {
for (const line of chunk) {
controller.enqueue(`data: ${line}\n\n`);
}
},
})
)
.pipeThrough(new TextEncoderStream());
// Once we start streaming, consider deleting the buffer when done.
// For a simple approach, we can rely on finalized and no more reads.
// Or we can let TTL cleanup handle it if multiple readers might come in.
return new Response(stream, {
headers: {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",
"x-trigger-relay-realtime-streams": "true",
},
});
}
async ingestData(
stream: ReadableStream<Uint8Array>,
runId: string,
streamId: string
): Promise<Response> {
const [localStream, fallbackStream] = stream.tee();
logger.debug("[RelayRealtimeStreams][ingestData] Ingesting data", { runId, streamId });
// Handle local buffering asynchronously and catch errors
this.handleLocalIngestion(localStream, runId, streamId).catch((err) => {
logger.error("[RelayRealtimeStreams][ingestData] Error in local ingestion:", { err });
});
// Forward to the fallback ingestor asynchronously and catch errors
return this.options.fallbackIngestor.ingestData(fallbackStream, runId, streamId);
}
/**
* Handles local buffering of the stream data.
* @param stream The readable stream to buffer.
* @param streamId The unique identifier for the stream.
*/
private async handleLocalIngestion(
stream: ReadableStream<Uint8Array>,
runId: string,
streamId: string
) {
this.createOrUpdateRelayedStream(`${runId}:${streamId}`, stream);
}
/**
* Retrieves an existing buffer or creates a new one for the given streamId.
* @param streamId The unique identifier for the stream.
*/
private createOrUpdateRelayedStream(
bufferKey: string,
stream: ReadableStream<Uint8Array>
): RelayedStreamRecord {
let record = this._buffers.get(bufferKey);
if (!record) {
record = {
stream,
createdAt: Date.now(),
lastAccessed: Date.now(),
finalized: false,
locked: false,
};
this._buffers.set(bufferKey, record);
} else {
record.lastAccessed = Date.now();
}
return record;
}
private cleanup() {
const now = Date.now();
logger.debug("[RelayRealtimeStreams][cleanup] Cleaning up old buffers", {
bufferCount: this._buffers.size,
});
for (const [key, record] of this._buffers.entries()) {
// If last accessed is older than ttl, clean up
if (now - record.lastAccessed > this.options.ttl) {
this.deleteBuffer(key);
}
}
logger.debug("[RelayRealtimeStreams][cleanup] Cleaned up old buffers", {
bufferCount: this._buffers.size,
});
}
private deleteBuffer(bufferKey: string) {
this._buffers.delete(bufferKey);
}
/**
* Waits for a buffer to be created within a specified timeout.
* @param streamId The unique identifier for the stream.
* @returns A promise that resolves to true if the buffer was created, false otherwise.
*/
private async waitForBuffer(bufferKey: string): Promise<RelayedStreamRecord | undefined> {
const timeout = this.waitForBufferTimeout;
const interval = this.waitForBufferInterval;
const maxAttempts = Math.ceil(timeout / interval);
let attempts = 0;
return new Promise<RelayedStreamRecord | undefined>((resolve) => {
const checkBuffer = () => {
attempts++;
if (this._buffers.has(bufferKey)) {
resolve(this._buffers.get(bufferKey));
return;
}
if (attempts >= maxAttempts) {
resolve(undefined);
return;
}
setTimeout(checkBuffer, interval);
};
checkBuffer();
});
}
// Don't forget to clear interval on shutdown if needed
close() {
clearInterval(this.cleanupInterval);
}
}
function initializeRelayRealtimeStreams() {
const service = new RelayRealtimeStreams({
ttl: 1000 * 60 * 5, // 5 minutes
cleanupInterval: 1000 * 60, // 1 minute
fallbackIngestor: v1RealtimeStreams,
fallbackResponder: v1RealtimeStreams,
});
signalsEmitter.on("SIGTERM", service.close.bind(service));
signalsEmitter.on("SIGINT", service.close.bind(service));
return service;
}
export const relayRealtimeStreams = singleton(
"relayRealtimeStreams",
initializeRelayRealtimeStreams
);
@@ -0,0 +1,265 @@
// app/realtime/S2RealtimeStreams.ts
import type { UnkeyCache } from "@internal/cache";
import { StreamIngestor, StreamResponder, StreamResponseOptions } from "./types";
import { Logger, LogLevel } from "@trigger.dev/core/logger";
import { randomUUID } from "node:crypto";
export type S2RealtimeStreamsOptions = {
// S2
basin: string; // e.g., "my-basin"
accessToken: string; // "Bearer" token issued in S2 console
streamPrefix?: string; // defaults to ""
// Read behavior
s2WaitSeconds?: number;
flushIntervalMs?: number; // how often to flush buffered chunks (default 200ms)
maxRetries?: number; // max number of retries for failed flushes (default 10)
logger?: Logger;
logLevel?: LogLevel;
accessTokenExpirationInMs?: number;
cache?: UnkeyCache<{
accessToken: string;
}>;
};
type S2IssueAccessTokenResponse = { access_token: string };
type S2AppendInput = { records: { body: string }[] };
type S2AppendAck = {
start: { seq_num: number; timestamp: number };
end: { seq_num: number; timestamp: number };
tail: { seq_num: number; timestamp: number };
};
export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
private readonly basin: string;
private readonly baseUrl: string;
private readonly token: string;
private readonly streamPrefix: string;
private readonly s2WaitSeconds: number;
private readonly flushIntervalMs: number;
private readonly maxRetries: number;
private readonly logger: Logger;
private readonly level: LogLevel;
private readonly accessTokenExpirationInMs: number;
private readonly cache?: UnkeyCache<{
accessToken: string;
}>;
constructor(opts: S2RealtimeStreamsOptions) {
this.basin = opts.basin;
this.baseUrl = `https://${this.basin}.b.aws.s2.dev/v1`;
this.token = opts.accessToken;
this.streamPrefix = opts.streamPrefix ?? "";
this.s2WaitSeconds = opts.s2WaitSeconds ?? 60;
this.flushIntervalMs = opts.flushIntervalMs ?? 200;
this.maxRetries = opts.maxRetries ?? 10;
this.logger = opts.logger ?? new Logger("S2RealtimeStreams", opts.logLevel ?? "info");
this.level = opts.logLevel ?? "info";
this.cache = opts.cache;
this.accessTokenExpirationInMs = opts.accessTokenExpirationInMs ?? 60_000 * 60 * 24; // 1 day
}
private toStreamName(runId: string, streamId: string): string {
return `${this.streamPrefix}/runs/${runId}/${streamId}`;
}
async initializeStream(
runId: string,
streamId: string
): Promise<{ responseHeaders?: Record<string, string> }> {
const id = randomUUID();
const accessToken = await this.getS2AccessToken(id);
return {
responseHeaders: {
"X-S2-Access-Token": accessToken,
"X-S2-Stream-Name": `/runs/${runId}/${streamId}`,
"X-S2-Basin": this.basin,
"X-S2-Flush-Interval-Ms": this.flushIntervalMs.toString(),
"X-S2-Max-Retries": this.maxRetries.toString(),
},
};
}
ingestData(
stream: ReadableStream<Uint8Array>,
runId: string,
streamId: string,
clientId: string,
resumeFromChunk?: number
): Promise<Response> {
throw new Error("S2 streams are written to S2 via the client, not from the server");
}
async appendPart(part: string, partId: string, runId: string, streamId: string): Promise<void> {
const s2Stream = this.toStreamName(runId, streamId);
this.logger.debug(`S2 appending to stream`, { part, stream: s2Stream });
const result = await this.s2Append(s2Stream, {
records: [{ body: JSON.stringify({ data: part, id: partId }) }],
});
this.logger.debug(`S2 append result`, { result });
}
getLastChunkIndex(runId: string, streamId: string, clientId: string): Promise<number> {
throw new Error("S2 streams are written to S2 via the client, not from the server");
}
// ---------- Serve SSE from S2 ----------
async streamResponse(
request: Request,
runId: string,
streamId: string,
signal: AbortSignal,
options?: StreamResponseOptions
): Promise<Response> {
const s2Stream = this.toStreamName(runId, streamId);
const startSeq = this.parseLastEventId(options?.lastEventId);
this.logger.info(`S2 streaming records from stream`, { stream: s2Stream, startSeq });
// Request SSE stream from S2 and return it directly
const s2Response = await this.s2StreamRecords(s2Stream, {
seq_num: startSeq ?? 0,
clamp: true,
wait: options?.timeoutInSeconds ?? this.s2WaitSeconds, // S2 will keep the connection open and stream new records
signal, // Pass abort signal so S2 connection is cleaned up when client disconnects
});
// Return S2's SSE response directly to the client
return s2Response;
}
// ---------- Internals: S2 REST ----------
private async s2Append(stream: string, body: S2AppendInput): Promise<S2AppendAck> {
// POST /v1/streams/{stream}/records (JSON)
const res = await fetch(`${this.baseUrl}/streams/${encodeURIComponent(stream)}/records`, {
method: "POST",
headers: {
Authorization: `Bearer ${this.token}`,
"Content-Type": "application/json",
"S2-Format": "raw", // UTF-8 JSON encoding (no base64 overhead) when your data is text. :contentReference[oaicite:8]{index=8}
},
body: JSON.stringify(body),
});
if (!res.ok) {
const text = await res.text().catch(() => "");
throw new Error(`S2 append failed: ${res.status} ${res.statusText} ${text}`);
}
return (await res.json()) as S2AppendAck;
}
private async getS2AccessToken(id: string): Promise<string> {
if (!this.cache) {
return this.s2IssueAccessToken(id);
}
const result = await this.cache.accessToken.swr(this.streamPrefix, async () => {
return this.s2IssueAccessToken(id);
});
if (!result.val) {
throw new Error("Failed to get S2 access token");
}
return result.val;
}
private async s2IssueAccessToken(id: string): Promise<string> {
// POST /v1/access-tokens
const res = await fetch(`https://aws.s2.dev/v1/access-tokens`, {
method: "POST",
headers: {
Authorization: `Bearer ${this.token}`,
"Content-Type": "application/json",
},
body: JSON.stringify({
id,
scope: {
basins: {
exact: this.basin,
},
ops: ["append", "create-stream"],
streams: {
prefix: this.streamPrefix,
},
},
expires_at: new Date(Date.now() + this.accessTokenExpirationInMs).toISOString(),
auto_prefix_streams: true,
}),
});
if (!res.ok) {
const text = await res.text().catch(() => "");
throw new Error(`S2 issue access token failed: ${res.status} ${res.statusText} ${text}`);
}
const data = (await res.json()) as S2IssueAccessTokenResponse;
return data.access_token;
}
private async s2StreamRecords(
stream: string,
opts: {
seq_num?: number;
clamp?: boolean;
wait?: number;
signal?: AbortSignal;
}
): Promise<Response> {
// GET /v1/streams/{stream}/records with Accept: text/event-stream for SSE streaming
const qs = new URLSearchParams();
if (opts.seq_num != null) qs.set("seq_num", String(opts.seq_num));
if (opts.clamp != null) qs.set("clamp", String(opts.clamp));
if (opts.wait != null) qs.set("wait", String(opts.wait));
const res = await fetch(`${this.baseUrl}/streams/${encodeURIComponent(stream)}/records?${qs}`, {
method: "GET",
headers: {
Authorization: `Bearer ${this.token}`,
Accept: "text/event-stream",
"S2-Format": "raw",
},
signal: opts.signal,
});
if (!res.ok) {
const text = await res.text().catch(() => "");
throw new Error(`S2 stream failed: ${res.status} ${res.statusText} ${text}`);
}
const headers = new Headers(res.headers);
headers.set("X-Stream-Version", "v2");
headers.set("Access-Control-Expose-Headers", "*");
return new Response(res.body, {
headers,
status: res.status,
statusText: res.statusText,
});
}
private parseLastEventId(lastEventId?: string): number | undefined {
if (!lastEventId) return undefined;
// tolerate formats like "1699999999999-5" (take leading digits)
const digits = lastEventId.split("-")[0];
const n = Number(digits);
return Number.isFinite(n) && n >= 0 ? n + 1 : undefined;
}
}
+19 -5
View File
@@ -1,21 +1,35 @@
import { AuthenticatedEnvironment } from "../apiAuth.server";
// Interface for stream ingestion
export interface StreamIngestor {
initializeStream(
runId: string,
streamId: string
): Promise<{ responseHeaders?: Record<string, string> }>;
ingestData(
stream: ReadableStream<Uint8Array>,
runId: string,
streamId: string
streamId: string,
clientId: string,
resumeFromChunk?: number
): Promise<Response>;
appendPart(part: string, partId: string, runId: string, streamId: string): Promise<void>;
getLastChunkIndex(runId: string, streamId: string, clientId: string): Promise<number>;
}
export type StreamResponseOptions = {
timeoutInSeconds?: number;
lastEventId?: string;
};
// Interface for stream response
export interface StreamResponder {
streamResponse(
request: Request,
runId: string,
streamId: string,
environment: AuthenticatedEnvironment,
signal: AbortSignal
signal: AbortSignal,
options?: StreamResponseOptions
): Promise<Response>;
}
@@ -1,6 +1,16 @@
import {
createCache,
createMemoryStore,
DefaultStatefulContext,
Namespace,
RedisCacheStore,
} from "@internal/cache";
import { env } from "~/env.server";
import { singleton } from "~/utils/singleton";
import { AuthenticatedEnvironment } from "../apiAuth.server";
import { RedisRealtimeStreams } from "./redisRealtimeStreams.server";
import { S2RealtimeStreams } from "./s2realtimeStreams.server";
import { StreamIngestor, StreamResponder } from "./types";
function initializeRedisRealtimeStreams() {
return new RedisRealtimeStreams({
@@ -13,7 +23,87 @@ function initializeRedisRealtimeStreams() {
...(env.REALTIME_STREAMS_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
keyPrefix: "tr:realtime:streams:",
},
inactivityTimeoutMs: env.REALTIME_STREAMS_INACTIVITY_TIMEOUT_MS,
});
}
export const v1RealtimeStreams = singleton("realtimeStreams", initializeRedisRealtimeStreams);
export function getRealtimeStreamInstance(
environment: AuthenticatedEnvironment,
streamVersion: string
): StreamIngestor & StreamResponder {
if (streamVersion === "v1") {
return v1RealtimeStreams;
} else {
if (env.REALTIME_STREAMS_S2_BASIN && env.REALTIME_STREAMS_S2_ACCESS_TOKEN) {
return new S2RealtimeStreams({
basin: env.REALTIME_STREAMS_S2_BASIN,
accessToken: env.REALTIME_STREAMS_S2_ACCESS_TOKEN,
streamPrefix: [
"org",
environment.organization.id,
"env",
environment.slug,
environment.id,
].join("/"),
logLevel: env.REALTIME_STREAMS_S2_LOG_LEVEL,
flushIntervalMs: env.REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS,
maxRetries: env.REALTIME_STREAMS_S2_MAX_RETRIES,
s2WaitSeconds: env.REALTIME_STREAMS_S2_WAIT_SECONDS,
accessTokenExpirationInMs: env.REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS,
cache: s2RealtimeStreamsCache,
});
}
throw new Error("Realtime streams v2 is required for this run but S2 configuration is missing");
}
}
export function determineRealtimeStreamsVersion(streamVersion?: string): "v1" | "v2" {
if (!streamVersion) {
return env.REALTIME_STREAMS_DEFAULT_VERSION;
}
if (
streamVersion === "v2" &&
env.REALTIME_STREAMS_S2_BASIN &&
env.REALTIME_STREAMS_S2_ACCESS_TOKEN
) {
return "v2";
}
return "v1";
}
const s2RealtimeStreamsCache = singleton(
"s2RealtimeStreamsCache",
initializeS2RealtimeStreamsCache
);
function initializeS2RealtimeStreamsCache() {
const ctx = new DefaultStatefulContext();
const redisCacheStore = new RedisCacheStore({
name: "s2-realtime-streams-cache",
connection: {
port: env.REALTIME_STREAMS_REDIS_PORT,
host: env.REALTIME_STREAMS_REDIS_HOST,
username: env.REALTIME_STREAMS_REDIS_USERNAME,
password: env.REALTIME_STREAMS_REDIS_PASSWORD,
enableAutoPipelining: true,
...(env.REALTIME_STREAMS_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
keyPrefix: "s2-realtime-streams-cache:",
},
useModernCacheKeyBuilder: true,
});
const memoryStore = createMemoryStore(5000, 0.001);
return createCache({
accessToken: new Namespace<string>(ctx, {
stores: [memoryStore, redisCacheStore],
fresh: Math.floor(env.REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS / 2),
stale: Math.floor(env.REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS / 2 + 60_000),
}),
});
}
@@ -43,6 +43,7 @@ const DEFAULT_ELECTRIC_COLUMNS = [
"outputType",
"runTags",
"error",
"realtimeStreams",
];
const RESERVED_COLUMNS = ["id", "taskIdentifier", "friendlyId", "status", "createdAt"];
+7 -3
View File
@@ -12,10 +12,14 @@ type SortType = {
userName?: string | null;
};
export function sortEnvironments<T extends SortType>(environments: T[]): T[] {
export function sortEnvironments<T extends SortType>(
environments: T[],
sortOrder?: RuntimeEnvironmentType[]
): T[] {
const order = sortOrder ?? environmentSortOrder;
return environments.sort((a, b) => {
const aIndex = environmentSortOrder.indexOf(a.type);
const bIndex = environmentSortOrder.indexOf(b.type);
const aIndex = order.indexOf(a.type);
const bIndex = order.indexOf(b.type);
const difference = aIndex - bIndex;
+12
View File
@@ -40,6 +40,10 @@ export const v3SpanParamsSchema = v3RunParamsSchema.extend({
spanParam: z.string(),
});
export const v3RunStreamParamsSchema = v3RunParamsSchema.extend({
streamKey: z.string(),
});
export const v3DeploymentParams = EnvironmentParamSchema.extend({
deploymentParam: z.string(),
});
@@ -459,6 +463,14 @@ export function branchesPath(
return `${v3EnvironmentPath(organization, project, environment)}/branches`;
}
export function concurrencyPath(
organization: OrgForPath,
project: ProjectForPath,
environment: EnvironmentForPath
) {
return `${v3EnvironmentPath(organization, project, environment)}/concurrency`;
}
export function regionsPath(
organization: OrgForPath,
project: ProjectForPath,
+1
View File
@@ -4,6 +4,7 @@ export const BuildSettingsSchema = z.object({
triggerConfigFilePath: z.string().optional(),
installCommand: z.string().optional(),
preBuildCommand: z.string().optional(),
useNativeBuildServer: z.boolean().optional(),
});
export type BuildSettings = z.infer<typeof BuildSettingsSchema>;
@@ -825,12 +825,16 @@ export async function resolveVariablesForEnvironment(
runtimeEnvironment: RuntimeEnvironmentForEnvRepo,
parentEnvironment?: RuntimeEnvironmentForEnvRepo
) {
const projectSecrets = await environmentVariablesRepository.getEnvironmentVariables(
let projectSecrets = await environmentVariablesRepository.getEnvironmentVariables(
runtimeEnvironment.projectId,
runtimeEnvironment.id,
parentEnvironment?.id
);
projectSecrets = renameVariables(projectSecrets, {
OTEL_RESOURCE_ATTRIBUTES: "CUSTOM_OTEL_RESOURCE_ATTRIBUTES",
});
const overridableTriggerVariables = await resolveOverridableTriggerVariables(runtimeEnvironment);
const builtInVariables =
@@ -853,6 +857,15 @@ export async function resolveVariablesForEnvironment(
return result;
}
function renameVariables(variables: EnvironmentVariable[], renameMap: Record<string, string>) {
return variables.map((variable) => {
return {
...variable,
key: renameMap[variable.key] ?? variable.key,
};
});
}
async function resolveOverridableTriggerVariables(
runtimeEnvironment: RuntimeEnvironmentForEnvRepo
) {
@@ -1185,6 +1198,14 @@ async function resolveCommonBuiltInVariables(
String(env.TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT)
),
},
{
key: "TRIGGER_WAIT_UNTIL_TIMEOUT_MS",
value: resolveBuiltInEnvironmentVariableOverrides(
"TRIGGER_WAIT_UNTIL_TIMEOUT_MS",
runtimeEnvironment,
String(env.WAIT_UNTIL_TIMEOUT_MS)
),
},
];
}
@@ -4,6 +4,7 @@ import type {
TaskEventDetailsV1Result,
TaskEventSummaryV1Result,
TaskEventV1Input,
TaskEventV2Input,
} from "@internal/clickhouse";
import { Attributes, startSpan, trace, Tracer } from "@internal/tracing";
import { createJsonErrorObject } from "@trigger.dev/core/v3/errors";
@@ -72,6 +73,18 @@ export type ClickhouseEventRepositoryConfig = {
maximumTraceSummaryViewCount?: number;
maximumTraceDetailedSummaryViewCount?: number;
maximumLiveReloadingSetting?: number;
/**
* Maximum age in milliseconds for start_time. If start_time is older than this threshold,
* it will be clamped to the current time when creating events.
* If not provided, no clamping will be done.
*/
startTimeMaxAgeMs?: number;
/**
* The version of the ClickHouse task_events table to use.
* - "v1": Uses task_events_v1 (partitioned by start_time)
* - "v2": Uses task_events_v2 (partitioned by inserted_at to avoid "too many parts" errors)
*/
version?: "v1" | "v2";
};
/**
@@ -81,13 +94,15 @@ export type ClickhouseEventRepositoryConfig = {
export class ClickhouseEventRepository implements IEventRepository {
private _clickhouse: ClickHouse;
private _config: ClickhouseEventRepositoryConfig;
private readonly _flushScheduler: DynamicFlushScheduler<TaskEventV1Input>;
private readonly _flushScheduler: DynamicFlushScheduler<TaskEventV1Input | TaskEventV2Input>;
private _tracer: Tracer;
private _version: "v1" | "v2";
constructor(config: ClickhouseEventRepositoryConfig) {
this._clickhouse = config.clickhouse;
this._config = config;
this._tracer = config.tracer ?? trace.getTracer("clickhouseEventRepo", "0.0.1");
this._version = config.version ?? "v1";
this._flushScheduler = new DynamicFlushScheduler({
batchSize: config.batchSize ?? 1000,
@@ -99,31 +114,90 @@ export class ClickhouseEventRepository implements IEventRepository {
memoryPressureThreshold: 10000,
loadSheddingThreshold: 10000,
loadSheddingEnabled: false,
isDroppableEvent: (event: TaskEventV1Input) => {
isDroppableEvent: (event: TaskEventV1Input | TaskEventV2Input) => {
// Only drop LOG events during load shedding
return event.kind === "DEBUG_EVENT";
},
});
}
get version() {
return this._version;
}
get maximumLiveReloadingSetting() {
return this._config.maximumLiveReloadingSetting ?? 1000;
}
async #flushBatch(flushId: string, events: TaskEventV1Input[]) {
/**
* Clamps a start time (in nanoseconds) to now if it's too far in the past.
* Returns the clamped value as a bigint.
*/
#clampStartTimeNanoseconds(startTimeNs: bigint): bigint {
if (!this._config.startTimeMaxAgeMs) {
return startTimeNs;
}
const nowNs = getNowInNanoseconds();
const maxAgeNs = BigInt(this._config.startTimeMaxAgeMs) * 1_000_000n; // ms to ns
const minAllowedStartTime = nowNs - maxAgeNs;
if (startTimeNs < minAllowedStartTime) {
return nowNs;
}
return startTimeNs;
}
/**
* Clamps a start time string (nanoseconds as string) to now if it's too far in the past.
* Returns the formatted string for ClickHouse.
*/
#clampAndFormatStartTime(startTimeNsString: string): string {
const startTimeNs = BigInt(startTimeNsString);
const clampedNs = this.#clampStartTimeNanoseconds(startTimeNs);
return formatClickhouseDate64NanosecondsEpochString(clampedNs.toString());
}
/**
* Clamps a Date start time to now if it's too far in the past.
*/
#clampStartTimeDate(startTime: Date): Date {
if (!this._config.startTimeMaxAgeMs) {
return startTime;
}
const now = new Date();
const minAllowedStartTime = new Date(now.getTime() - this._config.startTimeMaxAgeMs);
if (startTime < minAllowedStartTime) {
return now;
}
return startTime;
}
async #flushBatch(flushId: string, events: (TaskEventV1Input | TaskEventV2Input)[]) {
await startSpan(this._tracer, "flushBatch", async (span) => {
span.setAttribute("flush_id", flushId);
span.setAttribute("event_count", events.length);
span.setAttribute("version", this._version);
const firstEvent = events[0];
if (firstEvent) {
logger.debug("ClickhouseEventRepository.flushBatch first event", {
event: firstEvent,
version: this._version,
});
}
const [insertError, insertResult] = await this._clickhouse.taskEvents.insert(events, {
const insertFn =
this._version === "v2"
? this._clickhouse.taskEventsV2.insert
: this._clickhouse.taskEvents.insert;
const [insertError, insertResult] = await insertFn(events, {
params: {
clickhouse_settings: this.#getClickhouseInsertSettings(),
},
@@ -136,6 +210,7 @@ export class ClickhouseEventRepository implements IEventRepository {
logger.info("ClickhouseEventRepository.flushBatch Inserted batch into clickhouse", {
events: events.length,
insertResult,
version: this._version,
});
this.#publishToRedis(events);
@@ -155,7 +230,7 @@ export class ClickhouseEventRepository implements IEventRepository {
}
}
async #publishToRedis(events: TaskEventV1Input[]) {
async #publishToRedis(events: (TaskEventV1Input | TaskEventV2Input)[]) {
if (events.length === 0) return;
await tracePubSub.publish(events.map((e) => e.trace_id));
}
@@ -176,7 +251,7 @@ export class ClickhouseEventRepository implements IEventRepository {
project_id: event.projectId,
task_identifier: event.taskSlug,
run_id: event.runId,
start_time: formatClickhouseDate64NanosecondsEpochString(event.startTime.toString()),
start_time: this.#clampAndFormatStartTime(event.startTime.toString()),
duration: formatClickhouseUnsignedIntegerString(event.duration ?? 0),
trace_id: event.traceId,
span_id: event.spanId,
@@ -184,7 +259,10 @@ export class ClickhouseEventRepository implements IEventRepository {
message: event.message,
kind: this.createEventToTaskEventV1InputKind(event),
status: this.createEventToTaskEventV1InputStatus(event),
attributes: this.createEventToTaskEventV1InputAttributes(event.properties),
attributes: this.createEventToTaskEventV1InputAttributes(
event.properties,
event.resourceProperties
),
metadata: this.createEventToTaskEventV1InputMetadata(event),
expires_at: convertDateToClickhouseDateTime(
new Date(Date.now() + 365 * 24 * 60 * 60 * 1000) // 1 year
@@ -242,7 +320,7 @@ export class ClickhouseEventRepository implements IEventRepository {
project_id: event.projectId,
task_identifier: event.taskSlug,
run_id: event.runId,
start_time: formatClickhouseDate64NanosecondsEpochString(
start_time: this.#clampAndFormatStartTime(
convertDateToNanoseconds(spanEvent.time).toString()
),
duration: "0", // Events have no duration
@@ -278,7 +356,7 @@ export class ClickhouseEventRepository implements IEventRepository {
project_id: event.projectId,
task_identifier: event.taskSlug,
run_id: event.runId,
start_time: formatClickhouseDate64NanosecondsEpochString(
start_time: this.#clampAndFormatStartTime(
convertDateToNanoseconds(spanEvent.time).toString()
),
duration: "0", // Events have no duration
@@ -308,7 +386,7 @@ export class ClickhouseEventRepository implements IEventRepository {
project_id: event.projectId,
task_identifier: event.taskSlug,
run_id: event.runId,
start_time: formatClickhouseDate64NanosecondsEpochString(
start_time: this.#clampAndFormatStartTime(
convertDateToNanoseconds(spanEvent.time).toString()
),
duration: "0", // Events have no duration
@@ -342,7 +420,7 @@ export class ClickhouseEventRepository implements IEventRepository {
project_id: event.projectId,
task_identifier: event.taskSlug,
run_id: event.runId,
start_time: formatClickhouseDate64NanosecondsEpochString(
start_time: this.#clampAndFormatStartTime(
convertDateToNanoseconds(spanEvent.time).toString()
),
duration: "0", // Events have no duration
@@ -392,7 +470,24 @@ export class ClickhouseEventRepository implements IEventRepository {
return "OK";
}
private createEventToTaskEventV1InputAttributes(attributes: Attributes): Record<string, unknown> {
private createEventToTaskEventV1InputAttributes(
attributes: Attributes,
resourceAttributes?: Attributes
): Record<string, unknown> {
if (!attributes && !resourceAttributes) {
return {};
}
return {
...this.createAttributesToInputAttributes(attributes),
...this.createAttributesToInputAttributes(resourceAttributes, "$resource"),
};
}
private createAttributesToInputAttributes(
attributes: Attributes | undefined,
key?: string
): Record<string, unknown> {
if (!attributes) {
return {};
}
@@ -406,6 +501,12 @@ export class ClickhouseEventRepository implements IEventRepository {
const unflattenedAttributes = unflattenAttributes(publicAttributes);
if (unflattenedAttributes && typeof unflattenedAttributes === "object") {
if (key) {
return {
[key]: unflattenedAttributes,
};
}
return {
...unflattenedAttributes,
};
@@ -424,19 +525,24 @@ export class ClickhouseEventRepository implements IEventRepository {
private extractEntityFromAttributes(
attributes: Attributes
): { entityType: string; entityId?: string } | undefined {
): { entityType: string; entityId?: string; entityMetadata?: string } | undefined {
if (!attributes || typeof attributes !== "object") {
return undefined;
}
const entityType = attributes[SemanticInternalAttributes.ENTITY_TYPE];
const entityId = attributes[SemanticInternalAttributes.ENTITY_ID];
const entityMetadata = attributes[SemanticInternalAttributes.ENTITY_METADATA];
if (typeof entityType !== "string") {
return undefined;
}
return { entityType, entityId: entityId as string | undefined };
return {
entityType,
entityId: entityId as string | undefined,
entityMetadata: entityMetadata as string | undefined,
};
}
private addToBatch(events: TaskEventV1Input[] | TaskEventV1Input) {
@@ -482,7 +588,7 @@ export class ClickhouseEventRepository implements IEventRepository {
project_id: options.environment.projectId,
task_identifier: options.taskSlug,
run_id: options.attributes.runId,
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
start_time: this.#clampAndFormatStartTime(startTime.toString()),
duration: formatClickhouseUnsignedIntegerString(duration),
trace_id: traceId,
span_id: spanId,
@@ -583,7 +689,7 @@ export class ClickhouseEventRepository implements IEventRepository {
project_id: options.environment.projectId,
task_identifier: options.taskSlug,
run_id: options.attributes.runId,
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
start_time: this.#clampAndFormatStartTime(startTime.toString()),
duration: formatClickhouseUnsignedIntegerString(options.incomplete ? 0 : duration),
trace_id: traceId,
span_id: spanId,
@@ -617,7 +723,7 @@ export class ClickhouseEventRepository implements IEventRepository {
project_id: options.environment.projectId,
task_identifier: options.taskSlug,
run_id: options.attributes.runId,
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
start_time: this.#clampAndFormatStartTime(startTime.toString()),
duration: formatClickhouseUnsignedIntegerString(options.incomplete ? 0 : duration),
trace_id: traceId,
span_id: spanId,
@@ -655,7 +761,8 @@ export class ClickhouseEventRepository implements IEventRepository {
return;
}
const startTime = convertDateToNanoseconds(run.createdAt);
const clampedCreatedAt = this.#clampStartTimeDate(run.createdAt);
const startTime = convertDateToNanoseconds(clampedCreatedAt);
const expiresAt = convertDateToClickhouseDateTime(
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
);
@@ -705,7 +812,8 @@ export class ClickhouseEventRepository implements IEventRepository {
return;
}
const startTime = convertDateToNanoseconds(spanCreatedAt);
const clampedSpanCreatedAt = this.#clampStartTimeDate(spanCreatedAt);
const startTime = convertDateToNanoseconds(clampedSpanCreatedAt);
const expiresAt = convertDateToClickhouseDateTime(
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
);
@@ -747,7 +855,8 @@ export class ClickhouseEventRepository implements IEventRepository {
return;
}
const startTime = convertDateToNanoseconds(run.createdAt);
const clampedCreatedAt = this.#clampStartTimeDate(run.createdAt);
const startTime = convertDateToNanoseconds(clampedCreatedAt);
const expiresAt = convertDateToClickhouseDateTime(
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
);
@@ -795,7 +904,8 @@ export class ClickhouseEventRepository implements IEventRepository {
return;
}
const startTime = convertDateToNanoseconds(run.createdAt);
const clampedCreatedAt = this.#clampStartTimeDate(run.createdAt);
const startTime = convertDateToNanoseconds(clampedCreatedAt);
const expiresAt = convertDateToClickhouseDateTime(
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
);
@@ -843,7 +953,8 @@ export class ClickhouseEventRepository implements IEventRepository {
return;
}
const startTime = convertDateToNanoseconds(endTime ?? new Date());
const clampedEndTime = this.#clampStartTimeDate(endTime ?? new Date());
const startTime = convertDateToNanoseconds(clampedEndTime);
const expiresAt = convertDateToClickhouseDateTime(
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
);
@@ -887,7 +998,8 @@ export class ClickhouseEventRepository implements IEventRepository {
return;
}
const startTime = convertDateToNanoseconds(run.createdAt);
const clampedCreatedAt = this.#clampStartTimeDate(run.createdAt);
const startTime = convertDateToNanoseconds(clampedCreatedAt);
const expiresAt = convertDateToClickhouseDateTime(
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
);
@@ -927,9 +1039,15 @@ export class ClickhouseEventRepository implements IEventRepository {
endCreatedAt?: Date,
options?: { includeDebugLogs?: boolean }
): Promise<TraceSummary | undefined> {
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 1000);
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 60_000);
const endCreatedAtWithBuffer = endCreatedAt
? new Date(endCreatedAt.getTime() + 60_000)
: undefined;
const queryBuilder = this._clickhouse.taskEvents.traceSummaryQueryBuilder();
const queryBuilder =
this._version === "v2"
? this._clickhouse.taskEventsV2.traceSummaryQueryBuilder()
: this._clickhouse.taskEvents.traceSummaryQueryBuilder();
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("trace_id = {traceId: String}", { traceId });
@@ -937,12 +1055,20 @@ export class ClickhouseEventRepository implements IEventRepository {
startCreatedAt: convertDateToNanoseconds(startCreatedAtWithBuffer).toString(),
});
if (endCreatedAt) {
if (endCreatedAtWithBuffer) {
queryBuilder.where("start_time <= {endCreatedAt: String}", {
endCreatedAt: convertDateToNanoseconds(endCreatedAt).toString(),
endCreatedAt: convertDateToNanoseconds(endCreatedAtWithBuffer).toString(),
});
}
// For v2, add inserted_at filtering for partition pruning
if (this._version === "v2") {
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
});
// No upper bound on inserted_at - we want all events inserted up to now
}
if (options?.includeDebugLogs === false) {
queryBuilder.where("kind != {kind: String}", { kind: "DEBUG_EVENT" });
}
@@ -1027,7 +1153,10 @@ export class ClickhouseEventRepository implements IEventRepository {
): Promise<SpanDetail | undefined> {
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 1000);
const queryBuilder = this._clickhouse.taskEvents.spanDetailsQueryBuilder();
const queryBuilder =
this._version === "v2"
? this._clickhouse.taskEventsV2.spanDetailsQueryBuilder()
: this._clickhouse.taskEvents.spanDetailsQueryBuilder();
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("trace_id = {traceId: String}", { traceId });
@@ -1042,6 +1171,13 @@ export class ClickhouseEventRepository implements IEventRepository {
});
}
// For v2, add inserted_at filtering for partition pruning
if (this._version === "v2") {
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
});
}
queryBuilder.orderBy("start_time ASC");
const [queryError, records] = await queryBuilder.execute();
@@ -1082,8 +1218,20 @@ export class ClickhouseEventRepository implements IEventRepository {
}
let span: SpanDetail | undefined;
let earliestStartTime: Date | undefined;
for (const record of records) {
const recordStartTime = convertClickhouseDateTime64ToJsDate(record.start_time);
// Track the earliest start time across all records
if (
record.kind !== "ANCESTOR_OVERRIDE" &&
record.kind !== "SPAN_EVENT" &&
(!earliestStartTime || recordStartTime < earliestStartTime)
) {
earliestStartTime = recordStartTime;
}
if (!span) {
span = {
spanId: spanId,
@@ -1093,14 +1241,16 @@ export class ClickhouseEventRepository implements IEventRepository {
isPartial: true, // Partial by default, can only be set to false
isCancelled: false,
level: kindToLevel(record.kind),
startTime: convertClickhouseDateTime64ToJsDate(record.start_time),
startTime: recordStartTime,
duration: typeof record.duration === "number" ? record.duration : Number(record.duration),
events: [],
style: {},
properties: undefined,
resourceProperties: undefined,
entity: {
type: undefined,
id: undefined,
metadata: undefined,
},
metadata: {},
};
@@ -1118,7 +1268,7 @@ export class ClickhouseEventRepository implements IEventRepository {
// We need to add an event to the span
span.events.push({
name: record.message,
time: convertClickhouseDateTime64ToJsDate(record.start_time),
time: recordStartTime,
properties: parsedMetadata ?? {},
});
}
@@ -1140,6 +1290,12 @@ export class ClickhouseEventRepository implements IEventRepository {
span.entity = {
id: parsedMetadata.entity.entityId,
type: parsedMetadata.entity.entityType,
metadata:
"entityMetadata" in parsedMetadata.entity &&
parsedMetadata.entity.entityMetadata &&
typeof parsedMetadata.entity.entityMetadata === "string"
? parsedMetadata.entity.entityMetadata
: undefined,
};
}
@@ -1160,16 +1316,31 @@ export class ClickhouseEventRepository implements IEventRepository {
span.duration =
typeof record.duration === "number" ? record.duration : Number(record.duration);
} else {
span.startTime = convertClickhouseDateTime64ToJsDate(record.start_time);
span.message = record.message;
}
}
if (!span.properties && typeof record.attributes_text === "string") {
span.properties = this.#parseAttributes(record.attributes_text);
if (
(span.properties == null ||
(typeof span.properties === "object" && Object.keys(span.properties).length === 0)) &&
typeof record.attributes_text === "string"
) {
const parsedAttributes = this.#parseAttributes(record.attributes_text);
const resourceAttributes = parsedAttributes["$resource"];
// Remove the $resource key from the attributes
delete parsedAttributes["$resource"];
span.properties = parsedAttributes;
span.resourceProperties = resourceAttributes as Record<string, unknown> | undefined;
}
}
// Always use the earliest start time found across all records
if (span && earliestStartTime) {
span.startTime = earliestStartTime;
}
return span;
}
@@ -1311,8 +1482,20 @@ export class ClickhouseEventRepository implements IEventRepository {
}
let span: SpanSummary | undefined;
let earliestStartTime: Date | undefined;
for (const record of records) {
const recordStartTime = convertClickhouseDateTime64ToJsDate(record.start_time);
// Track the earliest start time across all records, except for ancestor overrides and span events
if (
record.kind !== "ANCESTOR_OVERRIDE" &&
record.kind !== "SPAN_EVENT" &&
(!earliestStartTime || recordStartTime < earliestStartTime)
) {
earliestStartTime = recordStartTime;
}
if (!span) {
span = {
id: spanId,
@@ -1327,7 +1510,7 @@ export class ClickhouseEventRepository implements IEventRepository {
isPartial: true, // Partial by default, can only be set to false
isCancelled: false,
isDebug: record.kind === "DEBUG_EVENT",
startTime: convertClickhouseDateTime64ToJsDate(record.start_time),
startTime: recordStartTime,
level: kindToLevel(record.kind),
events: [],
},
@@ -1354,7 +1537,7 @@ export class ClickhouseEventRepository implements IEventRepository {
// We need to add an event to the span
span.data.events.push({
name: record.message,
time: convertClickhouseDateTime64ToJsDate(record.start_time),
time: recordStartTime,
properties: parsedMetadata ?? {},
});
}
@@ -1380,12 +1563,16 @@ export class ClickhouseEventRepository implements IEventRepository {
span.data.duration =
typeof record.duration === "number" ? record.duration : Number(record.duration);
} else {
span.data.startTime = convertClickhouseDateTime64ToJsDate(record.start_time);
span.data.message = record.message;
}
}
}
// Always use the earliest start time found across all records
if (span && earliestStartTime) {
span.data.startTime = earliestStartTime;
}
return span;
}
@@ -1427,7 +1614,10 @@ export class ClickhouseEventRepository implements IEventRepository {
): Promise<TraceDetailedSummary | undefined> {
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 1000);
const queryBuilder = this._clickhouse.taskEvents.traceDetailedSummaryQueryBuilder();
const queryBuilder =
this._version === "v2"
? this._clickhouse.taskEventsV2.traceDetailedSummaryQueryBuilder()
: this._clickhouse.taskEvents.traceDetailedSummaryQueryBuilder();
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("trace_id = {traceId: String}", { traceId });
@@ -1441,6 +1631,13 @@ export class ClickhouseEventRepository implements IEventRepository {
});
}
// For v2, add inserted_at filtering for partition pruning
if (this._version === "v2") {
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
});
}
if (options?.includeDebugLogs === false) {
queryBuilder.where("kind != {kind: String}", { kind: "DEBUG_EVENT" });
}
@@ -1541,8 +1738,20 @@ export class ClickhouseEventRepository implements IEventRepository {
}
let span: SpanDetailedSummary | undefined;
let earliestStartTime: Date | undefined;
for (const record of records) {
const recordStartTime = convertClickhouseDateTime64ToJsDate(record.start_time);
// Track the earliest start time across all records
if (
record.kind !== "ANCESTOR_OVERRIDE" &&
record.kind !== "SPAN_EVENT" &&
(!earliestStartTime || recordStartTime < earliestStartTime)
) {
earliestStartTime = recordStartTime;
}
if (!span) {
span = {
id: spanId,
@@ -1556,7 +1765,7 @@ export class ClickhouseEventRepository implements IEventRepository {
isError: false,
isPartial: true, // Partial by default, can only be set to false
isCancelled: false,
startTime: convertClickhouseDateTime64ToJsDate(record.start_time),
startTime: recordStartTime,
level: kindToLevel(record.kind),
events: [],
},
@@ -1584,7 +1793,7 @@ export class ClickhouseEventRepository implements IEventRepository {
// We need to add an event to the span
span.data.events.push({
name: record.message,
time: convertClickhouseDateTime64ToJsDate(record.start_time),
time: recordStartTime,
properties: parsedMetadata ?? {},
});
}
@@ -1606,12 +1815,16 @@ export class ClickhouseEventRepository implements IEventRepository {
span.data.duration =
typeof record.duration === "number" ? record.duration : Number(record.duration);
} else {
span.data.startTime = convertClickhouseDateTime64ToJsDate(record.start_time);
span.data.message = record.message;
}
}
}
// Always use the earliest start time found across all records
if (span && earliestStartTime) {
span.data.startTime = earliestStartTime;
}
return span;
}
@@ -1625,7 +1838,10 @@ export class ClickhouseEventRepository implements IEventRepository {
): Promise<RunPreparedEvent[]> {
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 1000);
const queryBuilder = this._clickhouse.taskEvents.traceSummaryQueryBuilder();
const queryBuilder =
this._version === "v2"
? this._clickhouse.taskEventsV2.traceSummaryQueryBuilder()
: this._clickhouse.taskEvents.traceSummaryQueryBuilder();
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("trace_id = {traceId: String}", { traceId });
@@ -1640,6 +1856,13 @@ export class ClickhouseEventRepository implements IEventRepository {
});
}
// For v2, add inserted_at filtering for partition pruning
if (this._version === "v2") {
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
});
}
queryBuilder.where("kind != {kind: String}", { kind: "DEBUG_EVENT" });
queryBuilder.orderBy("start_time ASC");
@@ -8,7 +8,12 @@ export const clickhouseEventRepository = singleton(
initializeClickhouseRepository
);
function initializeClickhouseRepository() {
export const clickhouseEventRepositoryV2 = singleton(
"clickhouseEventRepositoryV2",
initializeClickhouseRepositoryV2
);
function getClickhouseClient() {
if (!env.EVENTS_CLICKHOUSE_URL) {
throw new Error("EVENTS_CLICKHOUSE_URL is not set");
}
@@ -16,12 +21,7 @@ function initializeClickhouseRepository() {
const url = new URL(env.EVENTS_CLICKHOUSE_URL);
url.searchParams.delete("secure");
const safeUrl = new URL(url.toString());
safeUrl.password = "redacted";
console.log("🗃️ Initializing Clickhouse event repository", { url: safeUrl.toString() });
const clickhouse = new ClickHouse({
return new ClickHouse({
url: url.toString(),
name: "task-events",
keepAlive: {
@@ -34,6 +34,22 @@ function initializeClickhouseRepository() {
},
maxOpenConnections: env.EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
});
}
function initializeClickhouseRepository() {
if (!env.EVENTS_CLICKHOUSE_URL) {
throw new Error("EVENTS_CLICKHOUSE_URL is not set");
}
const url = new URL(env.EVENTS_CLICKHOUSE_URL);
url.searchParams.delete("secure");
const safeUrl = new URL(url.toString());
safeUrl.password = "redacted";
console.log("🗃️ Initializing Clickhouse event repository (v1)", { url: safeUrl.toString() });
const clickhouse = getClickhouseClient();
const repository = new ClickhouseEventRepository({
clickhouse: clickhouse,
@@ -47,6 +63,41 @@ function initializeClickhouseRepository() {
waitForAsyncInsert: env.EVENTS_CLICKHOUSE_WAIT_FOR_ASYNC_INSERT === "1",
asyncInsertMaxDataSize: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE,
asyncInsertBusyTimeoutMs: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS,
startTimeMaxAgeMs: env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS,
version: "v1",
});
return repository;
}
function initializeClickhouseRepositoryV2() {
if (!env.EVENTS_CLICKHOUSE_URL) {
throw new Error("EVENTS_CLICKHOUSE_URL is not set");
}
const url = new URL(env.EVENTS_CLICKHOUSE_URL);
url.searchParams.delete("secure");
const safeUrl = new URL(url.toString());
safeUrl.password = "redacted";
console.log("🗃️ Initializing Clickhouse event repository (v2)", { url: safeUrl.toString() });
const clickhouse = getClickhouseClient();
const repository = new ClickhouseEventRepository({
clickhouse: clickhouse,
batchSize: env.EVENTS_CLICKHOUSE_BATCH_SIZE,
flushInterval: env.EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS,
maximumTraceSummaryViewCount: env.EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT,
maximumTraceDetailedSummaryViewCount:
env.EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT,
maximumLiveReloadingSetting: env.EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING,
insertStrategy: env.EVENTS_CLICKHOUSE_INSERT_STRATEGY,
waitForAsyncInsert: env.EVENTS_CLICKHOUSE_WAIT_FOR_ASYNC_INSERT === "1",
asyncInsertMaxDataSize: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE,
asyncInsertBusyTimeoutMs: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS,
version: "v2",
});
return repository;
@@ -783,6 +783,7 @@ export class EventRepository implements IEventRepository {
SemanticInternalAttributes.ENTITY_TYPE
),
id: rehydrateAttribute<string>(spanEvent.properties, SemanticInternalAttributes.ENTITY_ID),
metadata: undefined,
};
return {
@@ -53,6 +53,7 @@ export type CreateEventInput = Omit<
| "links"
> & {
properties: Attributes;
resourceProperties?: Attributes;
metadata: Attributes | undefined;
style: Attributes | undefined;
};
@@ -209,6 +210,7 @@ export type SpanDetail = {
events: SpanEvents; // Timeline events, SpanEvents component
style: TaskEventStyle; // Icons, variants, accessories (RunIcon, SpanTitle)
properties: Record<string, unknown> | string | number | boolean | null | undefined; // Displayed as JSON in span properties (CodeBlock)
resourceProperties?: Record<string, unknown> | string | number | boolean | null | undefined; // Displayed as JSON in span resource properties (CodeBlock)
// ============================================================================
// Entity & Relationships
@@ -217,6 +219,7 @@ export type SpanDetail = {
// Used for entity type switching in SpanEntity
type: string | undefined;
id: string | undefined;
metadata: string | undefined;
};
metadata: any; // Used by SpanPresenter for entity processing
@@ -1,6 +1,9 @@
import { env } from "~/env.server";
import { eventRepository } from "./eventRepository.server";
import { clickhouseEventRepository } from "./clickhouseEventRepositoryInstance.server";
import {
clickhouseEventRepository,
clickhouseEventRepositoryV2,
} from "./clickhouseEventRepositoryInstance.server";
import { IEventRepository, TraceEventOptions } from "./eventRepository.types";
import { $replica, prisma } from "~/db.server";
import { logger } from "~/services/logger.server";
@@ -10,6 +13,10 @@ import { getTaskEventStore } from "../taskEventStore.server";
export function resolveEventRepositoryForStore(store: string | undefined): IEventRepository {
const taskEventStore = store ?? env.EVENT_REPOSITORY_DEFAULT_STORE;
if (taskEventStore === "clickhouse_v2") {
return clickhouseEventRepositoryV2;
}
if (taskEventStore === "clickhouse") {
return clickhouseEventRepository;
}
@@ -22,6 +29,9 @@ export async function getEventRepository(
parentStore: string | undefined
): Promise<{ repository: IEventRepository; store: string }> {
if (typeof parentStore === "string") {
if (parentStore === "clickhouse_v2") {
return { repository: clickhouseEventRepositoryV2, store: "clickhouse_v2" };
}
if (parentStore === "clickhouse") {
return { repository: clickhouseEventRepository, store: "clickhouse" };
} else {
@@ -31,6 +41,10 @@ export async function getEventRepository(
const taskEventRepository = await resolveTaskEventRepositoryFlag(featureFlags);
if (taskEventRepository === "clickhouse_v2") {
return { repository: clickhouseEventRepositoryV2, store: "clickhouse_v2" };
}
if (taskEventRepository === "clickhouse") {
return { repository: clickhouseEventRepository, store: "clickhouse" };
}
@@ -42,6 +56,9 @@ export async function getV3EventRepository(
parentStore: string | undefined
): Promise<{ repository: IEventRepository; store: string }> {
if (typeof parentStore === "string") {
if (parentStore === "clickhouse_v2") {
return { repository: clickhouseEventRepositoryV2, store: "clickhouse_v2" };
}
if (parentStore === "clickhouse") {
return { repository: clickhouseEventRepository, store: "clickhouse" };
} else {
@@ -49,7 +66,9 @@ export async function getV3EventRepository(
}
}
if (env.EVENT_REPOSITORY_DEFAULT_STORE === "clickhouse") {
if (env.EVENT_REPOSITORY_DEFAULT_STORE === "clickhouse_v2") {
return { repository: clickhouseEventRepositoryV2, store: "clickhouse_v2" };
} else if (env.EVENT_REPOSITORY_DEFAULT_STORE === "clickhouse") {
return { repository: clickhouseEventRepository, store: "clickhouse" };
} else {
return { repository: eventRepository, store: getTaskEventStore() };
@@ -58,13 +77,17 @@ export async function getV3EventRepository(
async function resolveTaskEventRepositoryFlag(
featureFlags: Record<string, unknown> | undefined
): Promise<"clickhouse" | "postgres"> {
): Promise<"clickhouse" | "clickhouse_v2" | "postgres"> {
const flag = await flags({
key: FEATURE_FLAG.taskEventRepository,
defaultValue: env.EVENT_REPOSITORY_DEFAULT_STORE,
overrides: featureFlags,
});
if (flag === "clickhouse_v2") {
return "clickhouse_v2";
}
if (flag === "clickhouse") {
return "clickhouse";
}
@@ -75,6 +98,10 @@ async function resolveTaskEventRepositoryFlag(
const randomNumber = Math.random();
if (randomNumber < rolloutPercent) {
// Use the default store when rolling out (could be clickhouse or clickhouse_v2)
if (env.EVENT_REPOSITORY_DEFAULT_STORE === "clickhouse_v2") {
return "clickhouse_v2";
}
return "clickhouse";
}
}
@@ -100,6 +127,13 @@ export async function recordRunDebugLog(
error?: unknown;
}
> {
if (env.EVENT_REPOSITORY_DEBUG_LOGS_DISABLED) {
// drop debug events silently
return {
success: true,
};
}
return recordRunEvent(runId, message, {
...options,
attributes: {
+1 -1
View File
@@ -10,7 +10,7 @@ export const FEATURE_FLAG = {
const FeatureFlagCatalog = {
[FEATURE_FLAG.defaultWorkerInstanceGroupId]: z.string(),
[FEATURE_FLAG.runsListRepository]: z.enum(["clickhouse", "postgres"]),
[FEATURE_FLAG.taskEventRepository]: z.enum(["clickhouse", "postgres"]),
[FEATURE_FLAG.taskEventRepository]: z.enum(["clickhouse", "clickhouse_v2", "postgres"]),
};
type FeatureFlagKey = keyof typeof FeatureFlagCatalog;
@@ -6,6 +6,8 @@ import {
type Tag,
RepositoryNotFoundException,
GetAuthorizationTokenCommand,
PutLifecyclePolicyCommand,
PutImageTagMutabilityCommand,
} from "@aws-sdk/client-ecr";
import { STSClient, AssumeRoleCommand } from "@aws-sdk/client-sts";
import { tryCatch } from "@trigger.dev/core";
@@ -195,6 +197,22 @@ export function parseRegistryTags(tags: string): Tag[] {
.filter((tag): tag is Tag => tag !== null);
}
const untaggedImageExpirationPolicy = JSON.stringify({
rules: [
{
rulePriority: 1,
description: "Expire untagged images older than 3 days",
selection: {
tagStatus: "untagged",
countType: "sinceImagePushed",
countUnit: "days",
countNumber: 3,
},
action: { type: "expire" },
},
],
});
async function createEcrRepository({
repositoryName,
region,
@@ -213,7 +231,14 @@ async function createEcrRepository({
const result = await ecr.send(
new CreateRepositoryCommand({
repositoryName,
imageTagMutability: "IMMUTABLE",
imageTagMutability: "IMMUTABLE_WITH_EXCLUSION",
imageTagMutabilityExclusionFilters: [
{
// only the `cache` tag will be mutable, all other tags will be immutable
filter: "cache",
filterType: "WILDCARD",
},
],
encryptionConfiguration: {
encryptionType: "AES256",
},
@@ -227,9 +252,68 @@ async function createEcrRepository({
throw new Error(`Failed to create ECR repository: ${repositoryName}`);
}
// When the `cache` tag is mutated, the old cache images are untagged.
// This policy matches those images and expires them to avoid bloating the repository.
await ecr.send(
new PutLifecyclePolicyCommand({
repositoryName: result.repository.repositoryName,
registryId: result.repository.registryId,
lifecyclePolicyText: untaggedImageExpirationPolicy,
})
);
return result.repository;
}
async function updateEcrRepositoryCacheSettings({
repositoryName,
region,
accountId,
assumeRole,
}: {
repositoryName: string;
region: string;
accountId?: string;
assumeRole?: AssumeRoleConfig;
}): Promise<void> {
logger.debug("Updating ECR repository tag mutability to IMMUTABLE_WITH_EXCLUSION", {
repositoryName,
region,
});
const ecr = await createEcrClient({ region, assumeRole });
await ecr.send(
new PutImageTagMutabilityCommand({
repositoryName,
registryId: accountId,
imageTagMutability: "IMMUTABLE_WITH_EXCLUSION",
imageTagMutabilityExclusionFilters: [
{
// only the `cache` tag will be mutable, all other tags will be immutable
filter: "cache",
filterType: "WILDCARD",
},
],
})
);
// When the `cache` tag is mutated, the old cache images are untagged.
// This policy matches those images and expires them to avoid bloating the repository.
await ecr.send(
new PutLifecyclePolicyCommand({
repositoryName,
registryId: accountId,
lifecyclePolicyText: untaggedImageExpirationPolicy,
})
);
logger.debug("Successfully updated ECR repository to IMMUTABLE_WITH_EXCLUSION", {
repositoryName,
region,
});
}
async function getEcrRepository({
repositoryName,
region,
@@ -318,6 +402,22 @@ async function ensureEcrRepositoryExists({
if (existingRepo) {
logger.debug("ECR repository already exists", { repositoryName, region, existingRepo });
// check if the repository is missing the cache settings
if (existingRepo.imageTagMutability === "IMMUTABLE") {
const [updateError] = await tryCatch(
updateEcrRepositoryCacheSettings({ repositoryName, region, accountId, assumeRole })
);
if (updateError) {
logger.error("Failed to update ECR repository cache settings", {
repositoryName,
region,
updateError,
});
}
}
return {
repo: existingRepo,
repoCreated: false,
+1 -1
View File
@@ -29,7 +29,7 @@ export function isValidGitBranchName(branch: string): boolean {
return true;
}
export function sanitizeBranchName(ref: string): string | null {
export function sanitizeBranchName(ref: string | undefined): string | null {
if (!ref) return null;
if (ref.startsWith("refs/heads/")) return ref.substring("refs/heads/".length);
if (ref.startsWith("refs/remotes/")) return ref.substring("refs/remotes/".length);
+48 -1
View File
@@ -17,7 +17,10 @@ import {
} from "@trigger.dev/otlp-importer";
import { logger } from "~/services/logger.server";
import { ClickhouseEventRepository } from "./eventRepository/clickhouseEventRepository.server";
import { clickhouseEventRepository } from "./eventRepository/clickhouseEventRepositoryInstance.server";
import {
clickhouseEventRepository,
clickhouseEventRepositoryV2,
} from "./eventRepository/clickhouseEventRepositoryInstance.server";
import { generateSpanId } from "./eventRepository/common.server";
import { EventRepository, eventRepository } from "./eventRepository/eventRepository.server";
import type {
@@ -38,6 +41,7 @@ class OTLPExporter {
constructor(
private readonly _eventRepository: EventRepository,
private readonly _clickhouseEventRepository: ClickhouseEventRepository,
private readonly _clickhouseEventRepositoryV2: ClickhouseEventRepository,
private readonly _verbose: boolean,
private readonly _spanAttributeValueLengthLimit: number
) {
@@ -111,6 +115,10 @@ class OTLPExporter {
return this._clickhouseEventRepository;
}
if (store === "clickhouse_v2") {
return this._clickhouseEventRepositoryV2;
}
return this._eventRepository;
}
@@ -204,6 +212,24 @@ function convertLogsToCreateableEvents(
const resourceProperties = extractEventProperties(resourceAttributes);
const userDefinedResourceAttributes = truncateAttributes(
convertKeyValueItemsToMap(resourceAttributes ?? [], [], undefined, [
SemanticInternalAttributes.USAGE,
SemanticInternalAttributes.SPAN,
SemanticInternalAttributes.METADATA,
SemanticInternalAttributes.STYLE,
SemanticInternalAttributes.METRIC_EVENTS,
SemanticInternalAttributes.TRIGGER,
"process",
"sdk",
"service",
"ctx",
"cli",
"cloud",
]),
spanAttributeValueLengthLimit
);
const taskEventStore =
extractStringAttribute(resourceAttributes, [SemanticInternalAttributes.TASK_EVENT_STORE]) ??
env.EVENT_REPOSITORY_DEFAULT_STORE;
@@ -249,6 +275,7 @@ function convertLogsToCreateableEvents(
status: logLevelToEventStatus(log.severityNumber),
startTime: log.timeUnixNano,
properties,
resourceProperties: userDefinedResourceAttributes,
style: convertKeyValueItemsToMap(
pickAttributes(log.attributes ?? [], SemanticInternalAttributes.STYLE),
[]
@@ -285,6 +312,24 @@ function convertSpansToCreateableEvents(
const resourceProperties = extractEventProperties(resourceAttributes);
const userDefinedResourceAttributes = truncateAttributes(
convertKeyValueItemsToMap(resourceAttributes ?? [], [], undefined, [
SemanticInternalAttributes.USAGE,
SemanticInternalAttributes.SPAN,
SemanticInternalAttributes.METADATA,
SemanticInternalAttributes.STYLE,
SemanticInternalAttributes.METRIC_EVENTS,
SemanticInternalAttributes.TRIGGER,
"process",
"sdk",
"service",
"ctx",
"cli",
"cloud",
]),
spanAttributeValueLengthLimit
);
const taskEventStore =
extractStringAttribute(resourceAttributes, [SemanticInternalAttributes.TASK_EVENT_STORE]) ??
env.EVENT_REPOSITORY_DEFAULT_STORE;
@@ -336,6 +381,7 @@ function convertSpansToCreateableEvents(
events: spanEventsToEventEvents(span.events ?? []),
duration: span.endTimeUnixNano - span.startTimeUnixNano,
properties,
resourceProperties: userDefinedResourceAttributes,
style: convertKeyValueItemsToMap(
pickAttributes(span.attributes ?? [], SemanticInternalAttributes.STYLE),
[]
@@ -848,6 +894,7 @@ function initializeOTLPExporter() {
return new OTLPExporter(
eventRepository,
clickhouseEventRepository,
clickhouseEventRepositoryV2,
process.env.OTLP_EXPORTER_VERBOSE === "1",
process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT
? parseInt(process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, 10)
@@ -0,0 +1,97 @@
import { tryCatch } from "@trigger.dev/core";
import { ManageConcurrencyPresenter } from "~/presenters/v3/ManageConcurrencyPresenter.server";
import { BaseService } from "./baseService.server";
import { updateEnvConcurrencyLimits } from "../runQueue.server";
type Input = {
userId: string;
projectId: string;
organizationId: string;
environments: { id: string; amount: number }[];
};
type Result =
| {
success: true;
}
| {
success: false;
error: string;
};
export class AllocateConcurrencyService extends BaseService {
async call({ userId, projectId, organizationId, environments }: Input): Promise<Result> {
// fetch the current concurrency
const presenter = new ManageConcurrencyPresenter(this._prisma, this._replica);
const [error, result] = await tryCatch(
presenter.call({
userId,
projectId,
organizationId,
})
);
if (error) {
return {
success: false,
error: "Unknown error",
};
}
const previousExtra = result.environments.reduce(
(acc, e) => Math.max(0, e.maximumConcurrencyLimit - e.planConcurrencyLimit) + acc,
0
);
const requested = new Map(environments.map((e) => [e.id, e.amount]));
const newExtra = result.environments.reduce((acc, env) => {
const targetExtra = requested.has(env.id)
? Math.max(0, requested.get(env.id)!)
: Math.max(0, env.maximumConcurrencyLimit - env.planConcurrencyLimit);
return acc + targetExtra;
}, 0);
const change = newExtra - previousExtra;
const totalExtra = result.extraAllocatedConcurrency + change;
if (change > result.extraUnallocatedConcurrency) {
return {
success: false,
error: `You don't have enough unallocated concurrency available. You requested ${totalExtra} but only have ${result.extraUnallocatedConcurrency}.`,
};
}
for (const environment of environments) {
const existingEnvironment = result.environments.find((e) => e.id === environment.id);
if (!existingEnvironment) {
return {
success: false,
error: `Environment not found ${environment.id}`,
};
}
const newConcurrency = existingEnvironment.planConcurrencyLimit + environment.amount;
const updatedEnvironment = await this._prisma.runtimeEnvironment.update({
where: {
id: environment.id,
},
data: {
maximumConcurrencyLimit: newConcurrency,
},
include: {
project: true,
organization: true,
},
});
if (!updatedEnvironment.paused) {
await updateEnvConcurrencyLimits(updatedEnvironment);
}
}
return {
success: true,
};
}
}
@@ -0,0 +1,92 @@
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { BaseService } from "./baseService.server";
import { env } from "~/env.server";
import { createPresignedPost } from "@aws-sdk/s3-presigned-post";
import { S3Client } from "@aws-sdk/client-s3";
import { customAlphabet } from "nanoid";
import { errAsync, fromPromise } from "neverthrow";
const nanoid = customAlphabet("1234567890abcdefghijklmnopqrstuvwxyz", 24);
const objectStoreClient =
env.ARTIFACTS_OBJECT_STORE_ACCESS_KEY_ID &&
env.ARTIFACTS_OBJECT_STORE_SECRET_ACCESS_KEY &&
env.ARTIFACTS_OBJECT_STORE_BASE_URL
? new S3Client({
credentials: {
accessKeyId: env.ARTIFACTS_OBJECT_STORE_ACCESS_KEY_ID,
secretAccessKey: env.ARTIFACTS_OBJECT_STORE_SECRET_ACCESS_KEY,
},
region: env.ARTIFACTS_OBJECT_STORE_REGION,
endpoint: env.ARTIFACTS_OBJECT_STORE_BASE_URL,
forcePathStyle: true,
})
: new S3Client();
const artifactKeyPrefixByType = {
deployment_context: "deployments",
} as const;
const artifactBytesSizeLimitByType = {
deployment_context: 100 * 1024 * 1024, // 100MB
} as const;
export class ArtifactsService extends BaseService {
private readonly bucket = env.ARTIFACTS_OBJECT_STORE_BUCKET;
public createArtifact(
type: "deployment_context",
authenticatedEnv: AuthenticatedEnvironment,
contentLength?: number
) {
const limit = artifactBytesSizeLimitByType[type];
// this is just a validation using client-side data
// the actual limit will be enforced by S3
if (contentLength && contentLength > limit) {
return errAsync({
type: "artifact_size_exceeds_limit" as const,
contentLength,
sizeLimit: limit,
});
}
const uniqueId = nanoid();
const key = `${artifactKeyPrefixByType[type]}/${authenticatedEnv.project.externalRef}/${authenticatedEnv.slug}/${uniqueId}.tar.gz`;
return this.createPresignedPost(key, limit, contentLength).map((result) => ({
artifactKey: key,
uploadUrl: result.url,
uploadFields: result.fields,
expiresAt: result.expiresAt,
}));
}
private createPresignedPost(key: string, sizeLimit: number, contentLength?: number) {
if (!this.bucket) {
return errAsync({
type: "artifacts_bucket_not_configured" as const,
});
}
const ttlSeconds = 300; // 5 minutes
const expiresAt = new Date(Date.now() + ttlSeconds * 1000);
return fromPromise(
createPresignedPost(objectStoreClient, {
Bucket: this.bucket,
Key: key,
Conditions: [["content-length-range", 0, sizeLimit]],
Fields: {
"Content-Type": "application/gzip",
},
Expires: ttlSeconds,
}),
(error) => ({
type: "failed_to_create_presigned_post" as const,
cause: error,
})
).map((result) => ({
...result,
expiresAt,
}));
}
}
@@ -56,6 +56,7 @@ export type BatchTriggerTaskServiceOptions = {
traceContext?: Record<string, string | undefined>;
spanParentAsLink?: boolean;
oneTimeUseToken?: string;
realtimeStreamsVersion?: "v1" | "v2";
};
type RunItemData = {
@@ -851,6 +852,7 @@ export class BatchTriggerV3Service extends BaseService {
batchId: batch.friendlyId,
skipChecks: true,
runFriendlyId: task.runId,
realtimeStreamsVersion: options?.realtimeStreamsVersion,
}
);
@@ -173,6 +173,11 @@ function resetQueueConcurrencyLimit(db: PrismaClientOrTransaction, queue: TaskQu
}
function syncQueueConcurrencyToEngine(environment: AuthenticatedEnvironment, queue: TaskQueue) {
if (queue.paused) {
// Queue is paused, don't update Redis limits - keep at 0
return okAsync(queue);
}
if (typeof queue.concurrencyLimit === "number") {
return fromPromise(
updateQueueConcurrencyLimits(environment, queue.name, queue.concurrencyLimit),
@@ -361,14 +361,7 @@ async function createWorkerQueue(
const baseConcurrencyLimit =
typeof queue.concurrencyLimit === "number"
? Math.max(
Math.min(
queue.concurrencyLimit,
environment.maximumConcurrencyLimit,
environment.organization.maximumConcurrencyLimit
),
0
)
? Math.max(Math.min(queue.concurrencyLimit, environment.maximumConcurrencyLimit), 0)
: queue.concurrencyLimit;
const taskQueue = await upsertWorkerQueueRecord(
+192 -56
View File
@@ -1,13 +1,26 @@
import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { BaseService } from "./baseService.server";
import { errAsync, fromPromise, okAsync } from "neverthrow";
import { type WorkerDeployment } from "@trigger.dev/database";
import { logger, type GitMeta } from "@trigger.dev/core/v3";
import { errAsync, fromPromise, okAsync, type ResultAsync } from "neverthrow";
import { type WorkerDeployment, type Project } from "@trigger.dev/database";
import { logger, type GitMeta, type DeploymentEvent } from "@trigger.dev/core/v3";
import { TimeoutDeploymentService } from "./timeoutDeployment.server";
import { env } from "~/env.server";
import { createRemoteImageBuild } from "../remoteImageBuilder.server";
import { FINAL_DEPLOYMENT_STATUSES } from "./failDeployment.server";
import { generateRegistryCredentials } from "~/services/platform.v3.server";
import { enqueueBuild, generateRegistryCredentials } from "~/services/platform.v3.server";
import { AppendRecord, S2 } from "@s2-dev/streamstore";
import { createRedisClient } from "~/redis.server";
const S2_TOKEN_KEY_PREFIX = "s2-token:read:deployment-event-stream:project:";
const s2TokenRedis = createRedisClient("s2-token-cache", {
host: env.CACHE_REDIS_HOST,
port: env.CACHE_REDIS_PORT,
username: env.CACHE_REDIS_USERNAME,
password: env.CACHE_REDIS_PASSWORD,
tlsDisabled: env.CACHE_REDIS_TLS_DISABLED === "true",
clusterMode: env.CACHE_REDIS_CLUSTER_MODE_ENABLED === "1",
});
const s2 = env.S2_ENABLED === "1" ? new S2({ accessToken: env.S2_ACCESS_TOKEN }) : undefined;
export class DeploymentService extends BaseService {
/**
@@ -22,35 +35,11 @@ export class DeploymentService extends BaseService {
* @param friendlyId The friendly deployment ID.
* @param updates Optional deployment details to persist.
*/
public progressDeployment(
authenticatedEnv: AuthenticatedEnvironment,
friendlyId: string,
updates: Partial<Pick<WorkerDeployment, "contentHash" | "runtime"> & { git: GitMeta }>
) {
const getDeployment = () =>
fromPromise(
this._prisma.workerDeployment.findFirst({
where: {
friendlyId,
environmentId: authenticatedEnv.id,
},
select: {
status: true,
id: true,
},
}),
(error) => ({
type: "other" as const,
cause: error,
})
).andThen((deployment) => {
if (!deployment) {
return errAsync({ type: "deployment_not_found" as const });
}
return okAsync(deployment);
});
const validateDeployment = (deployment: Pick<WorkerDeployment, "id" | "status">) => {
if (deployment.status !== "PENDING" && deployment.status !== "INSTALLING") {
logger.warn(
@@ -134,7 +123,7 @@ export class DeploymentService extends BaseService {
})
);
return getDeployment()
return this.getDeployment(authenticatedEnv.projectId, friendlyId)
.andThen(validateDeployment)
.andThen((deployment) => {
if (deployment.status === "PENDING") {
@@ -160,30 +149,11 @@ export class DeploymentService extends BaseService {
friendlyId: string,
data?: Partial<Pick<WorkerDeployment, "canceledReason">>
) {
const getDeployment = () =>
fromPromise(
this._prisma.workerDeployment.findFirst({
where: {
friendlyId,
projectId: authenticatedEnv.projectId,
},
select: {
status: true,
id: true,
},
}),
(error) => ({
type: "other" as const,
cause: error,
})
).andThen((deployment) => {
if (!deployment) {
return errAsync({ type: "deployment_not_found" as const });
}
return okAsync(deployment);
});
const validateDeployment = (deployment: Pick<WorkerDeployment, "id" | "status">) => {
const validateDeployment = (
deployment: Pick<WorkerDeployment, "id" | "status" | "shortCode"> & {
environment: { project: { externalRef: string } };
}
) => {
if (FINAL_DEPLOYMENT_STATUSES.includes(deployment.status)) {
logger.warn("Attempted cancelling deployment in a final state", {
deployment,
@@ -194,7 +164,11 @@ export class DeploymentService extends BaseService {
return okAsync(deployment);
};
const cancelDeployment = (deployment: Pick<WorkerDeployment, "id">) =>
const cancelDeployment = (
deployment: Pick<WorkerDeployment, "id" | "shortCode"> & {
environment: { project: { externalRef: string } };
}
) =>
fromPromise(
this._prisma.workerDeployment.updateMany({
where: {
@@ -217,7 +191,7 @@ export class DeploymentService extends BaseService {
if (result.count === 0) {
return errAsync({ type: "deployment_cannot_be_cancelled" as const });
}
return okAsync({ id: deployment.id });
return okAsync({ deployment });
});
const deleteTimeout = (deployment: Pick<WorkerDeployment, "id">) =>
@@ -226,9 +200,25 @@ export class DeploymentService extends BaseService {
cause: error,
}));
return getDeployment()
return this.getDeployment(authenticatedEnv.projectId, friendlyId)
.andThen(validateDeployment)
.andThen(cancelDeployment)
.andThen(({ deployment }) =>
this.appendToEventLog(deployment.environment.project, deployment, [
{
type: "finalized",
data: {
result: "canceled",
message: data?.canceledReason ?? undefined,
},
},
])
.orElse((error) => {
logger.error("Failed to append event to deployment event log", { error });
return okAsync(deployment);
})
.map(() => deployment)
)
.andThen(deleteTimeout)
.map(() => undefined);
}
@@ -296,6 +286,142 @@ export class DeploymentService extends BaseService {
.andThen(generateCredentials);
}
public enqueueBuild(
authenticatedEnv: Pick<AuthenticatedEnvironment, "projectId">,
deployment: Pick<WorkerDeployment, "friendlyId">,
artifactKey: string,
options: {
skipPromotion?: boolean;
configFilePath?: string;
}
) {
return fromPromise(
enqueueBuild(authenticatedEnv.projectId, deployment.friendlyId, artifactKey, options),
(error) => ({
type: "failed_to_enqueue_build" as const,
cause: error,
})
);
}
public appendToEventLog(
project: Pick<Project, "externalRef">,
deployment: Pick<WorkerDeployment, "shortCode">,
events: DeploymentEvent[]
): ResultAsync<
undefined,
{ type: "s2_is_disabled" } | { type: "failed_to_append_to_event_log"; cause: unknown }
> {
if (env.S2_ENABLED !== "1" || !s2) {
return errAsync({ type: "s2_is_disabled" as const });
}
const basin = s2.basin(env.S2_DEPLOYMENT_LOGS_BASIN_NAME);
const stream = basin.stream(
`projects/${project.externalRef}/deployments/${deployment.shortCode}`
);
return fromPromise(
stream.append(events.map((event) => AppendRecord.make(JSON.stringify(event)))),
(error) => ({
type: "failed_to_append_to_event_log" as const,
cause: error,
})
).map(() => undefined);
}
public createEventStream(
project: Pick<Project, "externalRef">,
deployment: Pick<WorkerDeployment, "shortCode">
): ResultAsync<
{ basin: string; stream: string },
{ type: "s2_is_disabled" } | { type: "failed_to_create_event_stream"; cause: unknown }
> {
if (env.S2_ENABLED !== "1" || !s2) {
return errAsync({ type: "s2_is_disabled" as const });
}
const basin = s2.basin(env.S2_DEPLOYMENT_LOGS_BASIN_NAME);
return fromPromise(
basin.streams.create({
stream: `projects/${project.externalRef}/deployments/${deployment.shortCode}`,
}),
(error) => ({
type: "failed_to_create_event_stream" as const,
cause: error,
})
).map(({ name }) => ({
basin: basin.name,
stream: name,
}));
}
public getEventStreamAccessToken(
project: Pick<Project, "externalRef">
): ResultAsync<string, { type: "s2_is_disabled" } | { type: "other"; cause: unknown }> {
if (env.S2_ENABLED !== "1" || !s2) {
return errAsync({ type: "s2_is_disabled" as const });
}
const basinName = env.S2_DEPLOYMENT_LOGS_BASIN_NAME;
const redisKey = `${S2_TOKEN_KEY_PREFIX}${project.externalRef}`;
const getTokenFromCache = () =>
fromPromise(s2TokenRedis.get(redisKey), (error) => ({
type: "other" as const,
cause: error,
})).andThen((cachedToken) => {
if (!cachedToken) {
return errAsync({ type: "s2_token_cache_not_found" as const });
}
return okAsync(cachedToken);
});
const issueS2Token = () =>
fromPromise(
s2.accessTokens.issue({
id: `${project.externalRef}-${new Date().getTime()}`,
expires_at: new Date(Date.now() + 60 * 60 * 1000).toISOString(), // 1 hour
scope: {
ops: ["read"],
basins: {
exact: basinName,
},
streams: {
prefix: `projects/${project.externalRef}/deployments/`,
},
},
}),
(error) => ({
type: "other" as const,
cause: error,
})
).map(({ access_token }) => access_token);
const cacheToken = (token: string) =>
fromPromise(
s2TokenRedis.setex(
redisKey,
59 * 60, // slightly shorter than the token validity period
token
),
(error) => ({
type: "other" as const,
cause: error,
})
);
return getTokenFromCache().orElse(() =>
issueS2Token().andThen((token) =>
cacheToken(token)
.map(() => token)
.orElse((error) => {
logger.error("Failed to cache S2 token", { error });
return okAsync(token); // ignore the cache error
})
)
);
}
private getDeployment(projectId: string, friendlyId: string) {
return fromPromise(
this._prisma.workerDeployment.findFirst({
@@ -307,6 +433,16 @@ export class DeploymentService extends BaseService {
status: true,
id: true,
imageReference: true,
shortCode: true,
environment: {
include: {
project: {
select: {
externalRef: true,
},
},
},
},
},
}),
(error) => ({
@@ -1,7 +1,8 @@
import { PerformDeploymentAlertsService } from "./alerts/performDeploymentAlerts.server";
import { BaseService } from "./baseService.server";
import { logger } from "~/services/logger.server";
import { WorkerDeploymentStatus } from "@trigger.dev/database";
import { type WorkerDeploymentStatus } from "@trigger.dev/database";
import { DeploymentService } from "./deployment.server";
const FINAL_DEPLOYMENT_STATUSES: WorkerDeploymentStatus[] = [
"CANCELED",
@@ -31,6 +32,13 @@ export class DeploymentIndexFailed extends BaseService {
: {
id: maybeFriendlyId,
},
include: {
environment: {
include: {
project: true,
},
},
},
});
if (!deployment) {
@@ -66,6 +74,21 @@ export class DeploymentIndexFailed extends BaseService {
},
});
const deploymentService = new DeploymentService();
await deploymentService
.appendToEventLog(deployment.environment.project, failedDeployment, [
{
type: "finalized",
data: {
result: "failed",
message: error.message,
},
},
])
.orTee((error) => {
logger.error("Failed to append failed deployment event to event log", { error });
});
await PerformDeploymentAlertsService.enqueue(failedDeployment.id);
return failedDeployment;

Some files were not shown because too many files have changed in this diff Show More