From f8d5ffd285a8140ac7dfa951ab331bbfdf715bd6 Mon Sep 17 00:00:00 2001 From: Ben Senescu <44480372+bensenescu@users.noreply.github.com> Date: Sat, 13 Jun 2026 11:20:42 -0400 Subject: [PATCH] Scheduled rank checks run through DataForSEO's task queue (#268) --- .claude/skills/merge-ready/SKILL.md | 54 +++ .../rank-tracking/CheckConfirmModal.tsx | 3 +- .../rank-tracking/RankTrackingConfigModal.tsx | 3 + .../services/RankTrackingService.ts | 2 + src/server/lib/dataforseo/client.test.ts | 1 + src/server/lib/dataforseo/client.ts | 5 + src/server/lib/dataforseo/envelope.ts | 14 +- src/server/lib/dataforseo/index.ts | 4 + src/server/lib/dataforseo/serp.test.ts | 177 +++++++++ src/server/lib/dataforseo/serp.ts | 230 +++++++++++- src/server/workflows/RankCheckWorkflow.ts | 48 ++- src/server/workflows/rankCheckPaths.ts | 354 ++++++++++++++++-- src/shared/rank-tracking.ts | 25 +- 13 files changed, 847 insertions(+), 73 deletions(-) create mode 100644 .claude/skills/merge-ready/SKILL.md create mode 100644 src/server/lib/dataforseo/serp.test.ts diff --git a/.claude/skills/merge-ready/SKILL.md b/.claude/skills/merge-ready/SKILL.md new file mode 100644 index 0000000..c258569 --- /dev/null +++ b/.claude/skills/merge-ready/SKILL.md @@ -0,0 +1,54 @@ +--- +name: merge-ready +description: Take a branch from "code exists (or is about to)" to "ready for Ben's final review" — multi-axis subagent review with verified findings, fixes, ci:check, checkpoint commits, and an updated PR. Use whenever the user says a feature/fix/branch should be "merge ready", asks to get changes ready for review, or appends this to a build request ("build X and make it merge-ready"). +--- + +# Merge ready + +Drive the current work to the point where the only remaining step is Ben's own review and merge. The deliverable is a pushed branch with a clean `pnpm ci:check`, checkpoint commits along the way, and an open PR with a high-level description plus review instructions. + +**Never merge the PR. Ben always reviews last.** + +## 0. Figure out the starting point + +This skill composes with feature work — it is not only a review pass: + +- **Invoked alongside a build request** ("build X, make it merge-ready"): implement the feature/fix first, committing as you go, then continue below. The review phases cover _all_ changes on the branch vs `origin/main`, not just the last edit. +- **Invoked on existing work** ("make this branch merge-ready"): start directly at step 1. The scope is `git diff origin/main...HEAD` plus anything uncommitted. + +## 1. Sync with main + +- `git fetch origin main`. If the branch is behind, merge `origin/main` in and resolve conflicts (favor main's version for code this branch didn't intentionally change). +- **Checkpoint:** commit the merge before starting review, so conflict resolution is auditable separately from review fixes. + +## 2. Multi-axis subagent review + +Spawn independent review subagents **in parallel**, one per axis, each given the branch diff scope (`git diff origin/main...HEAD`) and repo access: + +1. **Unnecessary complexity** — thin wrappers, needless indirection, single-use abstractions, defensive guards for impossible states, dead config. This codebase deliberately stays simple. +2. **Security** — authz on new endpoints (org/project scoping), SSRF, injection, secrets handling, anything user-input-shaped reaching D1/R2/external APIs. +3. **Billing & metering** — ways a user could trigger DataForSEO/provider spend without being metered, charged-but-failed paths, retry/loop amplification, endpoints with unexpectedly high per-call user cost. Credits are billed via Autumn; uncounted spend is a revenue leak. +4. **Library & project idioms** — TanStack (Router/Query/Start) used idiomatically; patterns match how the rest of the codebase already does it (Result-pattern error handling at provider seams, db/schema conventions, existing component patterns). Flag novel patterns where an established one exists. +5. **Vibe-coded cruft** — leftover scaffolding, stale comments narrating the edit history, console.logs, TODO-without-owner, copy-pasted near-duplicates, files/exports nothing uses. + +Each reviewer returns findings with file:line, severity (`blocker` / `should-fix` / `nitpick`), and a one-line rationale. Tell reviewers explicitly: this is an early-stage product — do not chase theoretical edge cases; mark anything debatable as `nitpick`. + +## 3. Verify findings — never blindly accept + +For each `blocker` and `should-fix` finding, spawn verification subagents (in parallel) that adversarially check the finding against the actual code and verdict **APPLY / APPLY-MODIFIED / REJECT** with reasoning. Drop rejected findings. Nitpicks don't need verification — they're reported, not necessarily fixed. + +## 4. Fix, check, loop + +- Apply verified `blocker`/`should-fix` fixes. Apply nitpicks only when trivial and clearly right; otherwise list them in the PR for Ben to judge. +- **Checkpoint:** commit fixes in logical groups (e.g. one commit per axis or per concern) so the fix history is reviewable on its own. +- Run `pnpm ci:check` (prettier, knip, tsc, oxlint). Fix failures and re-run until clean. If a fix was substantial (not formatting/lint), run a quick re-review of just that change. +- Loop until ci:check passes and no verified findings remain unaddressed. + +## 5. Push and open/update the PR + +- Push the branch. Open a PR against `main` if one doesn't exist; otherwise update the existing PR's description. +- PR description requirements: + - **High-level** — what changed and why, written for a human skimming. No file paths, no per-file changelog. + - **How to review** — a short ordered guide: what to look at first, what the risky/judgment-call areas are, what was deliberately left out of scope. + - **Review notes** — unfixed nitpicks and any REJECT verdicts worth a second opinion, clearly labeled as such. +- Report back to Ben: PR link, one-paragraph summary, and anything that still needs his judgment. Do not merge. diff --git a/src/client/features/rank-tracking/CheckConfirmModal.tsx b/src/client/features/rank-tracking/CheckConfirmModal.tsx index 93a8f6e..aefc1e3 100644 --- a/src/client/features/rank-tracking/CheckConfirmModal.tsx +++ b/src/client/features/rank-tracking/CheckConfirmModal.tsx @@ -27,6 +27,7 @@ export function CheckConfirmModal({ keywordCount, devices, serpDepth, + "live", ); const dc = devicesCount(devices); const totalChecks = keywordCount * dc; @@ -66,7 +67,7 @@ export function CheckConfirmModal({

