Draft
Conversation
|
Contributor
WalkthroughReplaces module-level ClickHouse singletons with a new ClickhouseFactory that produces per-organization ClickHouse clients and event repositories. Adds OrganizationDataStore DB model and migration, ClickHouse secret schemas, an OrganizationDataStoresRegistry with a singleton instance that periodically reloads, admin UI/routes to manage data stores, and test helpers/fixtures. Refactors event repository interfaces and ClickHouse-backed implementations (metrics batching and sync behavior), updates many presenters/routes/services/tests to obtain organization-scoped clients, and introduces new clickhouse-related modules and integration tests. Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes 🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 16
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx (1)
137-159:⚠️ Potential issue | 🟠 MajorDon't let ClickHouse resolution bypass the deferred error path.
getClickhouseForOrganization()is now awaited beforetypeddefer(). If org datastore lookup or client initialization fails, the loader throws beforeTypedAwaitcan render itserrorElement, and the page shell won't stream until that lookup finishes. Fold the factory lookup intolistPromiseso it shares the existing deferred failure handling.♻️ Suggested shape
- const logsClickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "logs"); - const presenter = new LogsListPresenter($replica, logsClickhouse); - - const listPromise = presenter - .call(project.organizationId, environment.id, { - userId, - projectId: project.id, - tasks: tasks.length > 0 ? tasks : undefined, - runId, - search, - levels, - period, - from, - to, - defaultPeriod: "1h", - retentionLimitDays - }) + const listPromise = clickhouseFactory + .getClickhouseForOrganization(project.organizationId, "logs") + .then((logsClickhouse) => + new LogsListPresenter($replica, logsClickhouse).call(project.organizationId, environment.id, { + userId, + projectId: project.id, + tasks: tasks.length > 0 ? tasks : undefined, + runId, + search, + levels, + period, + from, + to, + defaultPeriod: "1h", + retentionLimitDays, + }) + ) .catch((error) => { if (error instanceof ServiceValidationError) { return { error: error.message }; } throw error; });🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/_app.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx around lines 137 - 159, The ClickHouse client lookup is awaited up front which can throw before the deferred error handling; instead, defer the factory call so failures are handled by the same promise chain: build listPromise by calling clickhouseFactory.getClickhouseForOrganization(...) inside the async chain used to create the LogsListPresenter and call presenter.call (keep references to clickhouseFactory.getClickhouseForOrganization, LogsListPresenter, presenter.call, and listPromise) so any errors from the org datastore/client init are caught by the existing .catch that handles ServiceValidationError and rethrows other errors, rather than throwing before TypedAwait renders its errorElement.apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts (1)
13-17:⚠️ Potential issue | 🟠 MajorDon't silently drop the connection-tuning fields.
Lines 13-17 still require
keepAliveEnabled,keepAliveIdleSocketTtl, andmaxOpenConnections, but Lines 82-119 no longer pass them anywhere. This endpoint will accept those values and return success even though they no longer affect replication behavior. Either wire them into the new factory path or remove them from the request schema before merge.One fix if these settings are no longer supported
const CreateRunReplicationServiceParams = z.object({ name: z.string(), - keepAliveEnabled: z.boolean(), - keepAliveIdleSocketTtl: z.number(), - maxOpenConnections: z.number(), maxFlushConcurrency: z.number(), flushIntervalMs: z.number(), flushBatchSize: z.number(),Also applies to: 82-93, 95-119
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts` around lines 13 - 17, The request schema CreateRunReplicationServiceParams still requires keepAliveEnabled, keepAliveIdleSocketTtl, and maxOpenConnections but the handler that constructs the replication service (the new factory path used in the run replication creation flow) no longer consumes these fields, so they are effectively ignored; either (A) thread these three fields into the replication service factory/constructor/creation call (the function that builds the replication client/service used in the handler) so the values are applied, making sure to map the names exactly, or (B) remove these fields from CreateRunReplicationServiceParams (and any validation/usage sites) so the API no longer accepts unused settings; locate the schema CreateRunReplicationServiceParams and the replication service factory/constructor function referenced in the create run replication handler to implement one of these fixes.
🧹 Nitpick comments (4)
apps/webapp/test/runsReplicationService.part2.test.ts (1)
26-27: Cover org-to-ClickHouse routing with a multi-store case.All of these setups hand the service a factory backed by a single
ClickHouseinstance. That means the suite still passes ifRunsReplicationServicesends the wrong organization ID into the factory, which is the new behavior this change depends on. Add one case that maps two orgs to different clients and asserts each org's rows land in the correct backend.Also applies to: 46-47, 155-156, 270-271, 391-392, 525-526, 632-633, 801-802, 926-927, 1139-1140
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/test/runsReplicationService.part2.test.ts` around lines 26 - 27, The tests currently instantiate RunsReplicationService with a TestReplicationClickhouseFactory backed by a single ClickHouse instance, so they don't catch org-to-ClickHouse routing bugs; update the affected test cases (instances where new RunsReplicationService is constructed with new TestReplicationClickhouseFactory(clickhouse)) to create a multi-store scenario: build two distinct Test ClickHouse clients (e.g., clickhouseA and clickhouseB), construct a TestReplicationClickhouseFactory that maps orgA -> clickhouseA and orgB -> clickhouseB, pass that factory into the RunsReplicationService, then insert rows for both orgs and assert each org's rows appear in the corresponding backend client (use the same helper assertions already used elsewhere in the tests). Ensure you change all occurrences referenced (the other construction sites of TestReplicationClickhouseFactory) so the suite validates org routing across multiple stores.apps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.ts (1)
94-130: Validate and typeconfigat this service boundary.These methods accept
config: anyand persist it directly. Any caller that skips route-level validation can store a payload that later makesloadFromDatabase()skip the datastore entirely. Parseconfighere with the schema forkind, and replaceanywith the inferred type.As per coding guidelines, "Use zod for validation in packages/core and apps/webapp."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.ts` around lines 94 - 130, addDataStore and updateDataStore accept config: any and persist it directly; validate and type the config at this service boundary using the zod schema for the DataStoreKind before storing secrets or writing to Prisma. Locate the methods addDataStore and updateDataStore and replace the loose any type with the inferred TypeScript type from the zod parser; run the appropriate zod.parse/partial-parse for the provided kind (e.g., schemaForKind(kind)) and throw or return a clear error on parse failure, then use the validated config when calling getSecretStore().setSecret(secretKey, ...) and when writing config to this._prisma.organizationDataStore.create/update; ensure loadFromDatabase behavior is preserved by storing only the validated shape (or a secured secretKey) and include error handling to surface validation failures.apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts (2)
336-389: Consider reducing duplication betweenclickhouseandclickhouse_v2cases.The two cases share ~95% identical configuration. Extracting common options would improve maintainability.
♻️ Proposed refactor
function buildEventRepository(store: string, clickhouse: ClickHouse): ClickhouseEventRepository { + const commonOptions = { + 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, + llmMetricsBatchSize: env.LLM_METRICS_BATCH_SIZE, + llmMetricsFlushInterval: env.LLM_METRICS_FLUSH_INTERVAL_MS, + llmMetricsMaxBatchSize: env.LLM_METRICS_MAX_BATCH_SIZE, + llmMetricsMaxConcurrency: env.LLM_METRICS_MAX_CONCURRENCY, + otlpMetricsBatchSize: env.METRICS_CLICKHOUSE_BATCH_SIZE, + otlpMetricsFlushInterval: env.METRICS_CLICKHOUSE_FLUSH_INTERVAL_MS, + otlpMetricsMaxConcurrency: env.METRICS_CLICKHOUSE_MAX_CONCURRENCY, + }; + switch (store) { case "clickhouse": { return new ClickhouseEventRepository({ - clickhouse, - batchSize: env.EVENTS_CLICKHOUSE_BATCH_SIZE, - // ... all the options + ...commonOptions, + startTimeMaxAgeMs: env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS, version: "v1", }); } case "clickhouse_v2": { return new ClickhouseEventRepository({ - clickhouse: clickhouse, - batchSize: env.EVENTS_CLICKHOUSE_BATCH_SIZE, - // ... all the options + ...commonOptions, version: "v2", }); } default: { throw new Error(`Unknown ClickHouse event repository store: ${store}`); } } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts` around lines 336 - 389, The switch in buildEventRepository duplicates nearly identical ClickhouseEventRepository construction for "clickhouse" and "clickhouse_v2"; refactor by extracting the shared options into a single commonOptions object (populate fields like clickhouse, batchSize, flushInterval, maximumTraceSummaryViewCount, maximumTraceDetailedSummaryViewCount, maximumLiveReloadingSetting, insertStrategy, waitForAsyncInsert, asyncInsertMaxDataSize, asyncInsertBusyTimeoutMs, startTimeMaxAgeMs (only present for v1), llmMetrics*, otlpMetrics*, etc.), then call new ClickhouseEventRepository({...commonOptions, version: "v1"}) for "clickhouse" and new ClickhouseEventRepository({...commonOptions, version: "v2"}) for "clickhouse_v2"); ensure you preserve startTimeMaxAgeMs presence only where required and keep explicit clickhouse property name to match the constructor.
164-179: Org-scoped clients use uniform configuration regardless ofclientType.Unlike the default clients where logs and replication have specialized settings (e.g.,
clickhouseSettingsfor logs,RUN_REPLICATION_*env vars for replication), all org-scoped clients use the same genericCLICKHOUSE_*configuration. If org-specific logs or replication clients require different settings (likemax_memory_usagefor logs queries), this could cause performance or resource issues.Consider whether org-scoped clients need type-specific configurations, or document that org data stores intentionally use simplified settings.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts` around lines 164 - 179, buildOrgClickhouseClient currently creates org-scoped ClickHouse clients with the same generic CLICKHOUSE_* config regardless of ClientType; update it to apply type-specific settings (e.g., when clientType === "logs" include clickhouseSettings like max_memory_usage or when clientType === "replication" honor RUN_REPLICATION_* env vars) or add a clear comment/docstring stating that org clients intentionally use simplified settings. Locate buildOrgClickhouseClient and branch on the ClientType argument to merge the same specialized options used for default clients (name already contains clientType) or document the design decision so callers know org clients lack type-specific tuning.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@apps/webapp/app/env.server.ts`:
- Around line 1304-1305: The env schema allows
ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS to be zero or negative, enabling
tight-loop reloads; update the zod schema for
ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS to enforce sensible bounds (e.g.,
.int().min(60000) to require at least 1 minute and optionally .max(86400000) to
cap at 24 hours) while keeping the existing default (5 * 60 * 1000); use
z.coerce.number().int().min(...).max(...).default(...) or a .refine validator if
you prefer custom logic so invalid env values fail validation.
In
`@apps/webapp/app/routes/_app.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx:
- Around line 248-253: The presenter is being given the same ClickHouse "query"
client twice; call clickhouseFactory.getClickhouseForOrganization a second time
with the logs-scoped role (e.g., "logs") to obtain a logs client and pass that
as the presenter’s third argument so ErrorGroupPresenter($replica, queryClient,
logsClient) receives a query-scoped client and a logs-scoped client respectively
(identify the factory call clickhouseFactory.getClickhouseForOrganization and
the constructor ErrorGroupPresenter to locate where to change).
In
`@apps/webapp/app/routes/_app.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx:
- Around line 71-72: The ModelRegistryPresenter is being instantiated with only
ClickHouse but it needs the Prisma replica client for methods like
getModelDetail(); update the instantiation where you call new
ModelRegistryPresenter(clickhouse) to pass the Prisma replica (named $replica)
as the second argument, ensure you import $replica from ~/db.server if not
already imported, and verify presenter methods (e.g., getModelDetail,
getUserMetrics) now use the provided clients.
In
`@apps/webapp/app/routes/_app.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx:
- Around line 105-106: Move the call to
clickhouseFactory.getClickhouseForOrganization(...) so it executes inside the
loader's existing try block before constructing the TestTaskPresenter;
specifically ensure the clickhouse lookup and assignment to clickhouse happen
inside the try that wraps presenter.call(), then instantiate presenter = new
TestTaskPresenter($replica, clickhouse) after the successful lookup so any
errors from getClickhouseForOrganization are caught, logged, and handled by the
existing redirect/path in the catch.
In `@apps/webapp/app/routes/admin.data-stores.tsx`:
- Line 31: Update the import of tryCatch in the admin.data-stores route to use
the package subpath export: replace the current root import of tryCatch (import
{ tryCatch } from "@trigger.dev/core") with the subpath import from
"@trigger.dev/core/utils" so the symbol tryCatch is imported via the utils
subpath.
- Around line 88-137: ClickhouseConnectionSchema.parse() is used directly in the
"add" and "update" action handlers and will throw on invalid ClickHouse URLs
before the surrounding tryCatch runs; replace both usages with
ClickhouseConnectionSchema.safeParse(connectionObject) and check the returned
.success flag, returning typedjson({ error: validation.error.message }, {
status: 400 }) (or similar) when validation fails, then pass the validated value
to organizationDataStoresRegistry.addDataStore / updateDataStore; locate the
parse call sites in the "add" branch (where ClickhouseConnectionSchema.parse is
called before calling organizationDataStoresRegistry.addDataStore) and the
"update" branch (conditional parse before
organizationDataStoresRegistry.updateDataStore) and apply the safeParse pattern
and explicit error handling there.
- Line 1: The component currently calls onOpenChange(false) during render when
an operation reports success (the direct call to onOpenChange in the
render/JSX); move this side-effect into a useEffect so it runs after render
instead: remove the onOpenChange(false) call from the render/JSX, add a
useEffect that watches the isSuccess (or equivalent success flag) and calls
onOpenChange(false) when it becomes true, and include onOpenChange and the
success flag in the effect dependency array to avoid stale closures. Ensure you
reference the same success state variable and the onOpenChange prop used in the
component (replace the inline call with the new effect).
In `@apps/webapp/app/routes/otel.v1.metrics.ts`:
- Line 10: The code eagerly awaits otlpExporter (const exporter = await
otlpExporter) before validating the request content-type, causing unnecessary
exporter startup and possible init failures for requests that should return 400;
move the await of otlpExporter so it only runs after the content-type check (the
branch that returns the unsupported-content-type response) so malformed/probe
traffic never triggers exporter initialization — locate the otlpExporter await
in the handler in otel.v1.metrics.ts and only resolve otlpExporter after the
content-type validation code path.
In `@apps/webapp/app/runEngine/services/triggerFailedTask.server.ts`:
- Around line 70-74: The current write selects repository/store via
getEventRepository but only records taskEventStore in the run model, which will
break reads if an org is re-pointed; update the write path in
triggerFailedTask.server.ts so that when calling getEventRepository you also
persist the concrete OrganizationDataStore identifier (e.g., the returned
repository/store metadata or an OrganizationDataStore enum/value) into the run
model alongside taskEventStore (or alternatively mark the org's datastore
assignment immutable until a migration flag is set); specifically, modify the
code that assigns repository/store and the run creation/update logic to save the
concrete datastore binding (referencing getEventRepository, repository, store,
taskEventStore, and OrganizationDataStore) so future reads resolve the original
backend used for those writes.
In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts`:
- Around line 313-334: getEventsClickhouseClient currently instantiates a new
ClickHouse on each call; change it to use the existing singleton pattern: create
a singleton holder (e.g., const defaultEventsClickhouseClient = singleton(() =>
new ClickHouse({...same options...}))) and have getEventsClickhouseClient return
defaultEventsClickhouseClient() (or inline return singleton(() => new
ClickHouse(...))()). Keep the same URL parsing and options
(url.searchParams.delete("secure"), keepAlive, compression, etc.) but ensure
only one ClickHouse instance is created and reused across calls.
- Around line 362-384: The clickhouse_v2 case building the
ClickhouseEventRepository is missing the startTimeMaxAgeMs config that the
clickhouse case passes; update the clickhouse_v2 branch where
ClickhouseEventRepository is constructed to include startTimeMaxAgeMs:
env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS so the same start-time
clamping/validation used by the v1 config is applied to v2 as well.
In
`@apps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.ts`:
- Around line 72-75: The loop that builds the lookup map silently overwrites
keys produced from `${orgId}:${row.kind}`, causing nondeterministic behavior
when an org is assigned to multiple stores; before calling lookup.set(key,
parsed) check lookup.has(key) and if present throw a clear error (or assert)
that includes the conflicting orgId, kind, and identifiers for both the existing
entry and the current row (e.g., existingParsed/id and parsed/id) so overlapping
assignments discovered during findMany() fail fast and surface which rows/IDs
conflict.
- Around line 94-157: The addDataStore, updateDataStore and deleteDataStore
methods only write to the DB but don't update the in-memory registry map
(_lookup), leaving routing stale; after each successful mutation (after the
secret write and after the prisma create/update/delete) call the registry's
method that repopulates the in-memory map used by get() (the same function the
background loader uses — e.g. this.reload() / this._reloadLookup() /
this._refreshLookup(), whichever exists) so _lookup is refreshed immediately and
subsequent get() calls see the new data.
In `@apps/webapp/app/services/queryService.server.ts`:
- Around line 278-279: The concurrency limiter is currently keyed by projectId
but the ClickHouse client and execution are organization-scoped (see
clickhouseFactory.getClickhouseForOrganization and executeTSQL), so update the
limiter acquire/release calls (where limiter.acquire and limiter.release are
used around query execution) to use organizationId instead of projectId so
concurrent queries are limited per organization rather than per project; ensure
any helper that constructs limiter keys or caches limiter instances uses
organizationId consistently in the same places referenced around the query
execution.
In `@apps/webapp/app/services/runsReplicationService.server.ts`:
- Around line 655-685: The current loop serializes inserts per ClickHouse group
because each await blocks the next; change it to run per-group flushes
concurrently with a bounded concurrency limiter (e.g., p-limit or an internal
semaphore) so slow/unhealthy ClickHouse targets don't stall unrelated orgs. For
each group (the items handled around sortTaskRunInserts, sortPayloadInserts,
combinedTaskRunInserts/combinedPayloadInserts and using this.#insertWithRetry
which calls this.#insertTaskRunInserts and this.#insertPayloadInserts with
flushId), create an async task that performs the two insertWithRetry calls,
records the first taskRunError/payloadError seen, and updates
_taskRunsInsertedCounter/_payloadsInsertedCounter when each insert succeeds;
schedule those tasks through the limiter and await Promise.all on the scheduled
tasks; ensure you keep pushing group.taskRunInserts and group.payloadInserts
into combined* before scheduling so semantics remain the same.
In `@apps/webapp/app/v3/otlpExporter.server.ts`:
- Around line 1207-1215: The initializer initializeOTLPExporter is reading
settings directly from process.env; change it to use the typed env export
instead: replace process.env.OTLP_EXPORTER_VERBOSE with
env.OTLP_EXPORTER_VERBOSE (coerce to "1" check as before) and replace
process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT with
env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT and parseInt that value,
falling back to 8192; ensure the existing env import is used (or add it if
missing) and keep clickhouseFactory and OTLPExporter usage unchanged.
---
Outside diff comments:
In
`@apps/webapp/app/routes/_app.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx:
- Around line 137-159: The ClickHouse client lookup is awaited up front which
can throw before the deferred error handling; instead, defer the factory call so
failures are handled by the same promise chain: build listPromise by calling
clickhouseFactory.getClickhouseForOrganization(...) inside the async chain used
to create the LogsListPresenter and call presenter.call (keep references to
clickhouseFactory.getClickhouseForOrganization, LogsListPresenter,
presenter.call, and listPromise) so any errors from the org datastore/client
init are caught by the existing .catch that handles ServiceValidationError and
rethrows other errors, rather than throwing before TypedAwait renders its
errorElement.
In `@apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts`:
- Around line 13-17: The request schema CreateRunReplicationServiceParams still
requires keepAliveEnabled, keepAliveIdleSocketTtl, and maxOpenConnections but
the handler that constructs the replication service (the new factory path used
in the run replication creation flow) no longer consumes these fields, so they
are effectively ignored; either (A) thread these three fields into the
replication service factory/constructor/creation call (the function that builds
the replication client/service used in the handler) so the values are applied,
making sure to map the names exactly, or (B) remove these fields from
CreateRunReplicationServiceParams (and any validation/usage sites) so the API no
longer accepts unused settings; locate the schema
CreateRunReplicationServiceParams and the replication service
factory/constructor function referenced in the create run replication handler to
implement one of these fixes.
---
Nitpick comments:
In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts`:
- Around line 336-389: The switch in buildEventRepository duplicates nearly
identical ClickhouseEventRepository construction for "clickhouse" and
"clickhouse_v2"; refactor by extracting the shared options into a single
commonOptions object (populate fields like clickhouse, batchSize, flushInterval,
maximumTraceSummaryViewCount, maximumTraceDetailedSummaryViewCount,
maximumLiveReloadingSetting, insertStrategy, waitForAsyncInsert,
asyncInsertMaxDataSize, asyncInsertBusyTimeoutMs, startTimeMaxAgeMs (only
present for v1), llmMetrics*, otlpMetrics*, etc.), then call new
ClickhouseEventRepository({...commonOptions, version: "v1"}) for "clickhouse"
and new ClickhouseEventRepository({...commonOptions, version: "v2"}) for
"clickhouse_v2"); ensure you preserve startTimeMaxAgeMs presence only where
required and keep explicit clickhouse property name to match the constructor.
- Around line 164-179: buildOrgClickhouseClient currently creates org-scoped
ClickHouse clients with the same generic CLICKHOUSE_* config regardless of
ClientType; update it to apply type-specific settings (e.g., when clientType ===
"logs" include clickhouseSettings like max_memory_usage or when clientType ===
"replication" honor RUN_REPLICATION_* env vars) or add a clear comment/docstring
stating that org clients intentionally use simplified settings. Locate
buildOrgClickhouseClient and branch on the ClientType argument to merge the same
specialized options used for default clients (name already contains clientType)
or document the design decision so callers know org clients lack type-specific
tuning.
In
`@apps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.ts`:
- Around line 94-130: addDataStore and updateDataStore accept config: any and
persist it directly; validate and type the config at this service boundary using
the zod schema for the DataStoreKind before storing secrets or writing to
Prisma. Locate the methods addDataStore and updateDataStore and replace the
loose any type with the inferred TypeScript type from the zod parser; run the
appropriate zod.parse/partial-parse for the provided kind (e.g.,
schemaForKind(kind)) and throw or return a clear error on parse failure, then
use the validated config when calling getSecretStore().setSecret(secretKey, ...)
and when writing config to this._prisma.organizationDataStore.create/update;
ensure loadFromDatabase behavior is preserved by storing only the validated
shape (or a secured secretKey) and include error handling to surface validation
failures.
In `@apps/webapp/test/runsReplicationService.part2.test.ts`:
- Around line 26-27: The tests currently instantiate RunsReplicationService with
a TestReplicationClickhouseFactory backed by a single ClickHouse instance, so
they don't catch org-to-ClickHouse routing bugs; update the affected test cases
(instances where new RunsReplicationService is constructed with new
TestReplicationClickhouseFactory(clickhouse)) to create a multi-store scenario:
build two distinct Test ClickHouse clients (e.g., clickhouseA and clickhouseB),
construct a TestReplicationClickhouseFactory that maps orgA -> clickhouseA and
orgB -> clickhouseB, pass that factory into the RunsReplicationService, then
insert rows for both orgs and assert each org's rows appear in the corresponding
backend client (use the same helper assertions already used elsewhere in the
tests). Ensure you change all occurrences referenced (the other construction
sites of TestReplicationClickhouseFactory) so the suite validates org routing
across multiple stores.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 3a31161c-3ac9-4af9-bcd2-f229a19f7a61
📒 Files selected for processing (69)
.cursor/mcp.json.server-changes/organization-scoped-clickhouse.mdCLAUDE.mdapps/webapp/app/env.server.tsapps/webapp/app/presenters/v3/ApiRunListPresenter.server.tsapps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.tsapps/webapp/app/presenters/v3/RunPresenter.server.tsapps/webapp/app/presenters/v3/RunTagListPresenter.server.tsapps/webapp/app/presenters/v3/SpanPresenter.server.tsapps/webapp/app/presenters/v3/TaskListPresenter.server.tsapps/webapp/app/presenters/v3/UsagePresenter.server.tsapps/webapp/app/presenters/v3/ViewSchedulePresenter.server.tsapps/webapp/app/presenters/v3/WaitpointPresenter.server.tsapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsxapps/webapp/app/routes/admin.api.v1.runs-replication.create.tsapps/webapp/app/routes/admin.api.v1.runs-replication.start.tsapps/webapp/app/routes/admin.data-stores.tsxapps/webapp/app/routes/admin.tsxapps/webapp/app/routes/api.v1.prompts.$slug.tsapps/webapp/app/routes/api.v1.prompts.$slug.versions.tsapps/webapp/app/routes/api.v1.prompts._index.tsapps/webapp/app/routes/otel.v1.logs.tsapps/webapp/app/routes/otel.v1.metrics.tsapps/webapp/app/routes/otel.v1.traces.tsapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsxapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.tsapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.tsapps/webapp/app/runEngine/concerns/traceEvents.server.tsapps/webapp/app/runEngine/services/triggerFailedTask.server.tsapps/webapp/app/services/admin/missingLlmModels.server.tsapps/webapp/app/services/clickhouse/clickhouseFactory.server.tsapps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.tsapps/webapp/app/services/clickhouseInstance.server.tsapps/webapp/app/services/dataStores/organizationDataStoreConfigSchemas.server.tsapps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.tsapps/webapp/app/services/dataStores/organizationDataStoresRegistryInstance.server.tsapps/webapp/app/services/queryService.server.tsapps/webapp/app/services/runsReplicationInstance.server.tsapps/webapp/app/services/runsReplicationService.server.tsapps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.tsapps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.tsapps/webapp/app/v3/eventRepository/eventRepository.server.tsapps/webapp/app/v3/eventRepository/eventRepository.types.tsapps/webapp/app/v3/eventRepository/index.server.tsapps/webapp/app/v3/otlpExporter.server.tsapps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.tsapps/webapp/app/v3/services/bulk/BulkActionV2.server.tsapps/webapp/test/clickhouseFactory.test.tsapps/webapp/test/organizationDataStoresRegistry.test.tsapps/webapp/test/runsBackfiller.test.tsapps/webapp/test/runsReplicationBenchmark.test.tsapps/webapp/test/runsReplicationService.part1.test.tsapps/webapp/test/runsReplicationService.part2.test.tsapps/webapp/test/utils/replicationUtils.tsapps/webapp/test/utils/testReplicationClickhouseFactory.tsapps/webapp/test/utils/tracing.tsinternal-packages/database/prisma/migrations/20260331212308_add_organization_data_stores/migration.sqlinternal-packages/database/prisma/schema.prisma
💤 Files with no reviewable changes (2)
- apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts
- apps/webapp/app/services/clickhouseInstance.server.ts
Comment on lines
+1304
to
+1305
| // Organization data stores registry | ||
| ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce.number().int().default(5 * 60 * 1000), // 5 minutes |
Contributor
There was a problem hiding this comment.
Validate reload interval bounds to prevent tight-loop reloads.
Line 1305 currently allows 0 or negative values. A bad env value can make the registry reload loop run too frequently and degrade service stability.
Suggested fix
- ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce.number().int().default(5 * 60 * 1000), // 5 minutes
+ ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce
+ .number()
+ .int()
+ .min(1_000)
+ .default(5 * 60 * 1000), // 5 minutes📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // Organization data stores registry | |
| ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce.number().int().default(5 * 60 * 1000), // 5 minutes | |
| // Organization data stores registry | |
| ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce | |
| .number() | |
| .int() | |
| .min(1_000) | |
| .default(5 * 60 * 1000), // 5 minutes |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/env.server.ts` around lines 1304 - 1305, The env schema
allows ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS to be zero or negative,
enabling tight-loop reloads; update the zod schema for
ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS to enforce sensible bounds (e.g.,
.int().min(60000) to require at least 1 minute and optionally .max(86400000) to
cap at 24 hours) while keeping the existing default (5 * 60 * 1000); use
z.coerce.number().int().min(...).max(...).default(...) or a .refine validator if
you prefer custom logic so invalid env values fail validation.
Comment on lines
+248
to
+253
| const clickhouseClient = await clickhouseFactory.getClickhouseForOrganization( | ||
| environment.organizationId, | ||
| "query" | ||
| ); | ||
|
|
||
| const presenter = new ErrorGroupPresenter($replica, clickhouseClient, clickhouseClient); |
Contributor
There was a problem hiding this comment.
Pass a logs-scoped client into the presenter's second ClickHouse slot.
Line 253 currently injects the "query" client twice. That keeps the error-side queries working, but any org that routes logs to a separate datastore will still read logs from the wrong backend on this page.
Suggested fix
- const clickhouseClient = await clickhouseFactory.getClickhouseForOrganization(
- environment.organizationId,
- "query"
- );
-
- const presenter = new ErrorGroupPresenter($replica, clickhouseClient, clickhouseClient);
+ const [queryClickhouseClient, logsClickhouseClient] = await Promise.all([
+ clickhouseFactory.getClickhouseForOrganization(environment.organizationId, "query"),
+ clickhouseFactory.getClickhouseForOrganization(environment.organizationId, "logs"),
+ ]);
+
+ const presenter = new ErrorGroupPresenter(
+ $replica,
+ queryClickhouseClient,
+ logsClickhouseClient
+ );📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| const clickhouseClient = await clickhouseFactory.getClickhouseForOrganization( | |
| environment.organizationId, | |
| "query" | |
| ); | |
| const presenter = new ErrorGroupPresenter($replica, clickhouseClient, clickhouseClient); | |
| const [queryClickhouseClient, logsClickhouseClient] = await Promise.all([ | |
| clickhouseFactory.getClickhouseForOrganization(environment.organizationId, "query"), | |
| clickhouseFactory.getClickhouseForOrganization(environment.organizationId, "logs"), | |
| ]); | |
| const presenter = new ErrorGroupPresenter( | |
| $replica, | |
| queryClickhouseClient, | |
| logsClickhouseClient | |
| ); |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@apps/webapp/app/routes/_app.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx
around lines 248 - 253, The presenter is being given the same ClickHouse "query"
client twice; call clickhouseFactory.getClickhouseForOrganization a second time
with the logs-scoped role (e.g., "logs") to obtain a logs client and pass that
as the presenter’s third argument so ErrorGroupPresenter($replica, queryClient,
logsClient) receives a query-scoped client and a logs-scoped client respectively
(identify the factory call clickhouseFactory.getClickhouseForOrganization and
the constructor ErrorGroupPresenter to locate where to change).
Comment on lines
+71
to
+72
| const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); | ||
| const presenter = new ModelRegistryPresenter(clickhouse); |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
presenter=$(fd 'ModelRegistryPresenter.server.ts' | head -n1)
factory=$(fd 'clickhouseFactory.server.ts' | head -n1)
echo "== ModelRegistryPresenter queries =="
rg -n -C4 'getModelDetail|getUserMetrics|getModelComparison|llm_models|llm_metrics' "$presenter"
echo
echo "== ClickHouse factory resolution =="
rg -n -C4 'getClickhouseForOrganization|OrganizationDataStore|standard' "$factory"Repository: triggerdotdev/trigger.dev
Length of output: 4487
🏁 Script executed:
fd 'ModelRegistryPresenter.server.ts' -x wc -l {}Repository: triggerdotdev/trigger.dev
Length of output: 137
🏁 Script executed:
fd 'ModelRegistryPresenter.server.ts' -x head -n 50 {}Repository: triggerdotdev/trigger.dev
Length of output: 1982
🏁 Script executed:
fd 'route.tsx' -path '*projects*$modelId*' -x cat -n {}Repository: triggerdotdev/trigger.dev
Length of output: 237
🏁 Script executed:
rg -n -A15 'constructor\(' apps/webapp/app/presenters/v3/ModelRegistryPresenter.server.ts | head -n 30Repository: triggerdotdev/trigger.dev
Length of output: 661
🏁 Script executed:
fd 'ModelRegistryPresenter' --type f -path '*$modelId*'Repository: triggerdotdev/trigger.dev
Length of output: 237
🏁 Script executed:
rg -l '\$modelId' apps/webapp/app/routes | head -n 5Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
rg -l 'ModelRegistryPresenter' apps/webapp/app/routes --type tsRepository: triggerdotdev/trigger.dev
Length of output: 407
🏁 Script executed:
rg -B5 -A10 'new ModelRegistryPresenter' apps/webapp/app/routesRepository: triggerdotdev/trigger.dev
Length of output: 7330
🏁 Script executed:
rg -B5 'this\._replica' apps/webapp/app/presenters/v3/ModelRegistryPresenter.server.ts | head -n 20Repository: triggerdotdev/trigger.dev
Length of output: 506
🏁 Script executed:
rg -B5 'class BasePresenter' apps/webapp/app/presenters/v3/basePresenter.server.ts | head -n 15Repository: triggerdotdev/trigger.dev
Length of output: 386
🏁 Script executed:
rg -A20 'class BasePresenter' apps/webapp/app/presenters/v3/basePresenter.server.tsRepository: triggerdotdev/trigger.dev
Length of output: 754
Pass the Prisma replica client when instantiating ModelRegistryPresenter.
The presenter requires both Prisma (for getModelDetail()) and ClickHouse (for getUserMetrics()). Currently instantiated with only the ClickHouse client, any call to getModelDetail() will crash when accessing this._replica.llmModel.findFirst(...).
Pass $replica as the second argument:
Diff
const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard");
- const presenter = new ModelRegistryPresenter(clickhouse);
+ const presenter = new ModelRegistryPresenter(clickhouse, $replica);
const model = await presenter.getModelDetail(modelId);Import $replica from ~/db.server if not already present.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@apps/webapp/app/routes/_app.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx
around lines 71 - 72, The ModelRegistryPresenter is being instantiated with only
ClickHouse but it needs the Prisma replica client for methods like
getModelDetail(); update the instantiation where you call new
ModelRegistryPresenter(clickhouse) to pass the Prisma replica (named $replica)
as the second argument, ensure you import $replica from ~/db.server if not
already imported, and verify presenter methods (e.g., getModelDetail,
getUserMetrics) now use the provided clients.
Comment on lines
+105
to
+106
| const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard"); | ||
| const presenter = new TestTaskPresenter($replica, clickhouse); |
Contributor
There was a problem hiding this comment.
Move the factory lookup back under the loader's try.
getClickhouseForOrganization() can now fail before presenter.call() runs. With it outside the try, those failures bypass the current logging + redirect path and surface as a 500 instead.
Proposed fix
- const clickhouse = await clickhouseFactory.getClickhouseForOrganization(project.organizationId, "standard");
- const presenter = new TestTaskPresenter($replica, clickhouse);
try {
+ const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
+ project.organizationId,
+ "standard"
+ );
+ const presenter = new TestTaskPresenter($replica, clickhouse);
const [result, regionsResult] = await Promise.all([
presenter.call({
userId: user.id,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@apps/webapp/app/routes/_app.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx
around lines 105 - 106, Move the call to
clickhouseFactory.getClickhouseForOrganization(...) so it executes inside the
loader's existing try block before constructing the TestTaskPresenter;
specifically ensure the clickhouse lookup and assignment to clickhouse happen
inside the try that wraps presenter.call(), then instantiate presenter = new
TestTaskPresenter($replica, clickhouse) after the successful lookup so any
errors from getClickhouseForOrganization are caught, logged, and handled by the
existing redirect/path in the catch.
| @@ -0,0 +1,400 @@ | |||
| import { useState } from "react"; | |||
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cd apps/webapp && find . -name "admin.data-stores.tsx" -type fRepository: triggerdotdev/trigger.dev
Length of output: 103
🏁 Script executed:
cat -n apps/webapp/app/routes/admin.data-stores.tsxRepository: triggerdotdev/trigger.dev
Length of output: 16350
Move the close-on-success update out of render.
Calling onOpenChange(false) during render (lines 307-310) triggers a render-phase state update on the parent. React will warn about this, and it gets noisy in Strict Mode.
🔧 Suggested fix
-import { useState } from "react";
+import { useEffect, useState } from "react";
...
- // Close dialog on success
- if (fetcher.data?.success && open) {
- onOpenChange(false);
- }
+ // Close dialog on success
+ useEffect(() => {
+ if (fetcher.data?.success && open) {
+ onOpenChange(false);
+ }
+ }, [fetcher.data?.success, open, onOpenChange]);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| import { useState } from "react"; | |
| import { useEffect, useState } from "react"; |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/routes/admin.data-stores.tsx` at line 1, The component
currently calls onOpenChange(false) during render when an operation reports
success (the direct call to onOpenChange in the render/JSX); move this
side-effect into a useEffect so it runs after render instead: remove the
onOpenChange(false) call from the render/JSX, add a useEffect that watches the
isSuccess (or equivalent success flag) and calls onOpenChange(false) when it
becomes true, and include onOpenChange and the success flag in the effect
dependency array to avoid stale closures. Ensure you reference the same success
state variable and the onOpenChange prop used in the component (replace the
inline call with the new effect).
Comment on lines
+72
to
+75
| for (const orgId of row.organizationIds) { | ||
| const key = `${orgId}:${row.kind}`; | ||
| lookup.set(key, parsed); | ||
| } |
Contributor
There was a problem hiding this comment.
Fail fast on overlapping org assignments.
lookup.set() silently overwrites any earlier ${orgId}:${row.kind} entry. Because findMany() is unordered here, an org that appears in two stores can end up with a nondeterministic effective route depending on row order.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.ts`
around lines 72 - 75, The loop that builds the lookup map silently overwrites
keys produced from `${orgId}:${row.kind}`, causing nondeterministic behavior
when an org is assigned to multiple stores; before calling lookup.set(key,
parsed) check lookup.has(key) and if present throw a clear error (or assert)
that includes the conflicting orgId, kind, and identifiers for both the existing
entry and the current row (e.g., existingParsed/id and parsed/id) so overlapping
assignments discovered during findMany() fail fast and surface which rows/IDs
conflict.
Comment on lines
+94
to
+157
| async addDataStore({ | ||
| key, | ||
| kind, | ||
| organizationIds, | ||
| config, | ||
| }: { | ||
| key: string; | ||
| kind: DataStoreKind; | ||
| organizationIds: string[]; | ||
| config: any; | ||
| }) { | ||
| const secretKey = this.#secretKey(key, kind); | ||
|
|
||
| const secretStore = getSecretStore("DATABASE", { prismaClient: this._prisma }); | ||
| await secretStore.setSecret(secretKey, config); | ||
|
|
||
| return this._prisma.organizationDataStore.create({ | ||
| data: { | ||
| key, | ||
| organizationIds, | ||
| kind: "CLICKHOUSE", | ||
| config: { version: 1, data: { secretKey } }, | ||
| }, | ||
| }); | ||
| } | ||
|
|
||
| async updateDataStore({ | ||
| key, | ||
| kind, | ||
| organizationIds, | ||
| config, | ||
| }: { | ||
| key: string; | ||
| kind: DataStoreKind; | ||
| organizationIds: string[]; | ||
| config?: any; | ||
| }) { | ||
| const secretKey = this.#secretKey(key, kind); | ||
|
|
||
| if (config) { | ||
| const secretStore = getSecretStore("DATABASE", { prismaClient: this._prisma }); | ||
| await secretStore.setSecret(secretKey, config); | ||
| } | ||
|
|
||
| return this._prisma.organizationDataStore.update({ | ||
| where: { | ||
| key, | ||
| }, | ||
| data: { | ||
| organizationIds, | ||
| kind: "CLICKHOUSE", | ||
| }, | ||
| }); | ||
| } | ||
|
|
||
| async deleteDataStore({ key, kind }: { key: string; kind: DataStoreKind }) { | ||
| const secretKey = this.#secretKey(key, kind); | ||
| const secretStore = getSecretStore("DATABASE", { prismaClient: this._prisma }); | ||
| await secretStore.deleteSecret(secretKey).catch(() => { | ||
| // Secret may not exist — proceed with deletion | ||
| }); | ||
|
|
||
| await this._prisma.organizationDataStore.delete({ where: { key } }); | ||
| } |
Contributor
There was a problem hiding this comment.
Refresh _lookup after mutations.
get() only reads the in-memory map, but these mutation methods only write to the database. On the singleton registry that means routing stays stale until the next background reload, so OTLP/replication can keep targeting the old ClickHouse instance after an admin change.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.ts`
around lines 94 - 157, The addDataStore, updateDataStore and deleteDataStore
methods only write to the DB but don't update the in-memory registry map
(_lookup), leaving routing stale; after each successful mutation (after the
secret write and after the prisma create/update/delete) call the registry's
method that repopulates the in-memory map used by get() (the same function the
background loader uses — e.g. this.reload() / this._reloadLookup() /
this._refreshLookup(), whichever exists) so _lookup is refreshed immediately and
subsequent get() calls see the new data.
Comment on lines
+278
to
+279
| const queryClickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "query"); | ||
| const result = await executeTSQL(queryClickhouse.reader, { |
Contributor
There was a problem hiding this comment.
Key the concurrency limiter by organizationId now that the backend is org-scoped.
Line 278 routes execution through an organization-scoped ClickHouse client, but the limiter in Lines 155-160 and 376-379 still acquires/releases by projectId. One organization with multiple projects can therefore exceed the advertised org limit and push more concurrent queries onto the same backend than intended.
⚠️ Proposed fix
const acquireResult = await queryConcurrencyLimiter.acquire({
- key: projectId,
+ key: organizationId,
requestId,
keyLimit: orgLimit,
globalLimit: GLOBAL_CONCURRENCY_LIMIT,
});
if (!acquireResult.success) {
const errorMessage =
acquireResult.reason === "key_limit"
- ? `You've exceeded your query concurrency of ${orgLimit} for this project. Please try again later.`
+ ? `You've exceeded your query concurrency of ${orgLimit} for this organization. Please try again later.`
: "We're experiencing a lot of queries at the moment. Please try again later.";
return { success: false, error: new QueryError(errorMessage, { query: options.query }) };
}
@@
} finally {
// Always release the concurrency slot
await queryConcurrencyLimiter.release({
- key: projectId,
+ key: organizationId,
requestId,
});
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/services/queryService.server.ts` around lines 278 - 279, The
concurrency limiter is currently keyed by projectId but the ClickHouse client
and execution are organization-scoped (see
clickhouseFactory.getClickhouseForOrganization and executeTSQL), so update the
limiter acquire/release calls (where limiter.acquire and limiter.release are
used around query execution) to use organizationId instead of projectId so
concurrent queries are limited per organization rather than per project; ensure
any helper that constructs limiter keys or caches limiter instances uses
organizationId consistently in the same places referenced around the query
execution.
Comment on lines
+655
to
+685
| for (const [clickhouse, group] of groups) { | ||
| sortTaskRunInserts(group.taskRunInserts); | ||
| sortPayloadInserts(group.payloadInserts); | ||
| combinedTaskRunInserts.push(...group.taskRunInserts); | ||
| combinedPayloadInserts.push(...group.payloadInserts); | ||
|
|
||
| const [trErr] = await this.#insertWithRetry( | ||
| (attempt) => this.#insertTaskRunInserts(clickhouse, group.taskRunInserts, attempt), | ||
| "task run inserts", | ||
| flushId | ||
| ); | ||
| if (trErr && !taskRunError) { | ||
| taskRunError = trErr; | ||
| } | ||
|
|
||
| const [plErr] = await this.#insertWithRetry( | ||
| (attempt) => this.#insertPayloadInserts(clickhouse, group.payloadInserts, attempt), | ||
| "payload inserts", | ||
| flushId | ||
| ); | ||
| if (plErr && !payloadError) { | ||
| payloadError = plErr; | ||
| } | ||
|
|
||
| if (!trErr) { | ||
| this._taskRunsInsertedCounter.add(group.taskRunInserts.length); | ||
| } | ||
| if (!plErr) { | ||
| this._payloadsInsertedCounter.add(group.payloadInserts.length); | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
Avoid serializing the per-ClickHouse-group flushes.
Each await here makes every org group wait behind the slowest ClickHouse target. If one org-specific service is unhealthy and spends a few seconds in retry/backoff, unrelated orgs in the same replication batch inherit that lag too. On this hot path, the groups should be flushed behind a bounded limiter instead of one-by-one.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/services/runsReplicationService.server.ts` around lines 655 -
685, The current loop serializes inserts per ClickHouse group because each await
blocks the next; change it to run per-group flushes concurrently with a bounded
concurrency limiter (e.g., p-limit or an internal semaphore) so slow/unhealthy
ClickHouse targets don't stall unrelated orgs. For each group (the items handled
around sortTaskRunInserts, sortPayloadInserts,
combinedTaskRunInserts/combinedPayloadInserts and using this.#insertWithRetry
which calls this.#insertTaskRunInserts and this.#insertPayloadInserts with
flushId), create an async task that performs the two insertWithRetry calls,
records the first taskRunError/payloadError seen, and updates
_taskRunsInsertedCounter/_payloadsInsertedCounter when each insert succeeds;
schedule those tasks through the limiter and await Promise.all on the scheduled
tasks; ensure you keep pushing group.taskRunInserts and group.payloadInserts
into combined* before scheduling so semantics remain the same.
Comment on lines
+1207
to
+1215
| async function initializeOTLPExporter() { | ||
| await clickhouseFactory.isReady(); | ||
| return new OTLPExporter({ | ||
| clickhouseFactory, | ||
| verbose: process.env.OTLP_EXPORTER_VERBOSE === "1", | ||
| spanAttributeValueLengthLimit: process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT | ||
| ? parseInt(process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, 10) | ||
| : 8192 | ||
| ); | ||
| } No newline at end of file | ||
| : 8192, | ||
| }); |
Contributor
There was a problem hiding this comment.
Use env for these new OTLP exporter settings.
This initializer now bypasses the webapp's typed env surface even though env is already imported in the file. Please route both settings through env instead of process.env.
As per coding guidelines, "Environment variables must be accessed via the env export from app/env.server.ts and never use process.env directly."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/v3/otlpExporter.server.ts` around lines 1207 - 1215, The
initializer initializeOTLPExporter is reading settings directly from
process.env; change it to use the typed env export instead: replace
process.env.OTLP_EXPORTER_VERBOSE with env.OTLP_EXPORTER_VERBOSE (coerce to "1"
check as before) and replace
process.env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT with
env.SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT and parseInt that value,
falling back to 8192; ensure the existing env import is used (or add it if
missing) and keep clickhouseFactory and OTLPExporter usage unchanged.
deepshekhardas
pushed a commit
to deepshekhardas/trigger.dev
that referenced
this pull request
Apr 7, 2026
matt-aitken
force-pushed
the
org-scoped-clickhouse
branch
from
a60f2cb to
6179dc7
Compare
The only way to get a ClickHouse client now is through the factory. Refactored all existing code to use that and pass in an org. The runReplication and otlpExporter are the hot paths here which need special attention in reviews.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts (1)
13-27:⚠️ Potential issue | 🟠 MajorDon't silently drop the connection-tuning params.
keepAliveEnabled,keepAliveIdleSocketTtl, andmaxOpenConnectionsare still required by the request schema, but this factory no longer passes them intoRunsReplicationService. That makes the admin API accept values it doesn't honor and can unexpectedly revert replication behavior to defaults.Proposed fix
function createRunReplicationService(params: CreateRunReplicationServiceParams) { const { name, + keepAliveEnabled, + keepAliveIdleSocketTtl, + maxOpenConnections, maxFlushConcurrency, flushIntervalMs, flushBatchSize, leaderLockTimeoutMs, leaderLockExtendIntervalMs, @@ const service = new RunsReplicationService({ clickhouseFactory, pgConnectionUrl: env.DATABASE_URL, serviceName: name, + keepAliveEnabled, + keepAliveIdleSocketTtl, + maxOpenConnections, slotName: env.RUN_REPLICATION_SLOT_NAME, publicationName: env.RUN_REPLICATION_PUBLICATION_NAME,Also applies to: 82-120
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts` around lines 13 - 27, The request schema (CreateRunReplicationServiceParams) requires connection-tuning params keepAliveEnabled, keepAliveIdleSocketTtl, and maxOpenConnections but the factory that constructs RunsReplicationService is not forwarding them, so the API accepts but ignores these values; update the factory that calls RunsReplicationService (the runs replication service constructor/factory) to include these three fields when building the service config/object (pass keepAliveEnabled, keepAliveIdleSocketTtl, maxOpenConnections from CreateRunReplicationServiceParams into RunsReplicationService), and ensure any defaulting/validation logic remains consistent with the schema so values are honored rather than dropped.
♻️ Duplicate comments (5)
apps/webapp/app/env.server.ts (1)
1304-1309:⚠️ Potential issue | 🟠 MajorConstrain registry reload interval to avoid tight-loop reloads.
ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MScurrently accepts0and negative values. Misconfiguration here can create a hot reload loop and unnecessary service load.Suggested fix
ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce .number() .int() + .min(60_000) + .max(86_400_000) .default(60 * 1000), // 1 minute🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/env.server.ts` around lines 1304 - 1309, ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS currently allows 0 or negative values which can cause tight reload loops; update its Zod schema to enforce a sensible positive minimum (e.g., replace .int().default(60 * 1000) with .int().min(1000).default(60 * 1000)) so values below 1000ms are rejected and configuration cannot set 0 or negative intervals.apps/webapp/app/routes/admin.data-stores.tsx (3)
305-307:⚠️ Potential issue | 🟠 MajorMove success-close logic out of render and gate it to a completed submission.
Line 305-307 and Line 375-377 perform state updates during render. This is a React render-phase side effect, and because
fetcher.data?.successis sticky, reopening can immediately close again unless you gate close-on-success to a submit→idle transition.Suggested fix pattern
-import { useState } from "react"; +import { useEffect, useRef, useState } from "react"; ... function EditButton({ name, organizationIds }: { name: string; organizationIds: string[] }) { const [open, setOpen] = useState(false); const fetcher = useFetcher<{ success?: boolean; error?: string }>(); + const wasSubmittingRef = useRef(false); const isSubmitting = fetcher.state !== "idle"; - - if (fetcher.data?.success && open) { - setOpen(false); - } + useEffect(() => { + if (fetcher.state === "submitting") wasSubmittingRef.current = true; + if (wasSubmittingRef.current && fetcher.state === "idle") { + if (fetcher.data?.success && open) setOpen(false); + wasSubmittingRef.current = false; + } + }, [fetcher.state, fetcher.data?.success, open]); ... function AddDataStoreDialog(...) { const fetcher = useFetcher<{ success?: boolean; error?: string }>(); + const wasSubmittingRef = useRef(false); const isSubmitting = fetcher.state !== "idle"; - - if (fetcher.data?.success && open) { - onOpenChange(false); - } + useEffect(() => { + if (fetcher.state === "submitting") wasSubmittingRef.current = true; + if (wasSubmittingRef.current && fetcher.state === "idle") { + if (fetcher.data?.success && open) onOpenChange(false); + wasSubmittingRef.current = false; + } + }, [fetcher.state, fetcher.data?.success, open, onOpenChange]);#!/bin/bash # Locate render-phase close-on-success blocks in this route rg -nP 'if \(fetcher\.data\?\.success && open\)\s*\{' apps/webapp/app/routes/admin.data-stores.tsx -A2 -B1Also applies to: 375-377
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/admin.data-stores.tsx` around lines 305 - 307, The render currently closes the modal by calling setOpen when fetcher.data?.success && open inside the component render; move this into a useEffect to avoid render-phase side effects and to only close after a completed submission transition (submit→idle). Add a useEffect that watches fetcher.state (and fetcher.data?.success) and records previous fetcher.state (or checks for fetcher.state === "idle" && previousState === "submitting") and then calls setOpen(false) only when fetcher.data?.success is true and the state has just transitioned to idle; update references to fetcher, open, and setOpen accordingly (replace the inline if in the render and the duplicate at the other location).
31-31:⚠️ Potential issue | 🟡 MinorUse
@trigger.dev/coresubpath import fortryCatch.Line 31 imports from the package root, which breaks the webapp subpath-import rule.
Suggested fix
-import { tryCatch } from "@trigger.dev/core"; +import { tryCatch } from "@trigger.dev/core/utils";#!/bin/bash # Verify root imports from `@trigger.dev/core` in this route rg -nP 'from "@trigger\.dev/core"' apps/webapp/app/routes/admin.data-stores.tsxAs per coding guidelines, "When importing from
@trigger.dev/corein the webapp, use subpath exports from the package.json instead of importing from the root path."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/admin.data-stores.tsx` at line 31, Replace the root-package import of tryCatch with the package subpath export: change the import line that reads import { tryCatch } from "@trigger.dev/core"; to use the subpath import (e.g. import { tryCatch } from "@trigger.dev/core/tryCatch";) so the webapp follows the subpath-import rule; update any nearby references if your build flags require exact subpath syntax.
96-97:⚠️ Potential issue | 🟠 MajorAvoid uncaught Zod throws in add/update URL validation paths.
Line 96 and Line 121 use
.parse(), which throws before thetryCatch(...)around registry calls. Invalid URL input can therefore bypass your 400 path and surface as a 500.Suggested fix
- const config = ClickhouseConnectionSchema.parse({ url: connectionUrl }); + const parsedConfig = ClickhouseConnectionSchema.safeParse({ url: connectionUrl }); + if (!parsedConfig.success) { + return typedjson( + { error: parsedConfig.error.issues.map((i) => i.message).join(", ") }, + { status: 400 } + ); + } + const config = parsedConfig.data; ... - const config = connectionUrl - ? ClickhouseConnectionSchema.parse({ url: connectionUrl }) - : undefined; + let config: { url: string } | undefined = undefined; + if (connectionUrl) { + const parsedConfig = ClickhouseConnectionSchema.safeParse({ url: connectionUrl }); + if (!parsedConfig.success) { + return typedjson( + { error: parsedConfig.error.issues.map((i) => i.message).join(", ") }, + { status: 400 } + ); + } + config = parsedConfig.data; + }#!/bin/bash # Verify all direct parse usages in this route rg -n 'ClickhouseConnectionSchema\.parse\(' apps/webapp/app/routes/admin.data-stores.tsxBased on learnings, when using Zod
safeParse, check.successbefore accessing.data.Also applies to: 120-122
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/admin.data-stores.tsx` around lines 96 - 97, Replace direct Zod throws by using ClickhouseConnectionSchema.safeParse instead of ClickhouseConnectionSchema.parse in the add/update URL validation paths (the places currently calling ClickhouseConnectionSchema.parse). Check the returned result.success and if false return the existing 400 error path (or construct the same validation error response) instead of letting Zod throw; then use result.data when success. Ensure validation happens before the tryCatch/registry calls so invalid input follows the 400 branch rather than bubbling as a 500.apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts (1)
362-383:⚠️ Potential issue | 🟠 MajorApply
startTimeMaxAgeMsto the v2 repository too.The v1 branch passes
startTimeMaxAgeMs, but the v2 branch omits it.ClickhouseEventRepositorystill uses that guard to clamp stalestart_timevalues before inserts, so v2 currently loses the same protection.🔧 Proposed fix
case "clickhouse_v2": { return 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, + startTimeMaxAgeMs: env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS, llmMetricsBatchSize: env.LLM_METRICS_BATCH_SIZE, llmMetricsFlushInterval: env.LLM_METRICS_FLUSH_INTERVAL_MS, llmMetricsMaxBatchSize: env.LLM_METRICS_MAX_BATCH_SIZE,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts` around lines 362 - 383, The v2 branch constructing ClickhouseEventRepository omits the startTimeMaxAgeMs option so the repository's clamping of stale start_time is skipped; add startTimeMaxAgeMs: env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS (or the existing env key used by v1) to the options passed in the "clickhouse_v2" case when creating new ClickhouseEventRepository so v2 receives the same guard as v1.
🧹 Nitpick comments (1)
apps/webapp/app/v3/eventRepository/index.server.ts (1)
59-72: Minor inefficiency: factory called for all stores including Postgres.The factory is called unconditionally on line 59-62, but for Postgres stores, the result is discarded and
eventRepositoryis returned on line 72. Consider moving the factory call inside the ClickHouse conditionals to avoid unnecessary work for Postgres-backed organizations.Suggested optimization
- const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( - taskEventStore, - organizationId - ); - if (taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE_V2) { + const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( + taskEventStore, + organizationId + ); return { repository: resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE_V2 }; } if (taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE) { + const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( + taskEventStore, + organizationId + ); return { repository: resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE }; } return { repository: eventRepository, store: EVENT_STORE_TYPES.POSTGRES };🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/v3/eventRepository/index.server.ts` around lines 59 - 72, The code currently calls clickhouseFactory.getEventRepositoryForOrganization unconditionally and discards its result for Postgres; move the call inside the ClickHouse branches so it's only invoked when taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE_V2 or === EVENT_STORE_TYPES.CLICKHOUSE. Concretely, keep the final return { repository: eventRepository, store: EVENT_STORE_TYPES.POSTGRES } for non-ClickHouse, and in each ClickHouse conditional call getEventRepositoryForOrganization to obtain { repository: resolvedRepository } and return { repository: resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE[_V2] } accordingly; ensure you still use the same symbols: clickhouseFactory.getEventRepositoryForOrganization, taskEventStore, EVENT_STORE_TYPES, eventRepository, and resolvedRepository.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@apps/webapp/app/routes/resources.runs`.$runParam.logs.download.ts:
- Around line 36-39: The code currently calls
resolveEventRepositoryForStore(run.taskEventStore, run.organizationId ?? "")
which masks missing TaskRun.organizationId and can route to a default repo;
change this to explicitly check run.organizationId (e.g., if
(!run.organizationId) throw new Error or return an appropriate HTTP error)
before calling resolveEventRepositoryForStore so tenant-scoped resolution fails
fast when organizationId is null/undefined; update the call sites that use
resolveEventRepositoryForStore and references to
run.taskEventStore/run.organizationId to follow the same explicit-null-check
pattern.
In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts`:
- Around line 164-178: The org-scoped builder buildOrgClickhouseClient currently
applies only the generic CLICKHOUSE_* env settings; update it to branch on the
clientType ("events", "replication", "logs", etc.) and apply the per-client
tuning (use EVENTS_CLICKHOUSE_* for events clients for
pool/compression/maxOpenConnections, use RUN_REPLICATION_* for replication
clients, and include the same clickhouseSettings/safety caps used by
initializeLogsClickhouseClient for logs). Locate buildOrgClickhouseClient and
add conditional logic to select the appropriate env vars and clickhouse options
(keepAlive, compression, maxOpenConnections, and clickhouseSettings) based on
clientType so org-scoped clients match the non-org default behavior.
- Around line 157-159: The cache key uses hashHostname which only hashes
parsed.hostname causing collisions across different org stores on the same host;
update hashHostname (or replace with a new helper used by _clientCache and
_eventRepositoryCache) to hash the full datastore identity (e.g., the complete
URL including origin, pathname and search/query, or a canonical JSON of
connection options such as host, database, credentials and path) instead of only
parsed.hostname so each tenant+database+params yields a unique key; ensure all
places that call hashHostname (including usages tied to _clientCache and
_eventRepositoryCache) are updated to use the new full-identity hash.
In `@apps/webapp/app/v3/runEngineHandlers.server.ts`:
- Around line 190-193: The code calls
resolveEventRepositoryForStore(run.taskEventStore, taskRun.organizationId ?? "")
which uses an empty string org ID when organizationId is missing; change this to
validate taskRun.organizationId (and any similar call sites such as the other
resolveEventRepositoryForStore usages) and bail out or throw if it's falsy
instead of passing "" — e.g., check taskRun.organizationId before calling
resolveEventRepositoryForStore, return an error/early exit (or propagate) when
missing, and only call resolveEventRepositoryForStore with the validated
non-empty organizationId to avoid writing events to the wrong backend.
In `@apps/webapp/app/v3/services/completeAttempt.server.ts`:
- Around line 166-169: The current calls to resolveEventRepositoryForStore (and
the two other repository resolution calls at the success, failure, and retry
paths) use a fallback organizationId of "" via
taskRunAttempt.taskRun.organizationId ?? "", which silently masks missing tenant
IDs; change each to explicitly guard for a missing organizationId on
taskRunAttempt.taskRun and throw or return an explicit error instead of using ""
— locate usages of resolveEventRepositoryForStore and the analogous repository
resolution functions near the success/failure/retry paths (also repeated around
the other occurrences) and replace the fallback with a check like if
(!taskRunAttempt.taskRun.organizationId) { throw new Error("Missing
organizationId for taskRunAttempt"); } before calling the resolver so the
resolver always receives a real organizationId.
In `@apps/webapp/test/clickhouseFactory.test.ts`:
- Around line 1-3: The test imports only describe and expect from vitest but
uses vi (e.g., vi.mock and vi.setConfig); update the import statement in
clickhouseFactory.test.ts to also import vi from "vitest" so calls like
vi.mock(...) and vi.setConfig(...) reference the vitest mock API correctly.
---
Outside diff comments:
In `@apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts`:
- Around line 13-27: The request schema (CreateRunReplicationServiceParams)
requires connection-tuning params keepAliveEnabled, keepAliveIdleSocketTtl, and
maxOpenConnections but the factory that constructs RunsReplicationService is not
forwarding them, so the API accepts but ignores these values; update the factory
that calls RunsReplicationService (the runs replication service
constructor/factory) to include these three fields when building the service
config/object (pass keepAliveEnabled, keepAliveIdleSocketTtl, maxOpenConnections
from CreateRunReplicationServiceParams into RunsReplicationService), and ensure
any defaulting/validation logic remains consistent with the schema so values are
honored rather than dropped.
---
Duplicate comments:
In `@apps/webapp/app/env.server.ts`:
- Around line 1304-1309: ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS currently
allows 0 or negative values which can cause tight reload loops; update its Zod
schema to enforce a sensible positive minimum (e.g., replace .int().default(60 *
1000) with .int().min(1000).default(60 * 1000)) so values below 1000ms are
rejected and configuration cannot set 0 or negative intervals.
In `@apps/webapp/app/routes/admin.data-stores.tsx`:
- Around line 305-307: The render currently closes the modal by calling setOpen
when fetcher.data?.success && open inside the component render; move this into a
useEffect to avoid render-phase side effects and to only close after a completed
submission transition (submit→idle). Add a useEffect that watches fetcher.state
(and fetcher.data?.success) and records previous fetcher.state (or checks for
fetcher.state === "idle" && previousState === "submitting") and then calls
setOpen(false) only when fetcher.data?.success is true and the state has just
transitioned to idle; update references to fetcher, open, and setOpen
accordingly (replace the inline if in the render and the duplicate at the other
location).
- Line 31: Replace the root-package import of tryCatch with the package subpath
export: change the import line that reads import { tryCatch } from
"@trigger.dev/core"; to use the subpath import (e.g. import { tryCatch } from
"@trigger.dev/core/tryCatch";) so the webapp follows the subpath-import rule;
update any nearby references if your build flags require exact subpath syntax.
- Around line 96-97: Replace direct Zod throws by using
ClickhouseConnectionSchema.safeParse instead of ClickhouseConnectionSchema.parse
in the add/update URL validation paths (the places currently calling
ClickhouseConnectionSchema.parse). Check the returned result.success and if
false return the existing 400 error path (or construct the same validation error
response) instead of letting Zod throw; then use result.data when success.
Ensure validation happens before the tryCatch/registry calls so invalid input
follows the 400 branch rather than bubbling as a 500.
In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts`:
- Around line 362-383: The v2 branch constructing ClickhouseEventRepository
omits the startTimeMaxAgeMs option so the repository's clamping of stale
start_time is skipped; add startTimeMaxAgeMs:
env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS (or the existing env key used by v1)
to the options passed in the "clickhouse_v2" case when creating new
ClickhouseEventRepository so v2 receives the same guard as v1.
---
Nitpick comments:
In `@apps/webapp/app/v3/eventRepository/index.server.ts`:
- Around line 59-72: The code currently calls
clickhouseFactory.getEventRepositoryForOrganization unconditionally and discards
its result for Postgres; move the call inside the ClickHouse branches so it's
only invoked when taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE_V2 or ===
EVENT_STORE_TYPES.CLICKHOUSE. Concretely, keep the final return { repository:
eventRepository, store: EVENT_STORE_TYPES.POSTGRES } for non-ClickHouse, and in
each ClickHouse conditional call getEventRepositoryForOrganization to obtain {
repository: resolvedRepository } and return { repository: resolvedRepository,
store: EVENT_STORE_TYPES.CLICKHOUSE[_V2] } accordingly; ensure you still use the
same symbols: clickhouseFactory.getEventRepositoryForOrganization,
taskEventStore, EVENT_STORE_TYPES, eventRepository, and resolvedRepository.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 8be3f360-595a-4400-b4e5-adcbb4c08b8f
📒 Files selected for processing (79)
.cursor/mcp.json.server-changes/organization-scoped-clickhouse.mdCLAUDE.mdapps/webapp/app/env.server.tsapps/webapp/app/presenters/v3/ApiRunListPresenter.server.tsapps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.tsapps/webapp/app/presenters/v3/RunPresenter.server.tsapps/webapp/app/presenters/v3/RunTagListPresenter.server.tsapps/webapp/app/presenters/v3/SpanPresenter.server.tsapps/webapp/app/presenters/v3/TaskListPresenter.server.tsapps/webapp/app/presenters/v3/UsagePresenter.server.tsapps/webapp/app/presenters/v3/ViewSchedulePresenter.server.tsapps/webapp/app/presenters/v3/WaitpointPresenter.server.tsapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsxapps/webapp/app/routes/admin.api.v1.runs-replication.create.tsapps/webapp/app/routes/admin.api.v1.runs-replication.start.tsapps/webapp/app/routes/admin.data-stores.tsxapps/webapp/app/routes/admin.tsxapps/webapp/app/routes/api.v1.prompts.$slug.tsapps/webapp/app/routes/api.v1.prompts.$slug.versions.tsapps/webapp/app/routes/api.v1.prompts._index.tsapps/webapp/app/routes/api.v1.runs.$runId.events.tsapps/webapp/app/routes/api.v1.runs.$runId.spans.$spanId.tsapps/webapp/app/routes/api.v1.runs.$runId.trace.tsapps/webapp/app/routes/otel.v1.logs.tsapps/webapp/app/routes/otel.v1.metrics.tsapps/webapp/app/routes/otel.v1.traces.tsapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsxapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.tsapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.tsapps/webapp/app/routes/resources.runs.$runParam.logs.download.tsapps/webapp/app/runEngine/concerns/traceEvents.server.tsapps/webapp/app/runEngine/services/triggerFailedTask.server.tsapps/webapp/app/services/admin/missingLlmModels.server.tsapps/webapp/app/services/clickhouse/clickhouseFactory.server.tsapps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.tsapps/webapp/app/services/clickhouseInstance.server.tsapps/webapp/app/services/dataStores/organizationDataStoreConfigSchemas.server.tsapps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.tsapps/webapp/app/services/dataStores/organizationDataStoresRegistryInstance.server.tsapps/webapp/app/services/queryService.server.tsapps/webapp/app/services/runsReplicationInstance.server.tsapps/webapp/app/services/runsReplicationService.server.tsapps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.tsapps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.tsapps/webapp/app/v3/eventRepository/eventRepository.server.tsapps/webapp/app/v3/eventRepository/eventRepository.types.tsapps/webapp/app/v3/eventRepository/index.server.tsapps/webapp/app/v3/otlpExporter.server.tsapps/webapp/app/v3/runEngineHandlers.server.tsapps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.tsapps/webapp/app/v3/services/bulk/BulkActionV2.server.tsapps/webapp/app/v3/services/cancelTaskRunV1.server.tsapps/webapp/app/v3/services/completeAttempt.server.tsapps/webapp/app/v3/services/crashTaskRun.server.tsapps/webapp/app/v3/services/expireEnqueuedRun.server.tsapps/webapp/app/v3/services/triggerTaskV1.server.tsapps/webapp/test/clickhouseFactory.test.tsapps/webapp/test/organizationDataStoresRegistry.test.tsapps/webapp/test/runsBackfiller.test.tsapps/webapp/test/runsReplicationBenchmark.test.tsapps/webapp/test/runsReplicationService.part1.test.tsapps/webapp/test/runsReplicationService.part2.test.tsapps/webapp/test/utils/replicationUtils.tsapps/webapp/test/utils/testReplicationClickhouseFactory.tsapps/webapp/test/utils/tracing.tsinternal-packages/database/prisma/migrations/20260331212308_add_organization_data_stores/migration.sqlinternal-packages/database/prisma/schema.prisma
💤 Files with no reviewable changes (2)
- apps/webapp/app/services/clickhouseInstance.server.ts
- apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts
✅ Files skipped from review due to trivial changes (18)
- .server-changes/organization-scoped-clickhouse.md
- apps/webapp/app/routes/admin.tsx
- .cursor/mcp.json
- CLAUDE.md
- apps/webapp/app/runEngine/services/triggerFailedTask.server.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsx
- apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts
- apps/webapp/app/presenters/v3/SpanPresenter.server.ts
- apps/webapp/test/runsReplicationService.part1.test.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsx
- apps/webapp/test/runsReplicationService.part2.test.ts
- apps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.ts
- apps/webapp/test/organizationDataStoresRegistry.test.ts
- apps/webapp/app/services/dataStores/organizationDataStoreConfigSchemas.server.ts
- internal-packages/database/prisma/migrations/20260331212308_add_organization_data_stores/migration.sql
- internal-packages/database/prisma/schema.prisma
- apps/webapp/app/v3/otlpExporter.server.ts
- apps/webapp/app/services/runsReplicationService.server.ts
🚧 Files skipped from review as they are similar to previous changes (27)
- apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsx
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsx
- apps/webapp/test/runsBackfiller.test.ts
- apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts
- apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx
- apps/webapp/app/routes/otel.v1.metrics.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsx
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsx
- apps/webapp/test/utils/tracing.ts
- apps/webapp/app/services/queryService.server.ts
- apps/webapp/app/routes/otel.v1.logs.ts
- apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts
- apps/webapp/app/presenters/v3/RunPresenter.server.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsx
- apps/webapp/app/routes/otel.v1.traces.ts
- apps/webapp/app/routes/api.v1.prompts._index.ts
- apps/webapp/test/runsReplicationBenchmark.test.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsx
- apps/webapp/app/services/admin/missingLlmModels.server.ts
- apps/webapp/app/services/runsReplicationInstance.server.ts
- apps/webapp/app/services/dataStores/organizationDataStoresRegistryInstance.server.ts
- apps/webapp/app/presenters/v3/TaskListPresenter.server.ts
- apps/webapp/test/utils/testReplicationClickhouseFactory.ts
- apps/webapp/app/routes/api.v1.prompts.$slug.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx
- apps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.ts
Comment on lines
+36
to
+39
| const eventRepository = resolveEventRepositoryForStore( | ||
| run.taskEventStore, | ||
| run.organizationId ?? "" | ||
| ); |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== TaskRun model organizationId definition =="
fd -i 'schema.prisma' --exec sh -c '
echo "--- $1 ---"
sed -n "/model TaskRun {/,/}/p" "$1" | rg -n "organizationId|taskEventStore|runtimeEnvironmentId"
' sh {}
echo
echo "== resolveEventRepositoryForStore callsites using empty-string fallback =="
rg -nP --type=ts 'resolveEventRepositoryForStore\(\s*[^,]+,\s*[^)]*\?\?\s*""\s*\)'
echo
echo "== resolveEventRepositoryForStore signature =="
rg -nP --type=ts 'export function resolveEventRepositoryForStore\('Repository: triggerdotdev/trigger.dev
Length of output: 510
🏁 Script executed:
cat -n apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts | head -60Repository: triggerdotdev/trigger.dev
Length of output: 156
🏁 Script executed:
rg -nA 5 'export function resolveEventRepositoryForStore' apps/Repository: triggerdotdev/trigger.dev
Length of output: 699
🏁 Script executed:
rg -nB 5 -A 10 'model TaskRun' internal-packages/database/prisma/schema.prismaRepository: triggerdotdev/trigger.dev
Length of output: 4886
🏁 Script executed:
rg -n 'organizationId' apps/webapp/app/routes/resources.runs.$runParam.logs.download.tsRepository: triggerdotdev/trigger.dev
Length of output: 164
🏁 Script executed:
find apps/webapp/app/routes -name "*logs.download*" -type fRepository: triggerdotdev/trigger.dev
Length of output: 133
🏁 Script executed:
rg -n 'organizationId' internal-packages/database/prisma/schema.prisma | grep -A 2 -B 2 'TaskRun'Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
sed -n '669,900p' internal-packages/database/prisma/schema.prisma | grep -n 'organizationId'Repository: triggerdotdev/trigger.dev
Length of output: 96
🏁 Script executed:
cat -n 'apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts'Repository: triggerdotdev/trigger.dev
Length of output: 4676
🏁 Script executed:
rg -B 15 -A 5 'resolveEventRepositoryForStore' 'apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts'Repository: triggerdotdev/trigger.dev
Length of output: 1341
🏁 Script executed:
rg -B 5 -A 20 'taskRun.findFirst' apps/webapp/app/routes/resources.runs.$runParam.logs.download.tsRepository: triggerdotdev/trigger.dev
Length of output: 164
🏁 Script executed:
# Check if organizationId is a required field in practice (is it set when TaskRun is created?)
rg -n 'organizationId:' internal-packages/database/prisma/schema.prisma | grep -A 5 -B 5 'TaskRun'Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Check how clickhouseFactory.getEventRepositoryForOrganizationSync handles empty string
rg -B 5 -A 15 'getEventRepositoryForOrganizationSync' apps/Repository: triggerdotdev/trigger.dev
Length of output: 7990
🏁 Script executed:
# Check TaskRun.organizationId nullability more thoroughly
sed -n '669,800p' internal-packages/database/prisma/schema.prisma | cat -nRepository: triggerdotdev/trigger.dev
Length of output: 4903
🏁 Script executed:
# Look for how TaskRun is created - does organizationId get set?
rg -n 'create.*TaskRun|prisma.taskRun.create' apps/webapp/app/services/ apps/webapp/app/v3/ | head -20Repository: triggerdotdev/trigger.dev
Length of output: 1093
🏁 Script executed:
# Check for any validation or cleanup of empty organizationId
rg -n 'organizationId.*""' apps/webapp/Repository: triggerdotdev/trigger.dev
Length of output: 1704
🏁 Script executed:
# Check if TaskRun.organizationId is always populated during creation
rg -B 10 -A 10 'data.*organizationId' apps/webapp/app/v3/ | grep -A 10 -B 10 'taskRun\|TaskRun' | head -40Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Check if there are any comments or TODOs about the organizationId fallback
rg -B 3 'organizationId ?? ""' apps/webapp/app/v3/eventRepository/Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Verify: does the query that fetches the run guarantee organizationId is set?
rg -B 20 'resolveEventRepositoryForStore' apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts | head -40Repository: triggerdotdev/trigger.dev
Length of output: 164
🏁 Script executed:
# Try a different approach - find the file directly
find apps/webapp -name "logs.download.ts" -type f -exec head -40 {} \;Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Check if Project-to-Organization relationship guarantees organizationId on TaskRun
rg -B 5 -A 5 'organizationId' internal-packages/database/prisma/schema.prisma | grep -A 10 -B 10 'Project\|Organization' | head -50Repository: triggerdotdev/trigger.dev
Length of output: 1334
🏁 Script executed:
# Check the query pattern - does fetchning a run through project->organization relationship mean organizationId should exist?
sed -n '17,30p' apps/webapp/app/routes/resources.runs.\$runParam.logs.download.tsRepository: triggerdotdev/trigger.dev
Length of output: 343
🏁 Script executed:
# Check how TaskRun gets created - specifically if organizationId is set
rg -B 5 -A 15 'prisma\.taskRun\.create' apps/webapp/app/v3/services/ | head -60Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Check if TaskRun uses projectId to derive organizationId, or if it's stored separately
rg -n 'organizationId.*projectId\|projectId.*organizationId' apps/webapp/app/v3/ | head -10Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Look at the Project model to understand the relationship
sed -n '/^model Project {/,/^}/p' internal-packages/database/prisma/schema.prisma | head -40Repository: triggerdotdev/trigger.dev
Length of output: 1563
🏁 Script executed:
# Check if organizationId is always set when TaskRun is created (by looking at actual creation code)
rg -B 20 'taskRun.*create' apps/webapp/app/v3/services/triggerV3.server.ts | head -80Repository: triggerdotdev/trigger.dev
Length of output: 156
🏁 Script executed:
# Look for any validation logic that checks organizationId after fetching
rg -B 5 -A 10 'if.*organizationId\|organizationId.*if' apps/webapp/app/v3/eventRepository/Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Check if the query always returns organizationId from the database (by looking at select clauses)
rg -B 5 'taskRun.findFirst.*where' apps/webapp/app/v3/eventRepository/ apps/webapp/app/services/ | head -30Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Find where TaskRun is created - search for actual create statements
find apps/webapp -name "*.server.ts" -o -name "*.server.tsx" | xargs rg -l 'taskRun.*create|prisma\.taskRun\.create' | head -5Repository: triggerdotdev/trigger.dev
Length of output: 340
🏁 Script executed:
# Look at one of the create implementations
fd -t f '*.server.ts' apps/webapp/app/v3/services | xargs grep -l 'create' | head -1 | xargs cat | head -100Repository: triggerdotdev/trigger.dev
Length of output: 359
🏁 Script executed:
# Check if organizationId is populated when TaskRun is created - search for data assignment
rg -B 5 -A 15 'taskRun.*create|create.*taskRun' apps/webapp/app/v3 | grep -A 15 'data.*{' | head -50Repository: triggerdotdev/trigger.dev
Length of output: 4326
🏁 Script executed:
# Check if organizationId is guaranteed to be set from the query - look at the fetch context
cat -n apps/webapp/app/routes/resources.runs.\$runParam.logs.download.ts | sed -n '17,35p'Repository: triggerdotdev/trigger.dev
Length of output: 551
🏁 Script executed:
# Now check if environment.organizationId is always present
rg -B 10 'organizationId: environment.organizationId' apps/webapp/app/v3/services/triggerTaskV1.server.ts | head -20Repository: triggerdotdev/trigger.dev
Length of output: 536
🏁 Script executed:
# Look for any code that might create TaskRun without organizationId
rg 'taskRun.*create|prisma\.taskRun\.create' -A 20 apps/webapp/app/v3/services/ | grep -v 'organizationId' | head -40Repository: triggerdotdev/trigger.dev
Length of output: 3896
🏁 Script executed:
# Check if the query implicitly guarantees organizationId through the project->organization relationship
# Does Project always have an organizationId?
sed -n '/^model Project {/,/^}/p' internal-packages/database/prisma/schema.prisma | grep -n organizationIdRepository: triggerdotdev/trigger.dev
Length of output: 218
🏁 Script executed:
# Check if clickhouseFactory handles empty string specially or if it causes issues
cat -n apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts | sed -n '60,120p'Repository: triggerdotdev/trigger.dev
Length of output: 2335
🏁 Script executed:
# See what getEventRepositoryForOrganizationSync does with an empty/invalid organizationId
rg -B 5 -A 20 'getEventRepositoryForOrganizationSync' apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts | head -50Repository: triggerdotdev/trigger.dev
Length of output: 1117
🏁 Script executed:
# Critical check: Can organizationId actually be NULL at runtime, or is it always set?
# Look for any code paths that create TaskRun without setting organizationId
rg -B 10 'taskRun\.create' apps/webapp/app/v3/ | grep -B 10 'organizationId' | grep -v 'organizationId' | tail -20Repository: triggerdotdev/trigger.dev
Length of output: 51
🏁 Script executed:
# Check if there are migrations or older code that might create TaskRun without organizationId
git log --oneline --all -- internal-packages/database/prisma/schema.prisma | head -5Repository: triggerdotdev/trigger.dev
Length of output: 392
🏁 Script executed:
# Final verification: Does the default Prisma select include organizationId?
rg -B 3 'taskRun.findFirst' apps/webapp/app/routes/resources.runs.\$runParam.logs.download.tsRepository: triggerdotdev/trigger.dev
Length of output: 240
Use explicit null checks for org-scoped repository resolution instead of silent empty-string fallback.
TaskRun.organizationId is optional in the schema (organizationId String?). Using run.organizationId ?? "" masks null values and silently routes to a default repository, risking cross-tenant data leakage. When the org ID is missing, the wrong event store may be accessed without raising an error.
Fail fast on missing org ID for tenant-scoped storage. Replace the fallback with an explicit check:
- const eventRepository = resolveEventRepositoryForStore(
- run.taskEventStore,
- run.organizationId ?? ""
- );
+ if (!run.organizationId) {
+ return new Response("Run organization is missing", { status: 500 });
+ }
+
+ const eventRepository = resolveEventRepositoryForStore(run.taskEventStore, run.organizationId);This pattern is also used in apps/webapp/app/v3/runEngineHandlers.server.ts, apps/webapp/app/v3/services/completeAttempt.server.ts, and apps/webapp/app/v3/eventRepository/eventRepository.server.ts.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| const eventRepository = resolveEventRepositoryForStore( | |
| run.taskEventStore, | |
| run.organizationId ?? "" | |
| ); | |
| if (!run.organizationId) { | |
| return new Response("Run organization is missing", { status: 500 }); | |
| } | |
| const eventRepository = resolveEventRepositoryForStore( | |
| run.taskEventStore, | |
| run.organizationId | |
| ); |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/routes/resources.runs`.$runParam.logs.download.ts around
lines 36 - 39, The code currently calls
resolveEventRepositoryForStore(run.taskEventStore, run.organizationId ?? "")
which masks missing TaskRun.organizationId and can route to a default repo;
change this to explicitly check run.organizationId (e.g., if
(!run.organizationId) throw new Error or return an appropriate HTTP error)
before calling resolveEventRepositoryForStore so tenant-scoped resolution fails
fast when organizationId is null/undefined; update the call sites that use
resolveEventRepositoryForStore and references to
run.taskEventStore/run.organizationId to follow the same explicit-null-check
pattern.
Comment on lines
+157
to
+159
| function hashHostname(url: string): string { | ||
| const parsed = new URL(url); | ||
| return createHash("sha256").update(parsed.hostname).digest("hex"); |
Contributor
There was a problem hiding this comment.
Use the full data-store identity in the cache key.
Hashing only parsed.hostname lets different org stores on the same host collide in _clientCache and _eventRepositoryCache. If two orgs use the same ClickHouse host with different databases/credentials/path/query params, the second org can reuse the first org’s client/repository and hit the wrong tenant store.
🔧 Proposed fix
-function hashHostname(url: string): string {
+function hashDataStore(url: string): string {
const parsed = new URL(url);
- return createHash("sha256").update(parsed.hostname).digest("hex");
+ parsed.searchParams.delete("secure");
+ return createHash("sha256").update(parsed.toString()).digest("hex");
}
@@
- const hostnameHash = hashHostname(dataStore.url);
- const cacheKey = `${hostnameHash}:${clientType}`;
+ const dataStoreHash = hashDataStore(dataStore.url);
+ const cacheKey = `${dataStoreHash}:${clientType}`;
@@
- const hostnameHash = hashHostname(dataStore.url);
- const cacheKey = `${hostnameHash}:events:${store}`;
+ const dataStoreHash = hashDataStore(dataStore.url);
+ const cacheKey = `${dataStoreHash}:events:${store}`;Also applies to: 230-236, 270-277
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts` around lines
157 - 159, The cache key uses hashHostname which only hashes parsed.hostname
causing collisions across different org stores on the same host; update
hashHostname (or replace with a new helper used by _clientCache and
_eventRepositoryCache) to hash the full datastore identity (e.g., the complete
URL including origin, pathname and search/query, or a canonical JSON of
connection options such as host, database, credentials and path) instead of only
parsed.hostname so each tenant+database+params yields a unique key; ensure all
places that call hashHostname (including usages tied to _clientCache and
_eventRepositoryCache) are updated to use the new full-identity hash.
Comment on lines
+164
to
+178
| function buildOrgClickhouseClient(url: string, clientType: ClientType): ClickHouse { | ||
| const parsed = new URL(url); | ||
| parsed.searchParams.delete("secure"); | ||
|
|
||
| return new ClickHouse({ | ||
| url: parsed.toString(), | ||
| name: `org-clickhouse-${clientType}`, | ||
| keepAlive: { | ||
| enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1", | ||
| idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS, | ||
| }, | ||
| logLevel: env.CLICKHOUSE_LOG_LEVEL, | ||
| compression: { request: true }, | ||
| maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS, | ||
| }); |
Contributor
There was a problem hiding this comment.
Preserve the per-client tuning for org-scoped clients.
buildOrgClickhouseClient() always uses the generic CLICKHOUSE_* settings, so org-scoped "events" clients lose the EVENTS_CLICKHOUSE_* pool/compression settings, "replication" loses the RUN_REPLICATION_* settings, and "logs" drops the extra clickhouseSettings safety caps from initializeLogsClickhouseClient(). That makes the org-scoped path behave differently from the default path on the exact hot paths this PR calls out.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts` around lines
164 - 178, The org-scoped builder buildOrgClickhouseClient currently applies
only the generic CLICKHOUSE_* env settings; update it to branch on the
clientType ("events", "replication", "logs", etc.) and apply the per-client
tuning (use EVENTS_CLICKHOUSE_* for events clients for
pool/compression/maxOpenConnections, use RUN_REPLICATION_* for replication
clients, and include the same clickhouseSettings/safety caps used by
initializeLogsClickhouseClient for logs). Locate buildOrgClickhouseClient and
add conditional logic to select the appropriate env vars and clickhouse options
(keepAlive, compression, maxOpenConnections, and clickhouseSettings) based on
clientType so org-scoped clients match the non-org default behavior.
Comment on lines
+190
to
+193
| const eventRepository = resolveEventRepositoryForStore( | ||
| run.taskEventStore, | ||
| taskRun.organizationId ?? "" | ||
| ); |
Contributor
There was a problem hiding this comment.
Avoid resolving org-scoped repositories with "" as the org ID.
If organizationId is ever missing here, this will still perform a lookup against an empty org key and can write attempt/cached events to the wrong backend instead of failing fast. Please guard and bail out, or plumb a real org ID into these paths.
Proposed fix
- const eventRepository = resolveEventRepositoryForStore(
- run.taskEventStore,
- taskRun.organizationId ?? ""
- );
+ if !taskRun.organizationId {
+ logger.error("[runAttemptFailed] Missing organizationId for task run", {
+ runId: run.id,
+ });
+ return;
+ }
+
+ const eventRepository = resolveEventRepositoryForStore(
+ run.taskEventStore,
+ taskRun.organizationId
+ );- const eventRepository = resolveEventRepositoryForStore(
- blockedRun.taskEventStore,
- blockedRun.organizationId ?? ""
- );
+ if (!blockedRun.organizationId) {
+ logger.error("[cachedRunCompleted] Missing organizationId for blocked run", {
+ blockedRunId,
+ });
+ return;
+ }
+
+ const eventRepository = resolveEventRepositoryForStore(
+ blockedRun.taskEventStore,
+ blockedRun.organizationId
+ );Also applies to: 294-297
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/v3/runEngineHandlers.server.ts` around lines 190 - 193, The
code calls resolveEventRepositoryForStore(run.taskEventStore,
taskRun.organizationId ?? "") which uses an empty string org ID when
organizationId is missing; change this to validate taskRun.organizationId (and
any similar call sites such as the other resolveEventRepositoryForStore usages)
and bail out or throw if it's falsy instead of passing "" — e.g., check
taskRun.organizationId before calling resolveEventRepositoryForStore, return an
error/early exit (or propagate) when missing, and only call
resolveEventRepositoryForStore with the validated non-empty organizationId to
avoid writing events to the wrong backend.
Comment on lines
+166
to
+169
| const eventRepository = resolveEventRepositoryForStore( | ||
| taskRunAttempt.taskRun.taskEventStore, | ||
| taskRunAttempt.taskRun.organizationId ?? "" | ||
| ); |
Contributor
There was a problem hiding this comment.
Remove ?? "" fallback for organization ID in all three repository resolutions.
These hot-path calls should not default to an empty tenant key. If organizationId is unexpectedly missing, handle it explicitly (guard + error) instead of silently resolving with "".
💡 Suggested refactor pattern
-const eventRepository = resolveEventRepositoryForStore(
- taskRunAttempt.taskRun.taskEventStore,
- taskRunAttempt.taskRun.organizationId ?? ""
-);
+const organizationId = taskRunAttempt.taskRun.organizationId;
+if (!organizationId) {
+ logger.error("[CompleteAttemptService] Missing organizationId on task run", {
+ runId: taskRunAttempt.taskRun.id,
+ taskEventStore: taskRunAttempt.taskRun.taskEventStore,
+ });
+ return "COMPLETED";
+}
+
+const eventRepository = resolveEventRepositoryForStore(
+ taskRunAttempt.taskRun.taskEventStore,
+ organizationId
+);Apply this consistently to the success, failure, and retry paths listed above.
Also applies to: 322-325, 547-550
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/v3/services/completeAttempt.server.ts` around lines 166 -
169, The current calls to resolveEventRepositoryForStore (and the two other
repository resolution calls at the success, failure, and retry paths) use a
fallback organizationId of "" via taskRunAttempt.taskRun.organizationId ?? "",
which silently masks missing tenant IDs; change each to explicitly guard for a
missing organizationId on taskRunAttempt.taskRun and throw or return an explicit
error instead of using "" — locate usages of resolveEventRepositoryForStore and
the analogous repository resolution functions near the success/failure/retry
paths (also repeated around the other occurrences) and replace the fallback with
a check like if (!taskRunAttempt.taskRun.organizationId) { throw new
Error("Missing organizationId for taskRunAttempt"); } before calling the
resolver so the resolver always receives a real organizationId.
Comment on lines
+1
to
+3
| import { describe, expect } from "vitest"; | ||
|
|
||
| vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} })); |
Contributor
There was a problem hiding this comment.
Missing vi import from vitest.
The test uses vi.mock and vi.setConfig but only imports describe and expect from vitest. Add vi to the import.
Proposed fix
-import { describe, expect } from "vitest";
+import { describe, expect, vi } from "vitest";📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| import { describe, expect } from "vitest"; | |
| vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} })); | |
| import { describe, expect, vi } from "vitest"; | |
| vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} })); |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/test/clickhouseFactory.test.ts` around lines 1 - 3, The test
imports only describe and expect from vitest but uses vi (e.g., vi.mock and
vi.setConfig); update the import statement in clickhouseFactory.test.ts to also
import vi from "vitest" so calls like vi.mock(...) and vi.setConfig(...)
reference the vitest mock API correctly.
6179dc7 to
66f294e
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (2)
apps/webapp/app/routes/admin.data-stores.tsx (2)
305-307:⚠️ Potential issue | 🟡 MinorClose these dialogs after submission completes, not during render.
Both components mutate state from render based on
fetcher.data?.success. Besides the React warning, the persisted success payload means the dialog will immediately close again on the next open. Gate the close behind a submit→idle transition in an effect/ref instead.🔧 One way to fix the pattern
-import { useState } from "react"; +import { useEffect, useRef, useState } from "react";const fetcher = useFetcher<{ success?: boolean; error?: string }>(); const isSubmitting = fetcher.state !== "idle"; + const wasSubmittingRef = useRef(false); - if (fetcher.data?.success && open) { - setOpen(false); - } + useEffect(() => { + const justFinished = wasSubmittingRef.current && fetcher.state === "idle"; + if (justFinished && fetcher.data?.success && open) { + setOpen(false); + } + wasSubmittingRef.current = fetcher.state !== "idle"; + }, [fetcher.state, fetcher.data?.success, open]);const fetcher = useFetcher<{ success?: boolean; error?: string }>(); const isSubmitting = fetcher.state !== "idle"; + const wasSubmittingRef = useRef(false); - // Close dialog on success - if (fetcher.data?.success && open) { - onOpenChange(false); - } + useEffect(() => { + const justFinished = wasSubmittingRef.current && fetcher.state === "idle"; + if (justFinished && fetcher.data?.success && open) { + onOpenChange(false); + } + wasSubmittingRef.current = fetcher.state !== "idle"; + }, [fetcher.state, fetcher.data?.success, open, onOpenChange]);Also applies to: 375-377
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/admin.data-stores.tsx` around lines 305 - 307, The render currently closes dialogs directly when fetcher.data?.success is truthy (e.g., the if (fetcher.data?.success && open) { setOpen(false); } lines), which must be moved into an effect that detects a submit→idle transition; implement a useEffect that tracks fetcher.state (with a ref for previous state) and when prevState === 'submitting' && fetcher.state === 'idle' && fetcher.data?.success is true, call setOpen(false); apply the same change to the other similar block (lines 375-377) so closing is only triggered after submission completes rather than during render or on subsequent re-renders.
31-31:⚠️ Potential issue | 🟡 MinorImport
tryCatchfrom the core utils subpath.This still pulls from the package root instead of the exported subpath used elsewhere in the webapp.
♻️ Proposed fix
-import { tryCatch } from "@trigger.dev/core"; +import { tryCatch } from "@trigger.dev/core/utils";As per coding guidelines, "Import from
@trigger.dev/coresubpaths only, never from the root."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/admin.data-stores.tsx` at line 31, Replace the root-package import of tryCatch with the core utils subpath: update the import statement that currently imports tryCatch from "@trigger.dev/core" to import from the package's utils subpath (e.g., "@trigger.dev/core/utils/tryCatch") so the tryCatch symbol is pulled from the exported utils subpath consistent with other imports.
🧹 Nitpick comments (5)
apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts (1)
15-17: Unused parameters in schema after refactoring to factory-based client.
keepAliveEnabled,keepAliveIdleSocketTtl, andmaxOpenConnectionsare parsed from the request but no longer used since the service now receivesclickhouseFactoryinstead of constructing its own ClickHouse client with these options.Consider removing these from the schema if they're no longer needed, or document if they're kept for future use.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts` around lines 15 - 17, Remove the now-unused keepAliveEnabled, keepAliveIdleSocketTtl, and maxOpenConnections fields from the Zod request schema used by the admin API runs-replication create route (the schema that currently parses those three fields) since the service now accepts clickhouseFactory instead of constructing a ClickHouse client; alternatively, if you intend to keep them for future use, add a clear comment above the schema explaining they are intentionally preserved and currently unused. Ensure any code that referenced these properties (parsing/validation logic for the route) is updated accordingly so no unused params remain.apps/webapp/app/v3/eventRepository/index.server.ts (1)
86-92: UseEVENT_STORE_TYPESconstants instead of string literals.Lines 86 and 90 use raw string literals
"clickhouse_v2"and"clickhouse"whileEVENT_STORE_TYPES.CLICKHOUSE_V2andEVENT_STORE_TYPES.CLICKHOUSEare defined and used elsewhere in this file (e.g., lines 64, 68). Consider using the constants for consistency.♻️ Proposed fix
- if (taskEventStore === "clickhouse_v2") { + if (taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE_V2) { return { repository: resolvedRepository, store: "clickhouse_v2" }; } - if (taskEventStore === "clickhouse") { + if (taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE) { return { repository: resolvedRepository, store: "clickhouse" }; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/v3/eventRepository/index.server.ts` around lines 86 - 92, Replace the raw string comparisons for taskEventStore with the existing EVENT_STORE_TYPES constants: change the conditions that check for "clickhouse_v2" and "clickhouse" to use EVENT_STORE_TYPES.CLICKHOUSE_V2 and EVENT_STORE_TYPES.CLICKHOUSE respectively, and return the same { repository: resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE_V2 } / { repository: resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE } so the code consistently uses the EVENT_STORE_TYPES enum instead of string literals.apps/webapp/app/presenters/v3/UsagePresenter.server.ts (1)
121-132: Remove unusedreplicaparameter.The
replicaparameter is declared but never used ingetTaskUsageByOrganization. Consider removing it to clean up the function signature.♻️ Proposed fix
async function getTaskUsageByOrganization( organizationId: string, startOfMonth: Date, - endOfMonth: Date, - replica: PrismaClientOrTransaction + endOfMonth: Date ) { const clickhouse = await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard");And update the call site at line 107-112:
const tasks = await getTaskUsageByOrganization( organizationId, startOfMonth, - endOfMonth, - this._replica + endOfMonth );🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/presenters/v3/UsagePresenter.server.ts` around lines 121 - 132, The function getTaskUsageByOrganization declares an unused parameter replica—remove replica from the function signature and any related type annotations so the function accepts (organizationId: string, startOfMonth: Date, endOfMonth: Date) only; update all call sites that pass a replica (e.g., where getTaskUsageByOrganization(...) is invoked) to stop supplying that argument, and run a quick search for references to getTaskUsageByOrganization to ensure callers and imports compile; no changes needed to the internal calls to clickhouseFactory.getClickhouseForOrganization or clickhouse.taskRuns.getTaskUsageByOrganization.apps/webapp/app/v3/runEngineHandlers.server.ts (2)
63-66: Inconsistent source fortaskEventStoreacross handlers.These handlers use
run.taskEventStorefrom the event payload, whilerunExpired(line 358) andrunCancelled(line 412) usetaskRun.taskEventStorefrom the database query. SincetaskRunis already fetched from the DB withtaskEventStoreselected, consider usingtaskRun.taskEventStoreconsistently across all handlers for a single source of truth.Suggested fix for consistency
const eventRepository = resolveEventRepositoryForStore( - run.taskEventStore, + taskRun.taskEventStore, taskRun.organizationId ?? organization.id );Apply this pattern to both
runSucceeded(lines 63-66) andrunFailed(lines 133-136).Also applies to: 133-136
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/v3/runEngineHandlers.server.ts` around lines 63 - 66, Use the DB-fetched taskRun.taskEventStore as the single source of truth instead of the event payload’s run.taskEventStore: in the runSucceeded and runFailed handlers replace references to run.taskEventStore with taskRun.taskEventStore (the same field used in runExpired and runCancelled) when calling resolveEventRepositoryForStore so all handlers consistently use taskRun.taskEventStore.
447-450: Consider extracting"taskEvent"to a named constant.The fallback
"taskEvent"appears to be a default store identifier. Per coding guidelines, sentinel/placeholder values should use named constants to avoid scattered magic strings and potential typos.Suggested refactor
+const DEFAULT_TASK_EVENT_STORE = "taskEvent"; + // ... in the handler: const eventRepository = resolveEventRepositoryForStore( - run.taskEventStore ?? "taskEvent", + run.taskEventStore ?? DEFAULT_TASK_EVENT_STORE, organization.id );🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/webapp/app/v3/runEngineHandlers.server.ts` around lines 447 - 450, Extract the magic fallback string "taskEvent" into a named constant (e.g., DEFAULT_TASK_EVENT_STORE or TASK_EVENT_STORE) and replace the inline literal in the call to resolveEventRepositoryForStore where run.taskEventStore ?? "taskEvent" is used; update any other usages of the same sentinel value to reference the constant to avoid duplicated magic strings and typos, and ensure the constant is exported/placed in a shared config/constants module if used across files.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@apps/webapp/app/routes/admin.data-stores.tsx`:
- Around line 91-94: The trimmed organizationIds array can end up empty (e.g.,
rawOrgIds=", , ") and should be rejected; after computing organizationIds from
rawOrgIds (the .split(",").map(s=>s.trim()).filter(Boolean) logic), validate
that organizationIds.length > 0 and return/throw a validation error (or add a
form error) if it’s zero so create/update is blocked; apply the same check to
the second occurrence of this logic (the block around lines 115-118) to ensure
normalized arrays are validated rather than the raw string.
---
Duplicate comments:
In `@apps/webapp/app/routes/admin.data-stores.tsx`:
- Around line 305-307: The render currently closes dialogs directly when
fetcher.data?.success is truthy (e.g., the if (fetcher.data?.success && open) {
setOpen(false); } lines), which must be moved into an effect that detects a
submit→idle transition; implement a useEffect that tracks fetcher.state (with a
ref for previous state) and when prevState === 'submitting' && fetcher.state ===
'idle' && fetcher.data?.success is true, call setOpen(false); apply the same
change to the other similar block (lines 375-377) so closing is only triggered
after submission completes rather than during render or on subsequent
re-renders.
- Line 31: Replace the root-package import of tryCatch with the core utils
subpath: update the import statement that currently imports tryCatch from
"@trigger.dev/core" to import from the package's utils subpath (e.g.,
"@trigger.dev/core/utils/tryCatch") so the tryCatch symbol is pulled from the
exported utils subpath consistent with other imports.
---
Nitpick comments:
In `@apps/webapp/app/presenters/v3/UsagePresenter.server.ts`:
- Around line 121-132: The function getTaskUsageByOrganization declares an
unused parameter replica—remove replica from the function signature and any
related type annotations so the function accepts (organizationId: string,
startOfMonth: Date, endOfMonth: Date) only; update all call sites that pass a
replica (e.g., where getTaskUsageByOrganization(...) is invoked) to stop
supplying that argument, and run a quick search for references to
getTaskUsageByOrganization to ensure callers and imports compile; no changes
needed to the internal calls to clickhouseFactory.getClickhouseForOrganization
or clickhouse.taskRuns.getTaskUsageByOrganization.
In `@apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts`:
- Around line 15-17: Remove the now-unused keepAliveEnabled,
keepAliveIdleSocketTtl, and maxOpenConnections fields from the Zod request
schema used by the admin API runs-replication create route (the schema that
currently parses those three fields) since the service now accepts
clickhouseFactory instead of constructing a ClickHouse client; alternatively, if
you intend to keep them for future use, add a clear comment above the schema
explaining they are intentionally preserved and currently unused. Ensure any
code that referenced these properties (parsing/validation logic for the route)
is updated accordingly so no unused params remain.
In `@apps/webapp/app/v3/eventRepository/index.server.ts`:
- Around line 86-92: Replace the raw string comparisons for taskEventStore with
the existing EVENT_STORE_TYPES constants: change the conditions that check for
"clickhouse_v2" and "clickhouse" to use EVENT_STORE_TYPES.CLICKHOUSE_V2 and
EVENT_STORE_TYPES.CLICKHOUSE respectively, and return the same { repository:
resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE_V2 } / { repository:
resolvedRepository, store: EVENT_STORE_TYPES.CLICKHOUSE } so the code
consistently uses the EVENT_STORE_TYPES enum instead of string literals.
In `@apps/webapp/app/v3/runEngineHandlers.server.ts`:
- Around line 63-66: Use the DB-fetched taskRun.taskEventStore as the single
source of truth instead of the event payload’s run.taskEventStore: in the
runSucceeded and runFailed handlers replace references to run.taskEventStore
with taskRun.taskEventStore (the same field used in runExpired and runCancelled)
when calling resolveEventRepositoryForStore so all handlers consistently use
taskRun.taskEventStore.
- Around line 447-450: Extract the magic fallback string "taskEvent" into a
named constant (e.g., DEFAULT_TASK_EVENT_STORE or TASK_EVENT_STORE) and replace
the inline literal in the call to resolveEventRepositoryForStore where
run.taskEventStore ?? "taskEvent" is used; update any other usages of the same
sentinel value to reference the constant to avoid duplicated magic strings and
typos, and ensure the constant is exported/placed in a shared config/constants
module if used across files.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 72fb2023-f0f8-4d55-80e6-70728d83f0fe
📒 Files selected for processing (79)
.cursor/mcp.json.server-changes/organization-scoped-clickhouse.mdCLAUDE.mdapps/webapp/app/env.server.tsapps/webapp/app/presenters/v3/ApiRunListPresenter.server.tsapps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.tsapps/webapp/app/presenters/v3/RunPresenter.server.tsapps/webapp/app/presenters/v3/RunTagListPresenter.server.tsapps/webapp/app/presenters/v3/SpanPresenter.server.tsapps/webapp/app/presenters/v3/TaskListPresenter.server.tsapps/webapp/app/presenters/v3/UsagePresenter.server.tsapps/webapp/app/presenters/v3/ViewSchedulePresenter.server.tsapps/webapp/app/presenters/v3/WaitpointPresenter.server.tsapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.$modelId/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsxapps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.test.tasks.$taskParam/route.tsxapps/webapp/app/routes/admin.api.v1.runs-replication.create.tsapps/webapp/app/routes/admin.api.v1.runs-replication.start.tsapps/webapp/app/routes/admin.data-stores.tsxapps/webapp/app/routes/admin.tsxapps/webapp/app/routes/api.v1.prompts.$slug.tsapps/webapp/app/routes/api.v1.prompts.$slug.versions.tsapps/webapp/app/routes/api.v1.prompts._index.tsapps/webapp/app/routes/api.v1.runs.$runId.events.tsapps/webapp/app/routes/api.v1.runs.$runId.spans.$spanId.tsapps/webapp/app/routes/api.v1.runs.$runId.trace.tsapps/webapp/app/routes/otel.v1.logs.tsapps/webapp/app/routes/otel.v1.metrics.tsapps/webapp/app/routes/otel.v1.traces.tsapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsxapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.tsapps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts.$promptSlug.generations.tsapps/webapp/app/routes/resources.runs.$runParam.logs.download.tsapps/webapp/app/runEngine/concerns/traceEvents.server.tsapps/webapp/app/runEngine/services/triggerFailedTask.server.tsapps/webapp/app/services/admin/missingLlmModels.server.tsapps/webapp/app/services/clickhouse/clickhouseFactory.server.tsapps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.tsapps/webapp/app/services/clickhouseInstance.server.tsapps/webapp/app/services/dataStores/organizationDataStoreConfigSchemas.server.tsapps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.tsapps/webapp/app/services/dataStores/organizationDataStoresRegistryInstance.server.tsapps/webapp/app/services/queryService.server.tsapps/webapp/app/services/runsReplicationInstance.server.tsapps/webapp/app/services/runsReplicationService.server.tsapps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.tsapps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.tsapps/webapp/app/v3/eventRepository/eventRepository.server.tsapps/webapp/app/v3/eventRepository/eventRepository.types.tsapps/webapp/app/v3/eventRepository/index.server.tsapps/webapp/app/v3/otlpExporter.server.tsapps/webapp/app/v3/runEngineHandlers.server.tsapps/webapp/app/v3/services/alerts/errorAlertEvaluator.server.tsapps/webapp/app/v3/services/bulk/BulkActionV2.server.tsapps/webapp/app/v3/services/cancelTaskRunV1.server.tsapps/webapp/app/v3/services/completeAttempt.server.tsapps/webapp/app/v3/services/crashTaskRun.server.tsapps/webapp/app/v3/services/expireEnqueuedRun.server.tsapps/webapp/app/v3/services/triggerTaskV1.server.tsapps/webapp/test/clickhouseFactory.test.tsapps/webapp/test/organizationDataStoresRegistry.test.tsapps/webapp/test/runsBackfiller.test.tsapps/webapp/test/runsReplicationBenchmark.test.tsapps/webapp/test/runsReplicationService.part1.test.tsapps/webapp/test/runsReplicationService.part2.test.tsapps/webapp/test/utils/replicationUtils.tsapps/webapp/test/utils/testReplicationClickhouseFactory.tsapps/webapp/test/utils/tracing.tsinternal-packages/database/prisma/migrations/20260331212308_add_organization_data_stores/migration.sqlinternal-packages/database/prisma/schema.prisma
💤 Files with no reviewable changes (2)
- apps/webapp/app/services/clickhouseInstance.server.ts
- apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts
✅ Files skipped from review due to trivial changes (18)
- apps/webapp/app/v3/services/expireEnqueuedRun.server.ts
- apps/webapp/app/routes/admin.tsx
- CLAUDE.md
- .server-changes/organization-scoped-clickhouse.md
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models._index/route.tsx
- apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors._index/route.tsx
- apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts
- apps/webapp/app/v3/services/completeAttempt.server.ts
- apps/webapp/test/runsReplicationService.part1.test.ts
- apps/webapp/app/env.server.ts
- apps/webapp/app/services/clickhouse/clickhouseSecretSchemas.server.ts
- apps/webapp/test/utils/testReplicationClickhouseFactory.ts
- internal-packages/database/prisma/migrations/20260331212308_add_organization_data_stores/migration.sql
- internal-packages/database/prisma/schema.prisma
- apps/webapp/test/organizationDataStoresRegistry.test.ts
- apps/webapp/app/services/dataStores/organizationDataStoreConfigSchemas.server.ts
- apps/webapp/test/runsReplicationService.part2.test.ts
🚧 Files skipped from review as they are similar to previous changes (34)
- .cursor/mcp.json
- apps/webapp/app/routes/admin.api.v1.runs-replication.start.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.models.compare/route.tsx
- apps/webapp/app/routes/otel.v1.traces.ts
- apps/webapp/app/v3/services/triggerTaskV1.server.ts
- apps/webapp/app/services/queryService.server.ts
- apps/webapp/app/routes/otel.v1.metrics.ts
- apps/webapp/test/utils/tracing.ts
- apps/webapp/test/utils/replicationUtils.ts
- apps/webapp/app/runEngine/concerns/traceEvents.server.ts
- apps/webapp/app/routes/api.v1.runs.$runId.events.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.prompts._index/route.tsx
- apps/webapp/app/presenters/v3/WaitpointPresenter.server.ts
- apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts
- apps/webapp/app/routes/api.v1.runs.$runId.trace.ts
- apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx
- apps/webapp/app/routes/api.v1.runs.$runId.spans.$spanId.ts
- apps/webapp/test/runsBackfiller.test.ts
- apps/webapp/app/routes/api.v1.prompts._index.ts
- apps/webapp/app/services/runsReplicationInstance.server.ts
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboards.$dashboardKey/route.tsx
- apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.errors.$fingerprint/route.tsx
- apps/webapp/app/routes/otel.v1.logs.ts
- apps/webapp/app/presenters/v3/RunTagListPresenter.server.ts
- apps/webapp/app/v3/eventRepository/eventRepository.types.ts
- apps/webapp/app/v3/eventRepository/eventRepository.server.ts
- apps/webapp/app/presenters/v3/RunPresenter.server.ts
- apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts
- apps/webapp/app/services/runsReplicationService.server.ts
- apps/webapp/app/services/admin/missingLlmModels.server.ts
- apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts
- apps/webapp/app/v3/services/crashTaskRun.server.ts
- apps/webapp/app/presenters/v3/TaskListPresenter.server.ts
- apps/webapp/app/services/dataStores/organizationDataStoresRegistry.server.ts
Comment on lines
+91
to
+94
| const organizationIds = rawOrgIds | ||
| .split(",") | ||
| .map((s) => s.trim()) | ||
| .filter(Boolean); |
Contributor
There was a problem hiding this comment.
Reject org lists that collapse to zero IDs after trimming.
Values like ", , " pass the current schema, become [], and then create/update an unusable data store with no organizations attached. Validate the normalized array, not just the raw string.
🛡️ Proposed fix
const organizationIds = rawOrgIds
.split(",")
.map((s) => s.trim())
.filter(Boolean);
+
+ if (organizationIds.length === 0) {
+ return typedjson({ error: "At least one organization ID is required" }, { status: 400 });
+ }Also applies to: 115-118
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/webapp/app/routes/admin.data-stores.tsx` around lines 91 - 94, The
trimmed organizationIds array can end up empty (e.g., rawOrgIds=", , ") and
should be rejected; after computing organizationIds from rawOrgIds (the
.split(",").map(s=>s.trim()).filter(Boolean) logic), validate that
organizationIds.length > 0 and return/throw a validation error (or add a form
error) if it’s zero so create/update is blocked; apply the same check to the
second occurrence of this logic (the block around lines 115-118) to ensure
normalized arrays are validated rather than the raw string.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
Comment on lines
+80
to
+84
| const taskEventStore = parentStore ?? (await resolveTaskEventRepositoryFlag(featureFlags)); | ||
| const { repository: resolvedRepository } = await clickhouseFactory.getEventRepositoryForOrganization( | ||
| taskEventStore, | ||
| organizationId | ||
| ); |
Contributor
There was a problem hiding this comment.
🔴 getEventRepository, getV3EventRepository, and getConfiguredEventRepository throw for "postgres" event store
The refactored getEventRepository (line 81), getV3EventRepository (line 102), and getConfiguredEventRepository (line 59) in index.server.ts unconditionally call clickhouseFactory.getEventRepositoryForOrganization(taskEventStore, organizationId) before checking whether the store is a ClickHouse variant. This call chain reaches buildEventRepository(store, ...) at clickhouseFactory.server.ts:336, which only handles "clickhouse" and "clickhouse_v2" — the default case throws Error: Unknown ClickHouse event repository store: postgres. Since the default for EVENT_REPOSITORY_DEFAULT_STORE is "postgres", this crashes all task triggering, trace event recording, and Logs page loads for any environment using the postgres event store. The old code checked the store type first and returned the Prisma-based eventRepository without going through the ClickHouse factory.
Prompt for agents
The functions `getEventRepository` (line 75), `getV3EventRepository` (line 97), and `getConfiguredEventRepository` (line 34) in `apps/webapp/app/v3/eventRepository/index.server.ts` all unconditionally call `clickhouseFactory.getEventRepositoryForOrganization(taskEventStore, organizationId)` before checking whether the resolved store is a clickhouse variant. When `taskEventStore` is "postgres", this reaches `buildEventRepository` in `apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts:336` which throws `Error: Unknown ClickHouse event repository store: postgres`.
The fix is to guard the factory call behind a clickhouse store check, similar to how `resolveEventRepositoryForStore` (line 24-32) already does it correctly:
1. In `getEventRepository`: check if `taskEventStore` is "clickhouse" or "clickhouse_v2" before calling the factory. If not, return `{ repository: eventRepository, store: getTaskEventStore() }` directly.
2. In `getV3EventRepository`: when `parentStore` is a string, check if it is a clickhouse variant before calling the factory. If not, return `{ repository: eventRepository, store: getTaskEventStore() }` directly.
3. In `getConfiguredEventRepository`: similarly guard the factory call.
Alternatively, `buildEventRepository` could handle "postgres" by returning a no-op or throwing a more descriptive error, but the proper fix is to not route postgres stores through the ClickHouse factory at all.
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
+248
to
+253
| const clickhouseClient = await clickhouseFactory.getClickhouseForOrganization( | ||
| environment.organizationId, | ||
| "query" | ||
| ); | ||
|
|
||
| const presenter = new ErrorGroupPresenter($replica, clickhouseClient, clickhouseClient); |
Contributor
There was a problem hiding this comment.
🔴 Error fingerprint route passes "query" client for both ErrorGroupPresenter constructor parameters, losing logs-specific ClickHouse settings
The ErrorGroupPresenter constructor at ErrorGroupPresenter.server.ts:96-99 expects two distinct ClickHouse clients: logsClickhouse (used for errors.listQueryBuilder(), errors.createOccurrencesByVersionQueryBuilder(), etc.) and clickhouse (used for NextRunListPresenter). The old code correctly passed logsClickhouseClient (with special settings like max_memory_usage, max_bytes_before_external_sort, max_threads) as the first parameter. The new code passes the same "query" client for both, losing all logs-specific query protections. The getDefaultLogsClickhouseClient function is even imported but never used, confirming this was an oversight.
Affected code
Line 253: const presenter = new ErrorGroupPresenter($replica, clickhouseClient, clickhouseClient);
Should be:
const logsClickhouse = await clickhouseFactory.getClickhouseForOrganization(environment.organizationId, "logs");
const queryClickhouse = await clickhouseFactory.getClickhouseForOrganization(environment.organizationId, "query");
const presenter = new ErrorGroupPresenter($replica, logsClickhouse, queryClickhouse);
Suggested change
| const clickhouseClient = await clickhouseFactory.getClickhouseForOrganization( | |
| environment.organizationId, | |
| "query" | |
| ); | |
| const presenter = new ErrorGroupPresenter($replica, clickhouseClient, clickhouseClient); | |
| const logsClickhouse = await clickhouseFactory.getClickhouseForOrganization( | |
| environment.organizationId, | |
| "logs" | |
| ); | |
| const queryClickhouse = await clickhouseFactory.getClickhouseForOrganization( | |
| environment.organizationId, | |
| "query" | |
| ); | |
| const presenter = new ErrorGroupPresenter($replica, logsClickhouse, queryClickhouse); |
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
+10
to
+12
| registry.loadFromDatabase().catch((err) => { | ||
| console.error("[OrganizationDataStoresRegistry] Failed to initialize", err); | ||
| }); |
Contributor
There was a problem hiding this comment.
🔴 isReady promise never resolves if initial loadFromDatabase() fails, causing service-wide hangs
In organizationDataStoresRegistryInstance.server.ts:10-12, if loadFromDatabase() throws (e.g., database unavailable at startup), the error is caught and logged but _readyResolve() in organizationDataStoresRegistry.server.ts:82 is never called. This means registry.isReady never resolves. Any code that calls await clickhouseFactory.isReady() (clickhouseFactory.server.ts:194-196) will hang forever — this includes OTLP exporter initialization (otlpExporter.server.ts:1208), runs replication start (runsReplicationInstance.server.ts:73), and the admin replication create route (admin.api.v1.runs-replication.create.ts:65). OTEL data ingestion and runs replication would become permanently unresponsive after a transient DB failure during startup.
Suggested change
| registry.loadFromDatabase().catch((err) => { | |
| console.error("[OrganizationDataStoresRegistry] Failed to initialize", err); | |
| }); | |
| registry.loadFromDatabase().catch((err) => { | |
| console.error("[OrganizationDataStoresRegistry] Failed to initialize, resolving with empty state", err); | |
| // Resolve isReady even on failure so dependent code doesn't hang forever. | |
| // The registry will return null for all orgs (defaulting to standard ClickHouse). | |
| if (!registry.isLoaded) { | |
| (registry as any)._loaded = true; | |
| (registry as any)._readyResolve(); | |
| } | |
| }); |
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
+164
to
+179
| function buildOrgClickhouseClient(url: string, clientType: ClientType): ClickHouse { | ||
| const parsed = new URL(url); | ||
| parsed.searchParams.delete("secure"); | ||
|
|
||
| return new ClickHouse({ | ||
| url: parsed.toString(), | ||
| name: `org-clickhouse-${clientType}`, | ||
| keepAlive: { | ||
| enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1", | ||
| idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS, | ||
| }, | ||
| logLevel: env.CLICKHOUSE_LOG_LEVEL, | ||
| compression: { request: true }, | ||
| maxOpenConnections: env.CLICKHOUSE_MAX_OPEN_CONNECTIONS, | ||
| }); | ||
| } |
Contributor
There was a problem hiding this comment.
🚩 Org-specific ClickHouse clients don't inherit specialized settings from default clients
The buildOrgClickhouseClient function at clickhouseFactory.server.ts:164-179 creates ClickHouse clients for org-specific data stores using generic settings (the standard CLICKHOUSE_* env vars). It does NOT replicate specialized settings that default clients have — for example, the logs client has max_memory_usage, max_bytes_before_external_sort, max_threads settings (clickhouseFactory.server.ts:58-69), and the replication client uses RUN_REPLICATION_* settings (clickhouseFactory.server.ts:140-150). When an org has a dedicated ClickHouse and a caller requests a "logs" or "replication" client type, they get a generic client without these query protections. This may be intentional (org-dedicated ClickHouse may not need the same guardrails), but could lead to surprising behavior differences between default and org-specific paths.
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
+94
to
+118
| async addDataStore({ | ||
| key, | ||
| kind, | ||
| organizationIds, | ||
| config, | ||
| }: { | ||
| key: string; | ||
| kind: DataStoreKind; | ||
| organizationIds: string[]; | ||
| config: any; | ||
| }) { | ||
| const secretKey = this.#secretKey(key, kind); | ||
|
|
||
| const secretStore = getSecretStore("DATABASE", { prismaClient: this._prisma }); | ||
| await secretStore.setSecret(secretKey, config); | ||
|
|
||
| return this._prisma.organizationDataStore.create({ | ||
| data: { | ||
| key, | ||
| organizationIds, | ||
| kind: "CLICKHOUSE", | ||
| config: { version: 1, data: { secretKey } }, | ||
| }, | ||
| }); | ||
| } |
Contributor
There was a problem hiding this comment.
🚩 addDataStore/updateDataStore don't call reload() — in-memory registry is stale until next periodic reload
After addDataStore, updateDataStore, or deleteDataStore in organizationDataStoresRegistry.server.ts, the in-memory _lookup map is NOT updated. The changes only persist to the database. The registry won't reflect changes until the next periodic reload interval (default 60s per ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS). For the admin UI at admin.data-stores.tsx, this means an admin could add a data store and for up to 60 seconds, traffic for the affected orgs would still route to the default ClickHouse. The admin page doesn't call reload() after mutations either. This is likely acceptable for the initial implementation but could surprise operators expecting immediate effect.
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
+186
to
+189
| /** ClickHouse clients keyed by hostname hash + clientType. */ | ||
| private readonly _clientCache = new Map<string, ClickHouse>(); | ||
| /** Event repositories keyed by hostname hash (stateful, must be reused). */ | ||
| private readonly _eventRepositoryCache = new Map<string, ClickhouseEventRepository>(); |
Contributor
There was a problem hiding this comment.
🚩 Client cache in ClickhouseFactory is never invalidated when data stores change
The _clientCache and _eventRepositoryCache in ClickhouseFactory (clickhouseFactory.server.ts:187-189) are populated when org-specific ClickHouse clients are first created but are never cleared when the underlying OrganizationDataStoresRegistry reloads. If an org's data store URL changes (e.g., during a ClickHouse migration), the old cached client/repository would continue to be used until the process restarts. The registry reload at organizationDataStoresRegistryInstance.server.ts:14-17 updates the _lookup map, but the factory's caches are separate. This is a cache staleness concern for long-running processes.
Was this helpful? React with 👍 or 👎 to provide feedback.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Added
OrganizationDataStorewhich allows orgs to have data stored in specific separate services.For now this is just used for ClickHouse. When using ClickHouse we get a client for the factory and pass in the org id.
Particular care has to be made with two hot-insert paths: