Feature/issue 54 orchestrator state persistence#265
Open
Danitello123 wants to merge 3 commits into
Open
Conversation
📝 WalkthroughWalkthroughThis PR adds three persistence-backed features across services: a Sequelize-managed Soroban transaction ledger (model, migration, helpers, tests) in the payments service; PostgreSQL workflow state persistence and recovery with optimistic concurrency in the orchestrator; and a Soroban RPC-polling escrow event listener with idempotent email/push dispatch in the notifications service. A shared migration runner script is also extended to apply versioned SQL files. ChangesPayments: Soroban Transaction Ledger
Orchestrator: Workflow State Persistence
Notifications: Escrow Event Listener
Migration Script
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
⚔️ Resolve merge conflicts
Comment Warning |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🧹 Nitpick comments (3)
tests/unit/src/payments-ledger.test.js (1)
45-74: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a duplicate-hash regression test for
logSubmission().These cases only cover the create path and terminal updates. The idempotency branch is
findOrCreate(...)=existing, so please add one case with an existingCONFIRMEDorFAILEDentry to lock in the expected no-downgrade behavior.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/src/payments-ledger.test.js` around lines 45 - 74, Add a regression test in payments-ledger.test.js for the idempotency branch in logSubmission() where findOrCreate returns an existing record: create or seed an entry with the same hash already in a terminal state like CONFIRMED or FAILED, call logSubmission() again, and assert it returns the existing entry without downgrading status or clearing fields like confirmedAt/errorDetails. Use the existing logSubmission() and updateLedgerStatus() test patterns to verify the no-downgrade behavior.apps/backend/notifications/src/escrowListener.test.ts (1)
64-108: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a failure-path regression here.
This only exercises the happy path, so it would not catch the two loss cases in the listener: checkpointing past a failed event and burning the idempotency key before a failed send. A focused test with
sendEmailorsendPushNotificationrejecting would lock that behavior down.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/backend/notifications/src/escrowListener.test.ts` around lines 64 - 108, Add a failure-path regression in escrowListener.test.ts around startEscrowEventListener to cover a rejected notification send. Mock either email.sendEmail or sendPushNotification to fail after the EscrowCreatedEvent is decoded, then assert the listener does not advance past the failed event and does not mark the idempotency entry as dispatched before a successful send. Use the existing mocks around idempotency.checkAndMarkDispatched, email.sendEmail, and the event listener setup to verify the retry-safe behavior.apps/backend/orchestrator/src/index.test.ts (1)
17-68: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a resumption test for startup recovery.
These cases validate the helper queries, but they never assert that startup reloads unfinished rows, skips terminal ones, and invokes
restorePurchaseWorkflowbeforestartHttpServer. That is the primary behavior introduced for issue#54.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/backend/orchestrator/src/index.test.ts` around lines 17 - 68, The current tests cover persistWorkflowState and recoverWorkflowState helpers, but they do not verify the startup resumption flow introduced for issue `#54`. Add a new test in index.test.ts around the startup path that mocks the workflow-loading query, asserts unfinished rows are reloaded while terminal rows are ignored, and checks that restorePurchaseWorkflow is called before startHttpServer. Use the existing symbols persistWorkflowState, recoverWorkflowState, restorePurchaseWorkflow, and startHttpServer to place the new assertions in the correct setup.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/backend/notifications/README.md`:
- Around line 15-21: Document the missing push notification environment
variables in the notifications README by adding VAPID_PUBLIC_KEY and
VAPID_PRIVATE_KEY to the env list. Update the section alongside the existing
notification settings so it matches the validation in push/index.ts and clearly
indicates both keys are required for the listener to start.
In `@apps/backend/notifications/src/escrowListener.ts`:
- Around line 57-67: The escrow event polling in `escrowListener.ts` advances
the Redis checkpoint too far after a single `server.getEvents` call, so later
events beyond the first 1000 can be skipped; update the polling logic around
`getEvents` and the cursor/Redis checkpoint handling to page through all results
until the requested ledger range is exhausted, and only advance the stored
cursor to the last consumed event or page boundary rather than `latestLedger`.
- Around line 193-213: The idempotency mark is being written too early in the
notification flow, so a failed send prevents retries. Update the escrow
notification path in escrowListener so the Redis dedupe key is only committed
after sendEmail/sendPushNotification succeeds, or explicitly remove/undo the
mark inside the failure path. Use the existing checkAndMarkDispatched helper and
the sendEmail/sendPushNotification branches to keep the retry behavior correct.
- Around line 74-125: The escrowListener processing loop is advancing the
checkpoint even when an event fails to parse or dispatch, which can permanently
skip notifications. Update the logic around the event iteration and the final
redis.set(lastProcessedKey, ...) so progress is only persisted after each
successfully handled event, or otherwise track and restore the last successful
cursor when dispatchEscrowNotification or parsing throws. Use the existing
escrowListener event loop, dispatchEscrowNotification, and lastProcessedKey
symbols to place the fix and ensure failures do not move the checkpoint past
unprocessed events.
In `@apps/backend/orchestrator/src/index.ts`:
- Around line 95-117: The startup recovery in orchestrator’s recovery flow is
failing open because a single outer try/catch around getPool, pool.query, and
restorePurchaseWorkflow lets the service keep starting even when recovery
breaks. Refactor the recovery logic in index.ts to fail fast on unrecoverable
query/DB errors and isolate per-row restore errors so one bad workflow row does
not stop the rest from being recovered; use the existing getPool, pool.query,
restorePurchaseWorkflow, and log.warn/log.error paths to make the recovery
outcome explicit.
- Around line 105-110: The restored MachineSnapshot in restorePurchaseWorkflow
is ignoring the persisted version and always starting at version 1, which makes
recovered workflows look new after restart. Update the snapshot construction to
use the loaded row.version value (the same one already copied into
context._dbVersion) so WorkflowSnapshot.version reflects the persisted state
when rebuilding the machine snapshot.
- Around line 41-63: Advance the in-memory context version after a successful
persist in the workflow save logic so `_dbVersion` stays in sync with the
database. Update the persist path in the `save` flow around `pool.query` to
capture the new row version for both the optimistic `UPDATE` branch and the
initial `INSERT ... ON CONFLICT DO UPDATE` branch, then assign that incremented
version back onto `context._dbVersion`. This ensures `expectedVersion` in the
same orchestrator code keeps working on later transitions and recovered
workflows do not retry with a stale version.
- Around line 128-137: The startup guard in orchestrator entrypoint logic is
still allowing import-time execution, which causes server startup when modules
like persistWorkflowState or recoverWorkflowState are imported. Update the
startup check in index.ts to use a true main-module gate around startup() and
remove the non-test fallback branch so only the actual entrypoint triggers
bootstrapping; keep the existing startup() catch/log handling tied to that
main-entry path.
In `@apps/backend/payments/events/index.ts`:
- Around line 213-219: The retry path in the transaction submission flow is
incorrectly overwriting an existing ledger row back to PENDING, which can
contradict a previously terminal state. In the duplicate-idempotency branch
inside the submission handler in events/index.ts, update the existing ledger
entry only with non-state fields like method/orderId/contractId and preserve
CONFIRMED/FAILED status plus confirmedAt/errorDetails when a row already exists.
Use the existing created check and ledgerEntry save path to ensure duplicates do
not downgrade a finalized record.
In `@apps/backend/payments/src/db.ts`:
- Line 6: The `databaseUrl` fallback currently hides a missing `DATABASE_URL`
and can send ledger writes to the wrong database. Update the `databaseUrl`
initialization in `db.ts` to require `process.env.DATABASE_URL` and fail
immediately if it is absent, matching the required configuration contract used
by the rest of the payments persistence setup.
In `@scripts/setup/migrate.js`:
- Around line 37-40: The duplicate-object handling in migrate.js is relying on
err.message text, which is too broad and can mask unrelated failures. Update the
catch block in the migration loop to inspect the SQLSTATE/code on the thrown
error instead, and only skip when the error indicates duplicate relations or
types/constraints (for example the cases handled by the migrate logic around the
existing file/relation check). Keep the existing warning path for those specific
duplicate-object errors and let all other errors continue to fail normally.
- Around line 29-43: Wrap each migration in a transaction inside the migrate
loop so a file is all-or-nothing. In scripts/setup/migrate.js, update the logic
around client.query(migrationSql) to begin a transaction before executing the
file, commit on success, and roll back on any failure before handling the error.
Keep the existing already-exists skip behavior in the catch path, but make sure
it applies only after a failed transaction and does not leave partially applied
statements behind.
---
Nitpick comments:
In `@apps/backend/notifications/src/escrowListener.test.ts`:
- Around line 64-108: Add a failure-path regression in escrowListener.test.ts
around startEscrowEventListener to cover a rejected notification send. Mock
either email.sendEmail or sendPushNotification to fail after the
EscrowCreatedEvent is decoded, then assert the listener does not advance past
the failed event and does not mark the idempotency entry as dispatched before a
successful send. Use the existing mocks around
idempotency.checkAndMarkDispatched, email.sendEmail, and the event listener
setup to verify the retry-safe behavior.
In `@apps/backend/orchestrator/src/index.test.ts`:
- Around line 17-68: The current tests cover persistWorkflowState and
recoverWorkflowState helpers, but they do not verify the startup resumption flow
introduced for issue `#54`. Add a new test in index.test.ts around the startup
path that mocks the workflow-loading query, asserts unfinished rows are reloaded
while terminal rows are ignored, and checks that restorePurchaseWorkflow is
called before startHttpServer. Use the existing symbols persistWorkflowState,
recoverWorkflowState, restorePurchaseWorkflow, and startHttpServer to place the
new assertions in the correct setup.
In `@tests/unit/src/payments-ledger.test.js`:
- Around line 45-74: Add a regression test in payments-ledger.test.js for the
idempotency branch in logSubmission() where findOrCreate returns an existing
record: create or seed an entry with the same hash already in a terminal state
like CONFIRMED or FAILED, call logSubmission() again, and assert it returns the
existing entry without downgrading status or clearing fields like
confirmedAt/errorDetails. Use the existing logSubmission() and
updateLedgerStatus() test patterns to verify the no-downgrade 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
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: f4dce224-eff7-45c1-86ee-6cb08bea8cd4
📒 Files selected for processing (15)
apps/backend/notifications/README.mdapps/backend/notifications/src/escrowListener.test.tsapps/backend/notifications/src/escrowListener.tsapps/backend/notifications/src/index.tsapps/backend/orchestrator/src/index.test.tsapps/backend/orchestrator/src/index.tsapps/backend/payments/README.mdapps/backend/payments/events/index.tsapps/backend/payments/package.jsonapps/backend/payments/src/db.tsapps/backend/payments/src/models/SorobanTransactionLedger.tsdatabase/migrations/004_soroban_transaction_ledger.sqldatabase/migrations/005_purchase_workflows.sqlscripts/setup/migrate.jstests/unit/src/payments-ledger.test.js
Comment on lines
+15
to
+21
| - `ESCROW_CONTRACT_ID`: The Soroban contract ID for the escrow contract, to listen for events. | ||
| - `SOROBAN_RPC_URL`: The Stellar RPC URL to poll for on-chain events. | ||
| - `REDIS_URL`: Redis connection URL, used for worker idempotency and deduplication. | ||
| - `DATABASE_URL`: PostgreSQL connection URL, used for wallet lookup adapter. | ||
| - `SENDGRID_API_KEY`: SendGrid API key for emails. | ||
| - `FROM_EMAIL`: Sender email address. | ||
| - `LOG_LEVEL`: Log level (e.g. info, debug). |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Document the VAPID keys too.
The new listener can dispatch push notifications, and apps/backend/notifications/push/index.ts:25-33 throws when VAPID_PUBLIC_KEY or VAPID_PRIVATE_KEY is missing. Omitting them from this env list makes the rollout instructions incomplete.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/notifications/README.md` around lines 15 - 21, Document the
missing push notification environment variables in the notifications README by
adding VAPID_PUBLIC_KEY and VAPID_PRIVATE_KEY to the env list. Update the
section alongside the existing notification settings so it matches the
validation in push/index.ts and clearly indicates both keys are required for the
listener to start.
Comment on lines
+57
to
+67
| const eventsResponse = await server.getEvents({ | ||
| startLedger, | ||
| filters: [ | ||
| { | ||
| type: "contract", | ||
| contractIds: [contractId], | ||
| topics: [["*"]], // Match any topics for this contract | ||
| }, | ||
| ], | ||
| limit: 1000, | ||
| }); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Paginate before advancing the Redis cursor.
getEvents is capped at limit: 1000, but the checkpoint jumps straight to latestLedger. If more than 1000 matching events land between polls, everything after the first page is skipped permanently. Advance the cursor only to the last consumed event/page and keep paging until the range is exhausted.
Also applies to: 124-125
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/notifications/src/escrowListener.ts` around lines 57 - 67, The
escrow event polling in `escrowListener.ts` advances the Redis checkpoint too
far after a single `server.getEvents` call, so later events beyond the first
1000 can be skipped; update the polling logic around `getEvents` and the
cursor/Redis checkpoint handling to page through all results until the requested
ledger range is exhausted, and only advance the stored cursor to the last
consumed event or page boundary rather than `latestLedger`.
Comment on lines
+74
to
+125
| try { | ||
| const topics = event.topic.map(t => scValToNative(t)); | ||
| if (topics[0] !== "escrow" || !topics[1]) continue; | ||
|
|
||
| const eventTypeRaw = topics[1]; | ||
| let eventType: EscrowContractEvent["eventType"]; | ||
|
|
||
| switch (eventTypeRaw) { | ||
| case "created": | ||
| eventType = "escrow_created"; | ||
| break; | ||
| case "resolved": | ||
| eventType = "escrow_released"; | ||
| break; | ||
| case "refunded": | ||
| eventType = "escrow_refunded"; | ||
| break; | ||
| case "disputed": | ||
| eventType = "escrow_disputed"; | ||
| break; | ||
| default: | ||
| continue; | ||
| } | ||
|
|
||
| const value = scValToNative(event.value); | ||
|
|
||
| if (eventTypeRaw === "resolved" && value && value.release_to_seller === false) { | ||
| eventType = "escrow_refunded"; | ||
| } | ||
| if (eventTypeRaw === "resolved" && value && value.release_to_seller === true) { | ||
| eventType = "escrow_released"; | ||
| } | ||
|
|
||
| const escrowEvent: EscrowContractEvent = { | ||
| contractId: event.contractId, | ||
| eventType, | ||
| orderId: value.order_id || value.orderId || "", | ||
| buyer: value.buyer || "", | ||
| merchant: value.seller || value.merchant || "", | ||
| amountStroops: value.amount ? value.amount.toString() : "0", | ||
| ledger: event.ledger, | ||
| txHash: event.txHash, | ||
| }; | ||
|
|
||
| await dispatchEscrowNotification(escrowEvent, `${event.txHash}-${event.id}`); | ||
| } catch (e) { | ||
| log.warn("Failed to parse or process event", { error: e, event }); | ||
| } | ||
| } | ||
|
|
||
| await redis.set(lastProcessedKey, latestLedger.toString()); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Don't checkpoint past failed events.
A parse/dispatch failure is only logged, then lastProcessedKey is still updated to latestLedger. That turns any transient SendGrid, VAPID, Redis, or wallet-lookup failure into a permanent missed notification. Persist progress only after each event succeeds, or rewind to the last successful cursor on failure.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/notifications/src/escrowListener.ts` around lines 74 - 125, The
escrowListener processing loop is advancing the checkpoint even when an event
fails to parse or dispatch, which can permanently skip notifications. Update the
logic around the event iteration and the final redis.set(lastProcessedKey, ...)
so progress is only persisted after each successfully handled event, or
otherwise track and restore the last successful cursor when
dispatchEscrowNotification or parsing throws. Use the existing escrowListener
event loop, dispatchEscrowNotification, and lastProcessedKey symbols to place
the fix and ensure failures do not move the checkpoint past unprocessed events.
Comment on lines
+193
to
+213
| const shouldSend = await checkAndMarkDispatched(redis, { | ||
| userId: target.userId, | ||
| channel: "email", | ||
| eventType: event.eventType, | ||
| eventId, | ||
| }); | ||
|
|
||
| if (shouldSend) { | ||
| tasks.push( | ||
| sendEmail({ | ||
| to: target.email, | ||
| subject, | ||
| templateName, | ||
| templateData: { | ||
| orderId: event.orderId, | ||
| amount: event.amountStroops, | ||
| }, | ||
| }).catch((err) => | ||
| log.error("Failed to send escrow email", { error: err, userId: target.userId }) | ||
| ) | ||
| ); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Mark idempotency after the channel send succeeds.
checkAndMarkDispatched writes the Redis NX key before sendEmail / sendPushNotification runs. If the downstream call fails, the dedupe key is already burned, so later polls cannot retry that notification. This needs a success-only mark or explicit cleanup on failure.
Also applies to: 219-247
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/notifications/src/escrowListener.ts` around lines 193 - 213, The
idempotency mark is being written too early in the notification flow, so a
failed send prevents retries. Update the escrow notification path in
escrowListener so the Redis dedupe key is only committed after
sendEmail/sendPushNotification succeeds, or explicitly remove/undo the mark
inside the failure path. Use the existing checkAndMarkDispatched helper and the
sendEmail/sendPushNotification branches to keep the retry behavior correct.
Comment on lines
+41
to
+63
| const expectedVersion = context._dbVersion as number | undefined; | ||
|
|
||
| if (expectedVersion !== undefined) { | ||
| const res = await pool.query( | ||
| `UPDATE purchase_workflows | ||
| SET state = $1, context = $2, version = version + 1, updated_at = CURRENT_TIMESTAMP | ||
| WHERE order_id = $3 AND version = $4`, | ||
| [state, JSON.stringify(context), orderId, expectedVersion] | ||
| ); | ||
| if (res.rowCount === 0) { | ||
| throw new Error(`Optimistic concurrency failure: workflow ${orderId}`); | ||
| } | ||
| } else { | ||
| await pool.query( | ||
| `INSERT INTO purchase_workflows (order_id, user_id, state, context, version) | ||
| VALUES ($1, $2, $3, $4, 1) | ||
| ON CONFLICT (order_id) DO UPDATE SET | ||
| state = EXCLUDED.state, | ||
| context = EXCLUDED.context, | ||
| version = purchase_workflows.version + 1, | ||
| updated_at = CURRENT_TIMESTAMP`, | ||
| [orderId, userId, state, JSON.stringify(context)] | ||
| ); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
Advance _dbVersion after each successful persist.
The live context never learns the new row version. Fresh workflows keep taking the upsert path with no optimistic check, and recovered workflows will fail on their second persisted transition because they still send the stale expected version.
Suggested fix
export async function persistWorkflowState(
orderId: string,
state: string,
context: Record<string, any>
): Promise<void> {
const pool = getPool();
const userId = context.userId;
const expectedVersion = context._dbVersion as number | undefined;
+ const persistedContext = { ...context };
+ delete persistedContext._dbVersion;
if (expectedVersion !== undefined) {
const res = await pool.query(
`UPDATE purchase_workflows
- SET state = $1, context = $2, version = version + 1, updated_at = CURRENT_TIMESTAMP
+ SET state = $1, context = $2, version = version + 1, updated_at = CURRENT_TIMESTAMP
WHERE order_id = $3 AND version = $4`,
- [state, JSON.stringify(context), orderId, expectedVersion]
+ [state, JSON.stringify(persistedContext), orderId, expectedVersion]
);
if (res.rowCount === 0) {
throw new Error(`Optimistic concurrency failure: workflow ${orderId}`);
}
+ context._dbVersion = expectedVersion + 1;
} else {
- await pool.query(
+ await pool.query(
`INSERT INTO purchase_workflows (order_id, user_id, state, context, version)
VALUES ($1, $2, $3, $4, 1)
ON CONFLICT (order_id) DO UPDATE SET
state = EXCLUDED.state,
context = EXCLUDED.context,
version = purchase_workflows.version + 1,
updated_at = CURRENT_TIMESTAMP`,
- [orderId, userId, state, JSON.stringify(context)]
+ [orderId, userId, state, JSON.stringify(persistedContext)]
);
+ context._dbVersion = 1;
}
}📝 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 expectedVersion = context._dbVersion as number | undefined; | |
| if (expectedVersion !== undefined) { | |
| const res = await pool.query( | |
| `UPDATE purchase_workflows | |
| SET state = $1, context = $2, version = version + 1, updated_at = CURRENT_TIMESTAMP | |
| WHERE order_id = $3 AND version = $4`, | |
| [state, JSON.stringify(context), orderId, expectedVersion] | |
| ); | |
| if (res.rowCount === 0) { | |
| throw new Error(`Optimistic concurrency failure: workflow ${orderId}`); | |
| } | |
| } else { | |
| await pool.query( | |
| `INSERT INTO purchase_workflows (order_id, user_id, state, context, version) | |
| VALUES ($1, $2, $3, $4, 1) | |
| ON CONFLICT (order_id) DO UPDATE SET | |
| state = EXCLUDED.state, | |
| context = EXCLUDED.context, | |
| version = purchase_workflows.version + 1, | |
| updated_at = CURRENT_TIMESTAMP`, | |
| [orderId, userId, state, JSON.stringify(context)] | |
| ); | |
| const expectedVersion = context._dbVersion as number | undefined; | |
| const persistedContext = { ...context }; | |
| delete persistedContext._dbVersion; | |
| if (expectedVersion !== undefined) { | |
| const res = await pool.query( | |
| `UPDATE purchase_workflows | |
| SET state = $1, context = $2, version = version + 1, updated_at = CURRENT_TIMESTAMP | |
| WHERE order_id = $3 AND version = $4`, | |
| [state, JSON.stringify(persistedContext), orderId, expectedVersion] | |
| ); | |
| if (res.rowCount === 0) { | |
| throw new Error(`Optimistic concurrency failure: workflow ${orderId}`); | |
| } | |
| context._dbVersion = expectedVersion + 1; | |
| } else { | |
| await pool.query( | |
| `INSERT INTO purchase_workflows (order_id, user_id, state, context, version) | |
| VALUES ($1, $2, $3, $4, 1) | |
| ON CONFLICT (order_id) DO UPDATE SET | |
| state = EXCLUDED.state, | |
| context = EXCLUDED.context, | |
| version = purchase_workflows.version + 1, | |
| updated_at = CURRENT_TIMESTAMP`, | |
| [orderId, userId, state, JSON.stringify(persistedContext)] | |
| ); | |
| context._dbVersion = 1; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/orchestrator/src/index.ts` around lines 41 - 63, Advance the
in-memory context version after a successful persist in the workflow save logic
so `_dbVersion` stays in sync with the database. Update the persist path in the
`save` flow around `pool.query` to capture the new row version for both the
optimistic `UPDATE` branch and the initial `INSERT ... ON CONFLICT DO UPDATE`
branch, then assign that incremented version back onto `context._dbVersion`.
This ensures `expectedVersion` in the same orchestrator code keeps working on
later transitions and recovered workflows do not retry with a stale version.
Comment on lines
+128
to
+137
| // Start only if not imported by tests | ||
| if (process.argv[1] && process.argv[1].endsWith("index.ts")) { | ||
| startup().catch((err) => { | ||
| log.error("Failed to start orchestrator", { error: err.message }); | ||
| process.exit(1); | ||
| }); | ||
| } else if (process.env.NODE_ENV !== 'test' && !process.argv[1]?.includes('vitest')) { | ||
| // ESM equivalent to require.main === module is checking argv sometimes, but we'll try to just start it. | ||
| startup().catch(console.error); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Guard startup behind the main entrypoint. The fallback branch still calls startup() for any non-test import, so importing persistWorkflowState or recoverWorkflowState also binds the HTTP port and starts recovery. Switch to a real main-module check and drop the import-time fallback.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/orchestrator/src/index.ts` around lines 128 - 137, The startup
guard in orchestrator entrypoint logic is still allowing import-time execution,
which causes server startup when modules like persistWorkflowState or
recoverWorkflowState are imported. Update the startup check in index.ts to use a
true main-module gate around startup() and remove the non-test fallback branch
so only the actual entrypoint triggers bootstrapping; keep the existing
startup() catch/log handling tied to that main-entry path.
Comment on lines
+213
to
+219
| if (!created) { | ||
| log.warn("Transaction submission ledger entry already exists, updating for retry", { hash }); | ||
| ledgerEntry.status = "PENDING"; | ||
| ledgerEntry.method = method; | ||
| if (orderId) ledgerEntry.orderId = orderId; | ||
| if (contractId) ledgerEntry.contractId = contractId; | ||
| await ledgerEntry.save(); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Don't downgrade an existing ledger row back to PENDING.
hash is the idempotency key here, so a duplicate submission for an already recorded transaction should not overwrite a terminal CONFIRMED/FAILED state. This branch currently does exactly that, and it leaves confirmedAt/errorDetails untouched, so the persisted row can become contradictory.
Suggested fix
if (!created) {
log.warn("Transaction submission ledger entry already exists, updating for retry", { hash });
- ledgerEntry.status = "PENDING";
- ledgerEntry.method = method;
- if (orderId) ledgerEntry.orderId = orderId;
- if (contractId) ledgerEntry.contractId = contractId;
- await ledgerEntry.save();
+ if (ledgerEntry.status === "PENDING") {
+ ledgerEntry.method = method;
+ if (orderId) ledgerEntry.orderId = orderId;
+ if (contractId) ledgerEntry.contractId = contractId;
+ await ledgerEntry.save();
+ }
}📝 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
| if (!created) { | |
| log.warn("Transaction submission ledger entry already exists, updating for retry", { hash }); | |
| ledgerEntry.status = "PENDING"; | |
| ledgerEntry.method = method; | |
| if (orderId) ledgerEntry.orderId = orderId; | |
| if (contractId) ledgerEntry.contractId = contractId; | |
| await ledgerEntry.save(); | |
| if (!created) { | |
| log.warn("Transaction submission ledger entry already exists, updating for retry", { hash }); | |
| if (ledgerEntry.status === "PENDING") { | |
| ledgerEntry.method = method; | |
| if (orderId) ledgerEntry.orderId = orderId; | |
| if (contractId) ledgerEntry.contractId = contractId; | |
| await ledgerEntry.save(); | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/payments/events/index.ts` around lines 213 - 219, The retry path
in the transaction submission flow is incorrectly overwriting an existing ledger
row back to PENDING, which can contradict a previously terminal state. In the
duplicate-idempotency branch inside the submission handler in events/index.ts,
update the existing ledger entry only with non-state fields like
method/orderId/contractId and preserve CONFIRMED/FAILED status plus
confirmedAt/errorDetails when a row already exists. Use the existing created
check and ledgerEntry save path to ensure duplicates do not downgrade a
finalized record.
|
|
||
| const log = createLogger("payments:db", process.env.LOG_LEVEL ?? "info"); | ||
|
|
||
| const databaseUrl = process.env.DATABASE_URL ?? "postgresql://delego:delego@localhost:5432/delego"; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Fail fast when DATABASE_URL is missing.
Falling back to a local DSN here means a misconfigured deployment will still start and write ledger rows into the wrong database. That breaks the persistence contract this PR is adding and contradicts the README, which treats DATABASE_URL as required.
Suggested fix
-const databaseUrl = process.env.DATABASE_URL ?? "postgresql://delego:delego@localhost:5432/delego";
+const databaseUrl = process.env.DATABASE_URL;
+if (!databaseUrl) {
+ throw new Error("DATABASE_URL must be set for `@delego/payments`");
+}📝 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 databaseUrl = process.env.DATABASE_URL ?? "postgresql://delego:delego@localhost:5432/delego"; | |
| const databaseUrl = process.env.DATABASE_URL; | |
| if (!databaseUrl) { | |
| throw new Error("DATABASE_URL must be set for `@delego/payments`"); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/payments/src/db.ts` at line 6, The `databaseUrl` fallback
currently hides a missing `DATABASE_URL` and can send ledger writes to the wrong
database. Update the `databaseUrl` initialization in `db.ts` to require
`process.env.DATABASE_URL` and fail immediately if it is absent, matching the
required configuration contract used by the rest of the payments persistence
setup.
Comment on lines
+29
to
+43
| for (const file of files) { | ||
| console.log(`[delego] db:migrate — running migration: ${file}...`); | ||
| const migrationPath = path.join(migrationsDir, file); | ||
| const migrationSql = fs.readFileSync(migrationPath, "utf8"); | ||
|
|
||
| try { | ||
| await client.query(migrationSql); | ||
| console.log(`[delego] db:migrate — migration ${file} applied successfully.`); | ||
| } catch (err) { | ||
| if (err.message && err.message.includes("already exists")) { | ||
| console.log(`[delego] db:migrate — warning: relation or type in ${file} already exists, skipping...`); | ||
| } else { | ||
| throw err; | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n## scripts/setup/migrate.js\n'
wc -l scripts/setup/migrate.js
sed -n '1,220p' scripts/setup/migrate.js
printf '\n## database/migrations/README.md\n'
wc -l database/migrations/README.md
sed -n '1,220p' database/migrations/README.md
printf '\n## migration files\n'
find database/migrations -maxdepth 1 -type f -name '*.sql' -print | sort
printf '\n## search for transaction handling / error handling in migration runner\n'
rg -n "BEGIN|COMMIT|ROLLBACK|already exists|migrate" scripts/setup/migrate.js database -g '!**/node_modules/**'Repository: DelegoLabs/Delego
Length of output: 4092
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n## migration file sizes\n'
for f in database/migrations/*.sql; do
[ -e "$f" ] || continue
printf '\n### %s (%s lines)\n' "$f" "$(wc -l < "$f")"
sed -n '1,220p' "$f"
done
printf '\n## initial schema size\n'
wc -l database/schema/001_initial.sql
sed -n '1,220p' database/schema/001_initial.sqlRepository: DelegoLabs/Delego
Length of output: 9815
Wrap each migration file in a transaction.
client.query(migrationSql) runs multi-statement migrations without rollback protection, so a later statement can fail after earlier DDL has already been applied. Because this loop also swallows "already exists" errors, a partially applied file can be treated as success and never fully re-run.
🧰 Tools
🪛 ast-grep (0.44.0)
[warning] 31-31: Filesystem path is not a string literal; a request-/variable-derived path can enable path traversal. Validate and normalize the path before use.
Context: fs.readFileSync(migrationPath, "utf8")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').
(detect-non-literal-fs-filename)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/setup/migrate.js` around lines 29 - 43, Wrap each migration in a
transaction inside the migrate loop so a file is all-or-nothing. In
scripts/setup/migrate.js, update the logic around client.query(migrationSql) to
begin a transaction before executing the file, commit on success, and roll back
on any failure before handling the error. Keep the existing already-exists skip
behavior in the catch path, but make sure it applies only after a failed
transaction and does not leave partially applied statements behind.
Comment on lines
+37
to
+40
| } catch (err) { | ||
| if (err.message && err.message.includes("already exists")) { | ||
| console.log(`[delego] db:migrate — warning: relation or type in ${file} already exists, skipping...`); | ||
| } else { |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the target file and nearby context.
git ls-files scripts/setup/migrate.js
wc -l scripts/setup/migrate.js
cat -n scripts/setup/migrate.js | sed -n '1,140p'
# Search for other migration error handling or SQLSTATE checks in the repo.
rg -n "already exists|42P07|42710|42701|err\.code|code ===|duplicate_object|duplicate_table|duplicate_column" scripts . -g '!node_modules' -g '!dist' -g '!build'Repository: DelegoLabs/Delego
Length of output: 3846
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# List migration files and summarize object-creation statements.
git ls-files 'database/schema/*.sql' 'database/migrations/*.sql'
printf '\n-- CREATE/ALTER summary --\n'
rg -n -i '^(create|alter|drop)\b' database/schema database/migrations
printf '\n-- file contents around CREATE statements --\n'
for f in database/schema/*.sql database/migrations/*.sql; do
[ -f "$f" ] || continue
echo
echo "### $f"
cat -n "$f" | sed -n '1,220p'
doneRepository: DelegoLabs/Delego
Length of output: 16509
Match duplicate-object errors by SQLSTATE, not message text. This migration can hit 42P07 for duplicate tables and 42710 for duplicate types/constraints; checking err.message.includes("already exists") is too broad and can hide unrelated failures.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/setup/migrate.js` around lines 37 - 40, The duplicate-object handling
in migrate.js is relying on err.message text, which is too broad and can mask
unrelated failures. Update the catch block in the migration loop to inspect the
SQLSTATE/code on the thrown error instead, and only skip when the error
indicates duplicate relations or types/constraints (for example the cases
handled by the migrate logic around the existing file/relation check). Keep the
existing warning path for those specific duplicate-object errors and let all
other errors continue to fail normally.
Contributor
|
Fix the conflicts |
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
2 participants
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.
closes #54
Summary
Type of change
Test plan
pnpm typecheckpnpm testChecklist
Summary by CodeRabbit
New Features
Bug Fixes
Documentation