diff --git a/.claude/commands/pr-address.md b/.claude/commands/pr-address.md new file mode 100644 index 0000000..2d2f35f --- /dev/null +++ b/.claude/commands/pr-address.md @@ -0,0 +1,159 @@ +--- +description: Triage PR review comments, fix the valid ones, then reply to each comment (manual commit + push) +allowed-tools: Bash(gh pr view:*), Bash(gh pr checks:*), Bash(gh pr comment:*), Bash(gh api:*), Bash(gh repo view:*), Bash(gh run view:*), Bash(gh run list:*), Bash(git diff:*), Bash(git log:*), Bash(git status:*), Bash(git add:*), Bash(git commit:*), Bash(git push:*), Bash(git rev-parse:*), Bash(git branch:*), Bash(pnpm:*) +argument-hint: [PR-number] +--- + +Triage PR review comments, fix the valid ones, then reply to each comment. The command will NOT commit or push — that's a manual step. Three confirmation gates: never proceed past one without my explicit "yes". + +Arguments: `$ARGUMENTS` — optional PR number. If omitted, use the PR for the current branch. + +## 1. Resolve PR (read-only) + +If `$ARGUMENTS` is a number, use that PR. Otherwise `gh pr view --json number,headRefName,state,title,url,baseRefName`. Stop if no PR exists for the current branch. + +Capture `` and `` from `gh repo view --json owner,name`. + +## 2. Fetch comments + checks (read-only) + +Comments — pull all three sources, retaining each comment's `id` (needed to reply): + +- General PR comments: `gh api repos///issues//comments` +- Review / line comments: `gh api repos///pulls//comments` +- Review summaries / approvals: `gh api repos///pulls//reviews` + +Skip any comment authored by the current `gh api user --jq .login` (don't re-engage your own past replies). + +Checks — run `gh pr checks` (exit code 8 means "not all complete", that's fine, parse the output anyway). For each check that finished with `failure`, fetch the tail of the failed job log: + +``` +gh run view --log-failed | tail -60 +``` + +Cap at 60 lines per failure; truncate longer with a note pointing at the check URL. + +## 3. Triage + +First, print a one-line check summary: `Checks: ✓ · ✗ · ⏳`. If everything is passing or pending, say so. + +Then classify each comment AND each failed check: + +- **valid** — concrete claim or actionable suggestion grounded in the diff. Will be fixed. Replies posted. +- **question** — clarifying question; no code change, but reply needed. +- **noise** — bot welcome / "review enabled" / auto-generated PR overview / "show summary per file" / known-flaky CI failure. No fix, no reply by default. +- **approved** — APPROVED review with no concrete request. Acknowledge with `👍 thanks`, no fix. +- **ci-failure** — a failed Action / check that points at a real problem in the diff (test/typecheck/lint/format). Will be fixed. NO reply (CI re-runs after push). + +For ci-failure rows use the check name as `AUTHOR` and the failed-job URL as the `id`. Put a one-line root cause (parsed from the failed log tail) in the summary. + +Show one table: + +``` +ID AUTHOR FILE:LINE CLASS ONE-LINE SUMMARY + <≤ 90 chars> +``` + +Then propose a fix plan for everything classified `valid` or `ci-failure`: + +``` +[id ] (:line) + → +``` + +Group related fixes (same file or shared root cause) into a single edit when natural. + +**GATE 1.** Ask exactly: `Triage + fix plan look right? (yes / edit / cancel)`. Wait. + +If `edit`: let me reclassify (e.g. "id 1234 is noise") or adjust the plan, then re-confirm. + +## 4. Apply fixes (only after Gate 1 = yes) + +1. If `git status --porcelain` is non-empty, STOP — don't mix the user's WIP into the fix commit. Tell me to stash/commit first. +2. Make minimal edits per the confirmed plan. Don't fix things that weren't reported. +3. Run validation, only scripts that exist in `package.json`: + ``` + pnpm typecheck + pnpm lint + pnpm test + pnpm format + ``` + If any fail, STOP and show the failure. Do not commit broken code, do not post replies promising a non-existent fix. +4. Show me the diff (`git diff`) for review. + +**GATE 2.** Ask: `Fixes look right? (yes / edit / cancel)`. Wait. + +On `yes`, do NOT commit and do NOT push. Instead: + +5. Print a suggested commit message I can copy-paste: + + ``` + fix: address PR review feedback + + addresses: , + ``` + +6. STOP. Wait for me to commit + push manually. When I reply with `pushed` (or equivalent), continue to section 5 below — not before. +7. Once I confirm, run `git rev-parse HEAD` to capture the new commit SHA. It goes in the reply text. + +## 5. Draft replies (only after Gate 2 = yes) + +For each row classified `valid`, `question`, or `approved`, write a reply. Skip `ci-failure` (CI re-runs after push) and `noise` (silence). + +Style: terse, one or two sentences max. Per-class template: + +- `valid` → `Fixed in : .` Optionally a brief follow-up if the fix isn't 1:1 to the suggestion. +- `question` → answer directly. Don't speculate. If the answer is "this is intentional because X", say that. +- `approved` → `👍 thanks`. +- `ci-failure` → no reply. +- `noise` → no reply. + +Show all drafts in one block, grouped by comment id. + +**GATE 3.** Ask: `Post these replies (and resolve threads for fix-landed rows)? (yes / edit / cancel)`. Wait. + +## 6. Post replies + resolve threads (only after Gate 3 = yes) + +### 6.1 Post replies + +- **Review / line comment** (came from `pulls//comments`): reply in-thread: + `gh api --method POST repos///pulls//comments//replies -f body=""` +- **General PR comment** (came from `issues//comments`): `gh pr comment --body ""`, quoting the original with `> @` for context. +- **Review summary** (came from `pulls//reviews`): no per-thread reply API. If a reply is genuinely warranted (rare), post a general comment @-mentioning the reviewer; otherwise skip. + +After each post, confirm it returned a 200/201. If any fail, list exactly which — don't claim success for a failed post. + +### 6.2 Resolve threads for fix-landed rows + +A row is "fix-landed" iff its reply starts with `Fixed in :` — i.e. the fix actually shipped in the commit. Mark each such review thread resolved. Do NOT resolve `noise`, `question`, `approved`, `ci-failure`, or deferred-fix `valid` threads. + +1. List the PR's review threads with their constituent comment ids: + ``` + gh api graphql -F owner= -F name= -F num= -f query='query($owner:String!,$name:String!,$num:Int!){repository(owner:$owner,name:$name){pullRequest(number:$num){reviewThreads(first:100){nodes{id isResolved comments(first:10){nodes{databaseId}}}}}}}' + ``` +2. For each fix-landed reply, find the thread whose `comments.databaseId` list includes the original comment id. If `isResolved` is already `true`, skip. Otherwise resolve: + ``` + gh api graphql -F threadId= -f query='mutation($threadId:ID!){resolveReviewThread(input:{threadId:$threadId}){thread{isResolved}}}' + ``` +3. Confirm each response shows `thread.isResolved == true`. List any failures by original comment id. + +General PR comments (issues/comments) and review summaries don't have line-thread structure — nothing to resolve there. + +## 7. Final summary + +Print: + +- `Triaged: Fixed: (review + ci) Replied: Resolved: Skipped (noise): ` +- Commit SHA(s) +- PR URL + +Stop. + +## Hard rules + +- Never edit code before Gate 1 confirms. +- Never run `git commit` or `git push`. Those are manual — print the suggested commit message, then wait for me to confirm `pushed`. +- Never post replies before Gate 3 confirms. +- Never promise a fix that didn't actually land. Validation must pass AND I must confirm `pushed` before drafting replies. +- Never address a comment that wasn't in the confirmed set (no scope creep). +- Never reply to your own past comments. +- Never resolve a thread for a deferred-fix `valid` row (reply was "Agreed, tracking" rather than "Fixed in "). Only resolve threads where the fix actually shipped in the commit. diff --git a/.claude/commands/pr-open.md b/.claude/commands/pr-open.md new file mode 100644 index 0000000..52ec4c5 --- /dev/null +++ b/.claude/commands/pr-open.md @@ -0,0 +1,53 @@ +--- +description: Open a GitHub PR for the current branch with an auto-generated title and body +allowed-tools: Bash(git diff:*), Bash(git log:*), Bash(git status:*), Bash(git rev-parse:*), Bash(git branch:*), Bash(git push:*), Bash(gh auth status:*), Bash(gh pr create:*), Bash(gh pr view:*), Bash(gh repo view:*) +argument-hint: [--draft] +--- + +Open a GitHub PR for the current branch. Generate the title + body, show them to me, confirm, then push + create. + +Arguments: `$ARGUMENTS` — if it contains `--draft`, create the PR as draft. + +## Pre-flight + +Stop with a clear message if any of these fail: + +1. `gh auth status` — gh must be authenticated. +2. `git rev-parse --abbrev-ref HEAD` — must not be on `main`/`master`. +3. Determine base branch: try `main`, fall back to `master`. If neither exists, stop. +4. `git log ..HEAD --oneline` — if empty, stop ("nothing to PR"). +5. `gh pr view --json url,state` on the current branch — if a PR already exists, print its URL and stop. Do NOT create a duplicate. +6. `git status --porcelain` — if there are uncommitted changes, warn me but do NOT commit them. + +## Generate + +Run `git diff ...HEAD` and `git log ..HEAD --oneline`, then produce: + +- **Title**: plain English sentence (NOT `: ` semantic format). Capitalized, present tense, ≤ 72 chars, no trailing period. +- **Body**: three sections, total ~15–25 lines, skim-friendly. + - `### What`: 2–4 short bullets. Lead with the _why_ (bug, gap, user need), then key changes. No opening paragraph. Skip anything obvious from the diff. + - `### How to test`: commands one per line, no prose, no "expected results". Use only scripts that exist in `package.json` (verify first). Prefix manual steps with `Recommended:` if you didn't run them. + - `### Security review`: default to single line `No security-impacting changes.` Expand to checklist only when an item is actually affected. Format when expanded: + - **Secrets / env vars:** + - **Auth / session:** + - **Network / API calls:** + - **Data handling / PII:** + - **Dependencies:** + + Omit lines that didn't change. + +## Confirm + +Show me the title and body in two separate fenced code blocks. Then ASK explicitly: `Open this PR? (yes / edit / cancel)`. Wait for my reply. Do NOT push or create the PR without my confirmation. + +If I say "edit", let me edit, then re-confirm. + +## Create + +On confirmation: + +1. Push the branch with `git push -u origin ` (skip if `git status -sb` shows the branch is already published and up to date). +2. Run `gh pr create --base --title "" --body "$(cat <<'EOF' ... EOF)"` — pass the body via heredoc so newlines survive. Add `--draft` if requested. +3. Print the PR URL that `gh` returns. + +Do not commit on my behalf. Do not push without confirmation. Do not skip the confirmation step. diff --git a/.claude/commands/pr-status.md b/.claude/commands/pr-status.md new file mode 100644 index 0000000..58b3d68 --- /dev/null +++ b/.claude/commands/pr-status.md @@ -0,0 +1,69 @@ +--- +description: Show CI checks (Actions) and comments for the current branch's PR (or a given PR number) +allowed-tools: Bash(gh pr view:*), Bash(gh pr checks:*), Bash(gh run list:*), Bash(gh run view:*), Bash(gh api:*), Bash(git rev-parse:*), Bash(git branch:*) +argument-hint: [PR-number] +--- + +Show CI status and comments for a PR. Read-only — never approves, merges, comments, or reruns anything. + +Arguments: `$ARGUMENTS` — optional PR number. If omitted, use the PR for the current branch. + +## Resolve PR + +1. If `$ARGUMENTS` is a number, use that PR. +2. Otherwise, run `git rev-parse --abbrev-ref HEAD` and `gh pr view --json number,headRefName,state,isDraft,mergeable,title,url,updatedAt`. If no PR exists for the current branch, stop and say so. + +## Checks (Actions) + +Run `gh pr checks <PR>` (or `gh pr checks` if using current branch). Show each check on one line: + +``` +<icon> <name> <conclusion> <elapsed> <url> +``` + +Use `✅` for success, `❌` for failure, `⏳` for in_progress/queued/pending, `⏭️` for skipped, `❔` for anything else. Sort failures first. + +If any check failed, ALSO run `gh run view <runId> --log-failed | tail -60` for the most recent failed run and print that log block under the check list (fenced as `text`). Cap at 60 lines — if longer, say `(truncated, see <url> for full log)`. + +## Comments + +Fetch both kinds: + +- General PR comments: `gh api repos/{owner}/{repo}/issues/<PR>/comments --jq '.[] | {user: .user.login, body: .body, at: .updated_at, url: .html_url}'` +- Review/line comments: `gh api repos/{owner}/{repo}/pulls/<PR>/comments --jq '.[] | {user: .user.login, body: .body, at: .updated_at, file: .path, line: (.line // .original_line), url: .html_url}'` +- Reviews (approvals/changes-requested): `gh api repos/{owner}/{repo}/pulls/<PR>/reviews --jq '.[] | {user: .user.login, state: .state, body: .body, at: .submitted_at}'` + +Print them merged in chronological order. One block per comment: + +``` +@<user> <state-or-tag> <relative-time> +[<file>:<line>] ← only if a line comment +<body — wrapped, max ~6 lines; truncate the rest with "…"> +``` + +Tags: `[review:APPROVED]`, `[review:CHANGES_REQUESTED]`, `[review:COMMENTED]`, `[line]`, `[comment]`. + +If there are zero comments, say `No comments.` and stop the section. + +## Output structure + +Print in this order, with one blank line between sections: + +``` +PR #<num> <state> <draft?> <mergeable?> +<title> +<url> +updated <relative-time> + +Checks: +<one line per check> + +[failed-log block if applicable] + +Comments (<count>): +<one block per comment> +``` + +Keep it skim-friendly. No prose preamble, no closing summary. Don't quote bodies in full if they're long. + +Do NOT post comments, approve, request changes, merge, close, or rerun any workflow. This command is read-only. diff --git a/.claude/settings.json b/.claude/settings.json index a1c1cfb..c3e573c 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -6,10 +6,7 @@ "Bash(pnpm typecheck:*)", "Bash(pnpm test:*)", "Bash(pnpm format:*)", - "Bash(pnpm --filter * test*)", - "Bash(pnpm --filter * lint*)", - "Bash(pnpm --filter * typecheck*)", - "Bash(pnpm --filter * format*)" + "Bash(pnpm --filter:*)" ] } } diff --git a/apps/operator/migrations/0003_amused_marvex.sql b/apps/operator/migrations/0003_amused_marvex.sql new file mode 100644 index 0000000..6eead07 --- /dev/null +++ b/apps/operator/migrations/0003_amused_marvex.sql @@ -0,0 +1 @@ +ALTER TABLE `schedules` ADD `claimed_at` text; \ No newline at end of file diff --git a/apps/operator/migrations/meta/0003_snapshot.json b/apps/operator/migrations/meta/0003_snapshot.json new file mode 100644 index 0000000..eb4812c --- /dev/null +++ b/apps/operator/migrations/meta/0003_snapshot.json @@ -0,0 +1,216 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "32d9e6de-4bae-4cd1-ae78-bdde98e2ec7d", + "prevId": "a0dafd22-6b6a-4b4d-96a5-1e5497150c5b", + "tables": { + "pending_actions": { + "name": "pending_actions", + "columns": { + "chat_id": { + "name": "chat_id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "action_type": { + "name": "action_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "payload": { + "name": "payload", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "schedules": { + "name": "schedules", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "chat_id": { + "name": "chat_id", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "schedule_type": { + "name": "schedule_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "hour": { + "name": "hour", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "minute": { + "name": "minute", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false, + "default": 0 + }, + "day_of_week": { + "name": "day_of_week", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "day_of_month": { + "name": "day_of_month", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "timezone": { + "name": "timezone", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'UTC'" + }, + "fixed_message": { + "name": "fixed_message", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "message_prompt": { + "name": "message_prompt", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "source_url": { + "name": "source_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "keywords": { + "name": "keywords", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "state_json": { + "name": "state_json", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "active": { + "name": "active", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": true + }, + "next_run_at": { + "name": "next_run_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "claimed_at": { + "name": "claimed_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "retry_count": { + "name": "retry_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "idx_schedules_next_run": { + "name": "idx_schedules_next_run", + "columns": ["active", "next_run_at"], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} diff --git a/apps/operator/migrations/meta/_journal.json b/apps/operator/migrations/meta/_journal.json index 4ee8b88..8f94b6a 100644 --- a/apps/operator/migrations/meta/_journal.json +++ b/apps/operator/migrations/meta/_journal.json @@ -22,6 +22,13 @@ "when": 1776496458476, "tag": "0002_gray_xorn", "breakpoints": true + }, + { + "idx": 3, + "version": "6", + "when": 1777183590996, + "tag": "0003_amused_marvex", + "breakpoints": true } ] } diff --git a/apps/operator/src/db/schema.ts b/apps/operator/src/db/schema.ts index 621b7d7..c3ff668 100644 --- a/apps/operator/src/db/schema.ts +++ b/apps/operator/src/db/schema.ts @@ -23,6 +23,7 @@ const schedules = sqliteTable( description: text("description").notNull(), active: int("active", { mode: "boolean" }).notNull().default(true), nextRunAt: text("next_run_at").notNull(), + claimedAt: text("claimed_at"), retryCount: int("retry_count").notNull().default(0), createdAt: text("created_at") .notNull() diff --git a/apps/operator/src/scheduled.ts b/apps/operator/src/scheduled.ts index 832d2b9..3aac59c 100644 --- a/apps/operator/src/scheduled.ts +++ b/apps/operator/src/scheduled.ts @@ -22,6 +22,24 @@ type Env = AppEnv["Bindings"]; const RESPONSE_SIZE_LIMIT = 1024 * 1024; // 1MB +const describeErrorCause = (error: unknown): string | undefined => { + if (!(error instanceof Error) || error.cause === undefined) { + return undefined; + } + const { cause } = error; + if (cause instanceof Error) { + return cause.message; + } + if (typeof cause === "string") { + return cause; + } + try { + return JSON.stringify(cause); + } catch { + return undefined; + } +}; + const handleScheduled = async ( _event: ScheduledEvent, env: Env, @@ -39,11 +57,13 @@ const handleScheduled = async ( } catch (error) { logger.error("failed to claim due schedules", { error: error instanceof Error ? error.message : String(error), + cause: describeErrorCause(error), }); return; } if (claimed.length === 0) { + logger.info("no due schedules"); return; } @@ -97,13 +117,19 @@ const handleScheduled = async ( logger.info("monitor keyword check — no matches, skipping", { scheduleId: schedule.id, }); - await scheduleService.updateState( - schedule.id, + const { lockLost } = await scheduleService.updateState( + schedule, JSON.stringify({ lastContent: `No keyword matches found in scraped content.${scrapeResult.truncated ? " Scraped content was truncated before keyword matching." : ""}`, lastScrapedAt: now.toISOString(), }) ); + if (lockLost) { + logger.warn( + "lock lost during monitor execution; state update skipped", + { scheduleId: schedule.id } + ); + } return; } scrapedContent = extractWindows(sourceContent, positions, 2000); @@ -141,13 +167,19 @@ const handleScheduled = async ( }); } - await scheduleService.updateState( - schedule.id, + const { lockLost } = await scheduleService.updateState( + schedule, JSON.stringify({ lastContent: analysis.newState, lastScrapedAt: now.toISOString(), }) ); + if (lockLost) { + logger.warn( + "lock lost during monitor execution; state update skipped", + { scheduleId: schedule.id } + ); + } return; } @@ -189,11 +221,17 @@ const handleScheduled = async ( ); // Handle failures — mark for retry + const completionTime = new Date(); for (let i = 0; i < results.length; i++) { const result = results[i]; const schedule = claimed[i]; if (result.status === "fulfilled") { - await scheduleService.markSuccess(schedule.id); + const { lockLost } = await scheduleService.markSuccess(schedule); + if (lockLost) { + logger.warn("lock lost before recording success", { + scheduleId: schedule.id, + }); + } } if (result.status === "rejected") { @@ -205,10 +243,16 @@ const handleScheduled = async ( : String(result.reason), }); - const { exhausted } = await scheduleService.markFailed( - schedule.id, - schedule.retryCount + const { exhausted, lockLost } = await scheduleService.markFailed( + schedule, + completionTime ); + if (lockLost) { + logger.warn("lock lost before recording failure", { + scheduleId: schedule.id, + }); + continue; + } if (exhausted) { logger.warn("schedule retries exhausted, skipping until next run", { scheduleId: schedule.id, diff --git a/apps/operator/src/services/schedule.ts b/apps/operator/src/services/schedule.ts index 8a739b5..397edf4 100644 --- a/apps/operator/src/services/schedule.ts +++ b/apps/operator/src/services/schedule.ts @@ -1,4 +1,4 @@ -import { and, eq, lte, sql } from "drizzle-orm"; +import { and, eq, isNull, lt, lte, or, sql } from "drizzle-orm"; import { drizzle } from "drizzle-orm/d1"; import type { DrizzleD1Database } from "drizzle-orm/d1"; import { z } from "zod"; @@ -11,6 +11,12 @@ type ScheduleType = (typeof SCHEDULE_TYPES)[number]; const MAX_ACTIVE_SCHEDULES = 20; const MAX_RETRIES = 3; +// A claim is treated as abandoned if it's older than this threshold, +// allowing another worker to re-claim the row after a crashed/timed-out run. +const STALE_LOCK_MS = 10 * 60 * 1000; + +type ClaimedSchedule = typeof schedules.$inferSelect & { claimedAt: string }; + const createScheduleSchema = z .object({ scheduleType: z.enum(SCHEDULE_TYPES), @@ -350,11 +356,20 @@ class ScheduleService { } /** - * Claim due schedules for a specific chat. Returns claimed rows - * with next_run_at already advanced to the next occurrence. + * Claim due schedules for a specific chat by acquiring a lock + * (claimed_at). next_run_at is NOT advanced here — that happens only + * after execution succeeds (or retries are exhausted), so an ambiguous + * D1 outcome can't silently drop an occurrence. Stale locks (older than + * STALE_LOCK_MS) are reclaimable to recover from crashed runs. */ - async claimDueSchedules(now: Date, allowedChatId: string) { + async claimDueSchedules( + now: Date, + allowedChatId: string + ): Promise<ClaimedSchedule[]> { const nowIso = now.toISOString(); + const staleBeforeIso = new Date( + now.getTime() - STALE_LOCK_MS + ).toISOString(); const chatIdNum = Number(allowedChatId); const dueRows = await this.db @@ -364,7 +379,11 @@ class ScheduleService { and( eq(schedules.active, true), lte(schedules.nextRunAt, nowIso), - eq(schedules.chatId, chatIdNum) + eq(schedules.chatId, chatIdNum), + or( + isNull(schedules.claimedAt), + lt(schedules.claimedAt, staleBeforeIso) + ) ) ); @@ -372,72 +391,124 @@ class ScheduleService { return []; } - // Claim each row by advancing next_run_at. - // The WHERE matches the row's actual stored nextRunAt so a concurrent - // cron invocation that already advanced it will get 0 changes. - const claimed: (typeof dueRows)[number][] = []; + const claimed: ClaimedSchedule[] = []; for (const row of dueRows) { - const nextRun = computeNextRun(row.scheduleType, row.timezone, now, { - hour: row.hour ?? undefined, - minute: row.minute ?? undefined, - dayOfWeek: row.dayOfWeek ?? undefined, - dayOfMonth: row.dayOfMonth ?? undefined, - }); + const lockGuard = + row.claimedAt === null + ? isNull(schedules.claimedAt) + : eq(schedules.claimedAt, row.claimedAt); + // Also guard on nextRunAt: if a concurrent worker completed this row + // between SELECT and UPDATE (advancing nextRunAt and clearing claimedAt + // back to NULL), the lockGuard alone would let us reclaim a row that's + // no longer due, causing duplicate execution. const result = await this.db .update(schedules) - .set({ nextRunAt: nextRun.toISOString() }) + .set({ claimedAt: nowIso }) .where( - and(eq(schedules.id, row.id), eq(schedules.nextRunAt, row.nextRunAt)) + and( + eq(schedules.id, row.id), + eq(schedules.nextRunAt, row.nextRunAt), + lockGuard + ) ); if (result.meta.changes > 0) { - claimed.push(row); + claimed.push({ ...row, claimedAt: nowIso }); } } return claimed; } - async markFailed(id: string, currentRetryCount: number) { - const newCount = currentRetryCount + 1; - if (newCount >= MAX_RETRIES) { - // All retries exhausted — reset retryCount so the next regular - // occurrence starts fresh. nextRunAt was already advanced to the - // next occurrence by claimDueSchedules, so leave it as-is. - await this.db - .update(schedules) - .set({ retryCount: 0 }) - .where(eq(schedules.id, id)); - return { exhausted: true }; + async markFailed(row: ClaimedSchedule, now: Date) { + const newCount = row.retryCount + 1; + const exhausted = newCount >= MAX_RETRIES; + + const update = exhausted + ? { + // Skip this slot, reset for the next regular occurrence. + // Compute next from row.nextRunAt (the slot we just attempted), + // not from `now`, so a long-running batch never silently jumps + // past intermediate occurrences for short-interval schedules. + nextRunAt: this.nextOccurrence(row).toISOString(), + retryCount: 0, + claimedAt: null, + } + : { + retryCount: newCount, + nextRunAt: new Date( + now.getTime() + newCount * 2 * 60 * 1000 + ).toISOString(), + claimedAt: null, + }; + + const result = await this.db + .update(schedules) + .set(update) + .where( + and(eq(schedules.id, row.id), eq(schedules.claimedAt, row.claimedAt)) + ); + + if (result.meta.changes === 0) { + return { exhausted: false, lockLost: true }; } + return { exhausted, lockLost: false }; + } - const backoffMs = newCount * 2 * 60 * 1000; - const nextRetry = new Date(Date.now() + backoffMs); - await this.db + async markSuccess(row: ClaimedSchedule) { + const result = await this.db .update(schedules) .set({ - retryCount: newCount, - nextRunAt: nextRetry.toISOString(), + nextRunAt: this.nextOccurrence(row).toISOString(), + retryCount: 0, + claimedAt: null, }) - .where(eq(schedules.id, id)); - return { exhausted: false }; + .where( + and(eq(schedules.id, row.id), eq(schedules.claimedAt, row.claimedAt)) + ); + return { lockLost: result.meta.changes === 0 }; } - async markSuccess(id: string) { - await this.db - .update(schedules) - .set({ retryCount: 0 }) - .where(eq(schedules.id, id)); + /** + * Compute the next occurrence relative to row.nextRunAt (the slot just + * processed), not relative to wall-clock `now`. This guarantees we always + * advance to the next strictly-later occurrence, so a long-running batch + * for a short-interval schedule (e.g. hourly) never skips intermediate + * occurrences. + */ + private nextOccurrence(row: ClaimedSchedule): Date { + return computeNextRun( + row.scheduleType, + row.timezone, + new Date(row.nextRunAt), + { + hour: row.hour ?? undefined, + minute: row.minute ?? undefined, + dayOfWeek: row.dayOfWeek ?? undefined, + dayOfMonth: row.dayOfMonth ?? undefined, + } + ); } - async updateState(id: string, stateJson: string) { + /** + * Lock-guarded state write. The UPDATE only succeeds if the caller still + * owns the lock (claimed_at matches the value they were given at claim + * time). Without this, a long-running worker that lost ownership to a + * stale-lock reclaim could overwrite a newer worker's state. Returns + * `{ lockLost: true }` so the caller can react instead of silently + * accepting a no-op. + */ + async updateState(row: ClaimedSchedule, stateJson: string) { const STATE_MAX_BYTES = 100 * 1024; if (new TextEncoder().encode(stateJson).byteLength > STATE_MAX_BYTES) { throw new Error("stateJson exceeds 100KB limit"); } - await this.db + const result = await this.db .update(schedules) .set({ stateJson }) - .where(eq(schedules.id, id)); + .where( + and(eq(schedules.id, row.id), eq(schedules.claimedAt, row.claimedAt)) + ); + return { lockLost: result.meta.changes === 0 }; } }