feat(runtime): add Workers runtime foundation#25
Merged
Conversation
Co-authored-by: monkeycode-ai <monkeycode-ai@chaitin.com>
Walkthrough本 PR 引入 Changes异步数据库抽象与 Cloudflare Worker 支持
Sequence Diagram(s)sequenceDiagram
participant Client as 外部请求
participant handleWorkerFetch as handleWorkerFetch
participant D1DatabaseAdapter as D1DatabaseAdapter
participant runMigration as runMigration
participant createWorkerTelegramWebhookHandler as WebhookHandler
participant Bot as Grammy Bot
Client->>handleWorkerFetch: fetch(request, env)
handleWorkerFetch->>D1DatabaseAdapter: new D1DatabaseAdapter(env.DB)
handleWorkerFetch->>runMigration: await migrate(db)
alt 路径 /healthz
handleWorkerFetch->>D1DatabaseAdapter: db.prepare("SELECT 1").get()
handleWorkerFetch-->>Client: JSON {ok: true}
else 路径 /telegram/webhook
handleWorkerFetch->>createWorkerTelegramWebhookHandler: 获取/创建 handler
createWorkerTelegramWebhookHandler->>createWorkerTelegramWebhookHandler: 校验 x-telegram-bot-api-secret-token
alt secret 不匹配
createWorkerTelegramWebhookHandler-->>Client: 403 Forbidden
else secret 匹配
createWorkerTelegramWebhookHandler->>Bot: webhookCallback(bot, "cloudflare-mod")
Bot-->>Client: 200 OK
end
else 其他路径
handleWorkerFetch-->>Client: 404 JSON
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 13
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
src/channels/telegram/commands.ts (1)
353-358: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift发送成功后的落库异常不能提示“重试发送”。
Line 353 之后任何状态写入异常都会落入 Line 356-358 的统一
catch。如果 TelegramsendMessage已经成功，管理员仍会收到“可重试 /draft send”，同一草稿就可能再次发给外部用户；若失败点落在前面的deps.deliveries.markSent(deliveryId)，当前实现还会把已送达的投递记成 failed。这里需要把“发送 Telegram”与“发送后状态同步”拆成两个错误路径：一旦外发成功，就不能再建议重试发送。🤖 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 `@src/channels/telegram/commands.ts` around lines 353 - 358, The unified catch in the draft send flow is mixing Telegram delivery failures with post-send persistence failures, which causes the reply to incorrectly suggest retrying /draft send after the message was already sent. Refactor the send path in commands.ts around the draft send handler so the external send step and the follow-up state writes are handled as separate error paths, using the existing draft send logic and helpers like deps.deliveries.markSent, deps.aiDrafts.markSent, and deps.audit.log to ensure that once Telegram send succeeds, the user gets a non-retryable sync-failure message instead of “重试发送”.src/runtime/config.ts (1)
99-118: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win将
node:fs/node:path从 Worker 可加载模块中拆出去
src/runtime/config.ts顶层直接依赖 Node 内置模块，而wrangler.toml的入口是src/runtime/worker.ts，它会直接导入这个文件；这会把.env读取路径一起带进 Worker 侧，存在 Cloudflare Workers 构建/运行失败的风险。建议把纯 schema/merge 逻辑和 Node-only 的.envloader 拆开，让 Worker 只引用前者。🤖 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 `@src/runtime/config.ts` around lines 99 - 118, The config module currently mixes Worker-safe schema/merge helpers with Node-only .env loading, so move the `loadEnv` implementation and any `node:fs`/`node:path` usage out of `src/runtime/config.ts` into a separate Node-only module. Keep `loadConfigFromSources`, `configIssues`, and `loadDatabaseConfig` in the Worker-imported path, and make `src/runtime/worker.ts` import only the pure config/schema helpers so Cloudflare Workers never bundle the .env reader.src/domain/conversations.ts (1)
427-450: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win不要在这里再次查询
last_insert_rowid()。
这里把INSERT、UPDATE和后续SELECT拆成了多个await；last_insert_rowid()依赖连接状态，期间同一连接上的其他写入会把它覆盖，可能取到别的消息行。run()在 Node SQLite 和 D1 适配器里都已经返回lastInsertRowid，直接保存这个 id 再查行，或改成单语句返回新行。🤖 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 `@src/domain/conversations.ts` around lines 427 - 450, The message insert flow in the conversation persistence logic should not query last_insert_rowid() after separate await calls, because the connection state can change before the SELECT runs. In the message-creation code around the INSERT/UPDATE sequence, capture the inserted row id directly from the result of run(), then use that saved id to fetch the message row, or switch to a single-statement return path. Update the logic in the message insert method so it relies on the run() return value instead of last_insert_rowid().Source: Path instructions
🤖 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 `@src/channels/telegram/commands.ts`:
- Around line 296-299: The delete flow in the Telegram command handler leaves a
stale conversation-to-Topic mapping if `ctx.api.deleteForumTopic` succeeds but
`deps.conversations.deleteConversationData` fails, causing later routing to a
non-existent Topic. Update the deletion sequence in this handler so the routable
state is removed or marked inactive before the external Topic delete, or make
the operation compensating/transactional so a partial failure cannot leave an
invalid mapping behind. Use the `deleteForumTopic`, `deps.audit.log`, and
`deps.conversations.deleteConversationData` steps as the main places to adjust.
In `@src/domain/ai-drafts.ts`:
- Around line 75-82: The draft insert flow in the ai-drafts logic currently does
an extra SELECT last_insert_rowid() after calling prepare(...).run(), which can
return the wrong id under shared-connection concurrency. Update the insert path
to use the lastInsertRowid returned directly from the run() result in the
ai-drafts method, and remove the follow-up SELECT so the later UPDATE ai_drafts
... WHERE id = ? targets the correct draft.
In `@src/domain/app-settings.ts`:
- Around line 32-37: The manual BEGIN/COMMIT transaction in app-settings
persistence is unsafe on a shared async Database connection because each await
in the loop can let unrelated writes interleave. Update the app-settings save
path in the code that performs the upsert loop to use a serialized database-side
transaction/batch API (or move the whole operation into a single
adapter-controlled critical section) so all key/value writes run atomically
without exposing the shared connection to interleaving.
In `@src/domain/conversations.ts`:
- Around line 318-356: The transaction logic in deleteConversationData and
resetConversation manually issues BEGIN/COMMIT across multiple awaited delete
calls, which allows other Database writes to interleave. Refactor these flows to
use a database-level transaction/batch abstraction provided by the adapter so
the deletions execute atomically without yielding between statements. Update the
transaction handling in the conversations data access code so the delete
sequence for deliveries, ai_drafts, conversation_tags, admin_notes,
telegram_topics/messages, and conversations is wrapped in a single
adapter-managed unit of work.
- Around line 171-203: The contact and conversation creation flow in the
conversation service is currently a non-atomic find-then-insert/load sequence,
which can race under concurrent webhooks and create duplicate rows or split
state. Refactor the logic around the contact lookup/creation and
`findLatestConversation` path to use an atomic upsert or a single transaction
with unique-key conflict handling so `contact` and `conversation` are created
and loaded exactly once. Keep the fix localized to the existing
contact/conversation creation method and preserve the final null checks after
the upsert/load step.
In `@src/domain/deliveries.ts`:
- Around line 26-33: `createPending()` in `Deliveries` is re-reading the row id
with a separate `SELECT last_insert_rowid()` after the insert, which can pick up
the wrong delivery if another write happens on the same connection. Update the
insert flow to use the `run()` result’s `lastInsertRowid` directly (or expose
that value through the DB adapter) and return that id from `createPending()`, so
`markSent()` and `markFailed()` keep targeting the correct record.
In `@src/runtime/main.ts`:
- Around line 79-86: Reuse the same settings snapshot for both validation and
config loading in main so the startup path does not read two different states.
Capture await settings.all() once in the main flow, pass that snapshot into
configIssues and loadConfigFromSources, and keep the existing
logger/lastRuntimeError behavior unchanged while ensuring the validated data is
the one used by loadConfigFromSources.
In `@src/runtime/web-console.ts`:
- Around line 342-352: When handling a successful login in web-console.ts, the
setup session created by the login flow remains in sessions even after
`WEB_CONSOLE_SETUP_TOKEN` is cleared, so a previously authorized setup session
can keep accessing privileged pages. Update the login/session flow around
`loginSessionKind`, `sessions.set`, and the password-save path so that saving a
new password invalidates the current setup session (or all setup sessions) and
forces re-login; if needed, change the auth helper to return both `{ token, kind
}` so the caller can revoke the exact session.
- Around line 790-807: Normalize the `page` query parameter before it is passed
into `listConversations`, `listFailedDeliveries`, `listAuditLogs`, and the
search branch in `web-console.ts`; the current `Math.max(1, Number(...))` still
allows `NaN` when `page` is non-numeric. Add a safe parsing step that falls back
to `1` for invalid values, then use that sanitized `pageNum` consistently when
calling the list/search helpers and rendering `renderConversationsBody`,
`renderDeliveriesBody`, and `renderAuditLogsBody` so downstream offset
calculations never receive `NaN`.
In `@src/runtime/worker.ts`:
- Around line 43-45: The handleWorkerFetch path is running migrate(db) on every
request, putting schema initialization into the hot fetch path for
handleWorkerFetch. Move migration out of per-request handling, either into a
separate deployment/initialization step or cache a single initialization Promise
at module scope keyed by the DB binding. Keep the fetch flow focused on request
handling so routes like /healthz and /telegram/webhook are not blocked by
repeated migration work.
In `@src/storage/migrations/0001_initial.ts`:
- Around line 133-134: The migration contract is now async, but some startup
paths still call migrate(...) without awaiting completion; update every migrate
caller to use await migrate(...) before any service initialization or database
access. Locate the migrate function and its downstream callers such as
handle.client startup or tool/bootstrap flows, and ensure the app does not
proceed to read/write the database until migration finishes.
In `@src/storage/migrations/runner.ts`:
- Around line 29-32: `addColumnIfMissing` 里 `PRAGMA table_info` 到 `ALTER TABLE`
之间存在并发竞态，可能在 Worker 的 fetch/scheduled 触发迁移时重复加列失败；请在 `ALTER TABLE`
执行失败后再次查询该表列信息，只有确认列已被其他并发迁移成功添加时才吞掉错误，否则继续抛出。重点修改
`src/storage/migrations/runner.ts` 中的 `addColumnIfMissing`
逻辑，保留现有幂等判断并补上失败后的二次确认。
In `@wrangler.toml`:
- Around line 8-11: The D1 configuration still uses a placeholder database_id,
so the Worker cannot bind to a real database after deployment. Update the
[[d1_databases]] entry in wrangler.toml by replacing the all-zero UUID with the
actual D1 database_id used by the DB binding, keeping the existing binding and
database_name values intact.
---
Outside diff comments:
In `@src/channels/telegram/commands.ts`:
- Around line 353-358: The unified catch in the draft send flow is mixing
Telegram delivery failures with post-send persistence failures, which causes the
reply to incorrectly suggest retrying /draft send after the message was already
sent. Refactor the send path in commands.ts around the draft send handler so the
external send step and the follow-up state writes are handled as separate error
paths, using the existing draft send logic and helpers like
deps.deliveries.markSent, deps.aiDrafts.markSent, and deps.audit.log to ensure
that once Telegram send succeeds, the user gets a non-retryable sync-failure
message instead of “重试发送”.
In `@src/domain/conversations.ts`:
- Around line 427-450: The message insert flow in the conversation persistence
logic should not query last_insert_rowid() after separate await calls, because
the connection state can change before the SELECT runs. In the message-creation
code around the INSERT/UPDATE sequence, capture the inserted row id directly
from the result of run(), then use that saved id to fetch the message row, or
switch to a single-statement return path. Update the logic in the message insert
method so it relies on the run() return value instead of last_insert_rowid().
In `@src/runtime/config.ts`:
- Around line 99-118: The config module currently mixes Worker-safe schema/merge
helpers with Node-only .env loading, so move the `loadEnv` implementation and
any `node:fs`/`node:path` usage out of `src/runtime/config.ts` into a separate
Node-only module. Keep `loadConfigFromSources`, `configIssues`, and
`loadDatabaseConfig` in the Worker-imported path, and make
`src/runtime/worker.ts` import only the pure config/schema helpers so Cloudflare
Workers never bundle the .env reader.
🪄 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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 7d03461f-c4df-498e-98d6-68d683d6006e
📒 Files selected for processing (26)
src/channels/telegram/bot.tssrc/channels/telegram/commands.tssrc/channels/telegram/factory.tssrc/channels/telegram/secrets.tssrc/channels/telegram/worker-webhook.tssrc/domain/ai-drafts.tssrc/domain/app-settings.tssrc/domain/audit.tssrc/domain/conversation-expiry.tssrc/domain/conversations.tssrc/domain/deliveries.tssrc/domain/retention.tssrc/ports/database.tssrc/runtime/config.tssrc/runtime/main.tssrc/runtime/maintenance.tssrc/runtime/web-console.tssrc/runtime/worker.tssrc/storage/client.tssrc/storage/d1.tssrc/storage/migrations/0001_initial.tssrc/storage/migrations/runner.tssrc/tools/check-telegram.tssrc/tools/retention-cleanup.tstest/core.test.tswrangler.toml
Comment on lines
296
to
299
| await ctx.api.deleteForumTopic(Number(topicContext.topic.managementChatId), topicContext.topic.messageThreadId); | ||
| deps.audit.log({ adminId, conversationId: conversation.id, action: "delete" }); | ||
| await deps.audit.log({ adminId, conversationId: conversation.id, action: "delete" }); | ||
| await deps.conversations.deleteConversationData(conversation.id); | ||
| return true; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
先删 Topic 再删库会留下失效路由。
Line 296 成功而 Line 298 失败时，数据库仍保留当前会话与 Topic 映射，但 Telegram Topic 已经不存在。后续入站消息还会命中这条失效映射，直接造成 Topic 错路由或投递失败。这里至少要先移除可路由状态，再执行外部删除，或者把删除流程做成可补偿操作。 As per path instructions, “任何可能导致非白名单成员代发、bot 自己消息回环、Topic 错路由或用户数据泄露的问题都应标为高优先级。”
🤖 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 `@src/channels/telegram/commands.ts` around lines 296 - 299, The delete flow in
the Telegram command handler leaves a stale conversation-to-Topic mapping if
`ctx.api.deleteForumTopic` succeeds but
`deps.conversations.deleteConversationData` fails, causing later routing to a
non-existent Topic. Update the deletion sequence in this handler so the routable
state is removed or marked inactive before the external Topic delete, or make
the operation compensating/transactional so a partial failure cannot leave an
invalid mapping behind. Use the `deleteForumTopic`, `deps.audit.log`, and
`deps.conversations.deleteConversationData` steps as the main places to adjust.
Source: Path instructions
Comment on lines
+75
to
82
| await this.db | ||
| .prepare( | ||
| `INSERT INTO ai_drafts (conversation_id, source_message_id, status, created_at, updated_at) | ||
| VALUES (?, ?, 'pending', ?, ?)`, | ||
| ) | ||
| .run(conversationId, sourceMessageId ?? null, timestamp, timestamp); | ||
| const row = this.db.prepare("SELECT last_insert_rowid() AS id").get() as { id: number }; | ||
| const row = (await this.db.prepare("SELECT last_insert_rowid() AS id").get()) as { id: number }; | ||
| const draftId = Number(row.id); |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '== file outline =='
ast-grep outline src/domain/ai-drafts.ts --view expanded || true
echo
echo '== relevant lines =='
wc -l src/domain/ai-drafts.ts
sed -n '1,240p' src/domain/ai-drafts.tsRepository: one-ea/InboxBridge
Length of output: 8899
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect ai-drafts implementation and database adapter usage.
echo '== ai-drafts file outline =='
ast-grep outline src/domain/ai-drafts.ts --view expanded || true
echo
echo '== ai-drafts content =='
wc -l src/domain/ai-drafts.ts
sed -n '1,240p' src/domain/ai-drafts.ts
echo
echo '== search for last_insert_rowid / INSERT RETURNING / db adapter =='
rg -n "last_insert_rowid|INSERT INTO ai_drafts|returning|prepare\\(" src -g '*.ts'Repository: one-ea/InboxBridge
Length of output: 18397
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Read the relevant file and surrounding code.
echo '== ai-drafts.ts =='
wc -l src/domain/ai-drafts.ts
sed -n '1,240p' src/domain/ai-drafts.ts
echo
echo '== usages of ai_drafts =='
rg -n "ai_drafts|last_insert_rowid|draftId" src -g '*.ts'Repository: one-ea/InboxBridge
Length of output: 10222
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the relevant module and database abstraction.
ast-grep outline src/domain/ai-drafts.ts --view expanded || true
echo
echo '== db wrapper references =='
rg -n "class .*Db|last_insert_rowid|better-sqlite3|d1|prepare\\(|run\\(|get\\(" src -g '*.ts'
echo
echo '== ai-drafts.ts excerpt =='
nl -ba src/domain/ai-drafts.ts | sed -n '1,220p'Repository: one-ea/InboxBridge
Length of output: 18881
🌐 Web query:
SQLite last_insert_rowid connection scoped concurrent async same connection D1 meta.last_row_id documentation
💡 Result:
In SQLite, the last_insert_rowid function returns the rowid of the most recent successful INSERT operation performed on the same database connection [1][2]. This value is connection-scoped, meaning it is isolated from other database connections [3][4]. If multiple threads or concurrent tasks use the same database connection to perform inserts, the value returned by last_insert_rowid is not thread-safe [5][6]. If one thread performs an INSERT, and a second thread performs another INSERT on the same connection before the first thread calls last_insert_rowid, the value returned to the first thread will be the rowid of the second thread's insertion (or potentially unpredictable data if accessed during the operation) [1][3][2]. To safely obtain the rowid in concurrent environments, you must either ensure each thread/task uses its own dedicated connection or implement a mutex/locking mechanism to synchronize access to the connection [3][5][7]. For Cloudflare D1, which is built on SQLite, the equivalent metadata for identifying the last inserted row is available in the D1Result object's meta property as meta.last_row_id [8]. Cloudflare D1 databases are designed to be single-threaded at the database instance level and process queries one at a time, queueing concurrent requests if necessary [9]. Because D1 manages its internal SQLite connections, you should rely on the meta.last_row_id returned directly within the D1Result of your specific query to ensure you receive the rowid associated with that exact execution [8][10].
Citations:
- 1: https://sqlite.org/c3ref/last_insert_rowid.html
- 2: https://www3.sqlite.org/matrix/c3ref/last_insert_rowid.html
- 3: https://sqlite.org/forum/info/c2b468f5688c4cb5
- 4: https://stackoverflow.com/questions/44054163/does-sqlites-last-insert-rowid-still-work-with-persistent-database-connection
- 5: https://sqlite.org/forum/info/5992bc7f1904b84d
- 6: https://stackoverflow.com/questions/29072109/is-sqlite-last-insert-rowid-atomic
- 7: https://stackoverflow.com/questions/31541436/thread-safe-last-insert-rowid-with-sqlite3-api-c
- 8: https://community.cloudflare.com/t/accessing-changes-lastrowid-fields-from-d1result-object/448712
- 9: https://developers.cloudflare.com/d1/platform/limits/
- 10: Add support for D1
meta.{last_row_id,changes}cloudflare/miniflare#500
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect database abstraction to see whether inserted ids can be read from run().
sed -n '1,220p' src/ports/database.ts
echo
sed -n '1,220p' src/storage/client.ts
echo
sed -n '1,220p' src/storage/d1.ts
echo
sed -n '1,120p' src/domain/deliveries.tsRepository: one-ea/InboxBridge
Length of output: 8217
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '== ports/database.ts =='
sed -n '1,220p' src/ports/database.ts
echo
echo '== storage/client.ts =='
sed -n '1,220p' src/storage/client.ts
echo
echo '== storage/d1.ts =='
sed -n '1,220p' src/storage/d1.ts
echo
echo '== deliveries.ts excerpt =='
sed -n '1,80p' src/domain/deliveries.tsRepository: one-ea/InboxBridge
Length of output: 7078
改用 run() 返回的 lastInsertRowid
这里先插入再单独 SELECT last_insert_rowid()；在共享连接的并发请求下，draftId 可能读到别的插入结果，后面的 UPDATE ai_drafts ... WHERE id = ? 就会写到错误草稿。run() 已经在 Node SQLite / D1 适配器里统一返回 lastInsertRowid，直接接住并移除这次查询即可。
🤖 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 `@src/domain/ai-drafts.ts` around lines 75 - 82, The draft insert flow in the
ai-drafts logic currently does an extra SELECT last_insert_rowid() after calling
prepare(...).run(), which can return the wrong id under shared-connection
concurrency. Update the insert path to use the lastInsertRowid returned directly
from the run() result in the ai-drafts method, and remove the follow-up SELECT
so the later UPDATE ai_drafts ... WHERE id = ? targets the correct draft.
Source: Path instructions
Comment on lines
+32
to
+37
| await this.db.exec("BEGIN"); | ||
| try { | ||
| for (const [key, value] of Object.entries(values)) statement.run(key, value, updatedAt); | ||
| this.db.exec("COMMIT"); | ||
| for (const [key, value] of Object.entries(values)) await statement.run(key, value, updatedAt); | ||
| await this.db.exec("COMMIT"); | ||
| } catch (error) { | ||
| this.db.exec("ROLLBACK"); | ||
| await this.db.exec("ROLLBACK"); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
不要在共享异步连接上手写 BEGIN/COMMIT。
BEGIN 后的每个 await statement.run(...) 都会让出执行权；同一 Database 实例上的其它写入可能插入到这笔事务里，导致 COMMIT/ROLLBACK 提交或回滚不属于本次配置保存的变更。请在数据库端口提供串行化的 transaction/batch API，或让适配器在同一临界区内执行整批 upsert。
As per path instructions, src/**/*.ts: “重点审查 ... 数据库写入逻辑。”
🤖 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 `@src/domain/app-settings.ts` around lines 32 - 37, The manual BEGIN/COMMIT
transaction in app-settings persistence is unsafe on a shared async Database
connection because each await in the loop can let unrelated writes interleave.
Update the app-settings save path in the code that performs the upsert loop to
use a serialized database-side transaction/batch API (or move the whole
operation into a single adapter-controlled critical section) so all key/value
writes run atomically without exposing the shared connection to interleaving.
Source: Path instructions
Comment on lines
+171
to
203
| let contact = await this.findContact(input.platform, input.externalUserId); | ||
|
|
||
| if (!contact) { | ||
| this.db | ||
| await this.db | ||
| .prepare( | ||
| `INSERT INTO contacts (platform, external_user_id, username, display_name, created_at, updated_at) | ||
| VALUES (?, ?, ?, ?, ?, ?)`, | ||
| ) | ||
| .run(input.platform, input.externalUserId, input.username ?? null, input.displayName ?? null, timestamp, timestamp); | ||
| contact = this.findContact(input.platform, input.externalUserId); | ||
| contact = await this.findContact(input.platform, input.externalUserId); | ||
| } else { | ||
| this.db | ||
| await this.db | ||
| .prepare("UPDATE contacts SET username = ?, display_name = ?, updated_at = ? WHERE id = ?") | ||
| .run(input.username ?? contact.username, input.displayName ?? contact.displayName, timestamp, contact.id); | ||
| contact = this.getContactSync(contact.id); | ||
| contact = await this.getContactSync(contact.id); | ||
| } | ||
|
|
||
| if (!contact) throw new Error("Failed to create or load contact."); | ||
|
|
||
| let conversation = this.findLatestConversation(contact.id); | ||
| let conversation = await this.findLatestConversation(contact.id); | ||
| if (!conversation) { | ||
| const expiresAt = this.defaultConversationRetentionDays === null ? null : addDaysIso(this.defaultConversationRetentionDays); | ||
| this.db | ||
| await this.db | ||
| .prepare( | ||
| `INSERT INTO conversations ( | ||
| contact_id, retention_days, expires_at, created_at, updated_at, last_message_at | ||
| ) VALUES (?, ?, ?, ?, ?, ?)`, | ||
| ) | ||
| .run(contact.id, this.defaultConversationRetentionDays, expiresAt, timestamp, timestamp, timestamp); | ||
| conversation = this.findLatestConversation(contact.id); | ||
| conversation = await this.findLatestConversation(contact.id); | ||
| } | ||
|
|
||
| if (!conversation) throw new Error("Failed to create or load conversation."); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
把联系人和会话创建改成原子 upsert。
当前是 find → await INSERT → find 的检查后写入流程；并发 webhook 可能把同一外部用户创建成重复 contact/conversation，或在唯一约束下让其中一个请求失败，后续 Topic/会话状态会分裂。请使用唯一键上的 upsert/返回行，或事务级串行化一次完成创建与加载。
As per path instructions, src/**/*.ts: “任何可能导致 ... Topic 错路由 ... 的问题都应标为高优先级。”
🤖 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 `@src/domain/conversations.ts` around lines 171 - 203, The contact and
conversation creation flow in the conversation service is currently a non-atomic
find-then-insert/load sequence, which can race under concurrent webhooks and
create duplicate rows or split state. Refactor the logic around the contact
lookup/creation and `findLatestConversation` path to use an atomic upsert or a
single transaction with unique-key conflict handling so `contact` and
`conversation` are created and loaded exactly once. Keep the fix localized to
the existing contact/conversation creation method and preserve the final null
checks after the upsert/load step.
Source: Path instructions
Comment on lines
318
to
356
| async deleteConversationData(conversationId: number): Promise<void> { | ||
| this.db.exec("BEGIN"); | ||
| await this.db.exec("BEGIN"); | ||
| try { | ||
| this.db | ||
| await this.db | ||
| .prepare( | ||
| `DELETE FROM deliveries | ||
| WHERE source_message_id IN (SELECT id FROM messages WHERE conversation_id = ?)`, | ||
| ) | ||
| .run(conversationId); | ||
| this.db.prepare("DELETE FROM ai_drafts WHERE conversation_id = ?").run(conversationId); | ||
| this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ?").run(conversationId); | ||
| this.db.prepare("DELETE FROM admin_notes WHERE conversation_id = ?").run(conversationId); | ||
| this.db.prepare("DELETE FROM telegram_topics WHERE conversation_id = ?").run(conversationId); | ||
| this.db.prepare("DELETE FROM messages WHERE conversation_id = ?").run(conversationId); | ||
| this.db.prepare("DELETE FROM conversations WHERE id = ?").run(conversationId); | ||
| this.db.exec("COMMIT"); | ||
| await this.db.prepare("DELETE FROM ai_drafts WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.prepare("DELETE FROM admin_notes WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.prepare("DELETE FROM telegram_topics WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.prepare("DELETE FROM messages WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.prepare("DELETE FROM conversations WHERE id = ?").run(conversationId); | ||
| await this.db.exec("COMMIT"); | ||
| } catch (error) { | ||
| this.db.exec("ROLLBACK"); | ||
| await this.db.exec("ROLLBACK"); | ||
| throw error; | ||
| } | ||
| } | ||
|
|
||
| async resetConversation(conversationId: number): Promise<void> { | ||
| this.db.exec("BEGIN"); | ||
| await this.db.exec("BEGIN"); | ||
| try { | ||
| this.db | ||
| await this.db | ||
| .prepare( | ||
| `DELETE FROM deliveries | ||
| WHERE source_message_id IN (SELECT id FROM messages WHERE conversation_id = ?)`, | ||
| ) | ||
| .run(conversationId); | ||
| this.db.prepare("DELETE FROM ai_drafts WHERE conversation_id = ?").run(conversationId); | ||
| this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ?").run(conversationId); | ||
| this.db.prepare("DELETE FROM admin_notes WHERE conversation_id = ?").run(conversationId); | ||
| this.db.prepare("DELETE FROM messages WHERE conversation_id = ?").run(conversationId); | ||
| this.db.exec("COMMIT"); | ||
| await this.db.prepare("DELETE FROM ai_drafts WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.prepare("DELETE FROM admin_notes WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.prepare("DELETE FROM messages WHERE conversation_id = ?").run(conversationId); | ||
| await this.db.exec("COMMIT"); | ||
| } catch (error) { | ||
| this.db.exec("ROLLBACK"); | ||
| await this.db.exec("ROLLBACK"); | ||
| throw error; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
不要用跨 await 的手写事务删除会话数据。
BEGIN 后连续等待多个删除语句时，共享 Database 上的其它写入可 interleave 到同一事务中，导致会话清理提交/回滚范围不确定。请通过数据库端口提供适配器级 transaction/batch，保证这些删除语句不可被其它请求插入。
As per path instructions, src/**/*.ts: “重点审查 ... 会话销毁和数据库写入逻辑。”
🤖 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 `@src/domain/conversations.ts` around lines 318 - 356, The transaction logic in
deleteConversationData and resetConversation manually issues BEGIN/COMMIT across
multiple awaited delete calls, which allows other Database writes to interleave.
Refactor these flows to use a database-level transaction/batch abstraction
provided by the adapter so the deletions execute atomically without yielding
between statements. Update the transaction handling in the conversations data
access code so the delete sequence for deliveries, ai_drafts, conversation_tags,
admin_notes, telegram_topics/messages, and conversations is wrapped in a single
adapter-managed unit of work.
Source: Path instructions
Comment on lines
+790
to
+807
| const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); | ||
| const convStatus = searchParams.get("status") ?? undefined; | ||
| const assignedTo = searchParams.get("assigned_to") ?? undefined; | ||
| const data = await options.listConversations({ page: pageNum, status: convStatus, assignedTo, pageSize: 50 }); | ||
| opsBody = renderConversationsBody(data, pageNum, convStatus, assignedTo); | ||
| } else if (tab === "deliveries") { | ||
| const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); | ||
| const data = await options.listFailedDeliveries({ page: pageNum, pageSize: 50 }); | ||
| opsBody = renderDeliveriesBody(data, pageNum); | ||
| } else if (tab === "audit") { | ||
| const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); | ||
| const adminId = searchParams.get("admin_id") ?? undefined; | ||
| const action = searchParams.get("action") ?? undefined; | ||
| const data = await options.listAuditLogs({ page: pageNum, adminId, action, pageSize: 50 }); | ||
| opsBody = renderAuditLogsBody(data, pageNum, adminId, action); | ||
| } else { | ||
| const query = searchParams.get("q") ?? ""; | ||
| const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
规范化 page 参数后再计算 offset。
Math.max(1, Number(...)) 遇到 ?page=abc 会得到 NaN，随后传给列表/搜索回调并用于 SQL OFFSET，运维页会退化成错误页。
建议调整
+ const pageFromParams = (): number => {
+ const raw = Number(searchParams.get("page") ?? "1");
+ return Number.isInteger(raw) && raw > 0 ? raw : 1;
+ };
+
let opsBody: string;
try {
@@
- const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1"));
+ const pageNum = pageFromParams();
@@
- const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1"));
+ const pageNum = pageFromParams();
@@
- const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1"));
+ const pageNum = pageFromParams();
@@
- const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1"));
+ const pageNum = pageFromParams();📝 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 pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); | |
| const convStatus = searchParams.get("status") ?? undefined; | |
| const assignedTo = searchParams.get("assigned_to") ?? undefined; | |
| const data = await options.listConversations({ page: pageNum, status: convStatus, assignedTo, pageSize: 50 }); | |
| opsBody = renderConversationsBody(data, pageNum, convStatus, assignedTo); | |
| } else if (tab === "deliveries") { | |
| const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); | |
| const data = await options.listFailedDeliveries({ page: pageNum, pageSize: 50 }); | |
| opsBody = renderDeliveriesBody(data, pageNum); | |
| } else if (tab === "audit") { | |
| const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); | |
| const adminId = searchParams.get("admin_id") ?? undefined; | |
| const action = searchParams.get("action") ?? undefined; | |
| const data = await options.listAuditLogs({ page: pageNum, adminId, action, pageSize: 50 }); | |
| opsBody = renderAuditLogsBody(data, pageNum, adminId, action); | |
| } else { | |
| const query = searchParams.get("q") ?? ""; | |
| const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); | |
| const pageFromParams = (): number => { | |
| const raw = Number(searchParams.get("page") ?? "1"); | |
| return Number.isInteger(raw) && raw > 0 ? raw : 1; | |
| }; | |
| const pageNum = pageFromParams(); | |
| const convStatus = searchParams.get("status") ?? undefined; | |
| const assignedTo = searchParams.get("assigned_to") ?? undefined; | |
| const data = await options.listConversations({ page: pageNum, status: convStatus, assignedTo, pageSize: 50 }); | |
| opsBody = renderConversationsBody(data, pageNum, convStatus, assignedTo); | |
| } else if (tab === "deliveries") { | |
| const pageNum = pageFromParams(); | |
| const data = await options.listFailedDeliveries({ page: pageNum, pageSize: 50 }); | |
| opsBody = renderDeliveriesBody(data, pageNum); | |
| } else if (tab === "audit") { | |
| const pageNum = pageFromParams(); | |
| const adminId = searchParams.get("admin_id") ?? undefined; | |
| const action = searchParams.get("action") ?? undefined; | |
| const data = await options.listAuditLogs({ page: pageNum, adminId, action, pageSize: 50 }); | |
| opsBody = renderAuditLogsBody(data, pageNum, adminId, action); | |
| } else { | |
| const query = searchParams.get("q") ?? ""; | |
| const pageNum = pageFromParams(); |
🤖 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 `@src/runtime/web-console.ts` around lines 790 - 807, Normalize the `page`
query parameter before it is passed into `listConversations`,
`listFailedDeliveries`, `listAuditLogs`, and the search branch in
`web-console.ts`; the current `Math.max(1, Number(...))` still allows `NaN` when
`page` is non-numeric. Add a safe parsing step that falls back to `1` for
invalid values, then use that sanitized `pageNum` consistently when calling the
list/search helpers and rendering `renderConversationsBody`,
`renderDeliveriesBody`, and `renderAuditLogsBody` so downstream offset
calculations never receive `NaN`.
Comment on lines
+43
to
+45
| export async function handleWorkerFetch(request: Request, env: WorkerEnv, options: WorkerRuntimeOptions = {}): Promise<Response> { | ||
| const db = new D1DatabaseAdapter(env.DB); | ||
| await migrate(db); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
不要在每个 fetch 请求里跑迁移。
Line 45 现在会让 /healthz、/telegram/webhook 甚至 404 都先阻塞在 migrate(db) 上，把 schema 初始化放进了 webhook 热路径。D1 一旦出现慢查询或锁冲突，收件和健康检查都会被一起拖慢。建议把迁移移到显式部署/初始化步骤，或至少在 module scope 按数据库绑定缓存一次初始化 Promise。
🤖 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 `@src/runtime/worker.ts` around lines 43 - 45, The handleWorkerFetch path is
running migrate(db) on every request, putting schema initialization into the hot
fetch path for handleWorkerFetch. Move migration out of per-request handling,
either into a separate deployment/initialization step or cache a single
initialization Promise at module scope keyed by the DB binding. Keep the fetch
flow focused on request handling so routes like /healthz and /telegram/webhook
are not blocked by repeated migration work.
Comment on lines
+133
to
+134
| export async function migrate(client: Database): Promise<void> { | ||
| await runMigration(client, migration); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
同步更新所有 migrate(...) 调用方为 await migrate(...)。
migrate 已变成异步契约，但提供的下游上下文仍有 migrate(handle.client) 形式；Node 运行时和工具脚本可能在迁移完成前开始读写数据库，触发缺表/缺列错误。请把这些启动路径改成等待迁移完成后再创建服务或处理请求。
As per path instructions: src/**/*.ts 重点审查数据库写入逻辑。
🤖 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 `@src/storage/migrations/0001_initial.ts` around lines 133 - 134, The migration
contract is now async, but some startup paths still call migrate(...) without
awaiting completion; update every migrate caller to use await migrate(...)
before any service initialization or database access. Locate the migrate
function and its downstream callers such as handle.client startup or
tool/bootstrap flows, and ensure the app does not proceed to read/write the
database until migration finishes.
Source: Path instructions
Comment on lines
+29
to
+32
| async function addColumnIfMissing(client: Database, table: string, column: string, definition: string): Promise<void> { | ||
| const rows = (await client.prepare(`PRAGMA table_info(${table})`).all()) as Array<{ name: string }>; | ||
| if (!rows.some((row) => row.name === column)) { | ||
| await client.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
补齐 ADD COLUMN 的并发幂等保护。
PRAGMA table_info 与 ALTER TABLE 之间存在 TOCTOU 窗口；Worker 的 fetch/scheduled 路径会触发迁移，首次部署或升级并发进入时，第二个请求可能因重复加列失败并中断请求。请在 ALTER 失败后重新确认列是否已存在，只忽略“已由并发迁移完成”的情况。
建议修复
async function addColumnIfMissing(client: Database, table: string, column: string, definition: string): Promise<void> {
- const rows = (await client.prepare(`PRAGMA table_info(${table})`).all()) as Array<{ name: string }>;
- if (!rows.some((row) => row.name === column)) {
+ const hasColumn = async (): Promise<boolean> => {
+ const rows = (await client.prepare(`PRAGMA table_info(${table})`).all()) as Array<{ name: string }>;
+ return rows.some((row) => row.name === column);
+ };
+ if (!(await hasColumn())) {
+ try {
await client.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`);
+ } catch (error) {
+ if (await hasColumn()) return;
+ throw error;
+ }
}
}As per path instructions: src/**/*.ts 重点审查数据库写入逻辑。
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| async function addColumnIfMissing(client: Database, table: string, column: string, definition: string): Promise<void> { | |
| const rows = (await client.prepare(`PRAGMA table_info(${table})`).all()) as Array<{ name: string }>; | |
| if (!rows.some((row) => row.name === column)) { | |
| await client.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); | |
| async function addColumnIfMissing(client: Database, table: string, column: string, definition: string): Promise<void> { | |
| const hasColumn = async (): Promise<boolean> => { | |
| const rows = (await client.prepare(`PRAGMA table_info(${table})`).all()) as Array<{ name: string }>; | |
| return rows.some((row) => row.name === column); | |
| }; | |
| if (!(await hasColumn())) { | |
| try { | |
| await client.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); | |
| } catch (error) { | |
| if (await hasColumn()) return; | |
| throw error; | |
| } | |
| } | |
| } |
🧰 Tools
🪛 OpenGrep (1.23.0)
[ERROR] 32-32: Dynamic command passed to child_process.exec/execSync. Use child_process.execFile or spawn with an argument array instead.
(coderabbit.command-injection.exec-js)
🤖 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 `@src/storage/migrations/runner.ts` around lines 29 - 32, `addColumnIfMissing`
里 `PRAGMA table_info` 到 `ALTER TABLE` 之间存在并发竞态，可能在 Worker 的 fetch/scheduled
触发迁移时重复加列失败；请在 `ALTER TABLE` 执行失败后再次查询该表列信息，只有确认列已被其他并发迁移成功添加时才吞掉错误，否则继续抛出。重点修改
`src/storage/migrations/runner.ts` 中的 `addColumnIfMissing`
逻辑，保留现有幂等判断并补上失败后的二次确认。
Source: Path instructions
Comment on lines
+8
to
+11
| [[d1_databases]] | ||
| binding = "DB" | ||
| database_name = "inboxbridge" | ||
| database_id = "00000000-0000-0000-0000-000000000000" |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
database_id 还是占位值。
全 0 UUID 不能指向真实的 D1 实例。按这份配置部署后，Worker 的 webhook 和定时任务都拿不到目标数据库绑定。请在合并前替换成实际的 D1 database_id。
🤖 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 `@wrangler.toml` around lines 8 - 11, The D1 configuration still uses a
placeholder database_id, so the Worker cannot bind to a real database after
deployment. Update the [[d1_databases]] entry in wrangler.toml by replacing the
all-zero UUID with the actual D1 database_id used by the DB binding, keeping the
existing binding and database_name values intact.
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.
Summary
Validation
Wiki
Summary by CodeRabbit
新功能
Bug Fixes
Refactor