lead-scraper/app/api/jobs/serp-enrich/route.ts

import { NextRequest, NextResponse } from "next/server";
import { prisma } from "@/lib/db";
import { getApiKey } from "@/lib/utils/apiKey";
import { isSocialOrDirectory } from "@/lib/utils/domains";
import { runGoogleSerpScraper, pollRunStatus, fetchDatasetItems } from "@/lib/services/apify";
import { bulkSearchDomains, type DecisionMakerCategory } from "@/lib/services/anymailfinder";
import { sinkLeadsToVault } from "@/lib/services/leadVault";

export async function POST(req: NextRequest) {
  try {
    const body = await req.json() as {
      query: string;
      maxPages: number;
      countryCode: string;
      languageCode: string;
      filterSocial: boolean;
      categories: DecisionMakerCategory[];
      selectedDomains?: string[];
    };

    const [apifyToken, anymailKey] = await Promise.all([getApiKey("apify"), getApiKey("anymailfinder")]);

    if (!apifyToken) return NextResponse.json({ error: "Apify API token not configured" }, { status: 400 });
    if (!anymailKey) return NextResponse.json({ error: "Anymailfinder API key not configured" }, { status: 400 });

    const job = await prisma.job.create({
      data: {
        type: "serp",
        status: "running",
        config: JSON.stringify(body),
        totalLeads: 0,
      },
    });

    runSerpEnrich(job.id, body, apifyToken, anymailKey).catch(console.error);

    return NextResponse.json({ jobId: job.id });
  } catch (err) {
    console.error("POST /api/jobs/serp-enrich error:", err);
    return NextResponse.json({ error: "Failed to start job" }, { status: 500 });
  }
}

async function runSerpEnrich(
  jobId: string,
  params: {
    query: string; maxPages: number; countryCode: string; languageCode: string;
    filterSocial: boolean; categories: DecisionMakerCategory[]; selectedDomains?: string[];
  },
  apifyToken: string,
  anymailKey: string
) {
  try {
    // 1. Run Apify SERP scraper
    const runId = await runGoogleSerpScraper(
      params.query, params.maxPages, params.countryCode, params.languageCode, apifyToken
    );

    // 2. Poll until complete
    let runStatus = "";
    let datasetId = "";
    while (runStatus !== "SUCCEEDED" && runStatus !== "FAILED" && runStatus !== "ABORTED") {
      await sleep(3000);
      const result = await pollRunStatus(runId, apifyToken);
      runStatus = result.status;
      datasetId = result.defaultDatasetId;
    }

    if (runStatus !== "SUCCEEDED") throw new Error(`Apify run ${runStatus}`);

    // 3. Fetch results
    let serpResults = await fetchDatasetItems(datasetId, apifyToken);

    // 4. Filter social/directories
    if (params.filterSocial) {
      serpResults = serpResults.filter(r => !isSocialOrDirectory(r.domain));
    }

    // 5. Deduplicate domains
    const seenDomains = new Set<string>();
    const uniqueResults = serpResults.filter(r => {
      if (!r.domain || seenDomains.has(r.domain)) return false;
      seenDomains.add(r.domain);
      return true;
    });

    // 6. Apply selectedDomains filter if provided
    const filteredResults = params.selectedDomains?.length
      ? uniqueResults.filter(r => params.selectedDomains!.includes(r.domain))
      : uniqueResults;

    const domains = filteredResults.map(r => r.domain);
    const serpMap = new Map(filteredResults.map(r => [r.domain, r]));

    await prisma.job.update({
      where: { id: jobId },
      data: { totalLeads: domains.length },
    });

    // 7. Enrich with Anymailfinder Bulk API
    const enrichResults = await bulkSearchDomains(
      domains,
      params.categories,
      anymailKey,
      async (_completed, total) => {
        await prisma.job.update({ where: { id: jobId }, data: { totalLeads: total } });
      }
    );

    // 8. Store results
    let emailsFound = 0;
    for (const result of enrichResults) {
      const serpData = serpMap.get(result.domain || "");
      const hasEmail = !!result.valid_email;
      if (hasEmail) emailsFound++;

      await prisma.leadResult.create({
        data: {
          jobId,
          companyName: serpData?.title || null,
          domain: result.domain || null,
          contactName: result.person_full_name || null,
          contactTitle: result.person_job_title || null,
          email: result.email || null,
          linkedinUrl: result.person_linkedin_url || null,
          source: JSON.stringify({
            url: serpData?.url,
            description: serpData?.description,
            position: serpData?.position,
            email_status: result.email_status,
          }),
        },
      });
    }

    await prisma.job.update({
      where: { id: jobId },
      data: { status: "complete", emailsFound, totalLeads: enrichResults.length },
    });

    // Sync to LeadVault
    await sinkLeadsToVault(
      enrichResults.map(r => {
        const serpData = serpMap.get(r.domain || "");
        return {
          domain: r.domain || null,
          companyName: serpData?.title || null,
          contactName: r.person_full_name || null,
          contactTitle: r.person_job_title || null,
          email: r.email || null,
          linkedinUrl: r.person_linkedin_url || null,
          serpTitle: serpData?.title || null,
          serpSnippet: serpData?.description || null,
          serpRank: serpData?.position ?? null,
          serpUrl: serpData?.url || null,
        };
      }),
      "serp",
      params.query,
      jobId,
    );
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    await prisma.job.update({
      where: { id: jobId },
      data: { status: "failed", error: message },
    });
  }
}

function sleep(ms: number) {
  return new Promise(r => setTimeout(r, ms));
}