Conversation
📝 WalkthroughWalkthroughThis PR introduces a comprehensive worker retry and dead-letter queue system. It adds configurable retry policies via environment variables, shared payload schemas with versioning, a new worker service for processing ingress jobs with transient/permanent error classification, and DLQ replay functionality for recovering failed jobs. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant API
participant IngressQueue as Ingress Queue<br/>(Redis/BullMQ)
participant Worker
participant Processor
participant DLQQueue as DLQ Queue<br/>(Redis/BullMQ)
Client->>API: POST webhook
activate API
API->>API: validateEnv & resolveWorkerRetryPolicy
API->>IngressQueue: createBullMqIngressQueue(redisUrl, retryPolicy)
API->>IngressQueue: enqueue(createIngressJobPayload({...}))
deactivate API
IngressQueue->>Worker: Job available
activate Worker
Worker->>Worker: runIngressJob with timeout
Worker->>Processor: Execute processor(payload)
activate Processor
alt Processor Success
Processor-->>Worker: resolved
Worker->>IngressQueue: Job complete
else Transient Error (retryable)
Processor-->>Worker: Error thrown
Worker->>Worker: classifyWorkerError → 'transient'
Worker->>Worker: Check attemptsMade < maxAttempts
alt Within retry limit
Worker->>IngressQueue: Retry (exponential backoff)
else Retries exhausted
Worker->>Worker: buildIngressDlqPayloadFromFailure
Worker->>DLQQueue: routeFailedIngressJobToDlq
DLQQueue->>DLQQueue: enqueue(dlqPayload)
end
else Permanent Error
Processor-->>Worker: Error thrown
Worker->>Worker: classifyWorkerError → 'permanent'
alt permanentMaxAttempts exceeded
Worker->>Worker: buildIngressDlqPayloadFromFailure
Worker->>DLQQueue: routeFailedIngressJobToDlq
DLQQueue->>DLQQueue: enqueue(dlqPayload)
else Still has attempts
Worker->>IngressQueue: Retry
end
end
deactivate Processor
deactivate Worker
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (4)
packages/shared/src/index.js (1)
140-146: Validatefailure.reasonandfailure.errorClassas a pair.The current guard accepts contradictory combinations like
transient_retries_exhaustedwitherrorClass: "permanent". Tightening that invariant here keeps DLQ records self-consistent for replay and reporting.💡 Suggested fix
if (failure.reason !== 'transient_retries_exhausted' && failure.reason !== 'permanent_retries_exhausted') { return false; } if (failure.errorClass !== 'transient' && failure.errorClass !== 'permanent') { return false; } + if ((failure.reason === 'transient_retries_exhausted' && failure.errorClass !== 'transient') || + (failure.reason === 'permanent_retries_exhausted' && failure.errorClass !== 'permanent')) { + return false; + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/shared/src/index.js` around lines 140 - 146, The current guards validate failure.reason and failure.errorClass independently, allowing inconsistent pairs (e.g., reason 'transient_retries_exhausted' with errorClass 'permanent'); update the logic around the failure.reason/failure.errorClass checks to enforce a consistent pair: if failure.reason startsWith('transient') or equals 'transient_retries_exhausted' then require failure.errorClass === 'transient', and if failure.reason startsWith('permanent') or equals 'permanent_retries_exhausted' then require failure.errorClass === 'permanent' (return false on mismatch); locate and modify the block using failure.reason and failure.errorClass to implement this paired validation.apps/api/src/index.js (1)
3-4: Use the shared JSON coercer on this enqueue path.This code now relies on
@wa-chat/sharedfor the payload contract, but it still bypasses the sharedcoerceJsonValueimplementation inpackages/shared/src/index.tsLines 107-131 by keeping a second copy locally. That makes payload semantics drift-prone: the next fix in the shared package will have to be duplicated here by hand.Also applies to: 457-460
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/api/src/index.js` around lines 3 - 4, The enqueue path duplicates JSON coercion instead of using the shared implementation; import and use the shared coerceJsonValue from `@wa-chat/shared` where the code builds the ingress payload (alongside INGRESS_JOB_NAME, INGRESS_QUEUE_NAME, createIngressJobPayload) and remove the local duplicate coercer; also search for the same local copy used in the other enqueue location (the later enqueue block) and replace it with the shared coerceJsonValue so payload semantics follow the single source of truth.apps/worker/src/dlq/replay.ts (1)
417-432: Minor inefficiency in job fetching logic.Line 429 uses
Math.max(args.limit * 5, args.limit) - 1, but sincelimitis always positive (validated earlier),limit * 5is always greater than or equal tolimit, makingMath.maxredundant.Additionally, fetching 5x the limit to account for filtered jobs may be inefficient for large queues. Consider documenting this trade-off or using pagination.
🔧 Simplify the expression
return dlqQueue.getJobs( ['wait', 'paused', 'prioritized', 'delayed'], 0, - Math.max(args.limit * 5, args.limit) - 1, + args.limit * 5 - 1, false, );🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/worker/src/dlq/replay.ts` around lines 417 - 432, The inspectJobs function currently calls dlqQueue.getJobs with an upper index computed as Math.max(args.limit * 5, args.limit) - 1 which is redundant because args.limit is validated positive; replace that expression with (args.limit * 5) - 1 (or better, implement simple pagination) and document the trade-off that we fetch 5× the requested limit to compensate for filtered/skip-prone jobs; update the call site in inspectJobs and any related comments to reflect the simplified expression and optional pagination note.apps/worker/src/queue/consumer.ts (1)
104-142: Consider documenting the timeout error classification behavior.
WorkerJobTimeoutErroris classified as transient (not in the permanent error list), meaning timeout failures will be retried. This seems intentional (transient network/service delays), but it's worth confirming this matches the desired behavior for job processing timeouts.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/worker/src/queue/consumer.ts` around lines 104 - 142, The runIngressJob function currently lets WorkerJobTimeoutError be classified as transient by classifyWorkerError (so timeouts are retried); add an explicit code comment in runIngressJob (near the try/catch and error-class handling) stating that WorkerJobTimeoutError is treated as transient and will be retried per policies.retry.permanentMaxAttempts, and include guidance that if timeouts should be treated as permanent instead the change must be made in classifyWorkerError or by adding an explicit branch in runIngressJob to handle WorkerJobTimeoutError as permanent; reference runIngressJob, classifyWorkerError, WorkerJobTimeoutError, and policies.retry.permanentMaxAttempts in the comment for clarity.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@apps/worker/package.json`:
- Line 10: Update the "dlq:replay" npm script to prebuild the workspace packages
before running the replay script: ensure it runs the same prebuild command used
by other scripts to build `@wa-chat/config` and `@wa-chat/shared` first, then
execute the node --import tsx src/dlq/replay.ts command so imports resolve from
the generated dist/ entrypoints for those packages.
In `@packages/shared/src/index.js`:
- Around line 67-81: createIngressJobPayload currently returns input.payload
untouched which can later fail assertIngressJobPayload or serialize
inconsistently; mirror the behavior of createIngressDlqJobPayload by
constructing the job object (using INGRESS_JOB_SCHEMA_VERSION, eventKey,
payload, receivedAt) and then validate it immediately with
assertIngressJobPayload(job) (or throw the assertion error) and optionally
deep-clone/normalize payload if createIngressDlqJobPayload does so; update
createIngressJobPayload to call assertIngressJobPayload before returning to
ensure the helper validates its own output.
- Around line 7-45: isJsonValue and coerceJsonValue currently accept non-plain
objects (Date, Map) and non-finite numbers (NaN, Infinity); update validations
so they only allow plain objects and finite numbers. Change isRecord to detect
plain objects (e.g., Object.getPrototypeOf(value) === Object.prototype or
value.constructor === Object) and use Number.isFinite to ensure numbers are
finite in isJsonValue and coerceJsonValue; for arrays still recurse, but when
encountering a non-plain object or a non-finite number, return false from
isJsonValue and return null from coerceJsonValue (instead of silently coercing),
referencing the functions isRecord, isJsonValue, and coerceJsonValue to locate
edits.
In `@packages/shared/src/index.ts`:
- Around line 146-147: The current checks use Date.parse which accepts non-ISO
formats; update isIngressJobPayload, createIngressJobPayload, and
isIngressDlqJobPayload to validate receivedAt with a strict ISO 8601 timestamp
pattern (e.g. regex that enforces YYYY-MM-DDThh:mm:ss(.sss)?(Z|±hh:mm)) instead
of relying solely on Date.parse, and only accept the value if it matches the
regex and parses to a valid Date; replace the `typeof value.receivedAt !==
'string' || Number.isNaN(Date.parse(value.receivedAt))` checks with a combined
regex.test(value.receivedAt) && !Number.isNaN(Date.parse(value.receivedAt))
guard so the code enforces strict ISO format while still ensuring it parses.
- Around line 66-67: Update isJsonValue, coerceJsonValue, and
createIngressJobPayload to reject non-JSON-serializable values: in isJsonValue
use Number.isFinite for numeric checks (so NaN and Infinity return false), treat
only plain objects (use the existing isRecord predicate) as valid object values
(so Date and other class instances are rejected), and ensure arrays and plain
objects are validated recursively; in coerceJsonValue mirror those rules by
rejecting non-plain objects instead of letting them pass through and only coerce
arrays/objects that consist entirely of valid JSON values; finally, validate the
final payload in createIngressJobPayload with isJsonValue before returning and
throw or return an error on invalid payloads.
---
Nitpick comments:
In `@apps/api/src/index.js`:
- Around line 3-4: The enqueue path duplicates JSON coercion instead of using
the shared implementation; import and use the shared coerceJsonValue from
`@wa-chat/shared` where the code builds the ingress payload (alongside
INGRESS_JOB_NAME, INGRESS_QUEUE_NAME, createIngressJobPayload) and remove the
local duplicate coercer; also search for the same local copy used in the other
enqueue location (the later enqueue block) and replace it with the shared
coerceJsonValue so payload semantics follow the single source of truth.
In `@apps/worker/src/dlq/replay.ts`:
- Around line 417-432: The inspectJobs function currently calls dlqQueue.getJobs
with an upper index computed as Math.max(args.limit * 5, args.limit) - 1 which
is redundant because args.limit is validated positive; replace that expression
with (args.limit * 5) - 1 (or better, implement simple pagination) and document
the trade-off that we fetch 5× the requested limit to compensate for
filtered/skip-prone jobs; update the call site in inspectJobs and any related
comments to reflect the simplified expression and optional pagination note.
In `@apps/worker/src/queue/consumer.ts`:
- Around line 104-142: The runIngressJob function currently lets
WorkerJobTimeoutError be classified as transient by classifyWorkerError (so
timeouts are retried); add an explicit code comment in runIngressJob (near the
try/catch and error-class handling) stating that WorkerJobTimeoutError is
treated as transient and will be retried per
policies.retry.permanentMaxAttempts, and include guidance that if timeouts
should be treated as permanent instead the change must be made in
classifyWorkerError or by adding an explicit branch in runIngressJob to handle
WorkerJobTimeoutError as permanent; reference runIngressJob,
classifyWorkerError, WorkerJobTimeoutError, and
policies.retry.permanentMaxAttempts in the comment for clarity.
In `@packages/shared/src/index.js`:
- Around line 140-146: The current guards validate failure.reason and
failure.errorClass independently, allowing inconsistent pairs (e.g., reason
'transient_retries_exhausted' with errorClass 'permanent'); update the logic
around the failure.reason/failure.errorClass checks to enforce a consistent
pair: if failure.reason startsWith('transient') or equals
'transient_retries_exhausted' then require failure.errorClass === 'transient',
and if failure.reason startsWith('permanent') or equals
'permanent_retries_exhausted' then require failure.errorClass === 'permanent'
(return false on mismatch); locate and modify the block using failure.reason and
failure.errorClass to implement this paired validation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 7ced5f86-5f40-4d7b-9ec2-c52419e4d72e
⛔ Files ignored due to path filters (8)
apps/api/src/index.d.ts.mapis excluded by!**/*.mapapps/api/src/index.js.mapis excluded by!**/*.mapapps/api/src/index.test.js.mapis excluded by!**/*.mappackage-lock.jsonis excluded by!**/package-lock.jsonpackages/config/src/index.d.ts.mapis excluded by!**/*.mappackages/config/src/index.js.mapis excluded by!**/*.mappackages/shared/src/index.d.ts.mapis excluded by!**/*.mappackages/shared/src/index.js.mapis excluded by!**/*.map
📒 Files selected for processing (24)
apps/api/.env.exampleapps/api/package.jsonapps/api/src/index.d.tsapps/api/src/index.jsapps/api/src/index.test.jsapps/api/src/index.test.tsapps/api/src/index.tsapps/api/tsconfig.jsonapps/worker/.env.exampleapps/worker/package.jsonapps/worker/src/dlq/replay.test.tsapps/worker/src/dlq/replay.tsapps/worker/src/index.test.tsapps/worker/src/index.tsapps/worker/src/queue/consumer.tspackages/config/package.jsonpackages/config/src/env.test.tspackages/config/src/index.d.tspackages/config/src/index.jspackages/config/src/index.tspackages/shared/src/index.d.tspackages/shared/src/index.jspackages/shared/src/index.tspackages/shared/tsconfig.json
| "build": "npm --workspace @wa-chat/config run build && tsc", | ||
| "build": "npm --workspace @wa-chat/config run build && npm --workspace @wa-chat/shared run build && tsc", | ||
| "dev": "tsx watch src/index.ts", | ||
| "dlq:replay": "node --import tsx src/dlq/replay.ts", |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== replay imports =="
sed -n '1,25p' apps/worker/src/dlq/replay.ts
echo
echo "== workspace package entrypoints =="
python <<'PY'
import json
from pathlib import Path
for manifest_path in [
Path("packages/config/package.json"),
Path("packages/shared/package.json"),
]:
data = json.loads(manifest_path.read_text())
print(f"--- {manifest_path} ---")
subset = {key: data.get(key) for key in ("name", "main", "module", "types", "exports")}
print(json.dumps(subset, indent=2))
print("dist exists:", (manifest_path.parent / "dist").exists())
print()
PYRepository: Ryanakml/ChatFuse
Length of output: 1418
Add prebuild step to dlq:replay to match other scripts.
The replay script imports @wa-chat/config and @wa-chat/shared, both of which have entrypoints configured to resolve through dist/ directories. Those directories don't currently exist on disk. Without prebuilding these workspace packages first, npm run dlq:replay will fail on a clean checkout. Add the same prebuild prefix you used for build, test, and typecheck:
"dlq:replay": "npm run build -w `@wa-chat/config` -w `@wa-chat/shared` && node --import tsx src/dlq/replay.ts",
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/worker/package.json` at line 10, Update the "dlq:replay" npm script to
prebuild the workspace packages before running the replay script: ensure it runs
the same prebuild command used by other scripts to build `@wa-chat/config` and
`@wa-chat/shared` first, then execute the node --import tsx src/dlq/replay.ts
command so imports resolve from the generated dist/ entrypoints for those
packages.
Comment on lines
+7
to
+45
| const isRecord = (value) => typeof value === 'object' && value !== null && !Array.isArray(value); | ||
| const isNonEmptyString = (value) => typeof value === 'string' && value.trim() !== ''; | ||
| const isNullableInteger = (value, options) => value === null || (typeof value === 'number' && Number.isInteger(value) && value >= options.minimum); | ||
| const isNullableFiniteNumber = (value) => value === null || (typeof value === 'number' && Number.isFinite(value)); | ||
| const isNullableJitter = (value) => value === null || (typeof value === 'number' && Number.isFinite(value) && value >= 0 && value <= 1); | ||
| export const isJsonValue = (value) => { | ||
| if (value === null || | ||
| typeof value === 'string' || | ||
| typeof value === 'number' || | ||
| typeof value === 'boolean') { | ||
| return true; | ||
| } | ||
| if (Array.isArray(value)) { | ||
| return value.every((entry) => isJsonValue(entry)); | ||
| } | ||
| if (!isRecord(value)) { | ||
| return false; | ||
| } | ||
| return Object.values(value).every((entry) => isJsonValue(entry)); | ||
| }; | ||
| export const coerceJsonValue = (value) => { | ||
| if (value === null || | ||
| typeof value === 'string' || | ||
| typeof value === 'number' || | ||
| typeof value === 'boolean') { | ||
| return value; | ||
| } | ||
| if (Array.isArray(value)) { | ||
| return value.map((entry) => coerceJsonValue(entry)); | ||
| } | ||
| if (!isRecord(value)) { | ||
| return null; | ||
| } | ||
| const jsonObject = {}; | ||
| for (const [key, entry] of Object.entries(value)) { | ||
| jsonObject[key] = coerceJsonValue(entry); | ||
| } | ||
| return jsonObject; | ||
| }; |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
In JavaScript, are NaNandInfinityvalid JSON numbers, and what doesJSON.stringifyemit for them? Also, doDateandMap instances count as plain JSON objects, or do they serialize differently?
💡 Result:
-
NaN/Infinityas JSON numbers: No—JSON numbers must match the JSON number grammar, and non-finite numeric values likeNaNandInfinityare not permitted in JSON. [1]
JSON.stringifyoutput: JavaScript’sJSON.stringifyserializesNaN,Infinity, and-Infinityas the JSON literalnull(e.g.,JSON.stringify(NaN) === "null"). [2] -
Dateinstances: ADateis not a “plain JSON object” in the sense of preserving its object structure/type. DuringJSON.stringify,Date.prototype.toJSON()is called, and it produces an ISO string (same format astoISOString()), so dates serialize as JSON strings. [3] -
Mapinstances: AMapis also not a plain JSON object forJSON.stringify.JSON.stringifyonly serializes an object’s own enumerable string-keyed properties;Mapentries aren’t stored that way, so aMaptypically stringifies as{}unless you transform it (e.g.,Object.fromEntries(map)orArray.from(map)) or provide a replacer. [4][5]
Sources
[1] RFC 8259 (JSON) — non-finite numbers not permitted: “Infinity and NaN are not permitted.” (datatracker.ietf.org)
[2] ECMAScript spec — JSON.stringify renders NaN/Infinity as "null". (tc39.es)
[3] MDN — Date.prototype.toJSON() returns an ISO string and is called by JSON.stringify. (developer.mozilla.org)
[4] ECMAScript spec — serialization uses enumerable own properties. (tc39.es)
[5] Map has no enumerable own properties for entries → typically {}. (stackoverflow.com)
Citations:
- 1: https://datatracker.ietf.org/doc/rfc8259/?utm_source=openai
- 2: https://tc39.es/ecma262/multipage/structured-data.html?utm_source=openai
- 3: https://developer.mozilla.org/docs/Web/JavaScript/Reference/Global_Objects/Date/tojson?utm_source=openai
- 4: https://tc39.es/ecma262/multipage/structured-data.html?utm_source=openai
- 5: https://stackoverflow.com/questions/65041179/json-stringifynew-map-prints-instead-of-for-a-particular-org?utm_source=openai
Restrict the JSON helpers to plain objects and finite numbers.
isJsonValue currently accepts NaN and Infinity, which are invalid in JSON and serialize to null during JSON.stringify. Additionally, isRecord accepts non-plain objects like Date and Map instances, which serialize to JSON strings and empty objects respectively—causing payloads to round-trip unpredictably. coerceJsonValue silently rewrites these values instead of rejecting them.
Suggested fix
-const isRecord = (value) => typeof value === 'object' && value !== null && !Array.isArray(value);
+const isRecord = (value) => {
+ if (typeof value !== 'object' || value === null || Array.isArray(value)) {
+ return false;
+ }
+ const prototype = Object.getPrototypeOf(value);
+ return prototype === Object.prototype || prototype === null;
+};
...
- typeof value === 'number' ||
+ (typeof value === 'number' && Number.isFinite(value)) ||
...
- typeof value === 'number' ||
+ (typeof value === 'number' && Number.isFinite(value)) ||📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| const isRecord = (value) => typeof value === 'object' && value !== null && !Array.isArray(value); | |
| const isNonEmptyString = (value) => typeof value === 'string' && value.trim() !== ''; | |
| const isNullableInteger = (value, options) => value === null || (typeof value === 'number' && Number.isInteger(value) && value >= options.minimum); | |
| const isNullableFiniteNumber = (value) => value === null || (typeof value === 'number' && Number.isFinite(value)); | |
| const isNullableJitter = (value) => value === null || (typeof value === 'number' && Number.isFinite(value) && value >= 0 && value <= 1); | |
| export const isJsonValue = (value) => { | |
| if (value === null || | |
| typeof value === 'string' || | |
| typeof value === 'number' || | |
| typeof value === 'boolean') { | |
| return true; | |
| } | |
| if (Array.isArray(value)) { | |
| return value.every((entry) => isJsonValue(entry)); | |
| } | |
| if (!isRecord(value)) { | |
| return false; | |
| } | |
| return Object.values(value).every((entry) => isJsonValue(entry)); | |
| }; | |
| export const coerceJsonValue = (value) => { | |
| if (value === null || | |
| typeof value === 'string' || | |
| typeof value === 'number' || | |
| typeof value === 'boolean') { | |
| return value; | |
| } | |
| if (Array.isArray(value)) { | |
| return value.map((entry) => coerceJsonValue(entry)); | |
| } | |
| if (!isRecord(value)) { | |
| return null; | |
| } | |
| const jsonObject = {}; | |
| for (const [key, entry] of Object.entries(value)) { | |
| jsonObject[key] = coerceJsonValue(entry); | |
| } | |
| return jsonObject; | |
| }; | |
| const isRecord = (value) => { | |
| if (typeof value !== 'object' || value === null || Array.isArray(value)) { | |
| return false; | |
| } | |
| const prototype = Object.getPrototypeOf(value); | |
| return prototype === Object.prototype || prototype === null; | |
| }; | |
| const isNonEmptyString = (value) => typeof value === 'string' && value.trim() !== ''; | |
| const isNullableInteger = (value, options) => value === null || (typeof value === 'number' && Number.isInteger(value) && value >= options.minimum); | |
| const isNullableFiniteNumber = (value) => value === null || (typeof value === 'number' && Number.isFinite(value)); | |
| const isNullableJitter = (value) => value === null || (typeof value === 'number' && Number.isFinite(value) && value >= 0 && value <= 1); | |
| export const isJsonValue = (value) => { | |
| if (value === null || | |
| typeof value === 'string' || | |
| (typeof value === 'number' && Number.isFinite(value)) || | |
| typeof value === 'boolean') { | |
| return true; | |
| } | |
| if (Array.isArray(value)) { | |
| return value.every((entry) => isJsonValue(entry)); | |
| } | |
| if (!isRecord(value)) { | |
| return false; | |
| } | |
| return Object.values(value).every((entry) => isJsonValue(entry)); | |
| }; | |
| export const coerceJsonValue = (value) => { | |
| if (value === null || | |
| typeof value === 'string' || | |
| (typeof value === 'number' && Number.isFinite(value)) || | |
| typeof value === 'boolean') { | |
| return value; | |
| } | |
| if (Array.isArray(value)) { | |
| return value.map((entry) => coerceJsonValue(entry)); | |
| } | |
| if (!isRecord(value)) { | |
| return null; | |
| } | |
| const jsonObject = {}; | |
| for (const [key, entry] of Object.entries(value)) { | |
| jsonObject[key] = coerceJsonValue(entry); | |
| } | |
| return jsonObject; | |
| }; |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/shared/src/index.js` around lines 7 - 45, isJsonValue and
coerceJsonValue currently accept non-plain objects (Date, Map) and non-finite
numbers (NaN, Infinity); update validations so they only allow plain objects and
finite numbers. Change isRecord to detect plain objects (e.g.,
Object.getPrototypeOf(value) === Object.prototype or value.constructor ===
Object) and use Number.isFinite to ensure numbers are finite in isJsonValue and
coerceJsonValue; for arrays still recurse, but when encountering a non-plain
object or a non-finite number, return false from isJsonValue and return null
from coerceJsonValue (instead of silently coercing), referencing the functions
isRecord, isJsonValue, and coerceJsonValue to locate edits.
Comment on lines
+67
to
+81
| export const createIngressJobPayload = (input) => { | ||
| const eventKey = input.eventKey.trim(); | ||
| if (!eventKey) { | ||
| throw new Error('Ingress job eventKey must be a non-empty string'); | ||
| } | ||
| const receivedAt = input.receivedAt ?? new Date().toISOString(); | ||
| if (Number.isNaN(Date.parse(receivedAt))) { | ||
| throw new Error('Ingress job receivedAt must be a valid ISO timestamp'); | ||
| } | ||
| return { | ||
| schemaVersion: INGRESS_JOB_SCHEMA_VERSION, | ||
| eventKey, | ||
| payload: input.payload, | ||
| receivedAt, | ||
| }; |
There was a problem hiding this comment.
Have the ingress builder validate its own output.
Unlike createIngressDlqJobPayload, this helper returns input.payload untouched. A caller can build an object that later fails assertIngressJobPayload or serializes differently once enqueued.
💡 Suggested fix
export const createIngressJobPayload = (input) => {
const eventKey = input.eventKey.trim();
if (!eventKey) {
throw new Error('Ingress job eventKey must be a non-empty string');
}
const receivedAt = input.receivedAt ?? new Date().toISOString();
if (Number.isNaN(Date.parse(receivedAt))) {
throw new Error('Ingress job receivedAt must be a valid ISO timestamp');
}
- return {
+ return assertIngressJobPayload({
schemaVersion: INGRESS_JOB_SCHEMA_VERSION,
eventKey,
payload: input.payload,
receivedAt,
- };
+ });
};📝 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 const createIngressJobPayload = (input) => { | |
| const eventKey = input.eventKey.trim(); | |
| if (!eventKey) { | |
| throw new Error('Ingress job eventKey must be a non-empty string'); | |
| } | |
| const receivedAt = input.receivedAt ?? new Date().toISOString(); | |
| if (Number.isNaN(Date.parse(receivedAt))) { | |
| throw new Error('Ingress job receivedAt must be a valid ISO timestamp'); | |
| } | |
| return { | |
| schemaVersion: INGRESS_JOB_SCHEMA_VERSION, | |
| eventKey, | |
| payload: input.payload, | |
| receivedAt, | |
| }; | |
| export const createIngressJobPayload = (input) => { | |
| const eventKey = input.eventKey.trim(); | |
| if (!eventKey) { | |
| throw new Error('Ingress job eventKey must be a non-empty string'); | |
| } | |
| const receivedAt = input.receivedAt ?? new Date().toISOString(); | |
| if (Number.isNaN(Date.parse(receivedAt))) { | |
| throw new Error('Ingress job receivedAt must be a valid ISO timestamp'); | |
| } | |
| return assertIngressJobPayload({ | |
| schemaVersion: INGRESS_JOB_SCHEMA_VERSION, | |
| eventKey, | |
| payload: input.payload, | |
| receivedAt, | |
| }); | |
| }; |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/shared/src/index.js` around lines 67 - 81, createIngressJobPayload
currently returns input.payload untouched which can later fail
assertIngressJobPayload or serialize inconsistently; mirror the behavior of
createIngressDlqJobPayload by constructing the job object (using
INGRESS_JOB_SCHEMA_VERSION, eventKey, payload, receivedAt) and then validate it
immediately with assertIngressJobPayload(job) (or throw the assertion error) and
optionally deep-clone/normalize payload if createIngressDlqJobPayload does so;
update createIngressJobPayload to call assertIngressJobPayload before returning
to ensure the helper validates its own output.
Comment on lines
+66
to
+67
| const isRecord = (value: unknown): value is Record<string, unknown> => | ||
| typeof value === 'object' && value !== null && !Array.isArray(value); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
wc -l packages/shared/src/index.tsRepository: Ryanakml/ChatFuse
Length of output: 93
🏁 Script executed:
cat -n packages/shared/src/index.ts | head -200Repository: Ryanakml/ChatFuse
Length of output: 6805
🏁 Script executed:
cat -n packages/shared/src/index.ts | tail -130Repository: Ryanakml/ChatFuse
Length of output: 3911
🏁 Script executed:
node <<'EOF'
// Test what Object.values returns for a Date object
const dateObj = new Date();
console.log("Object.values(new Date()):", Object.values(dateObj));
console.log("Length:", Object.values(dateObj).length);
// Test isRecord on Date
const isRecord = (value) => typeof value === 'object' && value !== null && !Array.isArray(value);
console.log("isRecord(new Date()):", isRecord(dateObj));
// Test NaN and Infinity
console.log("typeof NaN:", typeof NaN);
console.log("typeof Infinity:", typeof Infinity);
console.log("NaN === NaN:", NaN === NaN);
console.log("JSON.stringify(NaN):", JSON.stringify(NaN));
console.log("JSON.stringify(Infinity):", JSON.stringify(Infinity));
console.log("JSON.stringify({a: NaN}):", JSON.stringify({a: NaN}));
EOFRepository: Ryanakml/ChatFuse
Length of output: 283
🏁 Script executed:
rg "createIngressJobPayload|isJsonValue|coerceJsonValue" -A 3 --type ts --type tsx -g "!index.ts"Repository: Ryanakml/ChatFuse
Length of output: 88
🏁 Script executed:
node <<'EOF'
// Simulating the actual behavior
const isRecord = (value) => typeof value === 'object' && value !== null && !Array.isArray(value);
const isJsonValue = (value) => {
if (
value === null ||
typeof value === 'string' ||
typeof value === 'number' ||
typeof value === 'boolean'
) {
return true;
}
if (Array.isArray(value)) {
return value.every((entry) => isJsonValue(entry));
}
if (!isRecord(value)) {
return false;
}
return Object.values(value).every((entry) => isJsonValue(entry));
};
// Test cases from the review
console.log("=== Date Objects ===");
const dateObj = new Date();
console.log("isJsonValue(new Date()):", isJsonValue(dateObj));
console.log("\n=== NaN/Infinity ===");
console.log("isJsonValue(NaN):", isJsonValue(NaN));
console.log("isJsonValue(Infinity):", isJsonValue(Infinity));
console.log("\n=== Objects with NaN/Infinity ===");
console.log("isJsonValue({a: NaN, b: Infinity}):", isJsonValue({a: NaN, b: Infinity}));
console.log("\n=== What actually happens when serialized ===");
const payload = {a: NaN, b: Infinity, c: new Date()};
console.log("Original:", payload);
console.log("After JSON round-trip:", JSON.parse(JSON.stringify(payload)));
EOFRepository: Ryanakml/ChatFuse
Length of output: 445
🏁 Script executed:
rg "createIngressJobPayload|isJsonValue|coerceJsonValue" --type ts -g "!index.ts"Repository: Ryanakml/ChatFuse
Length of output: 1577
🏁 Script executed:
# Also check if there's any input validation before createIngressJobPayload is called
rg "createIngressJobPayload" -B 5 -A 2 --type tsRepository: Ryanakml/ChatFuse
Length of output: 8600
🏁 Script executed:
rg "createIngressJobPayload" -B 10 -A 5 apps/api/src/index.tsRepository: Ryanakml/ChatFuse
Length of output: 909
Fix isJsonValue and coerceJsonValue to reject non-JSON-serializable values.
Both functions currently accept values that cannot be safely serialized to JSON:
isJsonValueacceptsNaNandInfinity(no finite-number check)isJsonValueacceptsDateobjects and other non-plain objects (only checks!isRecord)coerceJsonValuehas the same issues, allowing these invalid values to pass through
When payloads containing NaN, Infinity, or Date objects reach downstream serialization (e.g., JSON.stringify), they are silently converted to null, causing data loss. Replace the number check with Number.isFinite(), reject non-plain objects in coerceJsonValue, and validate the payload before returning from createIngressJobPayload.
Also applies to: 86–105 (isJsonValue), 107–131 (coerceJsonValue)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/shared/src/index.ts` around lines 66 - 67, Update isJsonValue,
coerceJsonValue, and createIngressJobPayload to reject non-JSON-serializable
values: in isJsonValue use Number.isFinite for numeric checks (so NaN and
Infinity return false), treat only plain objects (use the existing isRecord
predicate) as valid object values (so Date and other class instances are
rejected), and ensure arrays and plain objects are validated recursively; in
coerceJsonValue mirror those rules by rejecting non-plain objects instead of
letting them pass through and only coerce arrays/objects that consist entirely
of valid JSON values; finally, validate the final payload in
createIngressJobPayload with isJsonValue before returning and throw or return an
error on invalid payloads.
Comment on lines
+146
to
+147
| if (typeof value.receivedAt !== 'string' || Number.isNaN(Date.parse(value.receivedAt))) { | ||
| return false; |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
wc -l packages/shared/src/index.tsRepository: Ryanakml/ChatFuse
Length of output: 93
🏁 Script executed:
head -200 packages/shared/src/index.ts | tail -60Repository: Ryanakml/ChatFuse
Length of output: 1524
🏁 Script executed:
sed -n '140,180p' packages/shared/src/index.tsRepository: Ryanakml/ChatFuse
Length of output: 1104
🏁 Script executed:
sed -n '270,290p' packages/shared/src/index.tsRepository: Ryanakml/ChatFuse
Length of output: 598
🏁 Script executed:
rg -i "iso.*timestamp|timestamp.*iso" packages/shared/src/index.ts -B 3 -A 3Repository: Ryanakml/ChatFuse
Length of output: 267
🏁 Script executed:
fd -t f "\.ts$" packages/shared/src | head -20Repository: Ryanakml/ChatFuse
Length of output: 120
🏁 Script executed:
rg "IngressJobPayload|IngressDlqJobPayload" packages/shared/src -B 2 -A 5Repository: Ryanakml/ChatFuse
Length of output: 10756
🏁 Script executed:
rg "receivedAt|failedAt" packages/shared/src -B 1 -A 1 | head -40Repository: Ryanakml/ChatFuse
Length of output: 1794
🏁 Script executed:
node <<'EOF'
// Test Date.parse behavior
const testCases = [
'2024-01-15T10:30:45.123Z', // ISO 8601 with milliseconds
'2024-01-15T10:30:45Z', // ISO 8601 without milliseconds
'2024-01-15', // Date only (non-ISO)
'01/15/2024', // US format
'15 January 2024', // Locale format
'Jan 15, 2024', // Another locale format
];
testCases.forEach(test => {
const parsed = Date.parse(test);
const isValid = !Number.isNaN(parsed);
console.log(`"${test}": ${isValid ? 'ACCEPTED' : 'REJECTED'}`);
});
EOFRepository: Ryanakml/ChatFuse
Length of output: 229
Use a strict ISO 8601 timestamp check instead of Date.parse.
Date.parse() accepts many non-ISO formats including date-only strings ("2024-01-15"), US formats ("01/15/2024"), and locale-dependent strings ("15 January 2024"), even though the error message claims "valid ISO timestamp". This looser validation breaks the contract, can disrupt lexical ordering of timestamp strings, and may cause parsing issues across different languages expecting strict ISO 8601 format.
Locations: 146-147 (isIngressJobPayload), 171-174 (createIngressJobPayload), 279-280 (isIngressDlqJobPayload)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/shared/src/index.ts` around lines 146 - 147, The current checks use
Date.parse which accepts non-ISO formats; update isIngressJobPayload,
createIngressJobPayload, and isIngressDlqJobPayload to validate receivedAt with
a strict ISO 8601 timestamp pattern (e.g. regex that enforces
YYYY-MM-DDThh:mm:ss(.sss)?(Z|±hh:mm)) instead of relying solely on Date.parse,
and only accept the value if it matches the regex and parses to a valid Date;
replace the `typeof value.receivedAt !== 'string' ||
Number.isNaN(Date.parse(value.receivedAt))` checks with a combined
regex.test(value.receivedAt) && !Number.isNaN(Date.parse(value.receivedAt))
guard so the code enforces strict ISO format while still ensuring it parses.
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
Release Notes
New Features
Configuration