-

${costUsd.toFixed(2)}

+

~${costUsd.toFixed(2)}

{isPending && }
diff --git a/src/client/features/rank-tracking/RankTrackingConfigModal.tsx b/src/client/features/rank-tracking/RankTrackingConfigModal.tsx index fc9338b..3ba9149 100644 --- a/src/client/features/rank-tracking/RankTrackingConfigModal.tsx +++ b/src/client/features/rank-tracking/RankTrackingConfigModal.tsx @@ -289,10 +289,13 @@ export function RankTrackingConfigModal({ {(() => { + // Scheduled checks run through the cheaper task queue; manual + // configs only ever pay the live price. const { costUsd: costPerKeyword } = estimateRankCheckCredits( 1, devices, serpDepth, + schedule === "manual" ? "live" : "queued", ); const checksPerMonth = schedule === "daily" ? 30 : 4; return ( diff --git a/src/server/features/rank-tracking/services/RankTrackingService.ts b/src/server/features/rank-tracking/services/RankTrackingService.ts index 595a925..c988c28 100644 --- a/src/server/features/rank-tracking/services/RankTrackingService.ts +++ b/src/server/features/rank-tracking/services/RankTrackingService.ts @@ -330,10 +330,12 @@ async function estimateCost(configId: string, projectId: string) { const config = await getValidatedConfig(configId, projectId); const keywordCount = await RankTrackingRepository.getKeywordCountForConfig(configId); + // Estimates the cost of a manual "check now", which always runs live. const { costUsd, costCredits } = estimateRankCheckCredits( keywordCount, config.devices, config.serpDepth, + "live", ); return { costUsd, diff --git a/src/server/lib/dataforseo/client.test.ts b/src/server/lib/dataforseo/client.test.ts index d6c373e..927eb4a 100644 --- a/src/server/lib/dataforseo/client.test.ts +++ b/src/server/lib/dataforseo/client.test.ts @@ -60,6 +60,7 @@ vi.mock("@/server/lib/dataforseo/labs", () => ({ vi.mock("@/server/lib/dataforseo/serp", () => ({ fetchLiveSerp: vi.fn(), fetchRankCheckSerp: vi.fn(), + postRankCheckTasks: vi.fn(), fetchLocalSerp: vi.fn(), })); vi.mock("@/server/lib/dataforseo/business", () => ({ diff --git a/src/server/lib/dataforseo/client.ts b/src/server/lib/dataforseo/client.ts index f00967e..054fd83 100644 --- a/src/server/lib/dataforseo/client.ts +++ b/src/server/lib/dataforseo/client.ts @@ -41,6 +41,7 @@ import { fetchLiveSerp, fetchLocalSerp, fetchRankCheckSerp, + postRankCheckTasks, } from "@/server/lib/dataforseo/serp"; import { fetchLighthouseResult } from "@/server/lib/dataforseo/lighthouse"; import { @@ -117,6 +118,10 @@ export function createDataforseoClient(customer: BillingCustomerContext) { serp: { live: meter(customer, fetchLiveSerp), rankCheck: meter(customer, fetchRankCheckSerp, "rank_tracking"), + // Posts up to 100 queued rank check tasks; one metered charge covers the + // whole batch (DataForSEO bills task_post at post time, collection is + // free). + rankCheckTaskPost: meter(customer, postRankCheckTasks, "rank_tracking"), local: meter(customer, fetchLocalSerp, "local_seo"), }, labs: { diff --git a/src/server/lib/dataforseo/envelope.ts b/src/server/lib/dataforseo/envelope.ts index e572736..14be62d 100644 --- a/src/server/lib/dataforseo/envelope.ts +++ b/src/server/lib/dataforseo/envelope.ts @@ -82,6 +82,14 @@ export function buildTaskBilling( return billing; } +/** DataForSEO's "No Search Results" (40501) — a successful empty result, not a failure. */ +export function isNoResultsTask(task: DataforseoTaskLike): boolean { + return ( + task.status_code === 40501 || + (task.status_message?.toLowerCase().includes("no search results") ?? false) + ); +} + type AssertOkOptions = { /** Maps a recognised access / billing failure to a product error. */ classify?: DataforseoErrorClassifier; @@ -124,11 +132,7 @@ export function assertOk( } if (task.status_code !== 20000) { - const isNoResults = - task.status_code === 40501 || - (task.status_message?.toLowerCase().includes("no search results") ?? - false); - if (treatNoResultsAsEmpty && isNoResults) return task; + if (treatNoResultsAsEmpty && isNoResultsTask(task)) return task; const message = task.status_message || "DataForSEO task failed"; const path = classifyPath ?? (task.path ? `/${task.path.join("/")}` : ""); diff --git a/src/server/lib/dataforseo/index.ts b/src/server/lib/dataforseo/index.ts index d7a73ff..29e3ace 100644 --- a/src/server/lib/dataforseo/index.ts +++ b/src/server/lib/dataforseo/index.ts @@ -18,8 +18,12 @@ export { } from "@/server/lib/dataforseo/google-ads"; export { + fetchRankCheckTaskResult, + MAX_TASKS_PER_POST, type SerpLiveItem, type RankCheckResult, + type RankCheckTaskInput, + type PostedRankCheckTask, } from "@/server/lib/dataforseo/serp"; export { diff --git a/src/server/lib/dataforseo/serp.test.ts b/src/server/lib/dataforseo/serp.test.ts new file mode 100644 index 0000000..75130f3 --- /dev/null +++ b/src/server/lib/dataforseo/serp.test.ts @@ -0,0 +1,177 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/server/lib/runtime-env", () => ({ + getRequiredEnvValue: vi.fn(async () => "test-api-key"), +})); + +import { + fetchRankCheckTaskResult, + postRankCheckTasks, +} from "@/server/lib/dataforseo/serp"; + +function parseDataforseoRequestBody(init: RequestInit | undefined): unknown { + const body = init?.body; + if (typeof body !== "string") { + throw new Error("Expected DataForSEO request body to be a string"); + } + return JSON.parse(body) as unknown; +} + +describe("rank check task queue", () => { + beforeEach(() => { + vi.restoreAllMocks(); + }); + + it("posts queued tasks, maps ids by tag, and sums cost over all entries", async () => { + const fetchMock = vi.fn().mockResolvedValue( + Response.json({ + status_code: 20000, + tasks: [ + { + id: "task-a", + status_code: 20100, + cost: 0.0006, + data: { tag: "kw-1:desktop" }, + }, + { + id: "task-b", + status_code: 20100, + cost: 0.0006, + data: { tag: "kw-1:mobile" }, + }, + { + id: "task-c", + status_code: 40006, + status_message: "Task Limit Exceeded", + cost: 0.0006, + data: { tag: "kw-2:desktop" }, + }, + ], + }), + ); + vi.stubGlobal("fetch", fetchMock); + + const result = await postRankCheckTasks({ + tasks: [ + { keyword: "alpha", keywordId: "kw-1", device: "desktop" }, + { keyword: "alpha", keywordId: "kw-1", device: "mobile" }, + { keyword: "beta", keywordId: "kw-2", device: "desktop" }, + ], + locationCode: 2840, + languageCode: "en", + depth: 20, + targetDomain: "example.com", + }); + + expect( + fetchMock.mock.calls.map(([url]) => + typeof url === "string" || url instanceof URL + ? url.toString() + : url.url, + ), + ).toEqual(["https://api.dataforseo.com/v3/serp/google/organic/task_post"]); + + // Every posted task asks DataForSEO to stop crawling at the target's + // organic listing — that is what cuts the actual crawl cost for ranking + // domains without false "not ranking" stops on sitelinks/PAA mentions. + const stopCrawl = { + stop_crawl_on_match: [ + { match_value: "example.com", match_type: "with_subdomains" }, + ], + find_targets_in: ["organic"], + }; + expect( + parseDataforseoRequestBody(fetchMock.mock.calls[0]?.[1]), + ).toMatchObject([stopCrawl, stopCrawl, stopCrawl]); + expect(result.data).toEqual([ + { + keyword: "alpha", + keywordId: "kw-1", + device: "desktop", + taskId: "task-a", + }, + { + keyword: "alpha", + keywordId: "kw-1", + device: "mobile", + taskId: "task-b", + }, + ]); + // The rejected entry's cost is still metered: a charge is a charge. + expect(result.billing.costUsd).toBeCloseTo(0.0018, 10); + expect(result.billing.path).toEqual([ + "v3", + "serp", + "google", + "organic", + "task_post", + ]); + }); + + it("reports a queued task still in progress as pending", async () => { + const fetchMock = vi.fn().mockResolvedValue( + Response.json({ + status_code: 20000, + tasks: [{ id: "task-a", status_code: 40602 }], + }), + ); + vi.stubGlobal("fetch", fetchMock); + + const outcome = await fetchRankCheckTaskResult({ + taskId: "task-a", + keywordId: "kw-1", + keyword: "alpha", + targetDomain: "example.com", + }); + + expect(outcome).toEqual({ status: "pending" }); + }); + + it("parses a completed queued task into a rank check result", async () => { + const fetchMock = vi.fn().mockResolvedValue( + Response.json({ + status_code: 20000, + tasks: [ + { + id: "task-a", + status_code: 20000, + cost: 0, + path: ["v3", "serp", "google", "organic", "task_get", "advanced"], + result: [ + { + items: [ + { + type: "organic", + rank_group: 3, + rank_absolute: 4, + domain: "www.example.com", + url: "https://www.example.com/page", + }, + ], + }, + ], + }, + ], + }), + ); + vi.stubGlobal("fetch", fetchMock); + + const outcome = await fetchRankCheckTaskResult({ + taskId: "task-a", + keywordId: "kw-1", + keyword: "alpha", + targetDomain: "example.com", + }); + + expect(outcome).toEqual({ + status: "completed", + result: { + keywordId: "kw-1", + keyword: "alpha", + position: 4, + url: "https://www.example.com/page", + serpFeatures: ["organic"], + }, + }); + }); +}); diff --git a/src/server/lib/dataforseo/serp.ts b/src/server/lib/dataforseo/serp.ts index 3f18c0b..450ab81 100644 --- a/src/server/lib/dataforseo/serp.ts +++ b/src/server/lib/dataforseo/serp.ts @@ -1,16 +1,45 @@ import { z } from "zod"; import { + SerpApiStopCrawlOnMatchInfo, SerpGoogleLocalFinderLiveAdvancedRequestInfo, SerpGoogleMapsLiveAdvancedRequestInfo, SerpGoogleOrganicLiveAdvancedRequestInfo, + SerpGoogleOrganicTaskPostRequestInfo, } from "dataforseo-client"; import { serpApi } from "@/server/lib/dataforseo/core"; import { assertOk, buildTaskBilling, + isNoResultsTask, parseTaskItems, type DataforseoApiResponse, } from "@/server/lib/dataforseo/envelope"; +import { AppError } from "@/server/lib/errors"; + +/** DataForSEO bills SERPs in pages of 10; depth outside 10-100 is rejected. */ +function clampSerpDepth(depth: number): number { + return Math.min(100, Math.max(10, depth)); +} + +/** + * Stop crawling SERP pages once the target domain is found — DataForSEO only + * bills the pages crawled, so a page-1 ranking at depth 20 costs one page + * instead of two. Matching is restricted to organic results and uses + * with_subdomains, mirroring buildRankCheckResult exactly: without + * find_targets_in, a sitelink or PAA mention could stop the crawl before the + * domain's organic listing and record a false "not ranking". + */ +function stopCrawlOnTarget(targetDomain: string) { + return { + stop_crawl_on_match: [ + new SerpApiStopCrawlOnMatchInfo({ + match_value: targetDomain, + match_type: "with_subdomains", + }), + ], + find_targets_in: ["organic"], + }; +} // Kept as a hand-written schema: the SDK's BaseSerpApiElementItem type omits // etv / estimated_paid_traffic_cost / backlinks_info / rank_changes, which we @@ -85,6 +114,28 @@ export interface RankCheckResult { serpFeatures: string[]; } +function buildRankCheckResult( + input: { keywordId: string; keyword: string; targetDomain: string }, + items: SerpLiveItem[], +): RankCheckResult { + const target = input.targetDomain.toLowerCase(); + const organicMatch = items.find((item) => { + if (item.type !== "organic" || item.domain == null) return false; + const domain = item.domain.toLowerCase(); + return domain === target || domain.endsWith(`.${target}`); + }); + + return { + keywordId: input.keywordId, + keyword: input.keyword, + position: organicMatch + ? (organicMatch.rank_absolute ?? organicMatch.rank_group ?? null) + : null, + url: organicMatch?.url ?? null, + serpFeatures: [...new Set(items.map((item) => item.type).filter(Boolean))], + }; +} + export async function fetchRankCheckSerp(input: { keyword: string; keywordId: string; @@ -94,7 +145,7 @@ export async function fetchRankCheckSerp(input: { targetDomain: string; depth: number; }): Promise> { - const depth = Math.min(100, Math.max(10, input.depth)); + const depth = clampSerpDepth(input.depth); const response = await serpApi().googleOrganicLiveAdvanced([ new SerpGoogleOrganicLiveAdvancedRequestInfo({ keyword: input.keyword, @@ -103,6 +154,7 @@ export async function fetchRankCheckSerp(input: { device: input.device, os: input.device === "desktop" ? "windows" : "android", depth, + ...stopCrawlOnTarget(input.targetDomain), }), ]); @@ -115,29 +167,169 @@ export async function fetchRankCheckSerp(input: { serpSnapshotItemSchema, ); - const target = input.targetDomain.toLowerCase(); - const organicMatch = items.find((item) => { - if (item.type !== "organic" || item.domain == null) return false; - const domain = item.domain.toLowerCase(); - return domain === target || domain.endsWith(`.${target}`); - }); - return { - data: { - keywordId: input.keywordId, - keyword: input.keyword, - position: organicMatch - ? (organicMatch.rank_absolute ?? organicMatch.rank_group ?? null) - : null, - url: organicMatch?.url ?? null, - serpFeatures: [ - ...new Set(items.map((item) => item.type).filter(Boolean)), - ], - }, + data: buildRankCheckResult(input, items), billing: buildTaskBilling(task), }; } +// --------------------------------------------------------------------------- +// Task-queue rank checks (scheduled runs). DataForSEO's standard queue costs +// ~30% of the live endpoint; tasks complete in ~5 minutes on average. The flow +// is task_post (charged) -> poll task_get (free) -> live fallback for +// stragglers, orchestrated by the rank check workflow. +// --------------------------------------------------------------------------- + +/** Max tasks DataForSEO accepts in a single task_post request. */ +export const MAX_TASKS_PER_POST = 100; + +export interface RankCheckTaskInput { + keyword: string; + keywordId: string; + device: "desktop" | "mobile"; +} + +export interface PostedRankCheckTask extends RankCheckTaskInput { + taskId: string; +} + +export async function postRankCheckTasks(input: { + tasks: RankCheckTaskInput[]; + locationCode: number; + languageCode: string; + depth: number; + targetDomain: string; +}): Promise> { + if (input.tasks.length === 0 || input.tasks.length > MAX_TASKS_PER_POST) { + throw new AppError( + "INTERNAL_ERROR", + `task_post accepts 1-${MAX_TASKS_PER_POST} tasks, got ${input.tasks.length}`, + ); + } + const depth = clampSerpDepth(input.depth); + const response = await serpApi().googleOrganicTaskPost( + input.tasks.map( + (task) => + new SerpGoogleOrganicTaskPostRequestInfo({ + keyword: task.keyword, + location_code: input.locationCode, + language_code: input.languageCode, + device: task.device, + os: task.device === "desktop" ? "windows" : "android", + depth, + // Queued tasks are billed provisionally at full depth at post time; + // task_get later reports the reduced actual cost when the crawl + // stopped early. We meter customers on the post-time amount — + // collection-time metering is a possible future optimization. + ...stopCrawlOnTarget(input.targetDomain), + // Echoed back on the response entry and task_get; used to map a + // DataForSEO task id back to our keyword without relying on order. + tag: `${task.keywordId}:${task.device}`, + }), + ), + ); + + if (!response || response.status_code !== 20000) { + throw new AppError( + "INTERNAL_ERROR", + response?.status_message || "DataForSEO task_post failed", + ); + } + + // One response entry per submitted task; accepted entries have status 20100 + // "Task Created" and their own cost (charged at post time). Cost is summed + // over every entry — accepted or not — so anything DataForSEO charged is + // metered. Rejected entries get no posted task; the workflow falls back to + // the live endpoint for any keyword/device pair missing from the result. + const byTag = new Map( + input.tasks.map((task) => [`${task.keywordId}:${task.device}`, task]), + ); + const posted: PostedRankCheckTask[] = []; + let costUsd = 0; + for (const entry of response.tasks ?? []) { + costUsd += entry.cost ?? 0; + const tag: unknown = entry.data?.tag; + const task = typeof tag === "string" ? byTag.get(tag) : undefined; + if (entry.status_code !== 20100 || !entry.id || !task) { + console.warn( + `dataforseo.task_post.rejected-entry (${entry.status_code}): ${entry.status_message}`, + ); + continue; + } + posted.push({ ...task, taskId: entry.id }); + } + + return { + data: posted, + billing: { + path: ["v3", "serp", "google", "organic", "task_post"], + costUsd, + }, + }; +} + +type RankCheckTaskOutcome = + | { status: "pending" } + | { status: "failed"; message: string } + | { status: "completed"; result: RankCheckResult }; + +// Task lifecycle codes meaning "not done yet": Task Created / Task Handed / +// Task In Queue. +const TASK_IN_PROGRESS_STATUS_CODES = new Set([20100, 40601, 40602]); + +/** + * Collect one queued task's result. Deliberately not metered and not wrapped + * in the billing envelope: collection is free (the task was charged at + * task_post), and the task_get response carries the task's settled cost + * (reduced when stop_crawl_on_match ended the crawl early) — running it + * through the metering seam would charge the customer twice. + */ +export async function fetchRankCheckTaskResult(input: { + taskId: string; + keywordId: string; + keyword: string; + targetDomain: string; +}): Promise { + const response = await serpApi().googleOrganicTaskGetAdvanced(input.taskId); + const task = response?.tasks?.[0]; + if (!response || response.status_code !== 20000 || !task) { + throw new AppError( + "INTERNAL_ERROR", + response?.status_message || "DataForSEO task_get failed", + ); + } + + if ( + task.status_code !== undefined && + TASK_IN_PROGRESS_STATUS_CODES.has(task.status_code) + ) { + return { status: "pending" }; + } + + if (task.status_code !== 20000) { + // "No Search Results" is valid for obscure/new keywords — same treatment + // as the live path's treatNoResultsAsEmpty. + if (!isNoResultsTask(task)) { + return { + status: "failed", + message: + task.status_message || `DataForSEO task failed (${task.status_code})`, + }; + } + return { + status: "completed", + result: buildRankCheckResult(input, []), + }; + } + + const items = parseTaskItems( + "google-organic-task-get-advanced", + task, + serpSnapshotItemSchema, + ); + return { status: "completed", result: buildRankCheckResult(input, items) }; +} + export async function fetchLocalSerp(input: { keyword: string; locationCoordinate?: string; diff --git a/src/server/workflows/RankCheckWorkflow.ts b/src/server/workflows/RankCheckWorkflow.ts index 50ebeb3..681c289 100644 --- a/src/server/workflows/RankCheckWorkflow.ts +++ b/src/server/workflows/RankCheckWorkflow.ts @@ -7,7 +7,11 @@ import { NonRetryableError } from "cloudflare:workflows"; import type { BillingCustomerContext } from "@/server/billing/subscription"; import { RankTrackingRepository } from "@/server/features/rank-tracking/repositories/RankTrackingRepository"; import { failRunIfActive } from "@/server/features/rank-tracking/services/rankCheckRunGuards"; -import { runLiveCheck } from "@/server/workflows/rankCheckPaths"; +import { + runLiveCheck, + runQueuedCheck, + type QueuedCheckStats, +} from "@/server/workflows/rankCheckPaths"; import { createDataforseoClient } from "@/server/lib/dataforseo"; import { captureServerEvent } from "@/server/lib/posthog"; import { AppError } from "@/server/lib/errors"; @@ -44,6 +48,7 @@ async function prepareRankCheckKeywords(input: { billingCustomer: BillingCustomerContext; devices: RankCheckParams["devices"]; serpDepth: number; + trigger: RankCheckParams["trigger"]; keywordIds?: string[]; }) { // If stale-cleanup marked our run failed before we got here, bail out @@ -72,12 +77,15 @@ async function prepareRankCheckKeywords(input: { throw new AppError("INTERNAL_ERROR", "No keywords to track"); } - // Verify the user has enough credits for the full check before starting + // Verify the user has enough credits for the full check before starting. + // Scheduled checks go through the cheaper task queue, so estimate at queued + // pricing — a live-price estimate would skip checks the user can afford. if (await isHostedServerAuthMode()) { const { costCredits } = estimateRankCheckCredits( trackingKeywords.length, input.devices, input.serpDepth, + input.trigger === "scheduled" ? "queued" : "live", ); const [monthlyCheck, topupCheck] = await Promise.all([ autumn.check({ @@ -119,6 +127,7 @@ async function finalizeRankCheckRun(input: { billingCustomer: BillingCustomerContext; trigger: RankCheckParams["trigger"]; batchError: string | null; + queueStats: QueuedCheckStats | null; }) { // If stale-cleanup already marked our run failed, don't overwrite that // decision with a completed status — a replacement run may already be @@ -168,6 +177,19 @@ async function finalizeRankCheckRun(input: { lastSkipReason: null, }); + // One-line summary per run so fallback rates are visible in Workers Logs. + // Keys match the PostHog event properties for log/event correlation. + const queueSummary = input.queueStats + ? ` queue_tasks=${input.queueStats.queueTasks} queue_collected=${input.queueStats.queueCollected} fallback_tasks=${input.queueStats.fallbackTasks} fallback_checked=${input.queueStats.fallbackChecked}` + : ""; + // Error text can echo vendor/user content — keep it one line and bounded. + const errorSummary = errorMessage + ? ` error="${errorMessage.replace(/\s+/g, " ").slice(0, 200)}"` + : ""; + console.log( + `[rank-check] ${input.runId} completed org=${input.billingCustomer.organizationId} project=${input.projectId} trigger=${input.trigger} keywords=${keywordsChecked}/${keywordsTotal}${queueSummary}${errorSummary}`, + ); + await captureServerEvent({ distinctId: input.billingCustomer.userId, event: "rank_tracking:check_complete", @@ -177,6 +199,14 @@ async function finalizeRankCheckRun(input: { status: "completed", trigger: input.trigger, keywords_checked: keywordsChecked, + ...(input.queueStats + ? { + queue_tasks: input.queueStats.queueTasks, + queue_collected: input.queueStats.queueCollected, + fallback_tasks: input.queueStats.fallbackTasks, + fallback_checked: input.queueStats.fallbackChecked, + } + : {}), }, }); } @@ -267,6 +297,7 @@ export class RankCheckWorkflow extends WorkflowEntrypoint< billingCustomer, devices, serpDepth, + trigger, keywordIds, }), ); @@ -276,9 +307,10 @@ export class RankCheckWorkflow extends WorkflowEntrypoint< console.log(`[rank-check] ${runId} loaded ${keywords.length} keywords`); let batchError: string | null = null; + let queueStats: QueuedCheckStats | null = null; try { - await runLiveCheck(step, { + const checkContext = { client, keywords, devices, @@ -287,7 +319,14 @@ export class RankCheckWorkflow extends WorkflowEntrypoint< locationCode, languageCode, runId, - }); + }; + // Scheduled checks use DataForSEO's task queue (~30% of live cost); + // manual checks stay on the live endpoint for instant results. + if (trigger === "scheduled") { + queueStats = await runQueuedCheck(step, checkContext); + } else { + await runLiveCheck(step, checkContext); + } } catch (error) { // Batch failure — snapshots for completed batches are already // persisted incrementally. Continue to finalization. @@ -303,6 +342,7 @@ export class RankCheckWorkflow extends WorkflowEntrypoint< billingCustomer, trigger, batchError, + queueStats, }), ); } catch (error) { diff --git a/src/server/workflows/rankCheckPaths.ts b/src/server/workflows/rankCheckPaths.ts index a12053d..a0b55ea 100644 --- a/src/server/workflows/rankCheckPaths.ts +++ b/src/server/workflows/rankCheckPaths.ts @@ -1,7 +1,15 @@ import type { WorkflowStep } from "cloudflare:workers"; import { RankTrackingRepository } from "@/server/features/rank-tracking/repositories/RankTrackingRepository"; -import type { createDataforseoClient } from "@/server/lib/dataforseo"; -import type { RankCheckResult } from "@/server/lib/dataforseo"; +import { + fetchRankCheckTaskResult, + MAX_TASKS_PER_POST, +} from "@/server/lib/dataforseo"; +import type { + createDataforseoClient, + PostedRankCheckTask, + RankCheckResult, + RankCheckTaskInput, +} from "@/server/lib/dataforseo"; import type { RankTrackingConfig } from "@/types/schemas/rank-tracking"; import { KEYWORDS_PER_BATCH } from "@/shared/rank-tracking"; @@ -42,6 +50,71 @@ interface CheckContext { runId: string; } +/** Expand keywords into one task input per keyword/device pair. */ +function expandToTaskInputs( + keywords: KeywordEntry[], + devices: RankTrackingConfig["devices"], +): RankCheckTaskInput[] { + const deviceList: Array<"desktop" | "mobile"> = + devices === "both" ? ["desktop", "mobile"] : [devices]; + return keywords.flatMap((kw) => + deviceList.map((device) => ({ + keyword: kw.keyword, + keywordId: kw.id, + device, + })), + ); +} + +// --------------------------------------------------------------------------- +// Step bodies. Each runs inside a single step.do: inputs are its parameters, +// the return value is what the workflow engine persists and replays. They must +// not touch any mutable state outside their arguments. +// --------------------------------------------------------------------------- + +/** + * Check keyword/device pairs against the live endpoint and persist snapshots. + * Per-call failures are logged and skipped (the metered client already charged + * or refused each call individually). Returns the snapshot count written. + */ +async function checkBatchLive( + ctx: CheckContext, + tasks: RankCheckTaskInput[], +): Promise { + const settled = await Promise.allSettled( + tasks.map((task) => + ctx.client.serp + .rankCheck({ + keyword: task.keyword, + keywordId: task.keywordId, + locationCode: ctx.locationCode, + languageCode: ctx.languageCode, + device: task.device, + targetDomain: ctx.domain, + depth: ctx.serpDepth, + }) + .then((r) => ({ ...r, device: task.device })), + ), + ); + const results: RankCheckResultWithDevice[] = []; + for (const outcome of settled) { + if (outcome.status === "fulfilled") { + results.push(outcome.value); + } else { + console.error( + `[rank-check] ${ctx.runId} live call failed:`, + outcome.reason, + ); + } + } + if (results.length > 0) { + await RankTrackingRepository.insertSnapshots( + mapResultsToSnapshotRows(ctx.runId, results), + ); + } + return results.length; +} + /** * Check keywords via Live API, parallel devices per keyword, real-time progress. * Snapshots are written incrementally after each batch so partial results @@ -52,53 +125,256 @@ export async function runLiveCheck( step: WorkflowStep, ctx: CheckContext, ): Promise { - const deviceList: Array<"desktop" | "mobile"> = - ctx.devices === "both" ? ["desktop", "mobile"] : [ctx.devices]; - let checked = 0; - for (let i = 0; i < ctx.keywords.length; i += KEYWORDS_PER_BATCH) { - const batch = ctx.keywords.slice(i, i + KEYWORDS_PER_BATCH); + const keywordBatch = ctx.keywords.slice(i, i + KEYWORDS_PER_BATCH); + const batchTasks = expandToTaskInputs(keywordBatch, ctx.devices); const batchIndex = Math.floor(i / KEYWORDS_PER_BATCH); + const keywordsChecked = i + keywordBatch.length; await step.do( `live-batch-${batchIndex}`, SINGLE_ATTEMPT_STEP_CONFIG, async () => { - const promises = batch.flatMap((kw) => - deviceList.map((device) => - ctx.client.serp - .rankCheck({ - keyword: kw.keyword, - keywordId: kw.id, - locationCode: ctx.locationCode, - languageCode: ctx.languageCode, - device, - targetDomain: ctx.domain, - depth: ctx.serpDepth, - }) - .then((r) => ({ ...r, device })), - ), - ); - const settled = await Promise.allSettled(promises); - const results: RankCheckResultWithDevice[] = []; - for (const outcome of settled) { - if (outcome.status === "fulfilled") { - results.push(outcome.value); - } else { - console.error("Rank check call failed:", outcome.reason); - } - } - checked += batch.length; + const written = await checkBatchLive(ctx, batchTasks); + // Progress for the UI; finalize recounts from the DB anyway. await RankTrackingRepository.updateRun(ctx.runId, { - keywordsChecked: checked, + keywordsChecked, }); - - if (results.length > 0) { - await RankTrackingRepository.insertSnapshots( - mapResultsToSnapshotRows(ctx.runId, results), - ); - } + return written; }, ); } } + +// Poll cadence for queued tasks. Standard-priority tasks complete in ~5 +// minutes on average, so the first check waits 4 minutes; cumulative waits are +// 4 / 6 / 8 / 10 / 12 / 15 minutes, after which stragglers fall back to the +// live endpoint. +const QUEUED_POLL_INTERVALS = [ + "4 minutes", + "2 minutes", + "2 minutes", + "2 minutes", + "2 minutes", + "3 minutes", +] as const; + +/** Concurrent task_get requests within a collect step. */ +const TASK_GET_CONCURRENCY = 25; + +/** Max task_get calls per collect round (per-invocation subrequest budget). */ +const TASK_GETS_PER_COLLECT = 500; + +// Collect steps may issue hundreds of task_get calls, so they get more room +// than SINGLE_ATTEMPT_STEP_CONFIG's 2-minute timeout. Unlike the metered +// steps, retrying is safe and free: task_get isn't charged and snapshot +// inserts are onConflictDoNothing. +const COLLECT_STEP_CONFIG = { + retries: { limit: 2, delay: "10 seconds" as const }, + timeout: "5 minutes" as const, +}; + +interface CollectRoundOutcome { + /** Snapshots written this round. */ + collected: number; + /** Tasks still in DataForSEO's queue — poll again next round. */ + stillPending: PostedRankCheckTask[]; + /** Tasks DataForSEO failed — route to the live fallback. */ + failed: PostedRankCheckTask[]; +} + +/** + * Fetch results for queued tasks (one free task_get each), persist completed + * snapshots, and update run progress. Transient task_get failures stay + * pending for the next round. + */ +async function collectQueuedRound( + ctx: CheckContext, + tasks: PostedRankCheckTask[], +): Promise { + const completed: RankCheckResultWithDevice[] = []; + const stillPending: PostedRankCheckTask[] = []; + const failed: PostedRankCheckTask[] = []; + + for (let i = 0; i < tasks.length; i += TASK_GET_CONCURRENCY) { + const chunk = tasks.slice(i, i + TASK_GET_CONCURRENCY); + const settled = await Promise.allSettled( + chunk.map((task) => + fetchRankCheckTaskResult({ + taskId: task.taskId, + keywordId: task.keywordId, + keyword: task.keyword, + targetDomain: ctx.domain, + }), + ), + ); + settled.forEach((result, index) => { + const task = chunk[index]; + if (result.status === "rejected") { + // Transient fetch failure — try again next round. + console.warn( + `[rank-check] ${ctx.runId} task_get failed:`, + result.reason, + ); + stillPending.push(task); + } else if (result.value.status === "pending") { + stillPending.push(task); + } else if (result.value.status === "failed") { + console.warn( + `[rank-check] ${ctx.runId} task ${task.taskId} failed: ${result.value.message}`, + ); + failed.push(task); + } else { + completed.push({ ...result.value.result, device: task.device }); + } + }); + } + + if (completed.length > 0) { + await RankTrackingRepository.insertSnapshots( + mapResultsToSnapshotRows(ctx.runId, completed), + ); + // Progress for the UI; finalize recounts from the DB anyway. + const snapshots = await RankTrackingRepository.getSnapshotsForRun( + ctx.runId, + ); + await RankTrackingRepository.updateRun(ctx.runId, { + keywordsChecked: new Set(snapshots.map((s) => s.trackingKeywordId)).size, + }); + } + + return { collected: completed.length, stillPending, failed }; +} + +/** Per-run accounting for the queued path, in keyword/device task units. */ +export interface QueuedCheckStats { + /** Tasks accepted into DataForSEO's queue. */ + queueTasks: number; + /** Task results collected from the queue within the polling window. */ + queueCollected: number; + /** Tasks routed to the live fallback (rejected, failed, or timed out). */ + fallbackTasks: number; + /** Fallback tasks that produced a snapshot. */ + fallbackChecked: number; +} + +/** + * Check keywords via DataForSEO's standard task queue (~30% of live cost). + * Posts every keyword/device pair as a queued task, then polls task_get for + * ~15 minutes, writing snapshots incrementally as tasks complete. Anything + * still unfinished after the polling window — plus tasks DataForSEO rejected + * or failed — gets one shot at the live endpoint so a run never hangs on a + * stuck queue. Billing happens at task_post (and per live-fallback call). + */ +export async function runQueuedCheck( + step: WorkflowStep, + ctx: CheckContext, +): Promise { + const taskInputs = expandToTaskInputs(ctx.keywords, ctx.devices); + + // Post all tasks to the queue, <=100 per request, one metered step each. + // A failed chunk must not abort the run — earlier chunks were already + // charged at DataForSEO, so their results have to be collected. The failed + // chunk's pairs go to the live fallback instead. + let pending: PostedRankCheckTask[] = []; + const fallback: RankCheckTaskInput[] = []; + for (let i = 0; i < taskInputs.length; i += MAX_TASKS_PER_POST) { + const chunk = taskInputs.slice(i, i + MAX_TASKS_PER_POST); + const postIndex = Math.floor(i / MAX_TASKS_PER_POST); + let posted: PostedRankCheckTask[]; + try { + posted = await step.do( + `post-tasks-${postIndex}`, + SINGLE_ATTEMPT_STEP_CONFIG, + async () => + ctx.client.serp.rankCheckTaskPost({ + tasks: chunk, + locationCode: ctx.locationCode, + languageCode: ctx.languageCode, + depth: ctx.serpDepth, + targetDomain: ctx.domain, + }), + ); + } catch (error) { + console.warn( + `[rank-check] ${ctx.runId} post-tasks-${postIndex} failed:`, + error, + ); + fallback.push(...chunk); + continue; + } + pending.push(...posted); + if (posted.length < chunk.length) { + const acceptedKeys = new Set( + posted.map((t) => `${t.keywordId}:${t.device}`), + ); + fallback.push( + ...chunk.filter((t) => !acceptedKeys.has(`${t.keywordId}:${t.device}`)), + ); + } + } + + const stats: QueuedCheckStats = { + queueTasks: pending.length, + queueCollected: 0, + fallbackTasks: 0, + fallbackChecked: 0, + }; + + // Poll until everything is collected or the ~15 minute window closes. A + // collect failure (past its retries) leaves that round's tasks pending for + // the next round — or the live fallback — instead of failing the run; the + // posted tasks are already paid for. + for ( + let round = 0; + round < QUEUED_POLL_INTERVALS.length && pending.length > 0; + round++ + ) { + await step.sleep(`wait-${round}`, QUEUED_POLL_INTERVALS[round]); + + // Cap task_gets per round so one collect step stays well inside the + // per-invocation subrequest limit at the 1000-keyword config ceiling. + const batch = pending.slice(0, TASK_GETS_PER_COLLECT); + const overflow = pending.slice(TASK_GETS_PER_COLLECT); + + let outcome: CollectRoundOutcome; + try { + outcome = await step.do(`collect-${round}`, COLLECT_STEP_CONFIG, () => + collectQueuedRound(ctx, batch), + ); + } catch (error) { + console.warn(`[rank-check] ${ctx.runId} collect-${round} failed:`, error); + continue; + } + + stats.queueCollected += outcome.collected; + pending = [...outcome.stillPending, ...overflow]; + fallback.push(...outcome.failed); + } + + // Live fallback: queued tasks that never finished, failed, or were rejected + // at post time. A straggler is double-billed (customer was metered the + // queued post cost and now the live call too — fractions of a cent). + // Progress isn't updated here; finalize recounts keywordsChecked from the + // DB. + const stragglers: RankCheckTaskInput[] = [...fallback, ...pending]; + stats.fallbackTasks = stragglers.length; + if (stragglers.length === 0) return stats; + + console.log( + `[rank-check] ${ctx.runId} live fallback for ${stragglers.length} task(s)`, + ); + + for (let i = 0; i < stragglers.length; i += KEYWORDS_PER_BATCH) { + const batch = stragglers.slice(i, i + KEYWORDS_PER_BATCH); + const batchIndex = Math.floor(i / KEYWORDS_PER_BATCH); + + stats.fallbackChecked += await step.do( + `fallback-batch-${batchIndex}`, + SINGLE_ATTEMPT_STEP_CONFIG, + () => checkBatchLive(ctx, batch), + ); + } + + return stats; +} diff --git a/src/shared/rank-tracking.ts b/src/shared/rank-tracking.ts index 42c78ea..73f114c 100644 --- a/src/shared/rank-tracking.ts +++ b/src/shared/rank-tracking.ts @@ -10,10 +10,22 @@ import type { RankTrackingConfig } from "@/types/schemas/rank-tracking"; // --------------------------------------------------------------------------- /** DataForSEO Live API: cost of first page (10 results) */ -const BASE_PAGE_COST_USD = 0.002; +const LIVE_BASE_PAGE_COST_USD = 0.002; /** DataForSEO Live API: cost of each additional page (75% of base) */ -const EXTRA_PAGE_COST_USD = 0.0015; +const LIVE_EXTRA_PAGE_COST_USD = 0.0015; + +/** DataForSEO task queue (standard priority): cost of first page (10 results) */ +const QUEUED_BASE_PAGE_COST_USD = 0.0006; + +/** DataForSEO task queue (standard priority): cost of each additional page (75% of base) */ +const QUEUED_EXTRA_PAGE_COST_USD = 0.00045; + +/** + * How a rank check reaches DataForSEO: "live" is the instant endpoint used for + * manual checks; "queued" is the cheaper task queue used for scheduled checks. + */ +type RankCheckMethod = "live" | "queued"; /** How many keywords are checked per batch */ export const KEYWORDS_PER_BATCH = 10; @@ -32,9 +44,11 @@ export const MAX_CONFIGS_PER_PROJECT = 20; // --------------------------------------------------------------------------- /** DataForSEO cost for a single SERP request at the given depth. */ -function costPerSerpAtDepth(depth: number): number { +function costPerSerpAtDepth(depth: number, method: RankCheckMethod): number { const pages = depth / 10; - return BASE_PAGE_COST_USD + (pages - 1) * EXTRA_PAGE_COST_USD; + return method === "queued" + ? QUEUED_BASE_PAGE_COST_USD + (pages - 1) * QUEUED_EXTRA_PAGE_COST_USD + : LIVE_BASE_PAGE_COST_USD + (pages - 1) * LIVE_EXTRA_PAGE_COST_USD; } export function depthToPages(depth: number): number { @@ -49,10 +63,11 @@ export function estimateRankCheckCredits( keywordCount: number, devices: RankTrackingConfig["devices"], depth: number, + method: RankCheckMethod, ) { const totalChecks = keywordCount * devicesCount(devices); const costUsd = roundUsdForBilling( - totalChecks * costPerSerpAtDepth(depth) * SEO_DATA_COST_MARKUP, + totalChecks * costPerSerpAtDepth(depth, method) * SEO_DATA_COST_MARKUP, ); const costCredits = Math.ceil(costUsd * AUTUMN_SEO_DATA_CREDITS_PER_USD); return { costUsd, costCredits };