diff --git a/src/server/features/audit/repositories/AuditRepository.ts b/src/server/features/audit/repositories/AuditRepository.ts index 221cd80..3dd1542 100644 --- a/src/server/features/audit/repositories/AuditRepository.ts +++ b/src/server/features/audit/repositories/AuditRepository.ts @@ -248,8 +248,8 @@ async function insertLighthouseResults( payloadSizeBytes: result.payloadSizeBytes ?? null, })), ); - // Upsert: a step retry can charge a second DataForSEO call whose result - // must not be silently dropped in favor of a failed first attempt. + // The persistence step is retryable after its paid provider result has been + // checkpointed, so repeated writes must stay idempotent. await executeInBatches(rows, (tx, row) => { const { id: _id, auditId: _auditId, ...dataColumns } = row; return tx.insert(auditLighthouseResults).values(row).onConflictDoUpdate({ diff --git a/src/server/features/audit/services/AuditService.limitTier.test.ts b/src/server/features/audit/services/AuditService.limitTier.test.ts new file mode 100644 index 0000000..c58e2b4 --- /dev/null +++ b/src/server/features/audit/services/AuditService.limitTier.test.ts @@ -0,0 +1,45 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { isHostedMock, hasManagedAccessMock, hasPaidPlanMock } = vi.hoisted( + () => ({ + isHostedMock: vi.fn(), + hasManagedAccessMock: vi.fn(), + hasPaidPlanMock: vi.fn(), + }), +); + +vi.mock("cloudflare:workers", () => ({ env: {} })); +vi.mock("@/server/lib/runtime-env", () => ({ + isHostedServerAuthMode: isHostedMock, +})); +vi.mock("@/server/billing/subscription", () => ({ + customerHasManagedAccess: hasManagedAccessMock, + customerHasPaidPlan: hasPaidPlanMock, +})); +vi.mock("@/server/features/audit/repositories/AuditRepository", () => ({ + AuditRepository: {}, +})); +vi.mock("@/server/features/audit/AuditScratchpad", () => ({ + getAuditScratchpad: vi.fn(), +})); +vi.mock("@/server/lib/audit/progress-kv", () => ({ AuditProgressKV: {} })); + +import { AuditService } from "@/server/features/audit/services/AuditService"; + +describe("resolveAuditLimitTier", () => { + beforeEach(() => { + vi.clearAllMocks(); + hasManagedAccessMock.mockResolvedValue(true); + hasPaidPlanMock.mockResolvedValue(true); + }); + + it("uses the uncapped self-hosted tier without consulting billing", async () => { + isHostedMock.mockResolvedValue(false); + + await expect(AuditService.resolveAuditLimitTier("org-1")).resolves.toBe( + "self_hosted", + ); + expect(hasManagedAccessMock).not.toHaveBeenCalled(); + expect(hasPaidPlanMock).not.toHaveBeenCalled(); + }); +}); diff --git a/src/server/features/audit/services/AuditService.ts b/src/server/features/audit/services/AuditService.ts index 459ec4c..a66b3d6 100644 --- a/src/server/features/audit/services/AuditService.ts +++ b/src/server/features/audit/services/AuditService.ts @@ -32,7 +32,7 @@ import { isHostedServerAuthMode } from "@/server/lib/runtime-env"; async function resolveAuditLimitTier( organizationId: string, ): Promise { - if (!(await isHostedServerAuthMode())) return "paid"; + if (!(await isHostedServerAuthMode())) return "self_hosted"; const [hasManagedAccess, hasPaidPlan] = await Promise.all([ customerHasManagedAccess(organizationId), customerHasPaidPlan(organizationId), diff --git a/src/server/features/audit/services/audit-capacity.test.ts b/src/server/features/audit/services/audit-capacity.test.ts index 238991c..f107821 100644 --- a/src/server/features/audit/services/audit-capacity.test.ts +++ b/src/server/features/audit/services/audit-capacity.test.ts @@ -47,4 +47,13 @@ describe("audit capacity helpers", () => { expect(freeAudit.pagesTotal).toBe(AUDIT_LIMITS.free.maxPagesPerAudit); expect(freeAudit.total).toBeLessThan(AUDIT_LIMITS.free.maxCapacityUnits); }); + + it("lifts only the cumulative self-hosted cap", () => { + expect(AUDIT_LIMITS.self_hosted.maxCapacityUnits).toBe( + Number.POSITIVE_INFINITY, + ); + expect(AUDIT_LIMITS.self_hosted.maxPagesPerAudit).toBe( + AUDIT_LIMITS.paid.maxPagesPerAudit, + ); + }); }); diff --git a/src/server/features/audit/services/audit-capacity.ts b/src/server/features/audit/services/audit-capacity.ts index 280d5b7..258e798 100644 --- a/src/server/features/audit/services/audit-capacity.ts +++ b/src/server/features/audit/services/audit-capacity.ts @@ -6,13 +6,14 @@ import { PAID_MAX_AUDIT_PAGES, } from "@/shared/audit-limits"; -export type AuditLimitTier = "free" | "paid"; +export type AuditLimitTier = "free" | "paid" | "self_hosted"; // The crawler runs on our Workers compute and isn't credit-metered, so these // per-tier bounds are the abuse control: free accounts cost nothing to create, // so they get one small audit at a time and a modest total budget. Paid gets // bounds sized for real sites rather than abuse (a payment method on file is -// the deterrent). Self-hosted deployments resolve to the paid tier. +// the deterrent). The cumulative bound is a hosted commercial policy, while +// the per-audit page limit is also a technical Workflow/database ceiling. export const AUDIT_LIMITS: Record< AuditLimitTier, { @@ -31,6 +32,11 @@ export const AUDIT_LIMITS: Record< maxCapacityUnits: 100_000, maxRunningAudits: Number.POSITIVE_INFINITY, }, + self_hosted: { + maxPagesPerAudit: PAID_MAX_AUDIT_PAGES, + maxCapacityUnits: Number.POSITIVE_INFINITY, + maxRunningAudits: Number.POSITIVE_INFINITY, + }, }; export function clampAuditMaxPages(maxPages?: number) { diff --git a/src/server/lib/audit/lighthouse.test.ts b/src/server/lib/audit/lighthouse.test.ts index 87e6807..89c19c3 100644 --- a/src/server/lib/audit/lighthouse.test.ts +++ b/src/server/lib/audit/lighthouse.test.ts @@ -1,14 +1,20 @@ -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +const createDataforseoClientMock = vi.hoisted(() => vi.fn()); vi.mock("@/server/lib/dataforseo", () => ({ - createDataforseoClient: vi.fn(), + createDataforseoClient: createDataforseoClientMock, })); vi.mock("@/server/lib/r2", () => ({ putTextToR2: vi.fn(), })); -import { selectLighthouseSample } from "./lighthouse"; +import { fetchLighthouseResult, selectLighthouseSample } from "./lighthouse"; + +afterEach(() => { + vi.clearAllMocks(); +}); describe("selectLighthouseSample", () => { it("includes a start page reached through a trailing-slash redirect", () => { @@ -42,4 +48,62 @@ describe("selectLighthouseSample", () => { expect(selected[0]).toBe("https://example.com/services"); }); + + it("does not sample another page from the start page's template", () => { + const selected = selectLighthouseSample( + [ + { url: "https://example.com/products/123", statusCode: 200 }, + { url: "https://example.com/products/456", statusCode: 200 }, + { url: "https://example.com/about", statusCode: 200 }, + ], + "https://example.com/products/123", + "auto", + ); + + expect(selected).toEqual([ + "https://example.com/products/123", + "https://example.com/about", + ]); + }); +}); + +describe("fetchLighthouseResult", () => { + const billingCustomer = { + userId: "user-1", + userEmail: "test@example.com", + organizationId: "org-1", + }; + + it("does not retry an ambiguous generic failure", async () => { + const live = vi + .fn() + .mockRejectedValueOnce(new Error("temporary failure")) + .mockResolvedValueOnce({ + scores: { + performance: 90, + accessibility: 91, + "best-practices": 92, + seo: 93, + }, + metrics: { + largestContentfulPaint: { numericValue: 1000 }, + cumulativeLayoutShift: { numericValue: 0.01 }, + interactionToNextPaint: { numericValue: 100 }, + serverResponseTime: { numericValue: 200 }, + }, + }); + createDataforseoClientMock.mockReturnValue({ + lighthouse: { live }, + }); + + const fetched = await fetchLighthouseResult( + "https://example.com/", + "page-1", + "desktop", + billingCustomer, + ); + + expect(live).toHaveBeenCalledOnce(); + expect(fetched.result.errorMessage).toBe("temporary failure"); + }); }); diff --git a/src/server/lib/audit/lighthouse.ts b/src/server/lib/audit/lighthouse.ts index 0590775..264b55e 100644 --- a/src/server/lib/audit/lighthouse.ts +++ b/src/server/lib/audit/lighthouse.ts @@ -22,99 +22,70 @@ type LighthouseFetchResult = { payloadJson: string | null; }; -async function fetchLighthouseResult( +export async function fetchLighthouseResult( url: string, pageId: string, strategy: "mobile" | "desktop", billingCustomer: BillingCustomerContext, ): Promise { - let lastError: Error | null = null; const dataforseo = createDataforseoClient(billingCustomer); + try { + const data = await dataforseo.lighthouse.live({ url, strategy }); - for (let attempt = 0; attempt < 3; attempt++) { - try { - if (attempt > 0) { - // Exponential backoff: 2s, 4s - await new Promise((resolve) => - setTimeout(resolve, 2000 * Math.pow(2, attempt - 1)), - ); - } - - const data = await dataforseo.lighthouse.live({ url, strategy }); - - return { - result: { - url, - pageId, - strategy, - performanceScore: data.scores.performance, - accessibilityScore: data.scores.accessibility, - bestPracticesScore: data.scores["best-practices"], - seoScore: data.scores.seo, - lcpMs: data.metrics.largestContentfulPaint.numericValue, - cls: data.metrics.cumulativeLayoutShift.numericValue, - inpMs: data.metrics.interactionToNextPaint.numericValue, - ttfbMs: data.metrics.serverResponseTime.numericValue, - }, - payloadJson: JSON.stringify(data), - }; - } catch (error) { - lastError = error instanceof Error ? error : new Error(String(error)); - console.warn( - `Lighthouse attempt ${attempt + 1} failed for ${url}:`, - lastError.message, - ); - } + return { + result: { + url, + pageId, + strategy, + performanceScore: data.scores.performance, + accessibilityScore: data.scores.accessibility, + bestPracticesScore: data.scores["best-practices"], + seoScore: data.scores.seo, + lcpMs: data.metrics.largestContentfulPaint.numericValue, + cls: data.metrics.cumulativeLayoutShift.numericValue, + inpMs: data.metrics.interactionToNextPaint.numericValue, + ttfbMs: data.metrics.serverResponseTime.numericValue, + }, + payloadJson: JSON.stringify(data), + }; + } catch (error) { + const failed = error instanceof Error ? error : new Error(String(error)); + console.error(`Lighthouse failed for ${url}:`, failed.message); + return { + result: { + url, + pageId, + strategy, + performanceScore: null, + accessibilityScore: null, + bestPracticesScore: null, + seoScore: null, + lcpMs: null, + cls: null, + inpMs: null, + ttfbMs: null, + errorMessage: failed.message, + }, + payloadJson: null, + }; } - - // All retries exhausted — return null scores - console.error( - `Lighthouse failed after 3 attempts for ${url}:`, - lastError?.message, - ); - return { - result: { - url, - pageId, - strategy, - performanceScore: null, - accessibilityScore: null, - bestPracticesScore: null, - seoScore: null, - lcpMs: null, - cls: null, - inpMs: null, - ttfbMs: null, - errorMessage: lastError?.message ?? "Lighthouse request failed", - }, - payloadJson: null, - }; } -export async function fetchAndStoreLighthouseResult(input: { - url: string; - pageId: string; - strategy: "mobile" | "desktop"; - billingCustomer: BillingCustomerContext; +export async function storeLighthouseResult(input: { projectId: string; auditId: string; + fetched: LighthouseFetchResult; }): Promise { - const fetched = await fetchLighthouseResult( - input.url, - input.pageId, - input.strategy, - input.billingCustomer, - ); - - if (!fetched.payloadJson) { - return fetched.result; + if (!input.fetched.payloadJson) { + return input.fetched.result; } - const key = `site-audit/${input.projectId}/${input.auditId}/${input.pageId}-${input.strategy}.json`; - const uploaded = await putTextToR2(key, fetched.payloadJson); + const { pageId, strategy } = input.fetched.result; + const key = `site-audit/${input.projectId}/${input.auditId}/${pageId}-${strategy}.json`; + const uploaded = await putTextToR2(key, input.fetched.payloadJson); return { - ...fetched.result, + ...input.fetched.result, r2Key: uploaded.key, payloadSizeBytes: uploaded.sizeBytes, }; @@ -153,6 +124,12 @@ export function selectLighthouseSample( // Group by URL template pattern const templateGroups = new Map(); + if (startPage) { + templateGroups.set( + detectUrlTemplate(new URL(startPage.url).pathname), + startPage, + ); + } for (const page of validPages) { if (selected.has(page.url)) continue; const template = detectUrlTemplate(new URL(page.url).pathname); diff --git a/src/server/lib/dataforseo/core.test.ts b/src/server/lib/dataforseo/core.test.ts new file mode 100644 index 0000000..f6af940 --- /dev/null +++ b/src/server/lib/dataforseo/core.test.ts @@ -0,0 +1,25 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/server/lib/runtime-env", () => ({ + getRequiredEnvValue: vi.fn().mockResolvedValue("encoded-credentials"), +})); + +import { onPageApi } from "@/server/lib/dataforseo/core"; + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("DataForSEO OnPage transport", () => { + it("does not retry a Lighthouse HTTP 5xx response", async () => { + const fetchMock = vi + .fn() + .mockResolvedValue(new Response("upstream failure", { status: 503 })); + vi.stubGlobal("fetch", fetchMock); + + await expect(onPageApi().lighthouseLiveJson([])).rejects.toMatchObject({ + code: "UPSTREAM_UNAVAILABLE", + }); + expect(fetchMock).toHaveBeenCalledOnce(); + }); +}); diff --git a/src/server/lib/dataforseo/core.ts b/src/server/lib/dataforseo/core.ts index 17fa97c..5bcd978 100644 --- a/src/server/lib/dataforseo/core.ts +++ b/src/server/lib/dataforseo/core.ts @@ -65,7 +65,10 @@ function formatDataforseoRequestPath(url: RequestInfo): string { * (which return HTTP 200) are handled downstream by {@link assertOk}. An * optional classifier maps recognised HTTP failures to product errors. */ -function createAuthenticatedFetch(classify?: DataforseoErrorClassifier) { +function createAuthenticatedFetch( + classify?: DataforseoErrorClassifier, + maxServerErrorRetries = DATAFORSEO_MAX_RETRIES, +) { return async (url: RequestInfo, init?: RequestInit): Promise => { const apiKey = await getRequiredEnvValue("DATAFORSEO_API_KEY"); const headers = new Headers(init?.headers); @@ -80,7 +83,7 @@ function createAuthenticatedFetch(classify?: DataforseoErrorClassifier) { if (response.ok) return response; // Transient upstream 5xx on an idempotent read -> back off and retry. - if (response.status >= 500 && attempt < DATAFORSEO_MAX_RETRIES) { + if (response.status >= 500 && attempt < maxServerErrorRetries) { await new Promise((resolve) => setTimeout(resolve, DATAFORSEO_RETRY_BACKOFF_MS * (attempt + 1)), ); @@ -116,8 +119,11 @@ function createAuthenticatedFetch(classify?: DataforseoErrorClassifier) { }; } -function http(classify?: DataforseoErrorClassifier) { - return { fetch: createAuthenticatedFetch(classify) }; +function http( + classify?: DataforseoErrorClassifier, + maxServerErrorRetries = DATAFORSEO_MAX_RETRIES, +) { + return { fetch: createAuthenticatedFetch(classify, maxServerErrorRetries) }; } // Per-section API factories. Each is created per-request so the auth secret is @@ -126,7 +132,9 @@ export const labsApi = () => new DataforseoLabsApi(API_BASE, http()); export const keywordsDataApi = () => new KeywordsDataApi(API_BASE, http()); export const serpApi = () => new SerpApi(API_BASE, http()); export const businessDataApi = () => new BusinessDataApi(API_BASE, http()); -export const onPageApi = () => new OnPageApi(API_BASE, http()); +// Lighthouse live is a billed, non-idempotent POST. A 5xx does not prove the +// provider skipped the charge, so this client must not replay it. +export const onPageApi = () => new OnPageApi(API_BASE, http(undefined, 0)); // Account/appendix data (spend, balance, rates). userData() is FREE ($0) and // read-only — do NOT wire it through metering. export const appendixApi = () => new AppendixApi(API_BASE, http()); diff --git a/src/server/lib/dataforseo/lighthouse.test.ts b/src/server/lib/dataforseo/lighthouse.test.ts new file mode 100644 index 0000000..201faed --- /dev/null +++ b/src/server/lib/dataforseo/lighthouse.test.ts @@ -0,0 +1,63 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { onPageApiMock, lighthouseLiveJson } = vi.hoisted(() => ({ + onPageApiMock: vi.fn(), + lighthouseLiveJson: vi.fn(), +})); + +vi.mock("dataforseo-client", () => ({ + OnPageLighthouseLiveJsonRequestInfo: class { + constructor(public input: unknown) {} + }, +})); + +vi.mock("@/server/lib/dataforseo/core", () => ({ + onPageApi: onPageApiMock, +})); + +import { DataforseoChargedTaskError } from "@/server/lib/dataforseo/envelope"; +import { fetchLighthouseResult } from "@/server/lib/dataforseo/lighthouse"; + +beforeEach(() => { + vi.clearAllMocks(); + onPageApiMock.mockReturnValue({ lighthouseLiveJson }); +}); + +describe("fetchLighthouseResult", () => { + it("carries billing metadata when parsing fails after a billed success", async () => { + lighthouseLiveJson.mockResolvedValue({ + status_code: 20000, + status_message: "Ok.", + tasks: [ + { + id: "task-1", + status_code: 20000, + status_message: "Ok.", + path: ["v3", "on_page", "lighthouse", "live", "json"], + cost: 0.00425, + result: [ + { + requestedUrl: "https://example.com/", + finalUrl: "https://example.com/", + categories: {}, + audits: {}, + }, + ], + }, + ], + }); + + const rejection = fetchLighthouseResult({ + url: "https://example.com/", + strategy: "mobile", + }); + + await expect(rejection).rejects.toBeInstanceOf(DataforseoChargedTaskError); + await expect(rejection).rejects.toMatchObject({ + billing: { + path: ["v3", "on_page", "lighthouse", "live", "json"], + costUsd: 0.00425, + }, + }); + }); +}); diff --git a/src/server/lib/dataforseo/lighthouse.ts b/src/server/lib/dataforseo/lighthouse.ts index f36d43e..3181e83 100644 --- a/src/server/lib/dataforseo/lighthouse.ts +++ b/src/server/lib/dataforseo/lighthouse.ts @@ -9,6 +9,7 @@ import { onPageApi } from "@/server/lib/dataforseo/core"; import { assertOk, buildTaskBilling, + DataforseoChargedTaskError, type DataforseoApiResponse, } from "@/server/lib/dataforseo/envelope"; @@ -24,9 +25,16 @@ export async function fetchLighthouseResult(input: { }), ]); - // assertOk handles status / charged-task billing; parse extracts the scores. + // Build the metering envelope before parsing. The provider has already + // charged a successful task, so a malformed payload must carry its billing + // metadata out to the metered client instead of looking retryable. const task = assertOk(response); - const data = parseDataforseoLighthousePayload(response, input); - - return { data, billing: buildTaskBilling(task) }; + const billing = buildTaskBilling(task); + try { + const data = parseDataforseoLighthousePayload(response, input); + return { data, billing }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw new DataforseoChargedTaskError(message, billing); + } } diff --git a/src/server/workflows/auditStepConfigs.ts b/src/server/workflows/auditStepConfigs.ts index 2386f53..8632b63 100644 --- a/src/server/workflows/auditStepConfigs.ts +++ b/src/server/workflows/auditStepConfigs.ts @@ -21,9 +21,18 @@ export const CRAWL_CHUNK_STEP: WorkflowStepConfig = { timeout: "5 minutes", }; -/** One Lighthouse batch: 20 DataForSEO calls, each with internal retries. */ -export const LIGHTHOUSE_BATCH_STEP: WorkflowStepConfig = { - retries: { limit: 1, delay: "10 seconds", backoff: "constant" }, +/** + * One Lighthouse URL (mobile + desktop). DataForSEO charges these calls, so a + * Workflow replay must never issue them again after the step starts. + */ +export const LIGHTHOUSE_FETCH_STEP: WorkflowStepConfig = { + retries: { limit: 0, delay: "1 second" }, + timeout: "5 minutes", +}; + +/** R2 + DB persistence is idempotent and safe to retry after the paid step. */ +export const LIGHTHOUSE_PERSIST_STEP: WorkflowStepConfig = { + retries: { limit: 3, delay: "5 seconds", backoff: "exponential" }, timeout: "5 minutes", }; diff --git a/src/server/workflows/siteAuditWorkflowPhases.test.ts b/src/server/workflows/siteAuditWorkflowPhases.test.ts new file mode 100644 index 0000000..abe3dc4 --- /dev/null +++ b/src/server/workflows/siteAuditWorkflowPhases.test.ts @@ -0,0 +1,176 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { + fetchLighthouseResultMock, + selectLighthouseSampleMock, + storeLighthouseResultMock, + pgStepMock, + getPagesForAuditMock, + insertLighthouseResultsMock, + updateAuditProgressMock, +} = vi.hoisted(() => ({ + fetchLighthouseResultMock: vi.fn(), + selectLighthouseSampleMock: vi.fn(), + storeLighthouseResultMock: vi.fn(), + pgStepMock: vi.fn(), + getPagesForAuditMock: vi.fn(), + insertLighthouseResultsMock: vi.fn(), + updateAuditProgressMock: vi.fn(), +})); + +vi.mock("@/server/lib/audit/lighthouse", () => ({ + fetchLighthouseResult: fetchLighthouseResultMock, + selectLighthouseSample: selectLighthouseSampleMock, + storeLighthouseResult: storeLighthouseResultMock, +})); +vi.mock("@/server/features/audit/repositories/AuditRepository", () => ({ + AuditRepository: { + getPagesForAudit: getPagesForAuditMock, + insertLighthouseResults: insertLighthouseResultsMock, + updateAuditProgress: updateAuditProgressMock, + }, +})); +vi.mock("@/server/features/audit/AuditScratchpad", () => ({ + getAuditScratchpad: vi.fn(), +})); +vi.mock("@/server/lib/audit/progress-kv", () => ({ AuditProgressKV: {} })); +vi.mock("@/server/lib/audit/discovery", () => ({ + discoverUrls: vi.fn(), + parseRobotsTxt: vi.fn(), +})); +vi.mock("@/server/lib/audit/issues/multipage", () => ({ + runMultipageChecks: vi.fn(), +})); +vi.mock("@/server/lib/posthog", () => ({ captureServerEvent: vi.fn() })); +vi.mock("@/server/workflows/siteAuditWorkflowCrawl", () => ({ + runCrawlPhase: vi.fn(), +})); +vi.mock("@/server/workflows/pgStep", () => ({ pgStep: pgStepMock })); + +import { runLighthousePhase } from "@/server/workflows/siteAuditWorkflowPhases"; + +describe("runLighthousePhase", () => { + beforeEach(() => { + vi.clearAllMocks(); + getPagesForAuditMock.mockResolvedValue([ + { + id: "page-1", + url: "https://example.com/", + statusCode: 200, + }, + ]); + selectLighthouseSampleMock.mockReturnValue(["https://example.com/"]); + fetchLighthouseResultMock.mockImplementation( + async (_url: string, pageId: string, strategy: "mobile" | "desktop") => ({ + result: { pageId, strategy }, + payloadJson: "{}", + }), + ); + storeLighthouseResultMock.mockImplementation( + async ({ fetched }: { fetched: { result: unknown } }) => fetched.result, + ); + insertLighthouseResultsMock.mockResolvedValue(undefined); + updateAuditProgressMock.mockResolvedValue(undefined); + }); + + it("does not replay paid calls when persistence retries", async () => { + let persistenceAttempts = 0; + let fetchRetryLimit: number | undefined; + let persistenceRetryLimit: number | undefined; + pgStepMock.mockImplementation( + async ( + _step: unknown, + name: string, + config: { retries?: { limit?: number } }, + callback: () => Promise, + ) => { + if (name === "lighthouse-fetch-1") { + fetchRetryLimit = config.retries?.limit; + } + if (name === "lighthouse-persist-1") { + persistenceRetryLimit = config.retries?.limit; + persistenceAttempts += 1; + try { + return await callback(); + } catch (error) { + if ((config.retries?.limit ?? 0) < 1) throw error; + persistenceAttempts += 1; + return callback(); + } + } + return callback(); + }, + ); + + updateAuditProgressMock + .mockResolvedValueOnce(undefined) + .mockRejectedValueOnce(new Error("progress unavailable")) + .mockResolvedValueOnce(undefined); + + // pgStep is mocked above, so the opaque WorkflowStep object is never read. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + await runLighthousePhase({} as never, { + auditId: "audit-1", + workflowInstanceId: "workflow-1", + billingCustomer: { + userId: "user-1", + userEmail: "test@example.com", + organizationId: "org-1", + }, + projectId: "project-1", + startUrl: "https://example.com/", + config: { maxPages: 50, lighthouseStrategy: "auto" }, + }); + + expect(fetchLighthouseResultMock).toHaveBeenCalledTimes(2); + expect(storeLighthouseResultMock).toHaveBeenCalledTimes(4); + expect(insertLighthouseResultsMock).toHaveBeenCalledTimes(2); + expect(persistenceAttempts).toBe(2); + expect(fetchRetryLimit).toBe(0); + expect(persistenceRetryLimit).toBe(3); + expect(updateAuditProgressMock).toHaveBeenLastCalledWith( + "audit-1", + "workflow-1", + { lighthouseCompleted: 2, lighthouseFailed: 0 }, + ); + }); + + it("does not replay paid calls for a cached legacy batch", async () => { + pgStepMock.mockImplementation( + async ( + _step: unknown, + name: string, + _config: unknown, + callback: () => Promise, + ) => { + if (name === "lighthouse-batch-1") { + return { completed: 2, failed: 0 }; + } + return callback(); + }, + ); + + // pgStep is mocked above, so the opaque WorkflowStep object is never read. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + await runLighthousePhase({} as never, { + auditId: "audit-1", + workflowInstanceId: "workflow-1", + billingCustomer: { + userId: "user-1", + userEmail: "test@example.com", + organizationId: "org-1", + }, + projectId: "project-1", + startUrl: "https://example.com/", + config: { maxPages: 50, lighthouseStrategy: "auto" }, + }); + + expect(pgStepMock).toHaveBeenCalledWith( + {}, + "lighthouse-batch-1", + expect.anything(), + expect.any(Function), + ); + expect(fetchLighthouseResultMock).not.toHaveBeenCalled(); + }); +}); diff --git a/src/server/workflows/siteAuditWorkflowPhases.ts b/src/server/workflows/siteAuditWorkflowPhases.ts index 00808ac..f0fd0a7 100644 --- a/src/server/workflows/siteAuditWorkflowPhases.ts +++ b/src/server/workflows/siteAuditWorkflowPhases.ts @@ -2,8 +2,9 @@ import type { WorkflowStep } from "cloudflare:workers"; import type { BillingCustomerContext } from "@/server/billing/subscription"; import { discoverUrls, parseRobotsTxt } from "@/server/lib/audit/discovery"; import { - fetchAndStoreLighthouseResult, + fetchLighthouseResult, selectLighthouseSample, + storeLighthouseResult, } from "@/server/lib/audit/lighthouse"; import { getOrigin, @@ -26,14 +27,19 @@ import { pgStep } from "@/server/workflows/pgStep"; import { DB_STEP, DISCOVERY_STEP, - LIGHTHOUSE_BATCH_STEP, + LIGHTHOUSE_FETCH_STEP, + LIGHTHOUSE_PERSIST_STEP, MULTIPAGE_CHECKS_STEP, } from "@/server/workflows/auditStepConfigs"; -const LIGHTHOUSE_URL_BATCH_SIZE = 10; +const LEGACY_LIGHTHOUSE_URL_BATCH_SIZE = 10; /** Frontier seeds per scratchpad RPC call. */ const SEED_RPC_BATCH = 2_000; +type LighthouseBatchBoundary = + | { schema: "retry-safe-v2" } + | { completed: number; failed: number }; + type AuditPhasesParams = { auditId: string; workflowInstanceId: string; @@ -166,7 +172,7 @@ type LighthousePhaseParams = { config: AuditConfig; }; -async function runLighthousePhase( +export async function runLighthousePhase( step: WorkflowStep, params: LighthousePhaseParams, ) { @@ -190,59 +196,88 @@ async function runLighthousePhase( let completedChecks = 0; let failedChecks = 0; - let lighthouseBatchIndex = 0; - - for (let i = 0; i < lighthouseWork.length; i += LIGHTHOUSE_URL_BATCH_SIZE) { - const batch = lighthouseWork.slice(i, i + LIGHTHOUSE_URL_BATCH_SIZE); - lighthouseBatchIndex += 1; - const priorCompleted = completedChecks; - const priorFailed = failedChecks; - - // Fetch, store (R2 + DB) and update progress inside one step. The step - // returns only counts; full results live in the DB. - const counts = await pgStep( + for ( + let batchStart = 0; + batchStart < lighthouseWork.length; + batchStart += LEGACY_LIGHTHOUSE_URL_BATCH_SIZE + ) { + const batchIndex = Math.floor( + batchStart / LEGACY_LIGHTHOUSE_URL_BATCH_SIZE, + ); + const boundary = await pgStep( step, - `lighthouse-batch-${lighthouseBatchIndex}`, - LIGHTHOUSE_BATCH_STEP, - async () => { - const perUrlResults = await Promise.all( - batch.map(async ({ url, pageId }) => { - const [mobileResult, desktopResult] = await Promise.all([ - fetchAndStoreLighthouseResult({ - url, - pageId, - strategy: "mobile", - billingCustomer, - projectId, - auditId, - }), - fetchAndStoreLighthouseResult({ - url, - pageId, - strategy: "desktop", - billingCustomer, - projectId, - auditId, - }), - ]); - return [mobileResult, desktopResult]; - }), - ); - const results = perUrlResults.flat(); - await AuditRepository.insertLighthouseResults(auditId, results); - - const failed = results.filter((result) => result.errorMessage).length; - const completed = results.length - failed; - await AuditRepository.updateAuditProgress(auditId, workflowInstanceId, { - lighthouseCompleted: priorCompleted + completed, - lighthouseFailed: priorFailed + failed, - }); - return { completed, failed }; - }, + `lighthouse-batch-${batchIndex + 1}`, + DB_STEP, + async (): Promise => ({ + schema: "retry-safe-v2", + }), ); - completedChecks += counts.completed; - failedChecks += counts.failed; + // Older deployments used this checkpoint name for a complete paid batch. + // If that cached shape replays under current code, all results and progress + // are already persisted; skip the batch instead of buying it again. + if ("completed" in boundary) { + completedChecks += boundary.completed; + failedChecks += boundary.failed; + continue; + } + + const batch = lighthouseWork.slice( + batchStart, + batchStart + LEGACY_LIGHTHOUSE_URL_BATCH_SIZE, + ); + for (const [batchOffset, { url, pageId }] of batch.entries()) { + const index = batchStart + batchOffset; + // The paid calls are checkpointed separately from all storage. With + // Workflow retries disabled, a later R2/DB/progress failure cannot replay + // DataForSEO. One URL groups its mobile + desktop checks into one compact + // checkpoint rather than returning a whole Lighthouse batch. + const fetched = await pgStep( + step, + `lighthouse-fetch-${index + 1}`, + LIGHTHOUSE_FETCH_STEP, + () => + Promise.all([ + fetchLighthouseResult(url, pageId, "mobile", billingCustomer), + fetchLighthouseResult(url, pageId, "desktop", billingCustomer), + ]), + ); + + const priorCompleted = completedChecks; + const priorFailed = failedChecks; + const counts = await pgStep( + step, + `lighthouse-persist-${index + 1}`, + LIGHTHOUSE_PERSIST_STEP, + async () => { + const results = await Promise.all( + fetched.map((result) => + storeLighthouseResult({ + projectId, + auditId, + fetched: result, + }), + ), + ); + await AuditRepository.insertLighthouseResults(auditId, results); + + const failed = results.filter((result) => result.errorMessage).length; + const completed = results.length - failed; + await AuditRepository.updateAuditProgress( + auditId, + workflowInstanceId, + { + lighthouseCompleted: priorCompleted + completed, + lighthouseFailed: priorFailed + failed, + }, + ); + return { completed, failed }; + }, + ); + + completedChecks += counts.completed; + failedChecks += counts.failed; + } } }