metatron-open-seo/src/server/features/audit/services/DataForSeoAuditService.ts

import { randomUUID } from "node:crypto";
import type { BillingCustomerContext } from "@/server/billing/subscription";
import { createDataforseoClient } from "@/server/lib/dataforseo";
import {
  getOnPageCrawlStatus,
  getOnPageRawHtml,
  listOnPagePages,
} from "@/server/lib/dataforseo/onpage";
import { runMultipageChecks } from "@/server/lib/audit/issues/multipage";
import { runPageReporters } from "@/server/lib/audit/issues/page-reporters";
import type { PageFetchClass } from "@/server/lib/audit/types";
import { buildAnalyzedPageResult } from "@/server/workflows/site-audit-workflow-helpers";
import { AuditRepository } from "@/server/features/audit/repositories/AuditRepository";
import { ActivityRepository } from "@/server/features/activity/ActivityRepository";
import { AppError } from "@/server/lib/errors";

// Fallback audit path for sites that block our own crawler: DataForSEO's
// OnPage API crawls from its own infrastructure (optionally rendering JS), so
// it gets past blocks a plain fetch() from our server can't. Capped at
// MONTHLY_QUOTA runs per project per calendar month — this is a paid,
// explicitly user-triggered action (a button, never an automatic fallback).
//
// Known gap vs. a native audit: broken-internal-link and orphan-page issues
// are not detected here. Those checks read the crawl's link graph out of the
// AuditScratchpad Durable Object, which only the native crawl populates.
// Every other per-page and cross-page check runs identically (same
// analyzeHtml/runPageReporters/runMultipageChecks the native path uses).

const MONTHLY_QUOTA = 5;
const DATAFORSEO_AUDIT_ACTION = "audit.dataforseo_report";
// Pages ingested (raw_html fetch + analyze + persist) per advanceDataForSeoAudit
// call, so one poll stays well inside a single request's time budget.
const INGEST_BATCH_SIZE = 15;

function currentMonthStart(): Date {
  const now = new Date();
  return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1));
}

function extractBareDomain(startUrl: string): string {
  // DataForSEO's `target` param wants a bare domain — no protocol, no www.
  const hostname = new URL(startUrl).hostname;
  return hostname.replace(/^www\./, "");
}

export async function getDataForSeoAuditQuota(input: {
  organizationId: string;
  projectId: string;
}): Promise<{ used: number; limit: number; remaining: number }> {
  const used = await ActivityRepository.countSince({
    organizationId: input.organizationId,
    action: DATAFORSEO_AUDIT_ACTION,
    targetId: input.projectId,
    since: currentMonthStart(),
  });
  return {
    used,
    limit: MONTHLY_QUOTA,
    remaining: Math.max(0, MONTHLY_QUOTA - used),
  };
}

export async function startDataForSeoAudit(
  billingCustomer: BillingCustomerContext,
  input: {
    projectId: string;
    startedByUserId: string;
    startUrl: string;
    maxCrawlPages: number;
  },
): Promise<{ auditId: string }> {
  const quota = await getDataForSeoAuditQuota({
    organizationId: billingCustomer.organizationId,
    projectId: input.projectId,
  });
  if (quota.remaining <= 0) {
    throw new AppError(
      "RATE_LIMITED",
      `This project has used all ${MONTHLY_QUOTA} DataForSEO audits available this month. It resets at the start of next month.`,
    );
  }

  const dataforseo = createDataforseoClient(billingCustomer);
  const { taskId } = await dataforseo.onPage.taskPost({
    target: extractBareDomain(input.startUrl),
    maxCrawlPages: input.maxCrawlPages,
  });

  const auditId = randomUUID();
  // No Cloudflare Workflow backs this audit; the id only needs to be stable
  // and unique so the repository's workflowInstanceId-guarded updates work.
  const workflowInstanceId = `dataforseo-${auditId}`;

  await AuditRepository.createAudit({
    id: auditId,
    projectId: input.projectId,
    startedByUserId: input.startedByUserId,
    startUrl: input.startUrl,
    workflowInstanceId,
    config: { maxPages: input.maxCrawlPages, lighthouseStrategy: "none" },
    pagesTotal: input.maxCrawlPages,
    lighthouseTotal: 0,
    crawlSource: "dataforseo",
    dataforseoTaskId: taskId,
  });

  // Counts toward the monthly quota immediately (the DataForSEO spend is
  // already committed at this point, regardless of how the crawl finishes).
  await ActivityRepository.record({
    context: {
      userId: billingCustomer.userId,
      userEmail: billingCustomer.userEmail,
      organizationId: billingCustomer.organizationId,
    },
    action: DATAFORSEO_AUDIT_ACTION,
    targetType: "project",
    targetId: input.projectId,
    targetLabel: input.startUrl,
    metadata: { maxCrawlPages: input.maxCrawlPages },
  });

  return { auditId };
}

type AdvanceResult = {
  status: "running" | "completed" | "failed";
  pagesCrawled: number;
  pagesTotal: number;
};

/**
 * Do one bounded unit of work on a DataForSEO-sourced audit and report
 * progress. Meant to be called repeatedly by the same polling loop the native
 * audit's progress UI already uses, until status is "completed" or "failed".
 */
