From f048dc3bd1c671e28aa4ceaaf22ca6663cdcb321 Mon Sep 17 00:00:00 2001 From: Ben Senescu <44480372+bensenescu@users.noreply.github.com> Date: Fri, 7 Aug 2026 13:53:30 -0400 Subject: [PATCH] fix(rank-tracking): drain due configs deterministically (clean-room alternative to #456) (#462) --- .agents/PAPERCUTS.md | 2 +- .../rank-tracking/RankTrackingDomainList.tsx | 6 + src/db/runBatch.ts | 2 +- src/server.ts | 7 +- src/server/billing/subscription.test.ts | 16 +- src/server/billing/subscription.ts | 18 +- .../RankTrackingRepository.query.test.ts | 278 ++++++++++++++ .../repositories/RankTrackingRepository.ts | 138 +++++-- .../services/scheduledRankChecks.test.ts | 354 ++++++++++++++++++ .../services/scheduledRankChecks.ts | 253 ++++++++++--- src/shared/rank-tracking.test.ts | 19 + src/shared/rank-tracking.ts | 7 + wrangler.jsonc | 2 +- 13 files changed, 1003 insertions(+), 99 deletions(-) create mode 100644 src/server/features/rank-tracking/repositories/RankTrackingRepository.query.test.ts create mode 100644 src/server/features/rank-tracking/services/scheduledRankChecks.test.ts diff --git a/.agents/PAPERCUTS.md b/.agents/PAPERCUTS.md index e14c70c..fec46cd 100644 --- a/.agents/PAPERCUTS.md +++ b/.agents/PAPERCUTS.md @@ -10,7 +10,7 @@ data, or sensitive paths. ## Open -- [ ] `2026-08-05T20:59:09Z` — `codex` — The documented `pnpm seed:rank-tracking` command fails before opening local D1 because `scripts/seed-rank-tracking.ts` imports the provider-aware `src/db/schema` barrel and plain `tsx` cannot load the resulting `cloudflare:workers` URL. Keep the seed script on dialect-local schema imports or run it through a Workers-compatible execution path. +- [ ] `2026-08-05T20:59:09Z` — `codex` — The documented `pnpm seed:rank-tracking` command fails before opening local D1 because `scripts/seed-rank-tracking.ts` imports the provider-aware `src/db/schema` barrel and plain `tsx` cannot load the resulting `cloudflare:workers` URL. Keep the seed script on dialect-local schema imports or run it through a Workers-compatible execution path. (Workaround: seed via raw SQL with `wrangler d1 execute DB --local`.) - [ ] `2026-08-01T16:28:36Z` — `claude` — web's pinned wrangler 4.71.0 fails `kv namespace create` with a bare "Authentication error [code: 10000]" even though the OAuth token has workers_kv write scope; wrangler@4.118.0 succeeds with identical auth. Fix: bump wrangler in web/package.json. - [ ] `2026-07-20T20:08:28Z` — `claude` — In a fresh git worktree, `oxlint --type-aware` crashes with `Cannot find module '@oxlint/binding-darwin-arm64'` — the platform-specific optional dep is missing from the worktree's node_modules while tsc/prettier work fine, and plain `pnpm install` reports up-to-date without restoring it; `pnpm install --force` (~22s) fixes it. Worth making the worktree-setup hook (or a documented step) run the forced install so lint doesn't die on fresh worktrees. - [ ] `2026-07-19T04:06:52Z` — `codex` — `pnpm --dir web build` fails with `vite: command not found` when `web/node_modules` is absent, despite the root toolchain being installed. Document or enforce the package-local install required before validating the `web/` subpackage. diff --git a/src/client/features/rank-tracking/RankTrackingDomainList.tsx b/src/client/features/rank-tracking/RankTrackingDomainList.tsx index 89eb362..6186dc5 100644 --- a/src/client/features/rank-tracking/RankTrackingDomainList.tsx +++ b/src/client/features/rank-tracking/RankTrackingDomainList.tsx @@ -234,6 +234,12 @@ function DomainRow({ Scheduled check skipped — insufficient credits

)} + {summary.lastSkipReason === "plan_required" && ( +

+ + Scheduled check skipped — paid plan required +

+ )}
{summary.keywordCount > 0 && ( diff --git a/src/db/runBatch.ts b/src/db/runBatch.ts index da9e069..3c19779 100644 --- a/src/db/runBatch.ts +++ b/src/db/runBatch.ts @@ -11,7 +11,7 @@ type BatchStatement = Parameters[0][number]; // D1 caps bound parameters at ~100 per statement; keep batches bounded so each // runBatch call stays under that limit. (Postgres allows far more, but the same // chunk size is harmless there.) -const DB_BATCH_SIZE = 100; +export const DB_BATCH_SIZE = 100; /** * Run a set of write statements atomically on either backend. diff --git a/src/server.ts b/src/server.ts index 0407171..9d476f3 100644 --- a/src/server.ts +++ b/src/server.ts @@ -189,7 +189,12 @@ export default { _ctx: ExecutionContext, ) { // Scope a per-request Postgres client for the cron run (no-op in D1 mode). - await withPgClient(() => runScheduledRankChecks(env)); + // Caught so a rank-tracking failure can't suppress the audit watchdog below. + try { + await withPgClient(() => runScheduledRankChecks(env)); + } catch (err) { + console.error("[cron] Scheduled rank checks failed:", err); + } // Watchdog: reconcile audits stuck in "running" whose workflow died // without reaching mark-failed (OOM/CPU kills, expired instances). await withPgClient(() => reconcileStaleAudits()); diff --git a/src/server/billing/subscription.test.ts b/src/server/billing/subscription.test.ts index 62e49c1..baf2291 100644 --- a/src/server/billing/subscription.test.ts +++ b/src/server/billing/subscription.test.ts @@ -63,10 +63,24 @@ describe("subscription billing", () => { }); }); - it("returns false when org lacks paid plan", async () => { + it("returns false without retrying when org lacks paid plan", async () => { checkMock.mockResolvedValue({ allowed: false }); await expect(customerHasPaidPlan("org_123")).resolves.toBe(false); + expect(checkMock).toHaveBeenCalledTimes(1); + }); + + it("recovers from a degraded negative read when retryDenied is set", async () => { + vi.useFakeTimers(); + checkMock + .mockResolvedValueOnce({ allowed: false }) + .mockResolvedValueOnce({ allowed: true }); + + const result = customerHasPaidPlan("org_123", { retryDenied: true }); + await vi.runAllTimersAsync(); + + await expect(result).resolves.toBe(true); + expect(checkMock).toHaveBeenCalledTimes(2); }); it("retries a missing monthly balance once", async () => { diff --git a/src/server/billing/subscription.ts b/src/server/billing/subscription.ts index d2e699f..a555254 100644 --- a/src/server/billing/subscription.ts +++ b/src/server/billing/subscription.ts @@ -63,13 +63,27 @@ export async function getOrCreateOrganizationCustomer( return { id: customer.id }; } -export async function customerHasPaidPlan(customerId: string) { +export async function customerHasPaidPlan( + customerId: string, + opts: { retryDenied?: boolean } = {}, +) { const result = await autumn.check({ customerId, featureId: AUTUMN_PAID_PLAN_FEATURE_ID, }); + if (result.allowed || !opts.retryDenied) return result.allowed; - return result.allowed; + // Autumn sometimes returns degraded entitlement data in a successful + // response (see the balance retry in getUsageCreditsRemaining). Where a + // false negative does lasting damage — the scheduler would advance a paying + // org's schedule and flag "plan_required" — callers opt into one re-check. + // Interactive deny paths skip it to stay fast for genuinely free users. + await new Promise((resolve) => setTimeout(resolve, 300)); + const retry = await autumn.check({ + customerId, + featureId: AUTUMN_PAID_PLAN_FEATURE_ID, + }); + return retry.allowed; } export async function customerHasManagedAccess(customerId: string) { diff --git a/src/server/features/rank-tracking/repositories/RankTrackingRepository.query.test.ts b/src/server/features/rank-tracking/repositories/RankTrackingRepository.query.test.ts new file mode 100644 index 0000000..dc343bc --- /dev/null +++ b/src/server/features/rank-tracking/repositories/RankTrackingRepository.query.test.ts @@ -0,0 +1,278 @@ +import { createClient, type Client } from "@libsql/client"; +import { drizzle } from "drizzle-orm/libsql"; +import { eq } from "drizzle-orm"; +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; +import { rankTrackingConfigs } from "@/db/schema"; +import type * as RankTrackingRepositoryModule from "./RankTrackingRepository"; + +// Real in-memory SQLite so the due-query ordering, the manual-interval filter, +// and claimDueConfig's compare-and-set run against actual SQL — the parts the +// mocked service tests can't see. + +vi.mock("cloudflare:workers", () => ({ + env: { DATABASE_PROVIDER: "d1" }, +})); + +let client: Client; +let testDb: ReturnType; +let RankTrackingRepository: typeof RankTrackingRepositoryModule.RankTrackingRepository; + +beforeAll(async () => { + client = createClient({ url: "file::memory:" }); + testDb = drizzle(client); + vi.doMock("@/db", () => ({ db: testDb })); + + await client.executeMultiple(` + CREATE TABLE projects ( + id TEXT PRIMARY KEY, + organization_id TEXT NOT NULL, + name TEXT NOT NULL, + domain TEXT, + location_code INTEGER NOT NULL DEFAULT 2840, + language_code TEXT NOT NULL DEFAULT 'en', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + archived_at TEXT + ); + CREATE TABLE rank_tracking_configs ( + id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + domain TEXT NOT NULL, + location_code INTEGER NOT NULL DEFAULT 2840, + language_code TEXT NOT NULL DEFAULT 'en', + devices TEXT NOT NULL DEFAULT 'both', + serp_depth INTEGER NOT NULL, + schedule_interval TEXT NOT NULL DEFAULT 'weekly', + location_name TEXT, + is_active INTEGER NOT NULL DEFAULT 1, + last_checked_at TEXT, + next_check_at TEXT, + last_skip_reason TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + CREATE TABLE rank_tracking_keywords ( + id TEXT PRIMARY KEY, + config_id TEXT NOT NULL, + keyword TEXT NOT NULL, + search_volume INTEGER, + keyword_difficulty INTEGER, + cpc REAL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + `); + + ({ RankTrackingRepository } = await import("./RankTrackingRepository")); +}); + +afterAll(() => { + client.close(); +}); + +beforeEach(async () => { + await client.executeMultiple(` + DELETE FROM rank_tracking_keywords; + DELETE FROM rank_tracking_configs; + DELETE FROM projects; + `); +}); + +const NOW = "2026-08-01T12:00:00.000Z"; + +async function seedProject(id: string, archivedAt: string | null = null) { + await client.execute({ + sql: "INSERT INTO projects (id, organization_id, name, archived_at) VALUES (?, ?, ?, ?)", + args: [id, `org_${id}`, id, archivedAt], + }); +} + +async function seedConfig(input: { + id: string; + projectId?: string; + nextCheckAt?: string | null; + scheduleInterval?: string; + isActive?: number; + lastSkipReason?: string | null; +}) { + await client.execute({ + sql: `INSERT INTO rank_tracking_configs + (id, project_id, domain, serp_depth, schedule_interval, is_active, next_check_at, last_skip_reason) + VALUES (?, ?, ?, 20, ?, ?, ?, ?)`, + args: [ + input.id, + input.projectId ?? "proj_1", + `${input.id}.example.com`, + input.scheduleInterval ?? "daily", + input.isActive ?? 1, + input.nextCheckAt === undefined + ? "2026-07-01T05:00:00.000Z" + : input.nextCheckAt, + input.lastSkipReason ?? null, + ], + }); +} + +async function getConfigRow(id: string) { + const rows = await testDb + .select({ + nextCheckAt: rankTrackingConfigs.nextCheckAt, + lastSkipReason: rankTrackingConfigs.lastSkipReason, + }) + .from(rankTrackingConfigs) + .where(eq(rankTrackingConfigs.id, id)); + return rows[0]; +} + +describe("getDueConfigsWithOrganization", () => { + it("returns due rows oldest-first and excludes manual, inactive, future, and archived rows", async () => { + await seedProject("proj_1"); + await seedProject("proj_archived", "2026-07-15T00:00:00.000Z"); + await seedConfig({ + id: "cfg_newer", + nextCheckAt: "2026-07-02T05:00:00.000Z", + }); + await seedConfig({ + id: "cfg_oldest", + nextCheckAt: "2026-07-01T05:00:00.000Z", + }); + // Same timestamp as cfg_oldest — id breaks the tie deterministically. + await seedConfig({ + id: "cfg_tie", + nextCheckAt: "2026-07-01T05:00:00.000Z", + }); + await seedConfig({ id: "cfg_manual", scheduleInterval: "manual" }); + await seedConfig({ id: "cfg_inactive", isActive: 0 }); + await seedConfig({ + id: "cfg_future", + nextCheckAt: "2027-01-01T05:00:00.000Z", + }); + await seedConfig({ id: "cfg_null_due", nextCheckAt: null }); + await seedConfig({ id: "cfg_archived", projectId: "proj_archived" }); + + const due = await RankTrackingRepository.getDueConfigsWithOrganization(NOW); + + expect(due.map((c) => c.id)).toEqual([ + "cfg_oldest", + "cfg_tie", + "cfg_newer", + ]); + expect(due[0].organizationId).toBe("org_proj_1"); + }); +}); + +describe("claimDueConfig", () => { + it("advances the schedule and writes the skip reason when the token matches", async () => { + await seedProject("proj_1"); + await seedConfig({ id: "cfg_1", nextCheckAt: "2026-07-01T05:00:00.000Z" }); + + const claimed = await RankTrackingRepository.claimDueConfig({ + configId: "cfg_1", + projectId: "proj_1", + observedNextCheckAt: "2026-07-01T05:00:00.000Z", + nextCheckAt: "2026-08-02T05:00:00.000Z", + lastSkipReason: "plan_required", + }); + + expect(claimed).toBe(true); + expect(await getConfigRow("cfg_1")).toEqual({ + nextCheckAt: "2026-08-02T05:00:00.000Z", + lastSkipReason: "plan_required", + }); + }); + + it("refuses when the observed token no longer matches, leaving the row untouched", async () => { + await seedProject("proj_1"); + await seedConfig({ + id: "cfg_1", + nextCheckAt: "2026-07-20T05:00:00.000Z", + lastSkipReason: "insufficient_credits", + }); + + const claimed = await RankTrackingRepository.claimDueConfig({ + configId: "cfg_1", + projectId: "proj_1", + observedNextCheckAt: "2026-07-01T05:00:00.000Z", + nextCheckAt: "2026-08-02T05:00:00.000Z", + lastSkipReason: null, + }); + + expect(claimed).toBe(false); + expect(await getConfigRow("cfg_1")).toEqual({ + nextCheckAt: "2026-07-20T05:00:00.000Z", + lastSkipReason: "insufficient_credits", + }); + }); + + it("refuses deactivated rows and preserves the skip reason when omitted", async () => { + await seedProject("proj_1"); + await seedConfig({ id: "cfg_off", isActive: 0 }); + await seedConfig({ + id: "cfg_restore", + nextCheckAt: "2026-08-02T05:00:00.000Z", + lastSkipReason: "insufficient_credits", + }); + + expect( + await RankTrackingRepository.claimDueConfig({ + configId: "cfg_off", + projectId: "proj_1", + observedNextCheckAt: "2026-07-01T05:00:00.000Z", + nextCheckAt: "2026-08-02T05:00:00.000Z", + }), + ).toBe(false); + + // Restore-shaped claim: no lastSkipReason key — the existing value stays. + expect( + await RankTrackingRepository.claimDueConfig({ + configId: "cfg_restore", + projectId: "proj_1", + observedNextCheckAt: "2026-08-02T05:00:00.000Z", + nextCheckAt: "2026-07-01T05:00:00.000Z", + }), + ).toBe(true); + expect(await getConfigRow("cfg_restore")).toEqual({ + nextCheckAt: "2026-07-01T05:00:00.000Z", + lastSkipReason: "insufficient_credits", + }); + }); +}); + +describe("getKeywordCountsForConfigs", () => { + it("groups counts by config and omits configs with no keywords", async () => { + await seedProject("proj_1"); + await seedConfig({ id: "cfg_a" }); + await seedConfig({ id: "cfg_b" }); + for (const [id, configId] of [ + ["kw1", "cfg_a"], + ["kw2", "cfg_a"], + ["kw3", "cfg_b"], + ] as const) { + await client.execute({ + sql: "INSERT INTO rank_tracking_keywords (id, config_id, keyword) VALUES (?, ?, ?)", + args: [id, configId, id], + }); + } + + const counts = await RankTrackingRepository.getKeywordCountsForConfigs([ + "cfg_a", + "cfg_b", + "cfg_missing", + ]); + + expect(counts).toEqual( + new Map([ + ["cfg_a", 2], + ["cfg_b", 1], + ]), + ); + expect(await RankTrackingRepository.getKeywordCountsForConfigs([])).toEqual( + new Map(), + ); + }); +}); diff --git a/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts b/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts index 9e80522..0cc8933 100644 --- a/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts +++ b/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts @@ -1,4 +1,15 @@ -import { and, count, desc, eq, inArray, isNull, lte, max } from "drizzle-orm"; +import { + and, + asc, + count, + desc, + eq, + inArray, + isNull, + lte, + max, + ne, +} from "drizzle-orm"; import type { InferInsertModel } from "drizzle-orm"; import { db } from "@/db"; import { @@ -8,7 +19,8 @@ import { rankTrackingKeywords, projects, } from "@/db/schema"; -import { executeInBatches } from "@/db/runBatch"; +import { DB_BATCH_SIZE, executeInBatches } from "@/db/runBatch"; +import type { RankTrackingSkipReason } from "@/shared/rank-tracking"; import { getLatestSnapshotsForKeywords, getSnapshotsBeforeDate, @@ -104,30 +116,81 @@ async function updateConfig( } async function getDueConfigsWithOrganization(nowIso: string) { - return db - .select({ - id: rankTrackingConfigs.id, - projectId: rankTrackingConfigs.projectId, - domain: rankTrackingConfigs.domain, - locationCode: rankTrackingConfigs.locationCode, - languageCode: rankTrackingConfigs.languageCode, - locationName: rankTrackingConfigs.locationName, - devices: rankTrackingConfigs.devices, - serpDepth: rankTrackingConfigs.serpDepth, - scheduleInterval: rankTrackingConfigs.scheduleInterval, - nextCheckAt: rankTrackingConfigs.nextCheckAt, - organizationId: projects.organizationId, + return ( + db + .select({ + id: rankTrackingConfigs.id, + projectId: rankTrackingConfigs.projectId, + domain: rankTrackingConfigs.domain, + locationCode: rankTrackingConfigs.locationCode, + languageCode: rankTrackingConfigs.languageCode, + locationName: rankTrackingConfigs.locationName, + devices: rankTrackingConfigs.devices, + serpDepth: rankTrackingConfigs.serpDepth, + scheduleInterval: rankTrackingConfigs.scheduleInterval, + nextCheckAt: rankTrackingConfigs.nextCheckAt, + organizationId: projects.organizationId, + }) + .from(rankTrackingConfigs) + .innerJoin(projects, eq(rankTrackingConfigs.projectId, projects.id)) + .where( + and( + eq(rankTrackingConfigs.isActive, true), + // A manual config can keep a stale non-null next_check_at; without this + // it would be selected every tick and never advanced. + ne(rankTrackingConfigs.scheduleInterval, "manual"), + lte(rankTrackingConfigs.nextCheckAt, nowIso), + isNull(projects.archivedAt), + ), + ) + // Oldest first so a large backlog drains in order instead of the same + // arbitrary rows filling every batch. `lte` already excludes NULL, so both + // ordering columns are non-null and SQLite/Postgres agree. + .orderBy( + asc(rankTrackingConfigs.nextCheckAt), + asc(rankTrackingConfigs.id), + ) + .limit(200) + ); +} + +/** + * Conditionally advance a due config's schedule, returning false when the + * config changed underneath us (manual edit, deactivation). + * + * `next_check_at` equality is the compare-and-set token. `schedule_interval` is + * deliberately absent from the predicate: every schedule edit rewrites + * `next_check_at` (updateConfig recomputes it, or nulls it for "manual"), so + * the timestamp check already detects interval changes. + * + * `lastSkipReason` is written only when the caller passes it — the restore + * path omits it so it can't clobber a reason the blocking run just wrote. + */ +async function claimDueConfig(input: { + configId: string; + projectId: string; + observedNextCheckAt: string; + nextCheckAt: string; + lastSkipReason?: RankTrackingSkipReason | null; +}): Promise { + const claimed = await db + .update(rankTrackingConfigs) + .set({ + nextCheckAt: input.nextCheckAt, + ...(input.lastSkipReason !== undefined && { + lastSkipReason: input.lastSkipReason, + }), }) - .from(rankTrackingConfigs) - .innerJoin(projects, eq(rankTrackingConfigs.projectId, projects.id)) .where( and( + eq(rankTrackingConfigs.id, input.configId), + eq(rankTrackingConfigs.projectId, input.projectId), eq(rankTrackingConfigs.isActive, true), - lte(rankTrackingConfigs.nextCheckAt, nowIso), - isNull(projects.archivedAt), + eq(rankTrackingConfigs.nextCheckAt, input.observedNextCheckAt), ), ) - .limit(50); + .returning({ id: rankTrackingConfigs.id }); + return claimed.length > 0; } // --------------------------------------------------------------------------- @@ -297,22 +360,7 @@ async function getConfigSummaries(projectId: string) { const configs = await getConfigsForProject(projectId); if (configs.length === 0) return []; - // Batch: keyword counts grouped by config - const kwCounts = await db - .select({ - configId: rankTrackingKeywords.configId, - value: count(), - }) - .from(rankTrackingKeywords) - .where( - inArray( - rankTrackingKeywords.configId, - configs.map((c) => c.id), - ), - ) - .groupBy(rankTrackingKeywords.configId); - - const kwCountMap = new Map(kwCounts.map((r) => [r.configId, r.value])); + const kwCountMap = await getKeywordCountsForConfigs(configs.map((c) => c.id)); // Subquery: latest startedAt per config const latestStarted = db @@ -395,6 +443,22 @@ async function getKeywordCountForConfig(configId: string) { return rows[0]?.value ?? 0; } +/** Keyword counts keyed by config id. Configs with no keywords are absent. */ +async function getKeywordCountsForConfigs(configIds: string[]) { + // Chunked so the IN list stays under D1's ~100 bound-parameter cap. + const counts = new Map(); + for (let i = 0; i < configIds.length; i += DB_BATCH_SIZE) { + const chunk = configIds.slice(i, i + DB_BATCH_SIZE); + const rows = await db + .select({ configId: rankTrackingKeywords.configId, value: count() }) + .from(rankTrackingKeywords) + .where(inArray(rankTrackingKeywords.configId, chunk)) + .groupBy(rankTrackingKeywords.configId); + for (const row of rows) counts.set(row.configId, row.value); + } + return counts; +} + export const RankTrackingRepository = { getConfigsForProject, getConfigById, @@ -402,6 +466,7 @@ export const RankTrackingRepository = { createConfig, updateConfig, getDueConfigsWithOrganization, + claimDueConfig, tryCreateRun, updateRun, getRunById, @@ -414,6 +479,7 @@ export const RankTrackingRepository = { removeKeywordsFromConfig, updateKeywordMetrics, getKeywordCountForConfig, + getKeywordCountsForConfigs, getConfigSummaries, getLatestSnapshotsForKeywords, getSnapshotsBeforeDate, diff --git a/src/server/features/rank-tracking/services/scheduledRankChecks.test.ts b/src/server/features/rank-tracking/services/scheduledRankChecks.test.ts new file mode 100644 index 0000000..94611e7 --- /dev/null +++ b/src/server/features/rank-tracking/services/scheduledRankChecks.test.ts @@ -0,0 +1,354 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +type DueConfigRow = { + id: string; + projectId: string; + domain: string; + locationCode: number; + languageCode: string; + locationName: string | null; + devices: "both" | "desktop" | "mobile"; + serpDepth: number; + scheduleInterval: "daily" | "weekly" | "monthly" | "manual"; + nextCheckAt: string | null; + organizationId: string; +}; + +type ClaimInput = { + configId: string; + projectId: string; + observedNextCheckAt: string; + nextCheckAt: string; + lastSkipReason?: string | null; +}; + +type BeginResult = + | { ok: true; runId: string } + | { ok: false; reason: string; blockingRunId: string | null }; + +// Typed so assertions can read `mock.calls` without tripping the type-aware +// lint rules on `any`. +const mocks = vi.hoisted(() => ({ + getDueConfigsWithOrganization: + vi.fn<(nowIso: string) => Promise>(), + getKeywordCountsForConfigs: + vi.fn<(configIds: string[]) => Promise>>(), + claimDueConfig: vi.fn<(input: ClaimInput) => Promise>(), + beginRankCheckRun: + vi.fn< + (input: { + config: DueConfigRow; + keywordsTotal: number; + trigger: string; + }) => Promise + >(), + customerHasPaidPlan: vi.fn<(organizationId: string) => Promise>(), + isHostedServerAuthMode: vi.fn<() => Promise>(), +})); + +vi.mock("cloudflare:workers", () => ({ env: {} })); +vi.mock( + "@/server/features/rank-tracking/repositories/RankTrackingRepository", + () => ({ + RankTrackingRepository: { + getDueConfigsWithOrganization: mocks.getDueConfigsWithOrganization, + getKeywordCountsForConfigs: mocks.getKeywordCountsForConfigs, + claimDueConfig: mocks.claimDueConfig, + }, + }), +); +vi.mock("@/server/features/rank-tracking/services/rankCheckRunGuards", () => ({ + beginRankCheckRun: mocks.beginRankCheckRun, +})); +vi.mock("@/server/billing/subscription", () => ({ + customerHasPaidPlan: mocks.customerHasPaidPlan, +})); +vi.mock("@/server/lib/runtime-env", () => ({ + isHostedServerAuthMode: mocks.isHostedServerAuthMode, +})); + +// oxlint-disable-next-line typescript/no-unsafe-type-assertion -- test double for the workflow binding +const testEnv = { RANK_CHECK_WORKFLOW: {} } as unknown as Env; + +function dueConfig(overrides: Partial = {}): DueConfigRow { + return { + id: "config_1", + projectId: "project_1", + domain: "acme.com", + locationCode: 2840, + languageCode: "en", + locationName: null, + devices: "both" as const, + serpDepth: 20, + scheduleInterval: "daily" as const, + nextCheckAt: "2026-01-01T00:00:00.000Z", + organizationId: "org_1", + ...overrides, + }; +} + +async function runTick() { + const { runScheduledRankChecks } = await import("./scheduledRankChecks"); + await runScheduledRankChecks(testEnv); +} + +describe("runScheduledRankChecks", () => { + beforeEach(() => { + vi.resetModules(); + vi.resetAllMocks(); + mocks.isHostedServerAuthMode.mockResolvedValue(true); + mocks.customerHasPaidPlan.mockResolvedValue(true); + mocks.claimDueConfig.mockResolvedValue(true); + mocks.beginRankCheckRun.mockResolvedValue({ ok: true, runId: "run_1" }); + mocks.getKeywordCountsForConfigs.mockResolvedValue( + new Map([["config_1", 5]]), + ); + mocks.getDueConfigsWithOrganization.mockResolvedValue([dueConfig()]); + }); + + it("advances a free config with plan_required instead of starting a workflow", async () => { + mocks.customerHasPaidPlan.mockResolvedValue(false); + + await runTick(); + + expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1); + expect(mocks.claimDueConfig).toHaveBeenCalledWith( + expect.objectContaining({ + configId: "config_1", + projectId: "project_1", + observedNextCheckAt: "2026-01-01T00:00:00.000Z", + lastSkipReason: "plan_required", + }), + ); + const [claim] = mocks.claimDueConfig.mock.calls[0]; + expect(new Date(claim.nextCheckAt).getTime()).toBeGreaterThan(Date.now()); + expect(mocks.beginRankCheckRun).not.toHaveBeenCalled(); + }); + + it("advances a zero-keyword config with no_keywords without consuming budget", async () => { + mocks.getDueConfigsWithOrganization.mockResolvedValue([ + dueConfig({ id: "config_empty" }), + dueConfig({ id: "config_2", nextCheckAt: "2026-01-02T00:00:00.000Z" }), + ]); + mocks.getKeywordCountsForConfigs.mockResolvedValue( + new Map([["config_2", 5]]), + ); + + await runTick(); + + expect(mocks.claimDueConfig).toHaveBeenCalledWith( + expect.objectContaining({ + configId: "config_empty", + lastSkipReason: "no_keywords", + }), + ); + // The empty config consumed no budget, so the paid one behind it still ran. + expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1); + expect(mocks.beginRankCheckRun).toHaveBeenCalledWith( + expect.objectContaining({ keywordsTotal: 5, trigger: "scheduled" }), + ); + }); + + it("claims a paid config and starts its workflow", async () => { + const logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); + + await runTick(); + + expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1); + expect(mocks.claimDueConfig).toHaveBeenCalledWith( + expect.objectContaining({ + configId: "config_1", + observedNextCheckAt: "2026-01-01T00:00:00.000Z", + lastSkipReason: null, + }), + ); + expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1); + // 5 keywords × both devices = 10 task units. + expect(logSpy).toHaveBeenCalledWith( + expect.objectContaining({ + event: "rank_tracking_scheduler_summary", + candidates: 1, + started: 1, + unitsStarted: 10, + stoppedByBudget: false, + configErrors: 0, + }), + ); + }); + + it("stops admitting configs once the task-unit budget is spent", async () => { + // 400 keywords × 2 devices = 800 units — admitted anyway as the tick's + // first start (oversized exemption), then exhausts the 200-unit budget. + mocks.getDueConfigsWithOrganization.mockResolvedValue([ + dueConfig({ id: "config_big" }), + dueConfig({ id: "config_next", nextCheckAt: "2026-01-02T00:00:00.000Z" }), + ]); + mocks.getKeywordCountsForConfigs.mockResolvedValue( + new Map([ + ["config_big", 400], + ["config_next", 5], + ]), + ); + + const logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); + + await runTick(); + + expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1); + expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1); + expect(mocks.customerHasPaidPlan).toHaveBeenCalledTimes(1); + expect(logSpy).toHaveBeenCalledWith( + expect.objectContaining({ stoppedByBudget: true, unitsStarted: 800 }), + ); + }); + + it("skips a config whose claim lost to a concurrent edit", async () => { + mocks.getDueConfigsWithOrganization.mockResolvedValue([ + dueConfig({ id: "config_edited" }), + dueConfig({ id: "config_2", nextCheckAt: "2026-01-02T00:00:00.000Z" }), + ]); + mocks.getKeywordCountsForConfigs.mockResolvedValue( + new Map([ + ["config_edited", 5], + ["config_2", 5], + ]), + ); + mocks.claimDueConfig.mockResolvedValueOnce(false); + const logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); + + await runTick(); + + expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1); + expect(mocks.beginRankCheckRun.mock.calls[0][0].config.id).toBe("config_2"); + expect(logSpy).toHaveBeenCalledWith( + expect.objectContaining({ concurrentChangeSkips: 1, started: 1 }), + ); + }); + + it("contains a per-config failure and still processes the rest of the tick", async () => { + mocks.getDueConfigsWithOrganization.mockResolvedValue([ + dueConfig({ id: "config_bad" }), + dueConfig({ id: "config_ok", nextCheckAt: "2026-01-02T00:00:00.000Z" }), + ]); + mocks.getKeywordCountsForConfigs.mockResolvedValue( + new Map([ + ["config_bad", 5], + ["config_ok", 5], + ]), + ); + mocks.claimDueConfig.mockRejectedValueOnce(new Error("db blip")); + // A tick with errors logs its summary at error level. + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); + + await runTick(); + + expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1); + expect(mocks.beginRankCheckRun.mock.calls[0][0].config.id).toBe( + "config_ok", + ); + expect(errorSpy).toHaveBeenCalledWith( + expect.objectContaining({ configErrors: 1, started: 1 }), + ); + }); + + it("defers a config whose units would overflow the remaining budget", async () => { + // First config fits (10 units); the second would overflow (800), so it is + // deferred rather than admitted just because budget remains. + mocks.getDueConfigsWithOrganization.mockResolvedValue([ + dueConfig({ id: "config_small" }), + dueConfig({ id: "config_big", nextCheckAt: "2026-01-02T00:00:00.000Z" }), + ]); + mocks.getKeywordCountsForConfigs.mockResolvedValue( + new Map([ + ["config_small", 5], + ["config_big", 400], + ]), + ); + const logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); + + await runTick(); + + expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1); + expect(mocks.beginRankCheckRun.mock.calls[0][0].config.id).toBe( + "config_small", + ); + expect(logSpy).toHaveBeenCalledWith( + expect.objectContaining({ stoppedByBudget: true, unitsStarted: 10 }), + ); + }); + + it("restores the original due time when a run is already active", async () => { + mocks.beginRankCheckRun.mockResolvedValue({ + ok: false, + reason: "already_running", + blockingRunId: "run_blocking", + }); + const logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); + + await runTick(); + + expect(mocks.claimDueConfig).toHaveBeenCalledTimes(2); + const [advance] = mocks.claimDueConfig.mock.calls[0]; + const [restore] = mocks.claimDueConfig.mock.calls[1]; + expect(restore).toEqual({ + configId: "config_1", + projectId: "project_1", + observedNextCheckAt: advance.nextCheckAt, + nextCheckAt: "2026-01-01T00:00:00.000Z", + }); + // Blocked configs leave no durable row state, so the summary names them. + expect(logSpy).toHaveBeenCalledWith( + expect.objectContaining({ + alreadyRunning: 1, + alreadyRunningConfigIds: ["config_1"], + }), + ); + }); + + it("leaves a config due when its plan check throws, and still processes other orgs", async () => { + mocks.getDueConfigsWithOrganization.mockResolvedValue([ + dueConfig({ id: "config_a1", organizationId: "org_a" }), + dueConfig({ + id: "config_a2", + organizationId: "org_a", + nextCheckAt: "2026-01-02T00:00:00.000Z", + }), + dueConfig({ + id: "config_b", + organizationId: "org_b", + nextCheckAt: "2026-01-03T00:00:00.000Z", + }), + ]); + mocks.getKeywordCountsForConfigs.mockResolvedValue( + new Map([ + ["config_a1", 5], + ["config_a2", 5], + ["config_b", 5], + ]), + ); + mocks.customerHasPaidPlan.mockImplementation(async (orgId: string) => { + if (orgId === "org_a") throw new Error("autumn down"); + return true; + }); + vi.spyOn(console, "error").mockImplementation(() => {}); + + await runTick(); + + // One Autumn call per org per tick — org_a's rejection is memoized. + expect(mocks.customerHasPaidPlan).toHaveBeenCalledTimes(2); + expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1); + expect(mocks.claimDueConfig).toHaveBeenCalledWith( + expect.objectContaining({ configId: "config_b" }), + ); + expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1); + }); + + it("makes no billing calls in self-hosted mode", async () => { + mocks.isHostedServerAuthMode.mockResolvedValue(false); + + await runTick(); + + expect(mocks.customerHasPaidPlan).not.toHaveBeenCalled(); + expect(mocks.beginRankCheckRun).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/server/features/rank-tracking/services/scheduledRankChecks.ts b/src/server/features/rank-tracking/services/scheduledRankChecks.ts index 8c64b9c..7f6f6be 100644 --- a/src/server/features/rank-tracking/services/scheduledRankChecks.ts +++ b/src/server/features/rank-tracking/services/scheduledRankChecks.ts @@ -4,90 +4,231 @@ import { customerHasPaidPlan } from "@/server/billing/subscription"; import { isHostedServerAuthMode } from "@/server/lib/runtime-env"; import { computeNextCheckAt, + devicesCount, isScheduledRankTrackingInterval, } from "@/shared/rank-tracking"; +// Work admitted per tick, in task units (keywords × devices). Admission +// control, not a hard rate limit: the first start of a tick is always +// admitted, so an oversized config can never starve. 200 units on the +// 5-minute cron ≈ 57,600 units/day, ~9× current steady-state demand. +const SCHEDULED_TASK_UNIT_BUDGET = 200; + +// Cap on the per-tick list of configs blocked by an active run. Blocked +// configs leave no durable trace on their row, so the summary names them. +const ALREADY_RUNNING_IDS_CAP = 20; + // Cron body for the `scheduled` Worker handler: start a rank-check run for every // config that's due. Wrapped in `withPgClient` at the entrypoint (server.ts). export async function runScheduledRankChecks(env: Env) { const nowIso = new Date().toISOString(); const dueConfigs = await RankTrackingRepository.getDueConfigsWithOrganization(nowIso); - const isHosted = await isHostedServerAuthMode(); + const keywordCounts = await RankTrackingRepository.getKeywordCountsForConfigs( + dueConfigs.map((config) => config.id), + ); + + // Function-local so it lives exactly one tick: at module scope this would be + // cross-invocation global state in Workers, and a rejection would be cached + // forever. Within a tick, a rejection staying memoized is intentional — one + // Autumn call per org, and that org's configs simply stay due. + const paidPlanChecks = new Map>(); + const checkPaidPlan = (organizationId: string) => { + let check = paidPlanChecks.get(organizationId); + if (!check) { + check = customerHasPaidPlan(organizationId, { retryDenied: true }); + paidPlanChecks.set(organizationId, check); + } + return check; + }; + + let unitsStarted = 0; + let started = 0; + let stoppedByBudget = false; + let skippedFree = 0; + let skippedNoKeywords = 0; + let concurrentChangeSkips = 0; + let alreadyRunning = 0; + const alreadyRunningConfigIds: string[] = []; + let planCheckErrors = 0; + let workflowStartErrors = 0; + let configErrors = 0; for (const config of dueConfigs) { + // Per-config containment: one bad row (e.g. malformed next_check_at, which + // sorts first and would head every scan) or transient DB error must not + // starve the rest of the tick or suppress the summary log. try { - // Skip configs whose org doesn't have a paid plan - if (isHosted && !(await customerHasPaidPlan(config.organizationId))) { - continue; - } - - // Skip configs with no keywords before advancing the schedule - const kwCount = await RankTrackingRepository.getKeywordCountForConfig( - config.id, - ); - if (kwCount === 0) { - console.log( - `[cron] Skipping config ${config.id} (${config.domain}) — no keywords`, - ); - // Still advance schedule so this config doesn't stay due forever - const skipInterval = isScheduledRankTrackingInterval( - config.scheduleInterval, - ) - ? config.scheduleInterval - : null; - if (skipInterval) { - await RankTrackingRepository.updateConfig( - config.id, - config.projectId, - { - nextCheckAt: computeNextCheckAt(skipInterval, config.nextCheckAt), - }, - ); - } - continue; - } - - // Advance nextCheckAt immediately to prevent retry storms if the run fails const interval = isScheduledRankTrackingInterval(config.scheduleInterval) ? config.scheduleInterval : null; - if (interval) { - await RankTrackingRepository.updateConfig(config.id, config.projectId, { - nextCheckAt: computeNextCheckAt(interval, config.nextCheckAt), - }); + // Unreachable: the due query excludes manual configs and NULL next check + // times. Narrow rather than assert so a query change can't produce a run + // with no schedule anchor. + if (!interval || !config.nextCheckAt) continue; + + const kwCount = keywordCounts.get(config.id) ?? 0; + const taskUnits = kwCount * devicesCount(config.devices); + // Projected stop: admit only what fits the budget. The first start of a + // tick is exempt so an oversized config can never starve, and zero-unit + // rows (no keywords) always advance. + if ( + started > 0 && + unitsStarted + taskUnits > SCHEDULED_TASK_UNIT_BUDGET + ) { + stoppedByBudget = true; + break; } - const result = await beginRankCheckRun({ - workflow: env.RANK_CHECK_WORKFLOW, - config, - projectId: config.projectId, - billingCustomer: { - userId: "system", - userEmail: "system@openseo.so", - organizationId: config.organizationId, - projectId: config.projectId, - }, - keywordsTotal: kwCount, - trigger: "scheduled", - workflowStartErrorMessage: "Failed to start scheduled workflow", - }); + const observedNextCheckAt = config.nextCheckAt; + const nextCheckAt = computeNextCheckAt(interval, observedNextCheckAt); - if (!result.ok) { - console.log( - `[cron] Skipping config ${config.id} (${config.domain}) — run already active`, + if (kwCount === 0) { + const claimed = await RankTrackingRepository.claimDueConfig({ + configId: config.id, + projectId: config.projectId, + observedNextCheckAt, + nextCheckAt, + lastSkipReason: "no_keywords", + }); + if (claimed) skippedNoKeywords++; + else concurrentChangeSkips++; + continue; + } + + // Self-hosted deployments treat every config as paid and make no Autumn + // calls at all. + let hasPaidPlan = true; + if (isHosted) { + try { + hasPaidPlan = await checkPaidPlan(config.organizationId); + } catch (err) { + // Never write nextCheckAt on an error: it is the schedule anchor, so + // an error write would permanently shift this config's slot and + // herd-sync configs after an outage. Leaving the row due is the retry. + console.error( + `[cron] Plan check failed for config ${config.id} (${config.domain}):`, + err, + ); + planCheckErrors++; + continue; + } + } + + if (!hasPaidPlan) { + const claimed = await RankTrackingRepository.claimDueConfig({ + configId: config.id, + projectId: config.projectId, + observedNextCheckAt, + nextCheckAt, + lastSkipReason: "plan_required", + }); + if (claimed) skippedFree++; + else concurrentChangeSkips++; + continue; + } + + // Claim the slot before starting. Clearing lastSkipReason here is what + // lets an upgraded org drop the "plan_required" badge — the workflow only + // writes null on a fully successful run. + const claimed = await RankTrackingRepository.claimDueConfig({ + configId: config.id, + projectId: config.projectId, + observedNextCheckAt, + nextCheckAt, + lastSkipReason: null, + }); + if (!claimed) { + concurrentChangeSkips++; + continue; + } + + let result; + try { + result = await beginRankCheckRun({ + workflow: env.RANK_CHECK_WORKFLOW, + config, + projectId: config.projectId, + billingCustomer: { + userId: "system", + userEmail: "system@openseo.so", + organizationId: config.organizationId, + projectId: config.projectId, + }, + keywordsTotal: kwCount, + trigger: "scheduled", + workflowStartErrorMessage: "Failed to start scheduled workflow", + }); + } catch (err) { + // Leave the schedule advanced: a systemic Workflows outage must not + // make hundreds of configs due again on the next tick. + workflowStartErrors++; + console.error( + `[cron] Failed to start scheduled rank check for config ${config.id} (${config.domain}):`, + err, ); - } else { + continue; + } + + if (result.ok) { + unitsStarted += taskUnits; + started++; + continue; + } + + alreadyRunning++; + if (alreadyRunningConfigIds.length < ALREADY_RUNNING_IDS_CAP) { + alreadyRunningConfigIds.push(config.id); + } + // Nothing was started, so give the slot back and retry next tick once the + // blocking run clears. A manual edit landing in between wins the CAS. + const restored = await RankTrackingRepository.claimDueConfig({ + configId: config.id, + projectId: config.projectId, + observedNextCheckAt: nextCheckAt, + nextCheckAt: observedNextCheckAt, + }); + if (!restored) { console.log( - `[cron] Started scheduled rank check ${result.runId} for config ${config.id} (${config.domain})`, + `[cron] Could not restore schedule for config ${config.id} (${config.domain}) — changed concurrently`, ); } } catch (err) { + configErrors++; console.error( `[cron] Error processing config ${config.id} (${config.domain}):`, err, ); } } + + // Oldest by the due query's next_check_at ASC ordering. + const oldestDue = dueConfigs[0]?.nextCheckAt; + // Object argument (not an interpolated string) so Workers Logs indexes the + // fields. Error level when anything failed, so ticks that need attention + // surface in error-filtered views. + const logSummary = + planCheckErrors + workflowStartErrors + configErrors > 0 + ? console.error + : console.log; + logSummary({ + event: "rank_tracking_scheduler_summary", + candidates: dueConfigs.length, + started, + unitsStarted, + budget: SCHEDULED_TASK_UNIT_BUDGET, + stoppedByBudget, + skippedFree, + skippedNoKeywords, + concurrentChangeSkips, + alreadyRunning, + alreadyRunningConfigIds, + planCheckErrors, + workflowStartErrors, + configErrors, + oldestDueAgeMs: oldestDue + ? Date.now() - new Date(oldestDue).getTime() + : null, + }); } diff --git a/src/shared/rank-tracking.test.ts b/src/shared/rank-tracking.test.ts index b90b8c3..1884e2b 100644 --- a/src/shared/rank-tracking.test.ts +++ b/src/shared/rank-tracking.test.ts @@ -91,4 +91,23 @@ describe("rank tracking schedules", () => { "2026-03-31T05:30:00.000Z", ); }); + + it("preserves the time-of-day anchor for heavily overdue daily schedules", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-03-10T12:00:00.000Z")); + + expect(computeNextCheckAt("daily", "2026-01-31T05:30:00.000Z")).toBe( + "2026-03-11T05:30:00.000Z", + ); + }); + + it("preserves the weekday and time anchor for heavily overdue weekly schedules", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-03-10T12:00:00.000Z")); + + // 2026-01-31 is a Saturday; every advance lands on a Saturday. + expect(computeNextCheckAt("weekly", "2026-01-31T05:30:00.000Z")).toBe( + "2026-03-14T05:30:00.000Z", + ); + }); }); diff --git a/src/shared/rank-tracking.ts b/src/shared/rank-tracking.ts index b673d13..03fb457 100644 --- a/src/shared/rank-tracking.ts +++ b/src/shared/rank-tracking.ts @@ -111,6 +111,13 @@ type ScheduledRankTrackingInterval = Exclude< "manual" >; +// Values written to rank_tracking_configs.last_skip_reason (free-form text in +// the schema; this union keeps writers and UI comparisons in sync). +export type RankTrackingSkipReason = + | "plan_required" + | "no_keywords" + | "insufficient_credits"; + export function estimateScheduledRankCheckCredits( keywordCount: number, devices: RankTrackingConfig["devices"], diff --git a/wrangler.jsonc b/wrangler.jsonc index 16013dd..162e07f 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -72,7 +72,7 @@ }, ], "triggers": { - "crons": ["*/15 * * * *"], + "crons": ["*/5 * * * *"], }, // This config serves local dev and Docker self-host only. All Cloudflare // deployments — previews, prod, self-host — go through Alchemy