Postgres db node - #71
Merged
Merged
Conversation
Contributor
📝 WalkthroughWalkthroughAdds two new node types (AGGREGATE, POSTGRES) with DB migrations and Prisma schema changes; implements full aggregate and Postgres subsystems (types, engines, executors, UI dialogs/nodes, realtime channels, TRPC routers, tests), credential UI updates, and related wiring across node registration and executor registry. Changes
Sequence Diagram(s)sequenceDiagram
participant User as User (UI)
participant Node as Node (Aggregate/Postgres)
participant Dialog as Config Dialog
participant TRPC as TRPC Router
participant DB as Prisma DB
participant Realtime as Inngest Realtime
User->>Node: Open settings (click/dblclick)
Node->>Dialog: Open with nodeId
Dialog->>TRPC: getByNodeId(nodeId)
TRPC->>DB: Read node config (includes workflow.userId)
DB-->>TRPC: Node config
TRPC-->>Dialog: Return config
User->>Dialog: Submit config
Dialog->>TRPC: upsert(nodeId, config)
TRPC->>DB: Upsert node row
DB-->>TRPC: Success
TRPC-->>Dialog: Confirm saved
Dialog->>Node: Update node.data
Node->>Realtime: Request token (getToken)
Realtime-->>Node: Token / status events
sequenceDiagram
participant Context as Execution Context
participant Executor as Aggregate/Postgres Executor
participant Prisma as Prisma DB
participant Engine as Engine (Aggregate/Postgres)
participant Realtime as Inngest Realtime
Context->>Executor: Execute node (nodeId, context)
Executor->>Prisma: Load node config (+ credential)
Prisma-->>Executor: Config (+ credential if any)
Executor->>Realtime: Publish "loading"
Executor->>Engine: Dispatch operation (query/aggregate)
Engine->>Prisma: Run SQL or compute aggregates (if needed)
Prisma-->>Engine: Result rows / data
Engine-->>Executor: Computed result
Executor->>Realtime: Publish "success"
Executor-->>Context: Return { variableName: result }
alt Error
Executor->>Realtime: Publish "error"
Executor-->>Context: Throw or return failure (continueOnFail)
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~110 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
⚔️ Resolve merge conflicts
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 |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e681cbaa3a
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| database: values.postgresDatabase, | ||
| user: values.postgresUser, | ||
| password: values.postgresPassword, | ||
| ssl: values.postgresSsl, |
There was a problem hiding this comment.
The credential form writes ssl as a boolean, but createConnection in postgres-engine.ts expects string modes ("disable" | "require" | "verify-full"). With the current payload, both true and false miss the first two branches and are treated like verify-full, so non-SSL/local databases and many self-signed setups will fail to connect even when users select the non-SSL option.
Useful? React with 👍 / 👎.
Comment on lines
+142
to
+146
| const payload = { | ||
| nodeId, | ||
| workflowId, | ||
| operation: values.operation as PostgresOperation, | ||
| credentialId: values.credentialId || null, |
There was a problem hiding this comment.
The dialog exposes operations like UPSERT, INSERT_MANY, and EXECUTE_TRANSACTION, but the submit payload only saves a small common subset and omits required fields such as conflictColumns, insertManyPath, and transactionStatements. Those values then fall back to router defaults ("[]"/empty), causing these operations to fail at execution time with validation errors, making multiple advertised operations unusable from the UI.
Useful? React with 👍 / 👎.
Comment on lines
+8
to
+12
| function quote(identifier: string): string { | ||
| if (identifier.includes(".")) { | ||
| return identifier.split(".").map(p => `"${p}"`).join(".") | ||
| } | ||
| return `"${identifier}"` |
There was a problem hiding this comment.
Identifier quoting currently just wraps input in double quotes without escaping internal " characters. If a table/column/schema name contains a quote, the generated SQL can break out of the identifier (syntax breakage or statement injection via appended SQL), so identifier handling is not actually safe despite being routed through quote().
Useful? React with 👍 / 👎.
![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: 13
Note
Due to the large number of review comments, Critical severity comments were prioritized as inline comments.
🟠 Major comments (14)
src/features/executions/components/postgres/dialog.tsx-140-169 (1)
140-169:⚠️ Potential issue | 🟠 MajorIncomplete payload may reset advanced configuration fields.
The
onFormSubmitpayload only includes basic fields. If a user configures advanced options (JOINs, orderBy, function args, JSON operations, etc.) through other means or API, saving from this dialog will reset those fields to defaults since they're not preserved in the payload.Either include all fields from
dbConfigin the payload, or explicitly only send fields that were actually modified.💡 Proposed fix: Preserve existing values
const onFormSubmit = async (values: PostgresFormValues) => { try { const payload = { nodeId, workflowId, operation: values.operation as PostgresOperation, credentialId: values.credentialId || null, tableName: values.tableName, schemaName: values.schemaName, variableName: values.variableName, query: values.query, queryParams: values.queryParams, selectColumns: values.selectColumns || "[]", whereConditions: values.whereConditions || "[]", insertData: values.insertData, updateData: values.updateData, limitRows: Number(values.limitRows), offsetRows: Number(values.offsetRows), returnData: values.returnData, continueOnFail: values.continueOnFail, + // Preserve existing advanced config + orderBy: (dbConfig?.orderBy as string) ?? "[]", + joins: (dbConfig?.joins as string) ?? "[]", + conflictColumns: (dbConfig?.conflictColumns as string) ?? "[]", + updateOnConflict: (dbConfig?.updateOnConflict as string) ?? "[]", + insertManyPath: (dbConfig?.insertManyPath as string) ?? "", + insertManyColumns: (dbConfig?.insertManyColumns as string) ?? "[]", + transactionStatements: (dbConfig?.transactionStatements as string) ?? "[]", + functionName: (dbConfig?.functionName as string) ?? "", + functionArgs: (dbConfig?.functionArgs as string) ?? "[]", + // ... other advanced fields }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/dialog.tsx` around lines 140 - 169, The onFormSubmit handler builds a payload with only basic fields which will overwrite and drop advanced DB configuration; modify onFormSubmit to merge the current dbConfig (or the form's full values object that contains advanced settings) into the payload before calling upsertMutation.mutateAsync so existing advanced fields (JOINs, orderBy, functionArgs, JSON ops, etc.) are preserved—e.g., locate onFormSubmit and the payload variable and construct the payload by spreading dbConfig (or fullValues) first and then the explicit overrides (nodeId, workflowId, operation, credentialId, etc.), making sure selectColumns/whereConditions and other optional arrays keep their existing values when empty.src/server/routers/postgres.router.ts-134-142 (1)
134-142:⚠️ Potential issue | 🟠 MajorMissing authorization check for realtime token generation.
The
getTokenprocedure generates a subscription token for anynodeIdwithout verifying the authenticated user owns the node. This could allow any authenticated user to subscribe to realtime status updates for any Postgres node.Compare to
getByNodeId(lines 30-31) anddelete(lines 128-129) which properly checknode.workflow.userId !== ctx.auth.user.id.🔒 Proposed fix: Add authorization check
getToken: protectedProcedure .input(z.object({ nodeId: z.string() })) .query(async ({ input, ctx }) => { + const node = await prisma.postgresNode.findUnique({ + where: { nodeId: input.nodeId }, + include: { workflow: { select: { userId: true } } }, + }) + if (!node) { + throw new TRPCError({ code: "NOT_FOUND" }) + } + if (node.workflow.userId !== ctx.auth.user.id) { + throw new TRPCError({ code: "UNAUTHORIZED" }) + } + const token = await getSubscriptionToken(inngest, { channel: postgresChannelName(input.nodeId), topics: ["status"], }) return { token } }),🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/routers/postgres.router.ts` around lines 134 - 142, getToken currently returns a realtime subscription token for any nodeId without verifying ownership; fix by loading the Postgres node (same way getByNodeId/delete do) and verify node.workflow.userId === ctx.auth.user.id before calling getSubscriptionToken and returning the token, and throw a forbidden/unauthorized error if the check fails; ensure you reference getToken, getSubscriptionToken, postgresChannelName, and the node.workflow.userId === ctx.auth.user.id check so the authorization logic mirrors getByNodeId/delete.src/server/routers/aggregate.router.ts-111-119 (1)
111-119:⚠️ Potential issue | 🟠 MajorGate realtime tokens by node ownership.
Lines 113-118 mint a subscription token for any
nodeIdwithout checking whether that node belongs toctx.auth.user.id. Any authenticated user who learns another node id can subscribe to that channel's status events.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/routers/aggregate.router.ts` around lines 111 - 119, getToken currently mints subscription tokens for any nodeId without verifying ownership; update the protectedProcedure query signature to accept ctx (i.e., async ({ input, ctx }) =>), lookup the node by input.nodeId (using your data access layer/Prisma) and verify its ownerId equals ctx.auth.user.id, throw an authorization error if not the owner, and only then call getSubscriptionToken with aggregateChannelName(input.nodeId) to return the token.src/server/routers/aggregate.router.ts-9-16 (1)
9-16:⚠️ Potential issue | 🟠 MajorValidate
multiOpsandgroupAggOpsagainst the operation schema before saving.Lines 78-85 only prove the strings are JSON. Payloads like
"{}"or"oops"will still persist, but the dialog later treats both fields asAggregateOp[]and calls.map()on them. The schema on Lines 9-16 is already here—usez.array(aggregateOpSchema)instead of syntax-only parsing.Also applies to: 76-86
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/routers/aggregate.router.ts` around lines 9 - 16, The multiOps and groupAggOps JSON string checks currently only verify they are JSON strings but not that their contents match the aggregate shape; validate them against the existing aggregateOpSchema by using z.array(aggregateOpSchema) (e.g., parse the incoming value with z.array(aggregateOpSchema).safeParse(...) or use z.preprocess to parse JSON then validate) instead of the current syntax-only parsing, and if validation fails return/throw an error before persisting; update the code paths that read multiOps and groupAggOps (where .map() is later called) to use the validated result so you never call .map() on invalid payloads.src/features/executions/components/postgres/types.ts-137-143 (1)
137-143:⚠️ Potential issue | 🟠 Major
CREATE_TABLEshould be inTABLE_OPS.The comment on Line 137 says this list covers operations that require
tableName, but Lines 138-143 omitCREATE_TABLEeven thoughsrc/features/executions/components/postgres/executor.ts:558-561rejects that operation without one. Any caller that relies onTABLE_OPSwill now classifyCREATE_TABLEincorrectly.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/types.ts` around lines 137 - 143, TABLE_OPS currently omits the CREATE_TABLE operation, causing callers to misclassify operations that require a tableName; add "CREATE_TABLE" to the TABLE_OPS array so CREATE_TABLE is treated as a table-name-required operation (update the TABLE_OPS constant where it's defined and ensure it now includes "CREATE_TABLE" alongside "DROP_TABLE" and the other PostgresOperation entries).src/features/executions/components/aggregate/dialog.tsx-57-78 (1)
57-78:⚠️ Potential issue | 🟠 MajorResync
opswhen thevalueprop changes.Lines 64-70 read
valueonly once. When saved config arrives from the query or the form is reset,MultiOpBuilderkeeps rendering its stale local array, so existingmultiOpsandgroupAggOpsdisappear in the editor.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/aggregate/dialog.tsx` around lines 57 - 78, MultiOpBuilder currently initializes its local ops state from the value prop only once (useState initializer) so it becomes stale when value changes; add a useEffect watching value that parses JSON from value (handling parse errors) and calls setOps(parsed) when the parsed array differs from current ops to resync the local state, while keeping the existing update callback (setOps and onChange) intact; reference the component MultiOpBuilder, state ops/setOps, props value/onChange, and the update callback to locate where to add the effect.src/features/executions/components/aggregate/dialog.tsx-224-258 (1)
224-258:⚠️ Potential issue | 🟠 MajorReset the form after
dbConfigresolves.Lines 233-257 recompute
merged, butuseFormonly consumesdefaultValueson the first render. BecausedbConfigis loaded asynchronously on Lines 224-229, persisted node settings never populate unless you callform.reset(...)when the query result changes.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/aggregate/dialog.tsx` around lines 224 - 258, The form's defaultValues are only applied on first render so asynchronous dbConfig never populates the fields; after trpc.aggregate.getByNodeId returns (dbConfig) compute the same merged defaults and call form.reset(...) to update the form state; add a useEffect that watches dbConfig and calls form.reset with the merged/defaulted AggregateNodeData (use the same merging logic used to build merged) so the useForm<AggregateNodeData> instance named form reflects persisted node settings when the query resolves.src/features/credentials/components/credential.tsx-57-62 (1)
57-62:⚠️ Potential issue | 🟠 MajorModel PostgreSQL SSL mode as the engine's string enum, not a boolean.
Line 62, Line 604, Line 790, and Lines 1669-1680 all treat SSL as
true/false, butsrc/features/executions/components/postgres/postgres-engine.ts:3-23branches on"disable" | "require" | "verify-full", andsrc/server/routers/postgres.router.ts:169-170passes the stored JSON straight through. With the current shape, both saved boolean values fall into the engine's fallback branch, so SSL is misconfigured during connection tests and executions.Also applies to: 594-610, 783-791, 1663-1687
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/credentials/components/credential.tsx` around lines 57 - 62, The credential schema currently models postgresSsl as a boolean which causes the engine (postgres-engine.ts) to fall back instead of using the expected string modes ("disable"|"require"|"verify-full"); change postgresSsl in the Zod schema (the symbol postgresSsl in credential.ts) from z.boolean().optional() to a string enum matching the engine's modes (e.g., z.enum(["disable","require","verify-full"]).optional()), update any UI form serialization/deserialization and defaulting logic that writes/reads postgresSsl so it stores one of those strings (not true/false), and ensure any code paths that previously assumed a boolean (connection test, execution setup, and server router passthrough) convert or validate values to the new enum shape before sending to postgres-engine.ts.src/features/executions/components/postgres/postgres-engine.ts-61-78 (1)
61-78:⚠️ Potential issue | 🟠 Major
maxRowsdoesn't actually cap memory usage.Lines 76-78 truncate only after
client.query()has already fetched the full result set. A large result set can still be fully materialized in memory before this slice runs, so this does not protect the worker from OOM.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/postgres-engine.ts` around lines 61 - 78, The current executeQuery implementation slices result.rows after client.query(), which still materializes the entire result set; change executeQuery to fetch rows incrementally using a server-side cursor/stream (e.g., pg-cursor or pg-query-stream) instead of client.query(), read up to maxRows from the cursor/stream into the rows array, close the cursor when done, and set rowCount to either the number of rows read or, if you need the true total, run a separate COUNT(*) query; replace the client.query usage in executeQuery and remove the post-query slice to ensure large results are never fully loaded into memory.src/features/executions/components/postgres/postgres-engine.ts-3-12 (1)
3-12:⚠️ Potential issue | 🟠 MajorDefault missing timeout fields before building the client config.
Line 11 and Line 12 make both timeouts required, but the new credential serializer in
src/features/credentials/components/credential.tsx:783-791stores only host/port/database/user/password/ssl. On those credentials, Lines 32-33 multiplyundefined, so every connection and test call is built with invalid timeout values unless you normalize defaults here.Also applies to: 25-33
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/postgres-engine.ts` around lines 3 - 12, PostgresConnectionConfig currently requires connectionTimeout and queryTimeout but credentials may omit them, causing multiplication of undefined when building the client; default the timeouts before constructing the client config (e.g., in the code that maps PostgresConnectionConfig into client options where you do connectionTimeout * 1000 or queryTimeout * 1000). Specifically, in the code that reads a PostgresConnectionConfig and builds the client (the function that multiplies timeouts into milliseconds), normalize with defaults like const connectionTimeoutSec = config.connectionTimeout ?? 10 and const queryTimeoutSec = config.queryTimeout ?? 30 (or other project-standard defaults) and use those variables in place of config.connectionTimeout/config.queryTimeout so the client always receives valid numeric timeout values.src/features/executions/components/postgres/types.ts-1-24 (1)
1-24:⚠️ Potential issue | 🟠 MajorDon't expose
COPY_FROMuntil the executor handles it.Line 19 and Line 129 advertise
COPY_FROMas a supported operation, butsrc/features/executions/components/postgres/executor.ts:71-780has no matching case and will fall through to the"Unknown operation"error. Either implement it or remove it from the public operation set for now.Also applies to: 111-135
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/types.ts` around lines 1 - 24, The PostgresOperation union currently lists "COPY_FROM" but the executor's operation dispatcher does not handle it, causing an "Unknown operation" error; either remove "COPY_FROM" from the PostgresOperation type or implement a matching handler in the Postgres executor switch/dispatcher. Locate the PostgresOperation type and either delete the "COPY_FROM" variant, or add a case for "COPY_FROM" in the executor's operation switch (the function that routes Postgres operations) that performs the expected copy-from logic or throws a clear not-yet-implemented error until full support is added.src/features/executions/components/postgres/query-builder.ts-65-70 (1)
65-70:⚠️ Potential issue | 🟠 MajorDirect interpolation of
j.onrequires caller to ensure sanitization.The
onclause is interpolated directly into SQL, which is necessary since join conditions contain column references that cannot be parameterized. However, ifj.onoriginates from user input, this is a SQL injection vector.Consider adding validation to ensure
ononly contains expected patterns (identifiers, operators, and literals), or document this as a caller responsibility.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/query-builder.ts` around lines 65 - 70, buildJoinClause currently interpolates the join condition via j.on with no validation, which can lead to SQL injection if caller-supplied; update buildJoinClause (or a helper used by it) to validate j.on against a strict allowed-pattern (e.g., only identifiers, dot notation, whitespace, allowed operators (=, <, >, <=, >=, !=, AND, OR, LIKE), numeric/string literals, and parentheses) and throw an error if it fails, or alternatively add a clear documented precondition on the JoinClause type requiring the caller to supply a pre-validated/safe on expression; reference the buildJoinClause function and the JoinClause.on property when adding the validation or documentation.src/features/executions/components/postgres/query-builder.ts-251-265 (1)
251-265: 🛠️ Refactor suggestion | 🟠 MajorDead code:
innerSqlvariable is defined but never used.Line 253 defines
innerSqlwhich is immediately overwritten at line 260. This appears to be leftover from a refactoring attempt. The comments on lines 255-257 suggest the original approach was abandoned.🧹 Proposed fix to remove dead code
export function buildJsonPathQuery(options: { schema: string; table: string; column: string; jsonPath: string; where: WhereCondition[] }): BuiltQuery { const tableRef = `${quote(options.schema)}.${quote(options.table)}` - let innerSql = `SELECT *, ${quote(options.column)} #>> $1::text[] AS json_value FROM ${tableRef} WHERE ${quote(options.column)} @? $2::jsonpath` - - // Actually, `#>>` expects text array and `@?` expects jsonpath. - // Let's modify to a generic form for JSONPath filtering - // PostgreSQL jsonpath: `jsonb_path_query(col, path)` or `col @? path` const { text: whereText, params: whereParams, nextIndex } = buildWhereClause(options.where, 3) let sql = `SELECT *, jsonb_path_query_array(${quote(options.column)}, $1::jsonpath) AS json_value FROM ${tableRef} WHERE ${quote(options.column)} @? $1::jsonpath`🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/query-builder.ts` around lines 251 - 265, Remove the unused dead variable and leftover comment in buildJsonPathQuery: delete the declaration of innerSql and its related outdated comments (the lines that construct SELECT ... #>> and the note about jsonpath) so only the active sql construction remains; ensure buildJsonPathQuery still returns the same { sql, params: [options.jsonPath, ...whereParams] } and that the call to buildWhereClause(nextIndex) remains untouched (reference function: buildJsonPathQuery, variable: innerSql).src/features/executions/components/postgres/query-builder.ts-294-310 (1)
294-310:⚠️ Potential issue | 🟠 MajorSQL injection risk:
c.defaultandc.referencesare directly interpolated.Lines 301 and 304 interpolate
defaultandreferencesvalues directly into the DDL. While DDL typically cannot use parameterized queries, these values require validation if they originate from user input.Additionally, line 296 produces a double space when
ifNotExistsis false:CREATE TABLE "schema".🔒 Proposed fix with spacing correction
export function buildCreateTable(options: { schema: string; table: string; columns: ColumnDefinition[]; ifNotExists: boolean }): BuiltQuery { const tableRef = `${quote(options.schema)}.${quote(options.table)}` - let sql = `CREATE TABLE ${options.ifNotExists ? "IF NOT EXISTS" : ""} ${tableRef} (` + let sql = `CREATE TABLE ${options.ifNotExists ? "IF NOT EXISTS " : ""}${tableRef} (`For the injection concerns, consider adding validation functions:
function validateDefaultValue(value: string): string { // Allow only safe patterns: literals, function calls, keywords const safePattern = /^(NULL|CURRENT_TIMESTAMP|NOW\(\)|'[^']*'|\d+(\.\d+)?|TRUE|FALSE)$/i if (!safePattern.test(value)) { throw new Error(`Unsafe default value: ${value}`) } return value }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/query-builder.ts` around lines 294 - 310, The buildCreateTable function currently interpolates c.default and c.references directly (risking SQL injection) and always inserts a space after CREATE TABLE causing a double space when options.ifNotExists is false; to fix, add strict validators (e.g., validateDefaultValue(value: string) and validateReference(ref: string)) and call them before using c.default and c.references in the column definition generation inside buildCreateTable, throwing on unsafe patterns and only allowing whitelisted literals/functions/identifiers, and change the CREATE TABLE assembly to conditionally include the "IF NOT EXISTS" token with surrounding spaces (e.g., compute ifNotExistsToken = options.ifNotExists ? "IF NOT EXISTS " : "") so no double space occurs when false; keep using quote(...) for schema/table and ensure ColumnDefinition fields are validated prior to concatenation.
🟡 Minor comments (2)
src/features/executions/components/aggregate/executor.ts-126-128 (1)
126-128:⚠️ Potential issue | 🟡 MinorSilent error swallowing may hide configuration issues.
When filter parsing fails in COUNT operation, the error is silently caught and all items are counted. While this provides a fallback, it can mask misconfigured
countFilterJSON, making debugging harder for users.Consider logging a warning or including a
filterParseErrorflag in the result.💡 Proposed improvement
} catch { - // If filter parsing fails, count all + // If filter parsing fails, count all (log for visibility) + console.warn(`Aggregate/COUNT: Invalid countFilter JSON, counting all items`) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/aggregate/executor.ts` around lines 126 - 128, The COUNT operation currently swallows errors when parsing countFilter (the empty catch after the filter parsing), which can hide misconfigured JSON; modify the catch to capture the error (e.g., catch (err)) and either log a warning using the existing logger or attach a flag like filterParseError: true to the COUNT result object so callers can detect parse failures; update the code path that parses countFilter (the try/catch block around parsing in the COUNT handling in executor.ts) to include the logger.warn(...) call with the error and/or set result.filterParseError before continuing to count all items.src/features/executions/components/postgres/dialog.tsx-91-111 (1)
91-111:⚠️ Potential issue | 🟡 MinorForm doesn't reset when
dbConfigloads asynchronously.The form
defaultValuesare set on initial mount, butdbConfigloads asynchronously from the query. If the dialog opens beforedbConfigis available, the form will usedefaultValuesand won't update whendbConfigarrives.Consider using
useEffectto callform.reset()whendbConfigchanges, or usevaluesprop (react-hook-form v7.43+) for reactive form values.💡 Proposed fix using reset
+import { useEffect } from "react" + export function PostgresDialog({ open, onOpenChange, onSubmit, defaultValues, nodeId, workflowId }: PostgresDialogProps) { const trpc = useTRPC() const { data: dbConfig } = useQuery( trpc.postgres.getByNodeId.queryOptions({ nodeId }, { enabled: !!nodeId }) ) // ... const form = useForm<PostgresFormValues>({ defaultValues: { /* ... */ } }) + + useEffect(() => { + if (dbConfig) { + form.reset({ + operation: dbConfig.operation ?? "EXECUTE_QUERY", + credentialId: dbConfig.credentialId ?? "", + // ... other fields + }) + } + }, [dbConfig, form])🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/dialog.tsx` around lines 91 - 111, The form is initialized once with defaultValues so it won't update when dbConfig loads asynchronously; update the component to watch for dbConfig changes and call form.reset(...) (from the useForm instance named form) inside a useEffect that runs when dbConfig or defaultValues change, passing the merged object (merged = { ...defaultValues, ...dbConfig }) mapped to PostgresFormValues types (operation, credentialId, variableName, schemaName, tableName, query, queryParams, selectColumns, whereConditions, insertData, updateData, continueOnFail, returnData, limitRows, offsetRows) and ensuring proper fallback/typing for each field, or alternatively switch to react-hook-form v7.43+ and use the reactive values prop if preferred.
🧹 Nitpick comments (7)
package.json (1)
67-67: Move@types/pgtodevDependencies.
@types/pgis compile-time only; keeping it in runtime dependencies increases production install footprint without runtime benefit.📦 Suggested dependency placement diff
"dependencies": { - "@types/pg": "^8.20.0", ... "pg": "^8.20.0", ... }, "devDependencies": { + "@types/pg": "^8.20.0", ... }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@package.json` at line 67, The package "@types/pg" is a TypeScript-only type package and should be moved out of runtime dependencies into devDependencies: edit package.json to remove the "@types/pg": "^8.20.0" entry from "dependencies" and add the same entry under "devDependencies" (preserve the version), then run your package manager (npm/yarn/pnpm) to update the lockfile so the change is reflected; ensure no runtime code imports rely on it at runtime.prisma/migrations/20260330130627_add_aggregate_node/migration.sql (1)
36-38: Minor: Redundant index onnodeId.Line 36 creates a unique index on
nodeId, which already enables efficient lookups. The additional regular index onnodeId(line 38) is redundant since the unique index serves the same purpose for queries.♻️ Proposed fix to remove redundant index
CREATE UNIQUE INDEX IF NOT EXISTS "AggregateNode_nodeId_key" ON "AggregateNode"("nodeId"); CREATE INDEX IF NOT EXISTS "AggregateNode_workflowId_idx" ON "AggregateNode"("workflowId"); -CREATE INDEX IF NOT EXISTS "AggregateNode_nodeId_idx" ON "AggregateNode"("nodeId");🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@prisma/migrations/20260330130627_add_aggregate_node/migration.sql` around lines 36 - 38, The migration creates both a unique index "AggregateNode_nodeId_key" and a regular index "AggregateNode_nodeId_idx" on AggregateNode("nodeId"), which is redundant; remove the CREATE INDEX IF NOT EXISTS "AggregateNode_nodeId_idx" ON "AggregateNode"("nodeId") statement so only the unique index "AggregateNode_nodeId_key" remains, and verify there are no other migration steps or code expecting the non-unique index name before applying the change.src/server/routers/postgres.router.ts (1)
144-180: Unnecessary fallback code based on Prisma misunderstanding.The fallback to
(prisma as any).credenial?.findFirst(lines 153-167) is dead code. Prisma generates client methods based on model names (prisma.credential), not mapped table names (the@@map("Credenial")directive). Theprisma.credential.findFirstcall on line 147 will work correctly regardless of the table name typo.This fallback will never execute since
prisma.credentialworks, and if it somehow did,(prisma as any).credenial(with the typo) wouldn't exist on the Prisma client anyway.♻️ Proposed simplification
testConnection: protectedProcedure .input(z.object({ credentialId: z.string() })) .mutation(async ({ input, ctx }) => { const credential = await prisma.credential.findFirst({ where: { id: input.credentialId, userId: ctx.auth.user.id } }) - // Using prisma.credential, fallback to type casting if schema still says 'Credenial' - // User noted: `prisma.credenial.findFirst` if Prisma schema typo is present. Let's use any if TS complains - // Let's use string keys for generic query if needed - if (!credential) { - // Fallback to "Credenial" if not fixed - const fallbackCred = await (prisma as any).credenial?.findFirst?.({ - where: { id: input.credentialId, userId: ctx.auth.user.id } - }) - if (!fallbackCred) { - throw new TRPCError({ code: "NOT_FOUND", message: "Credential not found" }) - } - const config = JSON.parse(decrypt(fallbackCred.value)) as PostgresConnectionConfig - const result = await testConnection(config) - if (!result.success) { - throw new TRPCError({ code: "BAD_REQUEST", message: result.error ?? "Connection failed" }) - } - return { latencyMs: result.latencyMs, success: true } + if (!credential) { + throw new TRPCError({ code: "NOT_FOUND", message: "Credential not found" }) } const config = JSON.parse(decrypt(credential.value)) as PostgresConnectionConfig const result = await testConnection(config)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/server/routers/postgres.router.ts` around lines 144 - 180, Remove the unnecessary fallback that tries to call (prisma as any).credenial?.findFirst and its surrounding branch inside the testConnection mutation: the Prisma client method is prisma.credential.findFirst and that will always be used; delete the entire fallback block (the if (!credential) { ... } branch that attempts fallbackCred, JSON.parse(decrypt(fallbackCred.value)), and the associated testConnection call) and instead throw NOT_FOUND when credential is missing, then continue using credential.value -> JSON.parse(decrypt(credential.value)) as PostgresConnectionConfig and call testConnection(config) as currently done for the primary path; keep TRPCError handling and the returned { latencyMs, success } behavior.src/features/executions/components/aggregate/executor.ts (1)
221-234: Minor: Duplicate standard deviation computation.
computeStdDev(values, true)is called twice: once forvalueand once forsampleStdDev. Consider computing once and reusing.♻️ Proposed optimization
case "STANDARD_DEVIATION": { if (!field) throw new NonRetriableError("Aggregate/STANDARD_DEVIATION requires a Field.") const values = extractNumericValues(inputData, field, nullHandling) + const sampleStdDev = computeStdDev(values, true) ?? 0 + const populationStdDev = computeStdDev(values, false) ?? 0 return { - value: roundTo(computeStdDev(values, true) ?? 0, roundDecimals), - sampleStdDev: roundTo(computeStdDev(values, true) ?? 0, roundDecimals), - populationStdDev: roundTo(computeStdDev(values, false) ?? 0, roundDecimals), + value: roundTo(sampleStdDev, roundDecimals), + sampleStdDev: roundTo(sampleStdDev, roundDecimals), + populationStdDev: roundTo(populationStdDev, roundDecimals), count: values.length,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/aggregate/executor.ts` around lines 221 - 234, In the STANDARD_DEVIATION case compute computeStdDev only once per mode and reuse the results instead of calling computeStdDev(values, true) twice; e.g., after extractNumericValues(...) assign const sample = computeStdDev(values, true) and const population = computeStdDev(values, false) and then use roundTo(sample ?? 0, roundDecimals) for both value and sampleStdDev and roundTo(population ?? 0, roundDecimals) for populationStdDev while keeping the rest of the returned object (count, operation, field, totalInput, timestamp) unchanged; update references to computeStdDev, roundTo, extractNumericValues and the STANDARD_DEVIATION return block accordingly.prisma/migrations/20260330140000_add_postgres_node/migration.sql (1)
52-57: Drop the duplicatenodeIdindex.Line 52 already creates a unique index on
"nodeId", so Lines 56-57 add the same access path twice. The extra index only adds write overhead and storage.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@prisma/migrations/20260330140000_add_postgres_node/migration.sql` around lines 52 - 57, The migration creates a unique index "PostgresNode_nodeId_key" on PostgresNode(nodeId) and then redundantly creates a non-unique "PostgresNode_nodeId_idx" on the same column; remove the duplicate CREATE INDEX statement for "PostgresNode_nodeId_idx" so only the unique index (PostgresNode_nodeId_key) remains while keeping the separate "PostgresNode_workflowId_idx" index intact.src/features/executions/components/aggregate/aggregate-engine.ts (1)
338-340: Redundant call tocomputeAverageon line 339.
computeAverage(vals)is invoked twice: once for the null check and again for rounding. For large datasets, this doubles the computation unnecessarily.♻️ Proposed fix to compute once
case "AVERAGE": - val = computeAverage(vals) !== null ? roundTo(computeAverage(vals)!, roundDecimals) : null + const avg = computeAverage(vals) + val = avg !== null ? roundTo(avg, roundDecimals) : null break🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/aggregate/aggregate-engine.ts` around lines 338 - 340, In the "AVERAGE" branch replace the double call to computeAverage(vals) by computing it once into a local variable (e.g., const avg = computeAverage(vals)), then set val = avg !== null ? roundTo(avg, roundDecimals) : null; update the case handling in the switch (case "AVERAGE") to use that local variable so computeAverage is not invoked twice.src/features/executions/components/postgres/query-builder.ts (1)
31-44: Hardcodedtext[]cast limits type flexibility for IN/NOT IN operations.The
::text[]cast on line 33 forces all array comparisons to use text type. This works for most cases due to PostgreSQL's type coercion, but may produce unexpected results for types likeuuid,timestamp, ornumericwhere string comparison semantics differ from native type comparison.Consider accepting an optional type hint in
WhereConditionto support proper typing when needed.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/postgres/query-builder.ts` around lines 31 - 44, The IN/NOT IN branch currently hardcodes ::text[] which can mis-type comparisons; add an optional type hint on the condition (e.g., cond.type or cond.valueType) and use it to build the SQL cast dynamically (e.g., ANY($${nextIndex}::${cond.type || "text"}[])); keep the existing parsing of cond.value into valArray but do not forcibly coerce element types in JS—let the DB cast handle them when a type hint is provided; update references in this block (op, part, params, nextIndex, cond.value) and ensure params.push(valArray) and nextIndex++ remain unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/features/executions/components/aggregate/actions.ts`:
- Around line 7-12: fetchAggregateRealtimeToken and the TRPC getToken procedure
mint realtime tokens for any nodeId without verifying ownership; add an
ownership check before calling getSubscriptionToken or returning a token: in
fetchAggregateRealtimeToken, load the node/workflow associated with nodeId using
the current session user id (ctx.session.user.id) and only call
getSubscriptionToken(…) with aggregateChannelName(nodeId) if the node's
workflow.userId === ctx.session.user.id (otherwise throw/return unauthorized);
mirror the same check in the TRPC protectedProcedure getToken implementation so
it first queries the node (or workflow) tied to the nodeId and validates
ctx.session.user.id matches workflow.userId before minting/returning the token.
In `@src/features/executions/components/aggregate/aggregate-engine.ts`:
- Around line 63-71: computeMin and computeMax risk a call-stack overflow by
using Math.min(...values) / Math.max(...values) with large arrays; replace the
spread usage with a safe iteration (e.g., Array.prototype.reduce or a simple for
loop) that returns null for empty arrays and otherwise computes the min/max by
scanning values to avoid creating huge argument lists; update the
implementations in computeMin and computeMax accordingly.
In `@src/features/executions/components/filter/executor.ts`:
- Line 250: The executor currently returns only the new binding ({
[variableName]: result! }) which overwrites the entire execution context; change
the return to merge into the existing context by returning an object that
spreads the incoming context and adds the new variable (follow the pattern used
by other executors such as http-request/executor.ts, merge/executor.ts,
loop/executor.ts), i.e., return { ...context, [variableName]: result! } so all
upstream bindings are preserved; update the function handling in executor.ts
where variableName and result are returned.
- Line 36: filterChannel and postgresChannel are exported as static channel
objects but are being invoked as functions (e.g.,
filterChannel(nodeId).status(...)) which causes a runtime TypeError; convert
both exports into factory functions like sortChannel (export const filterChannel
= (nodeId?) => ... ) so calling filterChannel(nodeId).status(...) and
postgresChannel(nodeId).status(...) is valid, then update the three call sites
in executor.ts that currently call filterChannel(nodeId) and any analogous
postgresChannel usages to use the factory functions; also adjust the executor
return to preserve prior context (use the pattern { ...context, [variableName]:
result } instead of returning only { [variableName]: result! }) to avoid
dropping earlier bindings.
In `@src/features/executions/components/postgres/actions.ts`:
- Around line 7-12: The token minting paths lack ownership checks: update
fetchPostgresRealtimeToken and the router's getToken handler to enforce the same
authorization as delete/getByNodeId by validating the node's workflow.userId
against the authenticated user before calling
getSubscriptionToken/postgresChannelName; ensure the server action obtains the
current user/session (or rejects unauthenticated requests), loads the node (by
nodeId), compares node.workflow.userId === ctx.auth.user.id (or
session.user.id), and throw an unauthorized error if they don't match so tokens
are only minted for node owners.
In `@src/features/executions/components/postgres/executor.ts`:
- Around line 812-826: The function parseWhereConditions currently swallows JSON
parse errors and returns an empty array; change it to throw a NonRetriableError
on parse failure so malformed WHERE payloads fail fast. In parseWhereConditions,
replace the catch branch with throwing new NonRetriableError('Invalid
whereConditions JSON' + optional error message) (or include the caught
error.message), keeping the successful path that maps values via r(c.value) and
r(c.value2 || ""). Ensure NonRetriableError is imported/available in the module
before using it.
- Around line 804-807: The executor currently returns only the new Postgres
variable (return { [variableName]: result! }) which discards the accumulated
WorkflowContext; update the return to merge the incoming context with the new
variable (preserve ...context and then add [variableName]: result) so the
NodeExecutor contract (NodeExecutor / WorkflowContext in
src/features/executions/types.ts) is honored and downstream nodes retain prior
variables; locate the return in executor.ts near
publish(postgresChannel(nodeId).status(...)) and replace the single-variable
return with a merged context including variableName and result.
In `@src/features/executions/components/postgres/query-builder.ts`:
- Around line 8-13: The quote function currently wraps identifiers with double
quotes but doesn't escape embedded double quotes, allowing invalid SQL/SQL
injection; update the quote(identifier: string) implementation to first replace
any instance of " in the identifier parts with two double quotes (e.g.,
identifierPart.replace(/"/g, '""')) before surrounding each part with quotes and
joining, and apply this to both dotted parts in the quote function so
identifiers like table"name become "table""name".
- Around line 178-201: The buildUpdate function currently allows updates with an
empty WHERE and must be protected similarly to buildDelete; update buildUpdate
(function name buildUpdate) to validate options.where and throw an error when
the where clause is empty (or undefined/length===0) to prevent accidental
full-table updates, unless a new explicit flag (e.g., allowFullTableUpdate?:
boolean) is provided on the options and set to true—mirror the same guard
logic/behavior used in buildDelete so the function throws by default and only
permits full-table updates when the flag is present and true.
In `@src/inngest/channels/aggregate.ts`:
- Around line 1-13: The runtime error comes from invoking aggregateChannel as a
function; aggregateChannel is a static channel object created by
channel(aggregateChannelName).addTopic(...). Update the executor calls that
currently call aggregateChannel(nodeId) to use the string factory
aggregateChannelName(nodeId) instead, or alternatively refactor this file to
expose a function that builds a channel per node (e.g., a factory function that
calls channel(aggregateChannelName(nodeId)).addTopic(...)); target the symbols
aggregateChannel (stop invoking it), aggregateChannelName (use it to build the
channel name), and AGGREGATE_CHANNEL (keep constant) when making the change.
In `@src/inngest/channels/filter.ts`:
- Around line 8-13: filterChannel was changed from a callable factory to a
static channel object, breaking callers that do
filterChannel(nodeId).status(...) and filterChannel() in functions.ts; restore
the callable pattern by making filterChannel a function that accepts an optional
nodeId (e.g., function filterChannel(nodeId?: string) { return
channel(filterChannelName(nodeId)).addTopic(...) } or equivalent) so callers in
src/features/executions/components/filter/executor.ts (uses
filterChannel(nodeId).status at lines referenced) and src/inngest/functions.ts
(invokes filterChannel()) work again; apply the same fix pattern to
aggregateChannel and postgresChannel if they are currently non-callable but used
as factories elsewhere.
In `@src/inngest/channels/postgres.ts`:
- Around line 1-13: The executor is calling postgresChannel(nodeId) but
postgresChannel is a channel object created by channel(...).addTopic(), not a
function; update the executor to call postgresChannelName(nodeId) wherever it
currently invokes postgresChannel(nodeId) (references found around the executor
usage) so it passes the string channel name, or alternatively change the export
to a function factory instead of postgresChannel if you prefer per-call
construction; specifically replace calls to postgresChannel(nodeId) with
postgresChannelName(nodeId) to match usage in node.tsx and actions.ts and keep
the exported postgresChannel and postgresChannelName definitions unchanged.
In `@src/server/routers/aggregate.router.ts`:
- Around line 68-74: The ownership check currently validates only the provided
workflowId via prisma.workflow.findUnique, but the upsert later targets an
aggregate node by nodeId alone (allowing a user to overwrite another user's
node); update the logic in aggregate.router to first load the existing aggregate
node (e.g., via prisma.aggregate.findUnique/findFirst by nodeId) and verify its
workflowId matches ctx.auth.user.id before performing an update, or change the
upsert uniqueness to use the composite key (workflowId, nodeId) so writes are
scoped to the caller's workflow; ensure checks reference the existing node
retrieval (prisma.aggregate.findUnique/findFirst) and the upsert call to prevent
cross-user overwrites.
---
Major comments:
In `@src/features/credentials/components/credential.tsx`:
- Around line 57-62: The credential schema currently models postgresSsl as a
boolean which causes the engine (postgres-engine.ts) to fall back instead of
using the expected string modes ("disable"|"require"|"verify-full"); change
postgresSsl in the Zod schema (the symbol postgresSsl in credential.ts) from
z.boolean().optional() to a string enum matching the engine's modes (e.g.,
z.enum(["disable","require","verify-full"]).optional()), update any UI form
serialization/deserialization and defaulting logic that writes/reads postgresSsl
so it stores one of those strings (not true/false), and ensure any code paths
that previously assumed a boolean (connection test, execution setup, and server
router passthrough) convert or validate values to the new enum shape before
sending to postgres-engine.ts.
In `@src/features/executions/components/aggregate/dialog.tsx`:
- Around line 57-78: MultiOpBuilder currently initializes its local ops state
from the value prop only once (useState initializer) so it becomes stale when
value changes; add a useEffect watching value that parses JSON from value
(handling parse errors) and calls setOps(parsed) when the parsed array differs
from current ops to resync the local state, while keeping the existing update
callback (setOps and onChange) intact; reference the component MultiOpBuilder,
state ops/setOps, props value/onChange, and the update callback to locate where
to add the effect.
- Around line 224-258: The form's defaultValues are only applied on first render
so asynchronous dbConfig never populates the fields; after
trpc.aggregate.getByNodeId returns (dbConfig) compute the same merged defaults
and call form.reset(...) to update the form state; add a useEffect that watches
dbConfig and calls form.reset with the merged/defaulted AggregateNodeData (use
the same merging logic used to build merged) so the useForm<AggregateNodeData>
instance named form reflects persisted node settings when the query resolves.
In `@src/features/executions/components/postgres/dialog.tsx`:
- Around line 140-169: The onFormSubmit handler builds a payload with only basic
fields which will overwrite and drop advanced DB configuration; modify
onFormSubmit to merge the current dbConfig (or the form's full values object
that contains advanced settings) into the payload before calling
upsertMutation.mutateAsync so existing advanced fields (JOINs, orderBy,
functionArgs, JSON ops, etc.) are preserved—e.g., locate onFormSubmit and the
payload variable and construct the payload by spreading dbConfig (or fullValues)
first and then the explicit overrides (nodeId, workflowId, operation,
credentialId, etc.), making sure selectColumns/whereConditions and other
optional arrays keep their existing values when empty.
In `@src/features/executions/components/postgres/postgres-engine.ts`:
- Around line 61-78: The current executeQuery implementation slices result.rows
after client.query(), which still materializes the entire result set; change
executeQuery to fetch rows incrementally using a server-side cursor/stream
(e.g., pg-cursor or pg-query-stream) instead of client.query(), read up to
maxRows from the cursor/stream into the rows array, close the cursor when done,
and set rowCount to either the number of rows read or, if you need the true
total, run a separate COUNT(*) query; replace the client.query usage in
executeQuery and remove the post-query slice to ensure large results are never
fully loaded into memory.
- Around line 3-12: PostgresConnectionConfig currently requires
connectionTimeout and queryTimeout but credentials may omit them, causing
multiplication of undefined when building the client; default the timeouts
before constructing the client config (e.g., in the code that maps
PostgresConnectionConfig into client options where you do connectionTimeout *
1000 or queryTimeout * 1000). Specifically, in the code that reads a
PostgresConnectionConfig and builds the client (the function that multiplies
timeouts into milliseconds), normalize with defaults like const
connectionTimeoutSec = config.connectionTimeout ?? 10 and const queryTimeoutSec
= config.queryTimeout ?? 30 (or other project-standard defaults) and use those
variables in place of config.connectionTimeout/config.queryTimeout so the client
always receives valid numeric timeout values.
In `@src/features/executions/components/postgres/query-builder.ts`:
- Around line 65-70: buildJoinClause currently interpolates the join condition
via j.on with no validation, which can lead to SQL injection if caller-supplied;
update buildJoinClause (or a helper used by it) to validate j.on against a
strict allowed-pattern (e.g., only identifiers, dot notation, whitespace,
allowed operators (=, <, >, <=, >=, !=, AND, OR, LIKE), numeric/string literals,
and parentheses) and throw an error if it fails, or alternatively add a clear
documented precondition on the JoinClause type requiring the caller to supply a
pre-validated/safe on expression; reference the buildJoinClause function and the
JoinClause.on property when adding the validation or documentation.
- Around line 251-265: Remove the unused dead variable and leftover comment in
buildJsonPathQuery: delete the declaration of innerSql and its related outdated
comments (the lines that construct SELECT ... #>> and the note about jsonpath)
so only the active sql construction remains; ensure buildJsonPathQuery still
returns the same { sql, params: [options.jsonPath, ...whereParams] } and that
the call to buildWhereClause(nextIndex) remains untouched (reference function:
buildJsonPathQuery, variable: innerSql).
- Around line 294-310: The buildCreateTable function currently interpolates
c.default and c.references directly (risking SQL injection) and always inserts a
space after CREATE TABLE causing a double space when options.ifNotExists is
false; to fix, add strict validators (e.g., validateDefaultValue(value: string)
and validateReference(ref: string)) and call them before using c.default and
c.references in the column definition generation inside buildCreateTable,
throwing on unsafe patterns and only allowing whitelisted
literals/functions/identifiers, and change the CREATE TABLE assembly to
conditionally include the "IF NOT EXISTS" token with surrounding spaces (e.g.,
compute ifNotExistsToken = options.ifNotExists ? "IF NOT EXISTS " : "") so no
double space occurs when false; keep using quote(...) for schema/table and
ensure ColumnDefinition fields are validated prior to concatenation.
In `@src/features/executions/components/postgres/types.ts`:
- Around line 137-143: TABLE_OPS currently omits the CREATE_TABLE operation,
causing callers to misclassify operations that require a tableName; add
"CREATE_TABLE" to the TABLE_OPS array so CREATE_TABLE is treated as a
table-name-required operation (update the TABLE_OPS constant where it's defined
and ensure it now includes "CREATE_TABLE" alongside "DROP_TABLE" and the other
PostgresOperation entries).
- Around line 1-24: The PostgresOperation union currently lists "COPY_FROM" but
the executor's operation dispatcher does not handle it, causing an "Unknown
operation" error; either remove "COPY_FROM" from the PostgresOperation type or
implement a matching handler in the Postgres executor switch/dispatcher. Locate
the PostgresOperation type and either delete the "COPY_FROM" variant, or add a
case for "COPY_FROM" in the executor's operation switch (the function that
routes Postgres operations) that performs the expected copy-from logic or throws
a clear not-yet-implemented error until full support is added.
In `@src/server/routers/aggregate.router.ts`:
- Around line 111-119: getToken currently mints subscription tokens for any
nodeId without verifying ownership; update the protectedProcedure query
signature to accept ctx (i.e., async ({ input, ctx }) =>), lookup the node by
input.nodeId (using your data access layer/Prisma) and verify its ownerId equals
ctx.auth.user.id, throw an authorization error if not the owner, and only then
call getSubscriptionToken with aggregateChannelName(input.nodeId) to return the
token.
- Around line 9-16: The multiOps and groupAggOps JSON string checks currently
only verify they are JSON strings but not that their contents match the
aggregate shape; validate them against the existing aggregateOpSchema by using
z.array(aggregateOpSchema) (e.g., parse the incoming value with
z.array(aggregateOpSchema).safeParse(...) or use z.preprocess to parse JSON then
validate) instead of the current syntax-only parsing, and if validation fails
return/throw an error before persisting; update the code paths that read
multiOps and groupAggOps (where .map() is later called) to use the validated
result so you never call .map() on invalid payloads.
In `@src/server/routers/postgres.router.ts`:
- Around line 134-142: getToken currently returns a realtime subscription token
for any nodeId without verifying ownership; fix by loading the Postgres node
(same way getByNodeId/delete do) and verify node.workflow.userId ===
ctx.auth.user.id before calling getSubscriptionToken and returning the token,
and throw a forbidden/unauthorized error if the check fails; ensure you
reference getToken, getSubscriptionToken, postgresChannelName, and the
node.workflow.userId === ctx.auth.user.id check so the authorization logic
mirrors getByNodeId/delete.
---
Minor comments:
In `@src/features/executions/components/aggregate/executor.ts`:
- Around line 126-128: The COUNT operation currently swallows errors when
parsing countFilter (the empty catch after the filter parsing), which can hide
misconfigured JSON; modify the catch to capture the error (e.g., catch (err))
and either log a warning using the existing logger or attach a flag like
filterParseError: true to the COUNT result object so callers can detect parse
failures; update the code path that parses countFilter (the try/catch block
around parsing in the COUNT handling in executor.ts) to include the
logger.warn(...) call with the error and/or set result.filterParseError before
continuing to count all items.
In `@src/features/executions/components/postgres/dialog.tsx`:
- Around line 91-111: The form is initialized once with defaultValues so it
won't update when dbConfig loads asynchronously; update the component to watch
for dbConfig changes and call form.reset(...) (from the useForm instance named
form) inside a useEffect that runs when dbConfig or defaultValues change,
passing the merged object (merged = { ...defaultValues, ...dbConfig }) mapped to
PostgresFormValues types (operation, credentialId, variableName, schemaName,
tableName, query, queryParams, selectColumns, whereConditions, insertData,
updateData, continueOnFail, returnData, limitRows, offsetRows) and ensuring
proper fallback/typing for each field, or alternatively switch to
react-hook-form v7.43+ and use the reactive values prop if preferred.
---
Nitpick comments:
In `@package.json`:
- Line 67: The package "@types/pg" is a TypeScript-only type package and should
be moved out of runtime dependencies into devDependencies: edit package.json to
remove the "@types/pg": "^8.20.0" entry from "dependencies" and add the same
entry under "devDependencies" (preserve the version), then run your package
manager (npm/yarn/pnpm) to update the lockfile so the change is reflected;
ensure no runtime code imports rely on it at runtime.
In `@prisma/migrations/20260330130627_add_aggregate_node/migration.sql`:
- Around line 36-38: The migration creates both a unique index
"AggregateNode_nodeId_key" and a regular index "AggregateNode_nodeId_idx" on
AggregateNode("nodeId"), which is redundant; remove the CREATE INDEX IF NOT
EXISTS "AggregateNode_nodeId_idx" ON "AggregateNode"("nodeId") statement so only
the unique index "AggregateNode_nodeId_key" remains, and verify there are no
other migration steps or code expecting the non-unique index name before
applying the change.
In `@prisma/migrations/20260330140000_add_postgres_node/migration.sql`:
- Around line 52-57: The migration creates a unique index
"PostgresNode_nodeId_key" on PostgresNode(nodeId) and then redundantly creates a
non-unique "PostgresNode_nodeId_idx" on the same column; remove the duplicate
CREATE INDEX statement for "PostgresNode_nodeId_idx" so only the unique index
(PostgresNode_nodeId_key) remains while keeping the separate
"PostgresNode_workflowId_idx" index intact.
In `@src/features/executions/components/aggregate/aggregate-engine.ts`:
- Around line 338-340: In the "AVERAGE" branch replace the double call to
computeAverage(vals) by computing it once into a local variable (e.g., const avg
= computeAverage(vals)), then set val = avg !== null ? roundTo(avg,
roundDecimals) : null; update the case handling in the switch (case "AVERAGE")
to use that local variable so computeAverage is not invoked twice.
In `@src/features/executions/components/aggregate/executor.ts`:
- Around line 221-234: In the STANDARD_DEVIATION case compute computeStdDev only
once per mode and reuse the results instead of calling computeStdDev(values,
true) twice; e.g., after extractNumericValues(...) assign const sample =
computeStdDev(values, true) and const population = computeStdDev(values, false)
and then use roundTo(sample ?? 0, roundDecimals) for both value and sampleStdDev
and roundTo(population ?? 0, roundDecimals) for populationStdDev while keeping
the rest of the returned object (count, operation, field, totalInput, timestamp)
unchanged; update references to computeStdDev, roundTo, extractNumericValues and
the STANDARD_DEVIATION return block accordingly.
In `@src/features/executions/components/postgres/query-builder.ts`:
- Around line 31-44: The IN/NOT IN branch currently hardcodes ::text[] which can
mis-type comparisons; add an optional type hint on the condition (e.g.,
cond.type or cond.valueType) and use it to build the SQL cast dynamically (e.g.,
ANY($${nextIndex}::${cond.type || "text"}[])); keep the existing parsing of
cond.value into valArray but do not forcibly coerce element types in JS—let the
DB cast handle them when a type hint is provided; update references in this
block (op, part, params, nextIndex, cond.value) and ensure params.push(valArray)
and nextIndex++ remain unchanged.
In `@src/server/routers/postgres.router.ts`:
- Around line 144-180: Remove the unnecessary fallback that tries to call
(prisma as any).credenial?.findFirst and its surrounding branch inside the
testConnection mutation: the Prisma client method is prisma.credential.findFirst
and that will always be used; delete the entire fallback block (the if
(!credential) { ... } branch that attempts fallbackCred,
JSON.parse(decrypt(fallbackCred.value)), and the associated testConnection call)
and instead throw NOT_FOUND when credential is missing, then continue using
credential.value -> JSON.parse(decrypt(credential.value)) as
PostgresConnectionConfig and call testConnection(config) as currently done for
the primary path; keep TRPCError handling and the returned { latencyMs, success
} behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
Comment on lines
+7
to
+12
| export async function fetchAggregateRealtimeToken(nodeId: string) { | ||
| try { | ||
| const token = await getSubscriptionToken(inngest, { | ||
| channel: aggregateChannelName(nodeId), | ||
| topics: ["status"], | ||
| }) |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# 1) Locate call sites
rg -n -C3 'fetchAggregateRealtimeToken\s*\(' src/features
# 2) Verify guard presence/absence in this action
rg -n -C5 'export async function fetchAggregateRealtimeToken|getSubscriptionToken|auth|session|userId|prisma' \
src/features/executions/components/aggregate/actions.ts
# 3) Compare with protected TRPC token route
rg -n -C5 'getToken|protectedProcedure|getSubscriptionToken|aggregateChannelName' \
src/server/routers/aggregate.router.tsRepository: Mayank-saraswal/n8n
Length of output: 4032
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check for middleware that might protect server actions
find . -name "middleware.ts" -o -name "middleware.js" | head -5
# Look for session/auth context in the actions file and related files
rg -n 'getSession|session|auth|userId' src/features/executions/components/aggregate/
# Check how other server actions handle authentication
rg -n '"use server"' -A 10 src/ | head -60
# Look for context/session utilities
rg -n 'export.*getSession|export.*useSession' src/ | head -10Repository: Mayank-saraswal/n8n
Length of output: 6166
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check aggregateChannelName function
rg -n 'aggregateChannelName' -B2 -A5 src/inngest/channels/aggregate.ts
# Check if nodeId is globally unique or per-workflow
rg -n 'nodeId.*String\(\)' src/lib/db.ts | head -20
# Look at Aggregate node schema to understand nodeId scope
rg -n 'model.*Aggregate' src/lib/db.ts -A 20
# Check if there's access control elsewhere that might protect node access
rg -n 'findUnique.*nodeId\|aggregate' src/server/routers/ | head -20Repository: Mayank-saraswal/n8n
Length of output: 433
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check how AggregateNode relates to users/workflows
rg -n 'aggregateNode.*findUnique\|aggregateNode.*findFirst' src/ -A 5 | head -40
# Verify that nodeId can be queried across different users
rg -n 'nodeId.*input.*workflow' src/server/routers/aggregate.router.ts -B2 -A2Repository: Mayank-saraswal/n8n
Length of output: 45
Add ownership validation to fetchAggregateRealtimeToken server action and getToken TRPC procedure.
The server action mints a realtime token for an arbitrary nodeId with no access control, allowing any authenticated user to obtain tokens for nodes they don't own. This bypasses the ownership checks enforced in other aggregate operations (getByNodeId, upsert, delete all verify ctx.session.user.id === node.workflow.userId). The TRPC getToken endpoint has the same issue—it uses protectedProcedure but skips the ownership check that should prevent cross-user token access.
Add a lookup to verify the node belongs to the current user before minting the token:
- Server action: verify the
nodeIdcorresponds to a node inctx.session.user.id's workflows - TRPC endpoint: apply the same ownership check as other aggregate procedures
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/aggregate/actions.ts` around lines 7 - 12,
fetchAggregateRealtimeToken and the TRPC getToken procedure mint realtime tokens
for any nodeId without verifying ownership; add an ownership check before
calling getSubscriptionToken or returning a token: in
fetchAggregateRealtimeToken, load the node/workflow associated with nodeId using
the current session user id (ctx.session.user.id) and only call
getSubscriptionToken(…) with aggregateChannelName(nodeId) if the node's
workflow.userId === ctx.session.user.id (otherwise throw/return unauthorized);
mirror the same check in the TRPC protectedProcedure getToken implementation so
it first queries the node (or workflow) tied to the nodeId and validates
ctx.session.user.id matches workflow.userId before minting/returning the token.
Comment on lines
+63
to
+71
| export function computeMin(values: number[]): number | null { | ||
| if (values.length === 0) return null | ||
| return Math.min(...values) | ||
| } | ||
|
|
||
| export function computeMax(values: number[]): number | null { | ||
| if (values.length === 0) return null | ||
| return Math.max(...values) | ||
| } |
Contributor
There was a problem hiding this comment.
Stack overflow risk with large arrays in computeMin and computeMax.
Math.min(...values) and Math.max(...values) spread all elements as function arguments. JavaScript engines have argument limits (typically 65K-125K), causing RangeError: Maximum call stack size exceeded for large datasets—a common scenario in aggregation operations.
🐛 Proposed fix using reduce for safe iteration
export function computeMin(values: number[]): number | null {
if (values.length === 0) return null
- return Math.min(...values)
+ return values.reduce((min, v) => v < min ? v : min, values[0]!)
}
export function computeMax(values: number[]): number | null {
if (values.length === 0) return null
- return Math.max(...values)
+ return values.reduce((max, v) => v > max ? v : max, values[0]!)
}📝 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
| export function computeMin(values: number[]): number | null { | |
| if (values.length === 0) return null | |
| return Math.min(...values) | |
| } | |
| export function computeMax(values: number[]): number | null { | |
| if (values.length === 0) return null | |
| return Math.max(...values) | |
| } | |
| export function computeMin(values: number[]): number | null { | |
| if (values.length === 0) return null | |
| return values.reduce((min, v) => v < min ? v : min, values[0]!) | |
| } | |
| export function computeMax(values: number[]): number | null { | |
| if (values.length === 0) return null | |
| return values.reduce((max, v) => v > max ? v : max, values[0]!) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/aggregate/aggregate-engine.ts` around
lines 63 - 71, computeMin and computeMax risk a call-stack overflow by using
Math.min(...values) / Math.max(...values) with large arrays; replace the spread
usage with a safe iteration (e.g., Array.prototype.reduce or a simple for loop)
that returns null for empty arrays and otherwise computes the min/max by
scanning values to avoid creating huge argument lists; update the
implementations in computeMin and computeMax accordingly.
| // ── Step 2: Execute filter ───────────────────────────────────────────────── | ||
| let result: Record<string, unknown> | ||
|
|
||
| await publish(filterChannel(nodeId).status({ nodeId, status: "loading" })) |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check how other executors call channel publishing to understand the expected pattern
rg -n "publish\(" --type=ts -C 3 src/features/executions/components/Repository: Mayank-saraswal/n8n
Length of output: 50375
🏁 Script executed:
#!/bin/bash
# Check the filterChannel export in the channels file
cat -n src/inngest/channels/filter.ts | head -20Repository: Mayank-saraswal/n8n
Length of output: 532
🏁 Script executed:
#!/bin/bash
# Read the filter executor to see all three problem lines
cat -n src/features/executions/components/filter/executor.ts | grep -A 2 -B 2 "filterChannel"Repository: Mayank-saraswal/n8n
Length of output: 938
🏁 Script executed:
#!/bin/bash
# Check return statements in multiple executors to understand context spread pattern
echo "=== postgres executor return ===" && sed -n '807p' src/features/executions/components/postgres/executor.ts
echo "=== sort executor return ===" && sed -n '194p' src/features/executions/components/sort/executor.ts
echo "=== media-upload executor return ===" && sed -n '55,57p' src/features/executions/components/media-upload/executor.ts
echo "=== filter executor returns ===" && grep -n "return {" src/features/executions/components/filter/executor.tsRepository: Mayank-saraswal/n8n
Length of output: 442
🏁 Script executed:
#!/bin/bash
# Check other transformation executors (switch, sort) to see their return patterns
echo "=== switch executor return ===" && sed -n '122p' src/features/executions/components/switch/executor.ts
echo "=== wait executor returns ===" && grep -n "return {" src/features/executions/components/wait/executor.ts
echo "=== filter executor structure ===" && wc -l src/features/executions/components/filter/executor.tsRepository: Mayank-saraswal/n8n
Length of output: 339
🏁 Script executed:
#!/bin/bash
# Check if filter executor is supposed to be a transformation that preserves context
head -30 src/features/executions/components/filter/executor.tsRepository: Mayank-saraswal/n8n
Length of output: 1201
🏁 Script executed:
#!/bin/bash
# Look for how filterChannel should be properly called by checking imports and actual channel function signatures
rg "filterChannelName|filterChannel" -B 2 -A 2 src/features/executions/components/filter/executor.tsRepository: Mayank-saraswal/n8n
Length of output: 798
🏁 Script executed:
#!/bin/bash
# Check how postgres and sort channels are defined to understand the nodeId pattern
cat -n src/inngest/channels/postgres.ts | head -20Repository: Mayank-saraswal/n8n
Length of output: 544
🏁 Script executed:
#!/bin/bash
# Check sort channel definition
cat -n src/inngest/channels/sort.ts | head -20Repository: Mayank-saraswal/n8n
Length of output: 524
🏁 Script executed:
#!/bin/bash
# Check if there's any other code that properly uses filterChannel
rg "filterChannel" src/ -B 1 -A 1 --type=tsRepository: Mayank-saraswal/n8n
Length of output: 3582
🏁 Script executed:
#!/bin/bash
# This is critical - postgres executor also calls postgresChannel(nodeId) but postgresChannel is a static object like filterChannel
# Let's verify if postgres executor actually works or has the same bug
grep -n "postgresChannel" src/features/executions/components/postgres/executor.ts | head -5Repository: Mayank-saraswal/n8n
Length of output: 374
🏁 Script executed:
#!/bin/bash
# Compare: sortChannel is a FUNCTION, but postgresChannel and filterChannel are static objects
# Check the function signature more clearly
echo "=== sortChannel (defined as function) ===" && cat src/inngest/channels/sort.ts | grep -A 5 "export const sortChannel"
echo ""
echo "=== postgresChannel (static object) ===" && cat src/inngest/channels/postgres.ts | grep -A 5 "export const postgresChannel"
echo ""
echo "=== filterChannel (static object) ===" && cat src/inngest/channels/filter.ts | grep -A 5 "export const filterChannel"Repository: Mayank-saraswal/n8n
Length of output: 954
🏁 Script executed:
#!/bin/bash
# Check if there's a discrepancy - maybe postgresChannel was recently refactored like filterChannel but executor wasn't updated
# Look at all three executors' usage patterns
echo "=== POSTGRES EXECUTOR USAGE ===" && grep "postgresChannel(" src/features/executions/components/postgres/executor.ts | head -3
echo ""
echo "=== SORT EXECUTOR USAGE ===" && grep "sortChannel(" src/features/executions/components/sort/executor.ts | head -3
echo ""
echo "=== FILTER EXECUTOR USAGE ===" && grep "filterChannel(" src/features/executions/components/filter/executor.ts | head -3Repository: Mayank-saraswal/n8n
Length of output: 847
🏁 Script executed:
#!/bin/bash
# Critical discovery: check if postgresChannel has the same bug
# The key is that sortChannel is a FUNCTION but postgresChannel and filterChannel are STATIC objects
# Yet all three executors call them the same way: channel(nodeId)
# Let's verify postgresChannel is actually static (not a function)
python3 << 'EOF'
import re
# Read filter.ts
with open('src/inngest/channels/filter.ts', 'r') as f:
filter_content = f.read()
# Read postgres.ts
with open('src/inngest/channels/postgres.ts', 'r') as f:
postgres_content = f.read()
# Read sort.ts
with open('src/inngest/channels/sort.ts', 'r') as f:
sort_content = f.read()
# Check if they're functions or objects
def is_function_export(content, name):
# Look for pattern: export const name = (params) => or export const name = function
pattern = rf'export const {name}\s*=\s*(?:\(|async\s|\w+\s+function)'
return bool(re.search(pattern, content))
print("filterChannel is function?", is_function_export(filter_content, 'filterChannel'))
print("postgresChannel is function?", is_function_export(postgres_content, 'postgresChannel'))
print("sortChannel is function?", is_function_export(sort_content, 'sortChannel'))
EOFRepository: Mayank-saraswal/n8n
Length of output: 160
🏁 Script executed:
#!/bin/bash
# The smoking gun - let me also check context preservation pattern by looking at what other transformation-like executors do
# Get all executor returns to see the pattern
for file in src/features/executions/components/*/executor.ts; do
name=$(basename -s /executor.ts "$(dirname "$file")")
lastreturn=$(grep -n "^ *return {" "$file" | tail -1)
if [ -n "$lastreturn" ]; then
linenum=$(echo "$lastreturn" | cut -d: -f1)
echo "=== $name ==="
sed -n "${linenum},$((linenum+3))p" "$file"
fi
done | head -50Repository: Mayank-saraswal/n8n
Length of output: 1270
filterChannel is not callable – critical runtime error affecting three locations.
filterChannel is exported as a static channel object in src/inngest/channels/filter.ts (lines 8–13), not as a function. Calling filterChannel(nodeId) at lines 36, 223, and 248 will throw TypeError: filterChannel is not a function at runtime.
Additionally, postgresChannel has the identical issue and suffers the same bug.
To fix, change the channel definitions to be factory functions (like sortChannel):
export const filterChannel = (nodeId?: string) =>
channel(filterChannelName(nodeId) as string).addTopic(
topic("status").type<{...}>()
)()Then update all three call sites from filterChannel(nodeId).status(...) to filterChannel(nodeId).status(...) (which will work once the factory function exists).
Regarding the return statement at line 250: most executors follow the pattern { ...context, [variableName]: ... } to preserve prior context bindings. The current return { [variableName]: result! } omits the context spread, which may cause loss of variables from earlier nodes—verify this aligns with execution engine expectations or add context spread for consistency.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/filter/executor.ts` at line 36,
filterChannel and postgresChannel are exported as static channel objects but are
being invoked as functions (e.g., filterChannel(nodeId).status(...)) which
causes a runtime TypeError; convert both exports into factory functions like
sortChannel (export const filterChannel = (nodeId?) => ... ) so calling
filterChannel(nodeId).status(...) and postgresChannel(nodeId).status(...) is
valid, then update the three call sites in executor.ts that currently call
filterChannel(nodeId) and any analogous postgresChannel usages to use the
factory functions; also adjust the executor return to preserve prior context
(use the pattern { ...context, [variableName]: result } instead of returning
only { [variableName]: result! }) to avoid dropping earlier bindings.
| } | ||
| await publish(filterChannel(nodeId).status({ nodeId, status: "success" })) | ||
|
|
||
| return { [variableName]: result! } |
Contributor
There was a problem hiding this comment.
Critical: Returning only the new binding overwrites entire context.
The return statement { [variableName]: result! } will replace the entire execution context rather than merging into it. Per the execution engine in src/inngest/functions.ts (lines 309-319):
context = await executor({...})This means all prior variable bindings from upstream nodes will be lost. Every other executor in the codebase follows the pattern { ...context, [variableName]: value } to preserve existing context (see http-request/executor.ts, merge/executor.ts, loop/executor.ts).
🐛 Proposed fix to preserve context
- return { [variableName]: result! }
+ return { ...context, [variableName]: result! }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/filter/executor.ts` at line 250, The
executor currently returns only the new binding ({ [variableName]: result! })
which overwrites the entire execution context; change the return to merge into
the existing context by returning an object that spreads the incoming context
and adds the new variable (follow the pattern used by other executors such as
http-request/executor.ts, merge/executor.ts, loop/executor.ts), i.e., return {
...context, [variableName]: result! } so all upstream bindings are preserved;
update the function handling in executor.ts where variableName and result are
returned.
Comment on lines
+7
to
+12
| export async function fetchPostgresRealtimeToken(nodeId: string) { | ||
| try { | ||
| const token = await getSubscriptionToken(inngest, { | ||
| channel: postgresChannelName(nodeId), | ||
| topics: ["status"], | ||
| }) |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# 1) Find who calls this server action (especially from client components)
rg -n -C3 'fetchPostgresRealtimeToken\s*\(' src/features
# 2) Inspect this action for auth/ownership checks
rg -n -C5 'export async function fetchPostgresRealtimeToken|getSubscriptionToken|auth|session|userId|prisma' \
src/features/executions/components/postgres/actions.ts
# 3) Compare with protected token endpoint implementation
rg -n -C5 'getToken|protectedProcedure|getSubscriptionToken|postgresChannelName' \
src/server/routers/postgres.router.tsRepository: Mayank-saraswal/n8n
Length of output: 4511
🏁 Script executed:
# Read the full actions.ts file to see all authorization logic
cat -n src/features/executions/components/postgres/actions.tsRepository: Mayank-saraswal/n8n
Length of output: 705
🏁 Script executed:
# Also check the full router getToken procedure to see if it has ownership checks
sed -n '134,142p' src/server/routers/postgres.router.tsRepository: Mayank-saraswal/n8n
Length of output: 373
🏁 Script executed:
# Search for what protectedProcedure provides and how it's initialized
rg -n -A10 'protectedProcedure.*=' src/trpc/init.ts | head -30Repository: Mayank-saraswal/n8n
Length of output: 379
🏁 Script executed:
# Check if server action can access session/auth context
rg -n 'getSession|auth|session|getCurrentUser|middleware' src/features/executions/components/postgres/actions.tsRepository: Mayank-saraswal/n8n
Length of output: 45
🏁 Script executed:
# Search for auth context in server actions in the codebase to understand the pattern
fd -e ts -e tsx src/features | xargs -I {} grep -l '"use server"' {} | head -5 | xargs -I {} rg -A5 -B5 'auth|session|userId' {}Repository: Mayank-saraswal/n8n
Length of output: 416
🏁 Script executed:
# Check what context is available in the router ctx parameter
rg -n -B10 'protectedProcedure.*use.*ctx' src/trpc/init.tsRepository: Mayank-saraswal/n8n
Length of output: 536
🏁 Script executed:
# Find how server actions access auth in this codebase
find src -name "*.ts" -o -name "*.tsx" | xargs grep -l '"use server"' | xargs grep -l 'auth\|session\|getCurrentUser' | head -3Repository: Mayank-saraswal/n8n
Length of output: 45
🏁 Script executed:
# Check createTRPCContext to understand ctx.auth structure
rg -n 'createTRPCContext' src/trpc/ -A20Repository: Mayank-saraswal/n8n
Length of output: 5406
🏁 Script executed:
# Search for patterns of auth checks in server actions
rg -n -B3 -A10 '"use server"' src --type ts --type tsx | grep -A10 'auth\|session\|userId' | head -40Repository: Mayank-saraswal/n8n
Length of output: 90
🏁 Script executed:
# Check if server actions can use auth session
rg -n 'auth\.api\.getSession\|auth()' src --type ts | head -20Repository: Mayank-saraswal/n8n
Length of output: 45
🏁 Script executed:
# Look for how other server actions handle auth
find src -name "*.ts" -path "*/actions.ts" | xargs grep -l '"use server"' | head -3 | xargs catRepository: Mayank-saraswal/n8n
Length of output: 1632
🏁 Script executed:
# Check the node.tsx component to see what user context it has
sed -n '80,95p' src/features/executions/components/postgres/node.tsxRepository: Mayank-saraswal/n8n
Length of output: 465
🏁 Script executed:
# Verify if nodes are scoped to users/workflows
rg -n 'postgresNode.*userId\|workflow.*userId' src/server/routers/postgres.router.ts -B2 -A2Repository: Mayank-saraswal/n8n
Length of output: 45
🏁 Script executed:
# Check what data the realtime subscription exposes
rg -n 'postgresChannelName\|publish.*postgres' src/inngestRepository: Mayank-saraswal/n8n
Length of output: 45
🏁 Script executed:
# Confirm the vulnerability: compare ownership checks across procedures
rg -n 'workflow\.userId.*ctx\.auth\.user\.id' src/server/routers/postgres.router.tsRepository: Mayank-saraswal/n8n
Length of output: 245
🏁 Script executed:
# Check if server action can access session/auth headers
rg -n 'headers\|getSession' src/features/executions/components/postgres/actions.tsRepository: Mayank-saraswal/n8n
Length of output: 45
🏁 Script executed:
# Verify the issue: other procedures ALL have ownership checks
rg -n -B5 'getByNodeId\|upsert\|delete' src/server/routers/postgres.router.ts | grep -A5 'protectedProcedure'Repository: Mayank-saraswal/n8n
Length of output: 45
Missing authorization and ownership checks before minting Postgres realtime tokens.
Both the server action and the router's getToken endpoint lack explicit node ownership verification. Any authenticated user can call either path with an arbitrary nodeId to obtain a subscription token, exposing realtime workflow status streams across users.
The server action has no auth mechanism. The router's getToken procedure wraps the token issuance in protectedProcedure (user login check only) but omits the explicit ownership check present in other procedures like delete and getByNodeId, which verify node.workflow.userId === ctx.auth.user.id before granting access.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/postgres/actions.ts` around lines 7 - 12,
The token minting paths lack ownership checks: update fetchPostgresRealtimeToken
and the router's getToken handler to enforce the same authorization as
delete/getByNodeId by validating the node's workflow.userId against the
authenticated user before calling getSubscriptionToken/postgresChannelName;
ensure the server action obtains the current user/session (or rejects
unauthenticated requests), loads the node (by nodeId), compares
node.workflow.userId === ctx.auth.user.id (or session.user.id), and throw an
unauthorized error if they don't match so tokens are only minted for node
owners.
Comment on lines
+178
to
+201
| export function buildUpdate(options: { schema: string; table: string; data: Record<string, unknown>; where: WhereCondition[]; returnData: boolean }): BuiltQuery { | ||
| const tableRef = `${quote(options.schema)}.${quote(options.table)}` | ||
| const keys = Object.keys(options.data) | ||
|
|
||
| let sql = `UPDATE ${tableRef} SET ` | ||
| const setParts: string[] = [] | ||
| const params: unknown[] = [] | ||
| let idx = 1 | ||
|
|
||
| keys.forEach(k => { | ||
| setParts.push(`${quote(k)}=$${idx++}`) | ||
| params.push(options.data[k]) | ||
| }) | ||
|
|
||
| sql += setParts.join(", ") | ||
|
|
||
| const { text: whereText, params: whereParams } = buildWhereClause(options.where, idx) | ||
| if (whereText) sql += ` ${whereText}` | ||
| params.push(...whereParams) | ||
|
|
||
| if (options.returnData) sql += " RETURNING *" | ||
|
|
||
| return { sql, params } | ||
| } |
Contributor
There was a problem hiding this comment.
Missing WHERE protection for UPDATE allows accidental mass updates.
Unlike buildDelete which throws when WHERE is empty (line 204-206), buildUpdate has no such safeguard. An UPDATE without WHERE conditions will modify every row in the table, which is rarely intentional and can cause severe data corruption.
🛡️ Proposed fix to add WHERE validation
export function buildUpdate(options: { schema: string; table: string; data: Record<string, unknown>; where: WhereCondition[]; returnData: boolean }): BuiltQuery {
+ if (!options.where || options.where.length === 0) {
+ throw new Error("UPDATE without WHERE conditions is not allowed.")
+ }
const tableRef = `${quote(options.schema)}.${quote(options.table)}`Alternatively, add an allowFullTableUpdate?: boolean flag if full-table updates are a legitimate use case.
📝 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
| export function buildUpdate(options: { schema: string; table: string; data: Record<string, unknown>; where: WhereCondition[]; returnData: boolean }): BuiltQuery { | |
| const tableRef = `${quote(options.schema)}.${quote(options.table)}` | |
| const keys = Object.keys(options.data) | |
| let sql = `UPDATE ${tableRef} SET ` | |
| const setParts: string[] = [] | |
| const params: unknown[] = [] | |
| let idx = 1 | |
| keys.forEach(k => { | |
| setParts.push(`${quote(k)}=$${idx++}`) | |
| params.push(options.data[k]) | |
| }) | |
| sql += setParts.join(", ") | |
| const { text: whereText, params: whereParams } = buildWhereClause(options.where, idx) | |
| if (whereText) sql += ` ${whereText}` | |
| params.push(...whereParams) | |
| if (options.returnData) sql += " RETURNING *" | |
| return { sql, params } | |
| } | |
| export function buildUpdate(options: { schema: string; table: string; data: Record<string, unknown>; where: WhereCondition[]; returnData: boolean }): BuiltQuery { | |
| if (!options.where || options.where.length === 0) { | |
| throw new Error("UPDATE without WHERE conditions is not allowed.") | |
| } | |
| const tableRef = `${quote(options.schema)}.${quote(options.table)}` | |
| const keys = Object.keys(options.data) | |
| let sql = `UPDATE ${tableRef} SET ` | |
| const setParts: string[] = [] | |
| const params: unknown[] = [] | |
| let idx = 1 | |
| keys.forEach(k => { | |
| setParts.push(`${quote(k)}=$${idx++}`) | |
| params.push(options.data[k]) | |
| }) | |
| sql += setParts.join(", ") | |
| const { text: whereText, params: whereParams } = buildWhereClause(options.where, idx) | |
| if (whereText) sql += ` ${whereText}` | |
| params.push(...whereParams) | |
| if (options.returnData) sql += " RETURNING *" | |
| return { sql, params } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/postgres/query-builder.ts` around lines
178 - 201, The buildUpdate function currently allows updates with an empty WHERE
and must be protected similarly to buildDelete; update buildUpdate (function
name buildUpdate) to validate options.where and throw an error when the where
clause is empty (or undefined/length===0) to prevent accidental full-table
updates, unless a new explicit flag (e.g., allowFullTableUpdate?: boolean) is
provided on the options and set to true—mirror the same guard logic/behavior
used in buildDelete so the function throws by default and only permits
full-table updates when the flag is present and true.
Comment on lines
+1
to
+13
| import { channel, topic } from "@inngest/realtime" | ||
|
|
||
| export const AGGREGATE_CHANNEL = "aggregate-execution" | ||
|
|
||
| export const aggregateChannelName = (nodeId?: string): string => | ||
| `${AGGREGATE_CHANNEL}${nodeId ? `-${nodeId}` : ""}` | ||
|
|
||
| export const aggregateChannel = channel(aggregateChannelName).addTopic( | ||
| topic("status").type<{ | ||
| nodeId: string | ||
| status: "loading" | "success" | "error" | ||
| }>() | ||
| ) |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check how aggregateChannel is used in the executor
rg -n "aggregateChannel" --type=ts -C 3 src/features/executions/components/aggregate/Repository: Mayank-saraswal/n8n
Length of output: 5113
Critical bug: aggregateChannel is called as a function but is not a function.
The executor attempts to call aggregateChannel(nodeId) on lines 56, 434, and 455, but aggregateChannel is a static channel object, not a function. This will cause a runtime error. The channel definition creates a static object via channel(aggregateChannelName).addTopic(...), which cannot be invoked with arguments.
The executor should use aggregateChannelName(nodeId) (the string function) instead, or the channel definition needs to be restructured to support dynamic nodeId at invocation time.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/inngest/channels/aggregate.ts` around lines 1 - 13, The runtime error
comes from invoking aggregateChannel as a function; aggregateChannel is a static
channel object created by channel(aggregateChannelName).addTopic(...). Update
the executor calls that currently call aggregateChannel(nodeId) to use the
string factory aggregateChannelName(nodeId) instead, or alternatively refactor
this file to expose a function that builds a channel per node (e.g., a factory
function that calls channel(aggregateChannelName(nodeId)).addTopic(...)); target
the symbols aggregateChannel (stop invoking it), aggregateChannelName (use it to
build the channel name), and AGGREGATE_CHANNEL (keep constant) when making the
change.
Comment on lines
+1
to
+13
| import { channel, topic } from "@inngest/realtime" | ||
|
|
||
| export const POSTGRES_CHANNEL = "postgres-execution" | ||
|
|
||
| export const postgresChannelName = (nodeId?: string): string => | ||
| `${POSTGRES_CHANNEL}${nodeId ? "-" + nodeId : ""}` | ||
|
|
||
| export const postgresChannel = channel(postgresChannelName).addTopic( | ||
| topic("status").type<{ | ||
| nodeId: string | ||
| status: "loading" | "success" | "error" | ||
| }>() | ||
| ) |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check how postgresChannel is used in the executor
rg -n "postgresChannel" --type=ts -C 3 src/features/executions/components/postgres/Repository: Mayank-saraswal/n8n
Length of output: 5196
Fix executor: postgresChannel is not callable.
The executor incorrectly invokes postgresChannel(nodeId) on lines 59, 784, and 804. postgresChannel is a channel object (created by channel().addTopic()), not a function. Either:
- Change executor to use
postgresChannelName(nodeId)instead, or - Change the channel definition to export a function that creates per-nodeId channels.
Compare with correct usage in node.tsx (line 86) and actions.ts (line 10), which both use postgresChannelName(nodeId).
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/inngest/channels/postgres.ts` around lines 1 - 13, The executor is
calling postgresChannel(nodeId) but postgresChannel is a channel object created
by channel(...).addTopic(), not a function; update the executor to call
postgresChannelName(nodeId) wherever it currently invokes
postgresChannel(nodeId) (references found around the executor usage) so it
passes the string channel name, or alternatively change the export to a function
factory instead of postgresChannel if you prefer per-call construction;
specifically replace calls to postgresChannel(nodeId) with
postgresChannelName(nodeId) to match usage in node.tsx and actions.ts and keep
the exported postgresChannel and postgresChannelName definitions unchanged.
Comment on lines
+68
to
+74
| const workflow = await prisma.workflow.findUnique({ | ||
| where: { id: input.workflowId }, | ||
| select: { userId: true }, | ||
| }) | ||
| if (!workflow || workflow.userId !== ctx.auth.user.id) { | ||
| throw new TRPCError({ code: "UNAUTHORIZED" }) | ||
| } |
Contributor
There was a problem hiding this comment.
Authorize the update path against the existing node, not just input.workflowId.
Lines 68-74 prove the caller owns the workflow id from the request, but Lines 90-93 upsert by nodeId alone. If that nodeId already belongs to another user's aggregate node, an authenticated caller can pass their own workflowId and overwrite the foreign row. Load the existing node first and enforce ownership on the update path, or make the uniqueness scope (workflowId, nodeId).
Also applies to: 90-94
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/server/routers/aggregate.router.ts` around lines 68 - 74, The ownership
check currently validates only the provided workflowId via
prisma.workflow.findUnique, but the upsert later targets an aggregate node by
nodeId alone (allowing a user to overwrite another user's node); update the
logic in aggregate.router to first load the existing aggregate node (e.g., via
prisma.aggregate.findUnique/findFirst by nodeId) and verify its workflowId
matches ctx.auth.user.id before performing an update, or change the upsert
uniqueness to use the composite key (workflowId, nodeId) so writes are scoped to
the caller's workflow; ensure checks reference the existing node retrieval
(prisma.aggregate.findUnique/findFirst) and the upsert call to prevent
cross-user overwrites.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/features/executions/components/filter/executor.ts`:
- Line 36: The code is calling filterChannel with an extra invocation
(filterChannel(nodeId)()) where filterChannel returns the channel object
directly; update all occurrences to call filterChannel(nodeId).status(...)
instead of filterChannel(nodeId)().status(...). Locate uses in the executor
module around the publish calls (search for publish(filterChannel(...))) and
replace the double-call pattern with a single call so
publish(filterChannel(nodeId).status({ nodeId, status: "loading" })) and the
analogous status/error/success calls use filterChannel(nodeId).status(...)
consistently.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: cf625306-b089-41ee-bed0-7e5aae2798a2
📒 Files selected for processing (2)
src/features/executions/components/filter/executor.tssrc/inngest/channels/filter.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- src/inngest/channels/filter.ts
| // ── Step 2: Execute filter ───────────────────────────────────────────────── | ||
| let result: Record<string, unknown> | ||
|
|
||
| await publish(filterChannel(nodeId)().status({ nodeId, status: "loading" })) |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check filterChannel definition to verify the expected call pattern
echo "=== filterChannel definition ==="
cat src/inngest/channels/filter.ts
echo ""
echo "=== Compare with sortChannel (known working pattern) ==="
cat src/inngest/channels/sort.ts | head -20Repository: Mayank-saraswal/n8n
Length of output: 951
Fix the double invocation pattern filterChannel(nodeId)() at lines 36, 223, and 248.
The filterChannel definition returns a channel object directly and does not have a trailing () like sortChannel does. The correct call pattern is filterChannel(nodeId).status(...) without the extra invocation.
Change:
- Line 36:
filterChannel(nodeId)().status(...)→filterChannel(nodeId).status(...) - Line 223: Same fix required
- Line 248: Same fix required
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/filter/executor.ts` at line 36, The code
is calling filterChannel with an extra invocation (filterChannel(nodeId)())
where filterChannel returns the channel object directly; update all occurrences
to call filterChannel(nodeId).status(...) instead of
filterChannel(nodeId)().status(...). Locate uses in the executor module around
the publish calls (search for publish(filterChannel(...))) and replace the
double-call pattern with a single call so publish(filterChannel(nodeId).status({
nodeId, status: "loading" })) and the analogous status/error/success calls use
filterChannel(nodeId).status(...) consistently.
Merged
Merged
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.
Summary by CodeRabbit
New Features
Bug Fixes