export async function advanceDataForSeoAudit(input: {
  auditId: string;
  projectId: string;
}): Promise<AdvanceResult> {
  const audit = await AuditRepository.getAuditForProject(
    input.auditId,
    input.projectId,
  );
  if (!audit || audit.crawlSource !== "dataforseo" || !audit.dataforseoTaskId) {
    throw new AppError("NOT_FOUND", "DataForSEO audit not found.");
  }
  if (audit.status !== "running") {
    return {
      status: audit.status,
      pagesCrawled: audit.pagesCrawled,
      pagesTotal: audit.pagesTotal,
    };
  }

  if (audit.currentPhase !== "analyzing") {
    return advanceCrawlPhase(audit);
  }
  return advanceAnalyzePhase(audit);
}

type AuditRow = NonNullable<
  Awaited<ReturnType<typeof AuditRepository.getAuditForProject>>
>;

// Always set by startDataForSeoAudit; a missing value means the audit row is
// corrupt, not a normal runtime state, so this throws rather than silently
// no-oping the guarded repository update below (an empty-string fallback
// would never match the WHERE clause and mask the failure instead).
function requireWorkflowInstanceId(audit: AuditRow): string {
  if (!audit.workflowInstanceId) {
    throw new AppError(
      "INTERNAL_ERROR",
      `Audit ${audit.id} is missing its workflowInstanceId.`,
    );
  }
  return audit.workflowInstanceId;
}

async function advanceCrawlPhase(audit: AuditRow): Promise<AdvanceResult> {
  const taskId = audit.dataforseoTaskId;
  if (!taskId) {
    throw new AppError(
      "INTERNAL_ERROR",
      "Audit is missing its DataForSEO task id.",
    );
  }
  const workflowInstanceId = requireWorkflowInstanceId(audit);

  const crawlStatus = await getOnPageCrawlStatus(taskId);
  if (!crawlStatus) {
    await AuditRepository.failAudit(audit.id, workflowInstanceId, {
      errorCode: "unknown",
      errorDetail: "DataForSEO task result expired before it was retrieved.",
      failedPhase: audit.currentPhase,
    });
    return {
      status: "failed",
      pagesCrawled: audit.pagesCrawled,
      pagesTotal: audit.pagesTotal,
    };
  }

  if (crawlStatus.progress === "in_progress") {
    await AuditRepository.updateAuditProgress(audit.id, workflowInstanceId, {
      pagesCrawled: crawlStatus.pagesCrawled,
    });
    return {
      status: "running",
      pagesCrawled: crawlStatus.pagesCrawled,
      pagesTotal: audit.pagesTotal,
    };
  }

  // Crawl finished on DataForSEO's side — switch to ingesting pages. The
  // pagesCrawled counter is repurposed as the ingest cursor for this phase.
  await AuditRepository.updateAuditProgress(audit.id, workflowInstanceId, {
    currentPhase: "analyzing",
    pagesCrawled: 0,
    pagesTotal: crawlStatus.pagesCrawled,
  });
  return {
    status: "running",
    pagesCrawled: 0,
    pagesTotal: crawlStatus.pagesCrawled,
  };
}

async function advanceAnalyzePhase(audit: AuditRow): Promise<AdvanceResult> {
  const taskId = audit.dataforseoTaskId;
  if (!taskId) {
    throw new AppError(
      "INTERNAL_ERROR",
      "Audit is missing its DataForSEO task id.",
    );
  }
  const workflowInstanceId = requireWorkflowInstanceId(audit);

  // Free read of already-billed crawl results; re-fetched each call to avoid
  // persisting a separate page list just for the ingest cursor.
  const allPages = await listOnPagePages(taskId);
  const offset = audit.pagesCrawled;
  const batch = allPages.slice(offset, offset + INGEST_BATCH_SIZE);

  if (batch.length === 0) {
    const issues = await runMultipageChecks({ auditId: audit.id });
    await AuditRepository.insertIssues(audit.id, issues);
    await AuditRepository.completeAudit(audit.id, workflowInstanceId, {
      pagesCrawled: allPages.length,
      pagesTotal: allPages.length,
    });
    return {
      status: "completed",
      pagesCrawled: allPages.length,
      pagesTotal: allPages.length,
    };
  }

  const analyzedPages = await Promise.all(
    batch.map(async (page) => {
      const html = await getOnPageRawHtml(taskId, page.url);
      const statusCode = page.status_code ?? 0;
      const fetchClass: PageFetchClass = html ? "ok" : "error";
      return buildAnalyzedPageResult({
        url: page.url,
        statusCode,
        body: html ?? "",
        fetchClass,
        responseTimeMs: 0,
        xRobotsTag: null,
        headerCanonicalUrl: null,
        crawlDepth: null,
        inSitemap: false,
      });
    }),
  );
  const issues = analyzedPages.flatMap((page) => runPageReporters(page));
  await AuditRepository.insertCrawledBatch(audit.id, analyzedPages, issues);

  const pagesCrawled = offset + batch.length;
  await AuditRepository.updateAuditProgress(audit.id, workflowInstanceId, {
    pagesCrawled,
  });

  return { status: "running", pagesCrawled, pagesTotal: allPages.length };
}