Design graph workflow schema and stand up D1 database with migration support - #1
Conversation
…ear JSON to new format - Design graph schema in `workflow.schema.json` with nodes (id, type, position, config) and edges (source, target, sourceHandle). - Implement Cloudflare D1 database support in `worker/src/index.js` and design database schema in `worker/schema.sql`. - Add D1 binding to `worker/wrangler.toml`. - Create a migration script `scripts/migrate-workflow.js` and convert `workflows/example.json` to the new format. - Update `scripts/validate.js` to validate graph structures, edge references, node configs, and run cycle detection (DAG validation). - Update `scripts/executor.js` to execute nodes as a graph and report execution results back to D1. Co-authored-by: aethelred-agent-factory <238771426+aethelred-agent-factory@users.noreply.github.com>
|
👋 Jules, reporting for duty! I'm here to lend a hand with this pull request. When you start a review, I'll add a 👀 emoji to each comment to let you know I've read it. I'll focus on feedback directed at me and will do my best to stay out of conversations between you and other bots or reviewers to keep the noise down. I'll push a commit with your requested changes shortly after. Please note there might be a delay between these steps, but rest assured I'm on the job! For more direct control, you can switch me to Reactive Mode. When this mode is on, I will only act on comments where you specifically mention me with New to Jules? Learn more at jules.google/docs. For security, I will only act on instructions from the user who triggered this task. |
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
📝 WalkthroughWalkthroughThe workflow model changes from linear steps to graph-based nodes and edges. Validation, migration, execution traversal, D1 persistence, GitHub dispatch, and execution reporting are updated to use the new format. ChangesGraph Workflow Execution
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant TelegramConfirm
participant D1
participant GitHubActions
participant Executor
participant WorkerAPI
TelegramConfirm->>D1: Save workflow graph and pending execution
TelegramConfirm->>GitHubActions: Dispatch execution_id and worker_url
GitHubActions->>Executor: Run workflow payload
Executor->>Executor: Traverse nodes and matching edges
Executor->>WorkerAPI: POST final status and execution log
WorkerAPI->>D1: Update execution record
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
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: 9
🤖 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 `@scripts/executor.js`:
- Around line 241-265: Secure the execution status callback by adding a
per-execution or shared secret to the workflow payload and requiring it in the
worker’s POST /executions/:id handler. Update the reporting flow around
reportUrl and the fetch call to send the secret in an authentication header, and
update the worker handler to validate that header before modifying execution
state; reject missing or invalid credentials.
- Around line 241-265: Update the execution-status POST in the
workerUrl/executionId reporting block to use the existing withTimeout helper,
passing the report fetch operation and the appropriate timeout configuration
used for other network calls in scripts/executor.js. Preserve the current
success, response-error, and catch logging behavior while ensuring a hung worker
response is bounded.
- Around line 141-226: The queue traversal around outgoing, incomingCount, and
the main execution loop must not run converging nodes on first arrival. Add
Kahn-style readiness tracking that counts each node’s incoming edge and only
queues it after every predecessor edge has either fired or been definitively
pruned according to success, failure, and always branch semantics; preserve the
existing fallback for graphs without start nodes and ensure unreachable/pruned
branches do not block valid joins.
In `@scripts/migrate-workflow.js`:
- Around line 36-63: Update the legacy migration flow around the oldData.steps
iteration to check the resulting node count against the 50-node limit before
reporting success. When steps exceed the limit, fail migration with a clear
error instead of emitting invalid graph output; preserve normal generation for
workflows within the limit and reuse the existing MAX_NODES symbol if available.
In `@scripts/validate.js`:
- Around line 76-78: Update the delay-node validation in the script’s
node-checking logic to reject numeric ms values outside the inclusive 0–300000
range, while preserving the existing type validation and error reporting style.
Ensure invalid negative or oversized delays are added to errors instead of being
accepted.
In `@worker/src/index.js`:
- Around line 175-193: Update validateWorkflow to perform DFS-based cycle
detection on the validated workflow edges before accepting it, matching the DAG
check in scripts/validate.js. Track node visitation states, report a validation
error when traversal encounters a node already on the current recursion path,
and preserve the existing node/edge validation behavior for acyclic workflows.
- Around line 138-173: Update validateWorkflow’s node loop to validate
node.position before accepting the workflow: require a position object with
numeric x and y values, and add validation errors for missing or malformed
coordinates. Keep handleConfirm’s existing position access safe by ensuring
invalid nodes are rejected before staging or persistence.
- Around line 11-44: Protect the POST /executions/:id branch before parsing or
updating data by requiring the established per-execution token or shared secret,
and reject unauthorized requests without touching executions. Update the body
handling so an empty or missing status cannot default the execution to success;
only accept an explicitly valid status when persisting through the executions
update query.
- Around line 287-330: Update the workflow persistence block around workflowId
and the D1 writes to use a stable, scoped, collision-resistant identifier rather
than only the normalized workflow.name, while preserving consistent references
across workflows, nodes, and edges. Build one ordered env.DB.batch call
containing the workflow upsert, node/edge deletes, and all node and edge
inserts, replacing the separate run and batch calls so the complete save is
atomic.
🪄 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 Plus
Run ID: 0816e1cc-8264-49ef-aeda-94912d0ec1a5
⛔ Files ignored due to path filters (1)
worker/package-lock.jsonis excluded by!**/package-lock.json
📒 Files selected for processing (9)
executions/hello-world-1784001389646.jsonscripts/executor.jsscripts/migrate-workflow.jsscripts/validate.jsworker/schema.sqlworker/src/index.jsworker/wrangler.tomlworkflows/example.jsonworkflows/workflow.schema.json
| // Build graph representations | ||
| const nodeMap = {}; | ||
| nodes.forEach(n => { nodeMap[n.id] = n; }); | ||
|
|
||
| // Outgoing edges per node | ||
| const outgoing = {}; | ||
| nodes.forEach(n => { outgoing[n.id] = []; }); | ||
| edges.forEach(e => { | ||
| if (outgoing[e.source]) { | ||
| outgoing[e.source].push(e); | ||
| } | ||
| }); | ||
|
|
||
| // Find start nodes (nodes with no incoming edges) | ||
| const incomingCount = {}; | ||
| nodes.forEach(n => { incomingCount[n.id] = 0; }); | ||
| edges.forEach(e => { | ||
| if (incomingCount[e.target] !== undefined) { | ||
| incomingCount[e.target]++; | ||
| } | ||
| }); | ||
|
|
||
| const startNodes = nodes.filter(n => incomingCount[n.id] === 0); | ||
|
|
||
| // If there's no node with 0 incoming count but we have nodes, default to the first node | ||
| let queue = []; | ||
| if (startNodes.length > 0) { | ||
| queue = startNodes.map(n => n.id); | ||
| } else if (nodes.length > 0) { | ||
| queue = [nodes[0].id]; | ||
| } | ||
|
|
||
| const executed = new Set(); | ||
| const nodeResults = {}; // maps nodeId to its execution entry | ||
|
|
||
| let index = 0; | ||
| while (queue.length > 0) { | ||
| const currentId = queue.shift(); | ||
| if (executed.has(currentId)) continue; | ||
|
|
||
| // Ensure all incoming dependency nodes that were actually reached have executed. | ||
| // However, in standard workflow branch execution, some incoming paths might never execute. | ||
| // We proceed to run the node. | ||
| const node = nodeMap[currentId]; | ||
| if (!node) continue; | ||
|
|
||
| let entry; | ||
| try { | ||
| const entry = await runStep(steps[i], i); | ||
| entry = await runNode(node, index++); | ||
| log.steps.push(entry); | ||
| if (entry.status === 'failed_ignored') overallStatus = 'partial_failure'; | ||
| nodeResults[currentId] = entry; | ||
| executed.add(currentId); | ||
| if (entry.status === 'failed_ignored') { | ||
| if (overallStatus === 'success') overallStatus = 'partial_failure'; | ||
| } | ||
| } catch (err) { | ||
| log.steps.push(err.entry || { index: i, status: 'failed', error: err.message }); | ||
| entry = err.entry || { id: currentId, index, status: 'failed', error: err.message }; | ||
| log.steps.push(entry); | ||
| nodeResults[currentId] = entry; | ||
| executed.add(currentId); | ||
| overallStatus = 'failed'; | ||
| break; // stop on first (non-ignored) failure | ||
| } | ||
|
|
||
| // Determine next nodes to queue based on edges | ||
| const outEdges = outgoing[currentId] || []; | ||
| outEdges.forEach(edge => { | ||
| const handle = edge.sourceHandle || 'success'; | ||
| const status = entry.status; // 'success', 'failed', 'failed_ignored' | ||
|
|
||
| let follow = false; | ||
| if (handle === 'always') { | ||
| follow = true; | ||
| } else if (handle === 'success') { | ||
| follow = (status === 'success' || status === 'failed_ignored'); | ||
| } else if (handle === 'failure') { | ||
| follow = (status === 'failed'); | ||
| } | ||
|
|
||
| if (follow) { | ||
| queue.push(edge.target); | ||
| } | ||
| }); | ||
|
|
||
| // If a node fails without continue_on_error and there is no failure branch, we abort the entire run. | ||
| if (entry.status === 'failed' && !outEdges.some(e => e.sourceHandle === 'failure' || e.sourceHandle === 'always')) { | ||
| break; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | 🏗️ Heavy lift
Join nodes can execute before all their predecessors finish.
The BFS queue only guards against re-running an already-executed node; it has no mechanism requiring all live predecessors of a node to complete before that node is dequeued and run. Whether a converging node executes early depends purely on how quickly each upstream branch happens to enqueue it.
Concrete trace: nodes X (start), A (start), B (A→B), C (X→C, B→C), node array order [X, A, B, C]:
- queue=
[X,A]→ runX, pushC→ queue=[A,C] - run
A, pushB→ queue=[C,B] - dequeue
C(front) and run it —B, one ofC's own predecessors, is still pending behind it in the queue.
For a graph model that explicitly supports success/failure/always branching (naturally producing diamond/join shapes when branches reconverge), this means a downstream node can run with incomplete upstream state, non-deterministically depending on node array order and which branch happens to resolve first.
A correct fix needs real in-degree tracking (Kahn's-algorithm style): only make a node eligible to run once every edge feeding it has either fired or been definitively pruned by the branch it belongs to (accounting for success/failure/always semantics), rather than "first arrival wins."
🤖 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/executor.js` around lines 141 - 226, The queue traversal around
outgoing, incomingCount, and the main execution loop must not run converging
nodes on first arrival. Add Kahn-style readiness tracking that counts each
node’s incoming edge and only queues it after every predecessor edge has either
fired or been definitively pruned according to success, failure, and always
branch semantics; preserve the existing fallback for graphs without start nodes
and ensure unreachable/pruned branches do not block valid joins.
| // If worker_url and executionId are provided, call back to Cloudflare D1 | ||
| if (workerUrl && executionId) { | ||
| const reportUrl = `${workerUrl.replace(/\/$/, '')}/executions/${executionId}`; | ||
| console.log(`Reporting execution status to ${reportUrl}...`); | ||
| try { | ||
| const response = await fetch(reportUrl, { | ||
| method: 'POST', | ||
| headers: { 'Content-Type': 'application/json' }, | ||
| body: JSON.stringify({ | ||
| status: overallStatus, | ||
| started_at: log.started_at, | ||
| finished_at: log.finished_at, | ||
| log: log, | ||
| }), | ||
| }); | ||
| if (response.ok) { | ||
| console.log('Execution successfully reported to D1.'); | ||
| } else { | ||
| console.error(`Failed to report execution to D1: ${response.status} ${await response.text()}`); | ||
| } | ||
| } catch (e) { | ||
| console.error(`Error reporting execution to D1: ${e.message}`); | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
Report endpoint accepts unauthenticated writes to execution state.
This POST sends status/log to ${worker_url}/executions/${execution_id} with no auth header. Per the worker/src/index.js context snippet, the receiving POST /executions/:id handler updates the executions row from the request body with no secret/token check at all (unlike /confirm, which validates a webhook secret). Since execution_id flows through the GitHub Actions client_payload and Actions run logs are visible to anyone who can view the workflow run, an unauthenticated party who obtains that ID could overwrite a legitimate execution's status/log.
Consider having the worker generate a per-execution secret (or reuse TELEGRAM_WEBHOOK_SECRET-style shared secret) passed via client_payload and required as a header on this report call.
🤖 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/executor.js` around lines 241 - 265, Secure the execution status
callback by adding a per-execution or shared secret to the workflow payload and
requiring it in the worker’s POST /executions/:id handler. Update the reporting
flow around reportUrl and the fetch call to send the secret in an authentication
header, and update the worker handler to validate that header before modifying
execution state; reject missing or invalid credentials.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Report POST has no timeout, unlike other network calls in this file.
The http node type wraps its fetch in an AbortController+timeoutMs, and delay uses the withTimeout helper, but this reporting call has neither. A hung/slow worker response would stall this step indefinitely (bounded only by the CI job's own timeout).
🔧 Proposed fix: reuse withTimeout
const response = await fetch(reportUrl, {
- method: 'POST',
- headers: { 'Content-Type': 'application/json' },
- body: JSON.stringify({
- status: overallStatus,
- started_at: log.started_at,
- finished_at: log.finished_at,
- log: log,
- }),
- });
+ const response = await withTimeout(fetch(reportUrl, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({
+ status: overallStatus,
+ started_at: log.started_at,
+ finished_at: log.finished_at,
+ log: log,
+ }),
+ }), 15_000, 'report');📝 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.
| // If worker_url and executionId are provided, call back to Cloudflare D1 | |
| if (workerUrl && executionId) { | |
| const reportUrl = `${workerUrl.replace(/\/$/, '')}/executions/${executionId}`; | |
| console.log(`Reporting execution status to ${reportUrl}...`); | |
| try { | |
| const response = await fetch(reportUrl, { | |
| method: 'POST', | |
| headers: { 'Content-Type': 'application/json' }, | |
| body: JSON.stringify({ | |
| status: overallStatus, | |
| started_at: log.started_at, | |
| finished_at: log.finished_at, | |
| log: log, | |
| }), | |
| }); | |
| if (response.ok) { | |
| console.log('Execution successfully reported to D1.'); | |
| } else { | |
| console.error(`Failed to report execution to D1: ${response.status} ${await response.text()}`); | |
| } | |
| } catch (e) { | |
| console.error(`Error reporting execution to D1: ${e.message}`); | |
| } | |
| } | |
| // If worker_url and executionId are provided, call back to Cloudflare D1 | |
| if (workerUrl && executionId) { | |
| const reportUrl = `${workerUrl.replace(/\/$/, '')}/executions/${executionId}`; | |
| console.log(`Reporting execution status to ${reportUrl}...`); | |
| try { | |
| const response = await withTimeout(fetch(reportUrl, { | |
| method: 'POST', | |
| headers: { 'Content-Type': 'application/json' }, | |
| body: JSON.stringify({ | |
| status: overallStatus, | |
| started_at: log.started_at, | |
| finished_at: log.finished_at, | |
| log: log, | |
| }), | |
| }), 15_000, 'report'); | |
| if (response.ok) { | |
| console.log('Execution successfully reported to D1.'); | |
| } else { | |
| console.error(`Failed to report execution to D1: ${response.status} ${await response.text()}`); | |
| } | |
| } catch (e) { | |
| console.error(`Error reporting execution to D1: ${e.message}`); | |
| } | |
| } |
🤖 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/executor.js` around lines 241 - 265, Update the execution-status POST
in the workerUrl/executionId reporting block to use the existing withTimeout
helper, passing the report fetch operation and the appropriate timeout
configuration used for other network calls in scripts/executor.js. Preserve the
current success, response-error, and catch logging behavior while ensuring a
hung worker response is bounded.
| const nodes = []; | ||
| const edges = []; | ||
|
|
||
| oldData.steps.forEach((step, idx) => { | ||
| const id = `node_${idx + 1}`; | ||
| const { type, ...restConfig } = step; | ||
|
|
||
| // Position them sequentially downwards | ||
| const position = { | ||
| x: 100, | ||
| y: 100 + idx * 150 | ||
| }; | ||
|
|
||
| nodes.push({ | ||
| id, | ||
| type, | ||
| position, | ||
| config: restConfig | ||
| }); | ||
|
|
||
| if (idx > 0) { | ||
| edges.push({ | ||
| source: `node_${idx}`, | ||
| target: id, | ||
| sourceHandle: 'success' | ||
| }); | ||
| } | ||
| }); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
No cap check for the new maxItems: 50 node limit.
Migrating a legacy workflow with more than 50 steps produces graph output that will fail workflow.schema.json's maxItems: 50 (and the same MAX_NODES constant in scripts/validate.js/worker/src/index.js), but this script reports success unconditionally (Line 73). A quick length check would surface the problem immediately instead of letting the user discover it later via a separate validation run.
🔧 Proposed fix
+const MAX_NODES = 50;
+
+if (oldData.steps.length > MAX_NODES) {
+ console.error(`Cannot migrate: ${oldData.steps.length} steps exceeds the graph format's max of ${MAX_NODES} nodes.`);
+ process.exit(1);
+}
+
const nodes = [];
const edges = [];📝 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.
| const nodes = []; | |
| const edges = []; | |
| oldData.steps.forEach((step, idx) => { | |
| const id = `node_${idx + 1}`; | |
| const { type, ...restConfig } = step; | |
| // Position them sequentially downwards | |
| const position = { | |
| x: 100, | |
| y: 100 + idx * 150 | |
| }; | |
| nodes.push({ | |
| id, | |
| type, | |
| position, | |
| config: restConfig | |
| }); | |
| if (idx > 0) { | |
| edges.push({ | |
| source: `node_${idx}`, | |
| target: id, | |
| sourceHandle: 'success' | |
| }); | |
| } | |
| }); | |
| const MAX_NODES = 50; | |
| if (oldData.steps.length > MAX_NODES) { | |
| console.error(`Cannot migrate: ${oldData.steps.length} steps exceeds the graph format's max of ${MAX_NODES} nodes.`); | |
| process.exit(1); | |
| } | |
| const nodes = []; | |
| const edges = []; |
🤖 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/migrate-workflow.js` around lines 36 - 63, Update the legacy
migration flow around the oldData.steps iteration to check the resulting node
count against the 50-node limit before reporting success. When steps exceed the
limit, fail migration with a clear error instead of emitting invalid graph
output; preserve normal generation for workflows within the limit and reuse the
existing MAX_NODES symbol if available.
| if (node.type === 'delay' && typeof config.ms !== 'number') { | ||
| errors.push(`node ${i} (${node.id || 'unnamed'}): "delay" config needs a numeric "ms"`); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Missing range check for delay node's ms.
Only type is checked here; workflow.schema.json constrains ms to 0–300000 and scripts/executor.js hard-caps at 5 minutes. A negative or absurdly large ms passes this validator but silently misbehaves (or gets clamped) at execution time, so the "OK" result here doesn't guarantee schema validity.
🔧 Proposed fix
- if (node.type === 'delay' && typeof config.ms !== 'number') {
- errors.push(`node ${i} (${node.id || 'unnamed'}): "delay" config needs a numeric "ms"`);
- }
+ if (node.type === 'delay') {
+ if (typeof config.ms !== 'number' || config.ms < 0 || config.ms > 300000) {
+ errors.push(`node ${i} (${node.id || 'unnamed'}): "delay" config needs a numeric "ms" between 0 and 300000`);
+ }
+ }📝 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.
| if (node.type === 'delay' && typeof config.ms !== 'number') { | |
| errors.push(`node ${i} (${node.id || 'unnamed'}): "delay" config needs a numeric "ms"`); | |
| } | |
| if (node.type === 'delay') { | |
| if (typeof config.ms !== 'number' || config.ms < 0 || config.ms > 300000) { | |
| errors.push(`node ${i} (${node.id || 'unnamed'}): "delay" config needs a numeric "ms" between 0 and 300000`); | |
| } | |
| } |
🤖 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/validate.js` around lines 76 - 78, Update the delay-node validation
in the script’s node-checking logic to reject numeric ms values outside the
inclusive 0–300000 range, while preserving the existing type validation and
error reporting style. Ensure invalid negative or oversized delays are added to
errors instead of being accepted.
| // Support execution feedback API: POST /executions/:id | ||
| if (request.method === 'POST' && url.pathname.startsWith('/executions/')) { | ||
| const execId = url.pathname.split('/').pop(); | ||
| try { | ||
| const body = await request.json(); | ||
| const { status, started_at, finished_at, log } = body; | ||
|
|
||
| if (env.DB) { | ||
| await env.DB.prepare( | ||
| `UPDATE executions | ||
| SET status = ?, started_at = ?, finished_at = ?, log = ? | ||
| WHERE id = ?` | ||
| ) | ||
| .bind( | ||
| status || 'success', | ||
| started_at || null, | ||
| finished_at || null, | ||
| log ? JSON.stringify(log) : null, | ||
| execId | ||
| ) | ||
| .run(); | ||
| } | ||
|
|
||
| return new Response(JSON.stringify({ ok: true }), { | ||
| status: 200, | ||
| headers: { 'Content-Type': 'application/json' }, | ||
| }); | ||
| } catch (err) { | ||
| return new Response(JSON.stringify({ ok: false, error: err.message }), { | ||
| status: 400, | ||
| headers: { 'Content-Type': 'application/json' }, | ||
| }); | ||
| } | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
git ls-files worker/src/index.js worker/schema.sql worker/src | sed 's#^`#FILE` #'
echo '--- index.js outline ---'
ast-grep outline worker/src/index.js --view expanded || true
echo '--- search auth/token refs ---'
rg -n "secret-token|repository_dispatch|executions|status \\|\\| 'success'|POST /executions|request.method === 'POST'" worker/src/index.js worker/src -S
echo '--- focused read ---'
nl -ba worker/src/index.js | sed -n '1,120p'Repository: quickerup/telegram-workflow-bot
Length of output: 1818
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- worker/src/index.js lines 1-120 ---'
sed -n '1,120p' worker/src/index.js
echo '--- secret/token/header searches ---'
rg -n "secret-token|Authorization|X-.*Token|timingSafeEqual|env\.|request\.headers|get\(" worker/src/index.js -S
echo '--- worker/src/index.js lines 120-430 ---'
sed -n '120,430p' worker/src/index.jsRepository: quickerup/telegram-workflow-bot
Length of output: 16240
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- worker/src/index.js lines 1-80 ---'
sed -n '1,80p' worker/src/index.js
echo '--- worker/src/index.js lines 270-430 ---'
sed -n '270,430p' worker/src/index.jsRepository: quickerup/telegram-workflow-bot
Length of output: 7651
/executions/:id needs auth. This branch runs before the Telegram secret-token gate, so anyone who knows an execution ID can overwrite status/log; status || 'success' also lets an empty body mark the run successful. Require a per-execution token or other shared secret before updating executions.
🤖 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 `@worker/src/index.js` around lines 11 - 44, Protect the POST /executions/:id
branch before parsing or updating data by requiring the established
per-execution token or shared secret, and reject unauthorized requests without
touching executions. Update the body handling so an empty or missing status
cannot default the execution to success; only accept an explicitly valid status
when persisting through the executions update query.
| const nodeIds = new Set(); | ||
| workflow.nodes.forEach((node, i) => { | ||
| if (!node || typeof node !== 'object') { | ||
| errors.push(`node ${i}: must be an object`); | ||
| return; | ||
| } | ||
| if (step.type === 'run' && typeof step.command !== 'string') { | ||
| errors.push(`step ${i}: "run" needs a string "command"`); | ||
| if (typeof node.id !== 'string' || !node.id.trim()) { | ||
| errors.push(`node ${i}: "id" must be a non-empty string`); | ||
| } else { | ||
| nodeIds.add(node.id); | ||
| } | ||
| if (step.type === 'http' && typeof step.url !== 'string') { | ||
| errors.push(`step ${i}: "http" needs a string "url"`); | ||
|
|
||
| if (!ALLOWED_STEP_TYPES.has(node.type)) { | ||
| errors.push(`node ${i}: type must be one of ${[...ALLOWED_STEP_TYPES].join(', ')}`); | ||
| return; | ||
| } | ||
|
|
||
| if (!node.config || typeof node.config !== 'object') { | ||
| errors.push(`node ${i}: missing or invalid "config" object`); | ||
| return; | ||
| } | ||
|
|
||
| const { config } = node; | ||
| if (node.type === 'run' && typeof config.command !== 'string') { | ||
| errors.push(`node ${i}: "run" config needs a string "command"`); | ||
| } | ||
| if (node.type === 'http' && typeof config.url !== 'string') { | ||
| errors.push(`node ${i}: "http" config needs a string "url"`); | ||
| } | ||
| if (step.type === 'delay' && typeof step.ms !== 'number') { | ||
| errors.push(`step ${i}: "delay" needs a numeric "ms"`); | ||
| if (node.type === 'delay' && typeof config.ms !== 'number') { | ||
| errors.push(`node ${i}: "delay" config needs a numeric "ms"`); | ||
| } | ||
| if (step.type === 'notify' && typeof step.message !== 'string') { | ||
| errors.push(`step ${i}: "notify" needs a string "message"`); | ||
| if (node.type === 'notify' && typeof config.message !== 'string') { | ||
| errors.push(`node ${i}: "notify" config needs a string "message"`); | ||
| } | ||
| }); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
validateWorkflow doesn't validate node.position, unlike scripts/validate.js.
scripts/validate.js (lines 56-62) checks that node.position.x/.y are numbers; this function — the actual gate for workflows submitted via Telegram — has no equivalent check. handleConfirm later dereferences node.position.x/.position.y unguarded (Lines 306-307) while building nodeStmts. A workflow missing position (or with a malformed one) passes this validator, gets staged, and throws mid-.map() on /confirm — after the DELETE FROM nodes/DELETE FROM edges calls have already committed, leaving that workflow's row pointing at zero nodes/edges (see the related atomicity comment below).
🔧 Proposed fix
if (!node.position || typeof node.position !== 'object') {
errors.push(`node ${i}: missing or invalid "config" object`);
return;
}
+ if (typeof node.position.x !== 'number' || typeof node.position.y !== 'number') {
+ errors.push(`node ${i}: "position" must have numeric "x" and "y"`);
+ }🤖 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 `@worker/src/index.js` around lines 138 - 173, Update validateWorkflow’s node
loop to validate node.position before accepting the workflow: require a position
object with numeric x and y values, and add validation errors for missing or
malformed coordinates. Keep handleConfirm’s existing position access safe by
ensuring invalid nodes are rejected before staging or persistence.
| if (Array.isArray(workflow.edges)) { | ||
| workflow.edges.forEach((edge, i) => { | ||
| if (!edge || typeof edge !== 'object') { | ||
| errors.push(`edge ${i}: must be an object`); | ||
| return; | ||
| } | ||
| if (typeof edge.source !== 'string' || !nodeIds.has(edge.source)) { | ||
| errors.push(`edge ${i}: source node "${edge.source}" does not exist`); | ||
| } | ||
| if (typeof edge.target !== 'string' || !nodeIds.has(edge.target)) { | ||
| errors.push(`edge ${i}: target node "${edge.target}" does not exist`); | ||
| } | ||
| if (edge.sourceHandle && !['success', 'failure', 'always'].includes(edge.sourceHandle)) { | ||
| errors.push(`edge ${i}: sourceHandle "${edge.sourceHandle}" is invalid`); | ||
| } | ||
| }); | ||
| } else { | ||
| errors.push('"edges" must be an array'); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
validateWorkflow doesn't check for cycles, unlike scripts/validate.js.
scripts/validate.js enforces DAG semantics via DFS cycle detection before treating a workflow as valid, but this is the function that actually gates workflows arriving over the Telegram webhook, and it has no equivalent check. A cyclic graph can be staged, confirmed, persisted to D1, and dispatched to scripts/executor.js for real execution on GitHub Actions — risking an unbounded traversal loop and a stuck/expensive CI run, since scripts/validate.js is only a local dev-time tool that isn't invoked by this worker.
Port the same DFS-based cycle check used in scripts/validate.js into this function before accepting the workflow.
🤖 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 `@worker/src/index.js` around lines 175 - 193, Update validateWorkflow to
perform DFS-based cycle detection on the validated workflow edges before
accepting it, matching the DAG check in scripts/validate.js. Track node
visitation states, report a validation error when traversal encounters a node
already on the current recursion path, and preserve the existing node/edge
validation behavior for acyclic workflows.
| const workflowId = workflow.name.replace(/[^a-z0-9_-]/gi, '_'); | ||
|
|
||
| // Insert/Replace workflow, nodes, and edges in D1 | ||
| if (env.DB) { | ||
| // We transactionally save or just execute sequentially since Cloudflare DB handles statements in order | ||
| await env.DB.prepare(`INSERT OR REPLACE INTO workflows (id, name) VALUES (?, ?)`).bind(workflowId, workflow.name).run(); | ||
|
|
||
| // Clear old nodes and edges | ||
| await env.DB.prepare(`DELETE FROM nodes WHERE workflow_id = ?`).bind(workflowId).run(); | ||
| await env.DB.prepare(`DELETE FROM edges WHERE workflow_id = ?`).bind(workflowId).run(); | ||
|
|
||
| // Insert new nodes | ||
| const nodeStmts = workflow.nodes.map(node => { | ||
| return env.DB.prepare( | ||
| `INSERT INTO nodes (id, workflow_id, type, position_x, position_y, config) VALUES (?, ?, ?, ?, ?, ?)` | ||
| ).bind( | ||
| node.id, | ||
| workflowId, | ||
| node.type, | ||
| node.position.x, | ||
| node.position.y, | ||
| JSON.stringify(node.config) | ||
| ); | ||
| }); | ||
|
|
||
| // Insert new edges | ||
| const edgeStmts = workflow.edges.map(edge => { | ||
| return env.DB.prepare( | ||
| `INSERT INTO edges (workflow_id, source, target, source_handle) VALUES (?, ?, ?, ?)` | ||
| ).bind( | ||
| workflowId, | ||
| edge.source, | ||
| edge.target, | ||
| edge.sourceHandle || 'success' | ||
| ); | ||
| }); | ||
|
|
||
| if (nodeStmts.length > 0) { | ||
| await env.DB.batch(nodeStmts); | ||
| } | ||
| if (edgeStmts.length > 0) { | ||
| await env.DB.batch(edgeStmts); | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== worker/src/index.js outline ==\n'
ast-grep outline worker/src/index.js --view expanded || true
printf '\n== excerpt around cited lines ==\n'
sed -n '250,360p' worker/src/index.js
printf '\n== schema and migration files ==\n'
git ls-files 'worker/*schema*' 'worker/**/schema.sql' 'scripts/*workflow*' 'worker/**/migrate*' 'worker/**/migration*' | sed 's#^`#FILE` #'
printf '\n== search for table definitions and workflowId usage ==\n'
rg -n "CREATE TABLE (workflows|nodes|edges)|PRIMARY KEY|UNIQUE|workflowId|workflow_id|node_1|node_2" worker scripts -S
printf '\n== show candidate schema files ==\n'
for f in $(git ls-files 'worker/**/schema.sql' 'worker/*schema*' 'scripts/*workflow*' 'worker/**/migrate*' 'worker/**/migration*'); do
echo
echo "### $f"
wc -l "$f"
doneRepository: quickerup/telegram-workflow-bot
Length of output: 6928
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== worker/src/index.js excerpt ==\n'
sed -n '280,340p' worker/src/index.js
printf '\n== files of interest ==\n'
git ls-files | rg '^(worker/src/index\.js|worker/.*schema\.sql|scripts/.*workflow.*|worker/.*migration.*|worker/.*migrate.*)$' || true
printf '\n== relevant definitions/usages ==\n'
rg -n "CREATE TABLE (workflows|nodes|edges)|PRIMARY KEY|UNIQUE|workflowId|workflow_id|node_1|node_2" worker scripts -SRepository: quickerup/telegram-workflow-bot
Length of output: 4293
🌐 Web query:
Cloudflare D1 batch atomic transaction separate .run() .batch() independent transactions documentation
💡 Result:
Cloudflare D1 does not currently support traditional SQL-level transactions (e.g., BEGIN TRANSACTION, COMMIT, ROLLBACK) [1][2][3]. Instead, D1 relies on the.batch API for grouping operations [4][5]. Regarding the behavior of.run and.batch:.run: This method executes a single prepared statement and returns a D1Result [6][7]. Each individual call to.run is independent and operates in auto-commit mode [4][5]..batch: This method sends multiple prepared statements to the database in a single network round trip [4][5]. While earlier documentation or community discussions may have contained conflicting information regarding atomicity [2][8][5], current Cloudflare D1 documentation specifies that batched statements are executed as a single transaction [4][9]. If a statement within the batch fails, the entire sequence is aborted or rolled back [9]. However,.batch does not allow you to use the results of one statement to inform the logic of a subsequent statement within the same batch (i.e., you cannot interleave JavaScript logic with SQL execution within a single transaction) [1][2][3]. In summary: - Independent Transactions: D1 does not support arbitrary, multi-round-trip JavaScript-controlled transactions [1][2]. -.run: Executes a single statement; operates independently [4][5]. -.batch: Executes a group of statements as a single atomic transaction [4][9]. It is intended for reducing latency by grouping independent queries into one round trip, not for implementing complex, logic-dependent transactions [2][3][5].
Citations:
- 1: https://www.answeroverflow.com/m/1385144670375383072?focus=1385144670375383072
- 2: https://www.answeroverflow.com/m/1243531489535197265
- 3: https://www.answeroverflow.com/m/1402580672937459803
- 4: https://developers.cloudflare.com/d1/worker-api/d1-database/
- 5: https://registry.npmjs.org/%40cloudflare%2Fd1
- 6: https://github.com/cloudflare/cloudflare-docs/blob/production/src/content/docs/d1/worker-api/return-object.mdx
- 7: https://developers.cloudflare.com/d1/best-practices/query-d1/index.md
- 8: 🚀 Feature Request: D1 Transaction support cloudflare/workers-sdk#2733
- 9: https://www.answeroverflow.com/m/1290665043809144844?focus=1290665043809144844
🌐 Web query:
Cloudflare D1 batch atomicity independent transactions run batch documentation
💡 Result:
In Cloudflare D1, the batch() API executes multiple SQL statements within a single atomic transaction [1][2][3]. According to official Cloudflare D1 documentation, if any statement within the batch fails, the database will abort or roll back the entire sequence of statements [1][2]. Key characteristics of the batch() operation include: - Atomicity: All statements in the batch are executed as a single transaction [3][4]. If one statement fails, the others are rolled back [1][2]. - Performance: Batching reduces network latency by executing multiple statements in a single round trip to the database [1][2]. - Sequential Execution: The statements are executed and committed sequentially, non-concurrently, in the order they are provided [1][2]. While early documentation or legacy sources may have contained conflicting information regarding transaction support [5], current authoritative Cloudflare D1 documentation confirms that batched statements are treated as SQL transactions [1][2]. Note that batch() is the primary mechanism for atomic multi-statement operations in D1, as D1 does not currently support explicit SQL BEGIN TRANSACTION or COMMIT statements [6].
Citations:
- 1: https://5dc1e11d.preview.developers.cloudflare.com/d1/worker-api/d1-database/
- 2: https://developers.cloudflare.com/d1/llms-full.txt
- 3: https://www.answeroverflow.com/m/1290665043809144844?focus=1290665043809144844
- 4: https://www.answeroverflow.com/m/1192109327939547157
- 5: https://registry.npmjs.org/%40cloudflare%2Fd1
- 6: https://www.answeroverflow.com/m/1402580672937459803
Make the D1 save atomic and use a collision-free workflow ID. worker/src/index.js:287-330
- The workflow, node, and edge writes are split across separate D1 calls, so a failure after the deletes can leave the workflow row updated while its nodes/edges are gone. Put the whole save into one
batch(). workflowIdcomes only fromworkflow.name, so normalized-name collisions can overwrite another workflow’s data and execution history. Use a stable scoped ID instead.
🤖 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 `@worker/src/index.js` around lines 287 - 330, Update the workflow persistence
block around workflowId and the D1 writes to use a stable, scoped,
collision-resistant identifier rather than only the normalized workflow.name,
while preserving consistent references across workflows, nodes, and edges. Build
one ordered env.DB.batch call containing the workflow upsert, node/edge deletes,
and all node and edge inserts, replacing the separate run and batch calls so the
complete save is atomic.
1 participant
This PR introduces graph-based workflow capabilities to the bot. Workflows are now represented as Directed Acyclic Graphs (DAGs) defined by a set of nodes and edges (supporting success/failure branching). Staging a workflow is now backed by a Cloudflare D1 database (with tables for workflows, nodes, edges, and executions). A migration script converts legacy linear JSON workflows to the new graph format. Execution logs are posted back from the GitHub Actions runner directly to the D1 database.
PR created automatically by Jules for task 9450620653231500155 started by @aethelred-agent-factory
Summary by CodeRabbit
New Features
Bug Fixes
Documentation