Scheduled rank checks run through DataForSEO's task queue (#268)
This commit is contained in:
Ben Senescu
GitHub
parent
5f03992b78
commit
f8d5ffd285
54
.claude/skills/merge-ready/SKILL.md
Normal file
54
.claude/skills/merge-ready/SKILL.md
Normal file
@ -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.
|
||||
@ -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({
|
||||
</p>
|
||||
</div>
|
||||
<div className="text-right">
|
||||
<p className="font-mono font-semibold">${costUsd.toFixed(2)}</p>
|
||||
<p className="font-mono font-semibold">~${costUsd.toFixed(2)}</p>
|
||||
{isPending && <Loader2 className="size-3 animate-spin ml-auto" />}
|
||||
</div>
|
||||
</button>
|
||||
|
||||
@ -289,10 +289,13 @@ export function RankTrackingConfigModal({
|
||||
</div>
|
||||
|
||||
{(() => {
|
||||
// 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 (
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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", () => ({
|
||||
|
||||
@ -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: {
|
||||
|
||||
@ -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<T extends DataforseoTaskLike>(
|
||||
}
|
||||
|
||||
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("/")}` : "");
|
||||
|
||||
@ -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 {
|
||||
|
||||
177
src/server/lib/dataforseo/serp.test.ts
Normal file
177
src/server/lib/dataforseo/serp.test.ts
Normal file
@ -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<typeof fetch>().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<typeof fetch>().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<typeof fetch>().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"],
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
@ -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<DataforseoApiResponse<RankCheckResult>> {
|
||||
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<DataforseoApiResponse<PostedRankCheckTask[]>> {
|
||||
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<RankCheckTaskOutcome> {
|
||||
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;
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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<number> {
|
||||
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<void> {
|
||||
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<CollectRoundOutcome> {
|
||||
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<QueuedCheckStats> {
|
||||
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;
|
||||
}
|
||||
|
||||
@ -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 };
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user