huginn/app/Services/CommonCrawl/CommonCrawlIndexService.php

<?php

namespace App\Services\CommonCrawl;

use App\Models\Backlink;
use App\Models\Site;
use App\Models\SitePage;
use App\Support\CommonCrawl\WarcRecordFetcher;
use App\Support\RootDomain;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Http;

/**
 * Talks to the Common Crawl CDX index, groups raw capture records into sites/pages, and
 * upserts them. Fetching a capture's actual page content (title/description/keywords/
 * body text) from its WARC file is delegated to WarcRecordFetcher.
 */
class CommonCrawlIndexService
{
    private const COLLINFO_URL = 'https://index.commoncrawl.org/collinfo.json';

    private const DATA_BASE_URL = 'https://data.commoncrawl.org/';

    public function __construct(private readonly WarcRecordFetcher $warcFetcher) {}

    public function resolveLatestCollection(): ?string
    {
        $response = Http::withUserAgent(WarcRecordFetcher::DEFAULT_USER_AGENT)
            ->timeout(30)
            ->retry(3, 2000)
            ->get(self::COLLINFO_URL);

        if (! $response->successful()) {
            return null;
        }

        return $response->json('0.id');
    }

    /**
     * Fetch one CDX index page. Returns an array with a 'status' of:
     *  - 'ok' — 'lines' holds the raw JSON-per-line capture records
     *  - 'end' — matchType=domain/prefix ranges are exhausted, stop paging
     *  - 'not_found' — no captures at all for this query
     *  - 'error' — 'httpStatus'/'body' describe what went wrong
     */
    public function fetchIndexPage(string $collection, string $url, string $matchType, int $limit, int $page, int $pageSize): array
    {
        $response = Http::withUserAgent(WarcRecordFetcher::DEFAULT_USER_AGENT)
            ->timeout(60)
            ->retry(3, 2000)
            ->get("https://index.commoncrawl.org/{$collection}-index", [
                'url' => $url,
                'output' => 'json',
                'matchType' => $matchType,
                'limit' => $limit,
                'page' => $page,
                'pageSize' => $pageSize,
            ]);

        if ($response->status() === 400) {
            return ['status' => 'end'];
        }

        if ($response->status() === 404) {
            return ['status' => 'not_found'];
        }

        if (! $response->successful()) {
            return ['status' => 'error', 'httpStatus' => $response->status(), 'body' => $response->body()];
        }

        return ['status' => 'ok', 'lines' => array_values(array_filter(explode("\n", trim($response->body()))))];
    }

    /**
     * Parse one CDX page's worth of lines and group into sites (by domain) and, within
     * each site, into pages (by SURT urlkey) — different forms of the same page
     * (http/https, www/non-www, different capture timestamps) collapse to one page row.
     *
     * Returns ['captureCount' => int, 'sites' => Collection<{domain, pages}>].
     */
    public function groupIntoSites(array $lines, string $collection): array
    {
        $captures = collect($lines)
            ->map(fn (string $line) => json_decode($line, true))
            ->filter(fn ($record) => is_array($record) && ! empty($record['urlkey']))
            ->map(fn (array $record) => $this->mapCapture($record, $collection));

        $sites = $captures
            ->groupBy('domain')
            ->map(fn (Collection $domainCaptures) => $domainCaptures
                ->groupBy('urlkey')
                ->map(fn (Collection $pageCaptures) => $pageCaptures->sort(fn ($a, $b) => $this->compareCaptures($a, $b))->first())
                ->values())
            ->map(fn (Collection $pages, string $domain) => ['domain' => $domain, 'pages' => $pages])
            ->values();

        return ['captureCount' => $captures->count(), 'sites' => $sites];
    }

    /**
     * Upsert one site (and its pages) from a group produced by groupIntoSites(). When
     * $withContent is set, fetches each distinct capture's HTML from its WARC file to
     * fill in title/description/keywords/content — one extra request per capture,
     * memoized in $contentCache across the whole run.
     *
     * Returns ['site' => Site, 'wasNewSite' => bool, 'pagesCreated' => int, 'pagesUpdated' => int].
     */
    public function saveSiteGroup(array $group, string $collection, bool $withContent, array &$contentCache): array
    {
        /** @var Collection $pages */
        $pages = $group['pages'];

        $site = Site::firstOrNew(['collection' => $collection, 'domain' => $group['domain']]);
        $wasNewSite = ! $site->exists;

        $homepage = $pages->sort(fn ($a, $b) => $this->compareCaptures($a, $b))->first();
        $site->url = $homepage['url'];
        $site->root_domain = RootDomain::of($group['domain']);

        if ($withContent) {
            $content = $this->fetchContent($homepage, $contentCache);

            if (! empty($content)) {
                $site->fill([
                    'title' => $content['title'],
                    'description' => $content['description'],
                    'keywords' => $content['keywords'],
                    'server' => $content['server'],
                    'powered_by' => $content['powered_by'],
                    'generator' => $content['generator'],
                ]);
            }
        }

        $site->save();

        $pagesCreated = 0;
        $pagesUpdated = 0;

        foreach ($pages as $page) {
            $attributes = $page;
            unset($attributes['collection'], $attributes['domain']);
            $attributes['urlkey_hash'] = sha1($page['urlkey']);

            $links = [];

            if ($withContent) {
                $content = $this->fetchContent($page, $contentCache);
                $links = $content['links'] ?? [];
                unset($content['links']);
                $attributes = array_merge($attributes, $content);
            }

            $result = SitePage::updateOrCreate(
                ['site_id' => $site->id, 'urlkey_hash' => $attributes['urlkey_hash']],
                $attributes,
            );

            $result->wasRecentlyCreated ? $pagesCreated++ : $pagesUpdated++;

            if ($withContent) {
                $this->saveBacklinks($result, $site, $links);
            }
        }

        return [
            'site' => $site,
            'wasNewSite' => $wasNewSite,
            'pagesCreated' => $pagesCreated,
            'pagesUpdated' => $pagesUpdated,
        ];
    }

    /**
     * Replace a page's recorded outbound links wholesale — simpler and more correct
     * than diffing against what was there before, since a page's actual link set can
     * change between crawls.
     */
    private function saveBacklinks(SitePage $sitePage, Site $site, array $links): void
    {
        $sitePage->links()->delete();

        if (empty($links)) {
            return;
        }

        $now = now();

        $rows = collect($links)
            ->unique('url')
            ->map(fn (array $link) => [
                'site_id' => $site->id,
                'site_page_id' => $sitePage->id,
                'target_url' => $link['url'],
                'target_url_hash' => sha1($link['url']),
                'target_domain' => $link['domain'],
                'anchor_text' => $link['anchor_text'],
                'rel' => $link['rel'],
                'created_at' => $now,
                'updated_at' => $now,
            ])
            ->values()
            ->all();

        Backlink::insert($rows);
    }

    /**
     * List every CDX shard file for a collection, straight from the bulk data storage
     * (data.commoncrawl.org) rather than the index.commoncrawl.org query API — this is
     * what lets bulk ingestion work even when that API is offline. Returns paths
     * relative to DATA_BASE_URL, or null if the paths file couldn't be fetched/parsed.
     */
    public function listCdxShardPaths(string $collection): ?array
    {
        $response = Http::withUserAgent(WarcRecordFetcher::DEFAULT_USER_AGENT)
            ->timeout(60)
            ->retry(3, 2000)
            ->get(self::DATA_BASE_URL."crawl-data/{$collection}/cc-index.paths.gz");

        if (! $response->successful()) {
            return null;
        }

        $decompressed = @gzdecode($response->body());

        if ($decompressed === false) {
            return null;
        }

        $paths = array_filter(explode("\n", trim($decompressed)));

        return array_values(array_filter(
            $paths,
            fn (string $path) => str_contains($path, '/indexes/cdx-') && str_ends_with($path, '.gz')
        ));
    }

    /**
     * Download and decompress one CDX shard file, returning its raw lines (each
     * "<urlkey> <timestamp> <json>", not yet parsed — see parseShardLine()). The whole
     * shard is pulled into memory at once, which can be sizeable (shards run up to a
     * couple hundred MB compressed); null on download/decompress failure.
     */
    public function fetchShardLines(string $path): ?array
    {
        $response = Http::withUserAgent(WarcRecordFetcher::DEFAULT_USER_AGENT)
            ->timeout(300)
            ->retry(2, 3000)
            ->get(self::DATA_BASE_URL.$path);

        if (! $response->successful()) {
            return null;
        }

        $decompressed = @gzdecode($response->body());

        if ($decompressed === false) {
            return null;
        }

        return array_values(array_filter(explode("\n", trim($decompressed))));
    }

    /**
     * A raw CDX shard line is "<urlkey> <timestamp> <json>", with urlkey/timestamp
     * outside the JSON blob — unlike the CDX API's output, which is one self-contained
     * JSON object per line. Re-shape it into that same self-contained form so it can
     * flow through groupIntoSites() unchanged. Returns null for malformed lines.
     */
    public function parseShardLine(string $line): ?string
    {
        if (! preg_match('/^(\S+) (\d{14}) (\{.*\})$/', $line, $matches)) {
            return null;
        }

        $record = json_decode($matches[3], true);

        if (! is_array($record)) {
            return null;
        }

        $record['urlkey'] = $matches[1];
        $record['timestamp'] = $matches[2];

        return json_encode($record);
    }

    private function fetchContent(array $page, array &$contentCache): array
    {
        $contentCache[$page['record_hash']] ??= $this->warcFetcher->fetchMetadata($page);

        return $contentCache[$page['record_hash']];
    }

    /**
     * Sort captures best-first: a live HTML homepage beats a live HTML subpage, which
     * beats any other 200, which beats everything else; ties broken by the most recent
     * capture timestamp. Used both to pick a site's homepage and to pick which capture
     * of a given page (across timestamps) to keep.
     */
    private function compareCaptures(array $a, array $b): int
    {
        $scoreDiff = $this->captureScore($b) <=> $this->captureScore($a);

        return $scoreDiff !== 0 ? $scoreDiff : strcmp($b['timestamp'], $a['timestamp']);
    }

    private function captureScore(array $attributes): int
    {
        $isHtml200 = ($attributes['status'] ?? null) === 200 && ($attributes['mime'] ?? null) === 'text/html';

        return match (true) {
            $isHtml200 && in_array($this->pathFromUrlkey($attributes['urlkey']), ['/', ''], true) => 3,
            $isHtml200 => 2,
            ($attributes['status'] ?? null) === 200 => 1,
            default => 0,
        };
    }

    private function mapCapture(array $record, string $collection): array
    {
        $filename = $record['filename'] ?? null;
        $offset = isset($record['offset']) ? (int) $record['offset'] : null;
        $status = isset($record['status']) && is_numeric($record['status']) ? (int) $record['status'] : null;

        return [
            'collection' => $collection,
            'urlkey' => $record['urlkey'],
            'domain' => $this->domainFromUrlkey($record['urlkey']),
            'timestamp' => $record['timestamp'] ?? '',
            'url' => $record['url'] ?? '',
            'mime' => $record['mime'] ?? null,
            'mime_detected' => $record['mime-detected'] ?? null,
            'status' => $status,
            'digest' => $record['digest'] ?? null,
            'length' => isset($record['length']) ? (int) $record['length'] : null,
            'offset' => $offset,
            'filename' => $filename,
            'languages' => $record['languages'] ?? null,
            'charset' => $record['charset'] ?? null,
            'redirect' => $record['redirect'] ?? null,
            'raw' => $record,
            'record_hash' => sha1($collection.'|'.$filename.'|'.$offset),
        ];
    }

    /**
     * The urlkey is a SURT key, e.g. "com,example)/path" — reversed, comma-separated
     * host labels (with a leading "www." already stripped by Common Crawl's
     * canonicalization) followed by the path after ")". Reversing the host labels back
     * gives the plain domain, which is what pages are grouped into sites by.
     */
    private function domainFromUrlkey(string $urlkey): string
    {
        $hostPart = strstr($urlkey, ')', true);
        $hostPart = $hostPart === false ? $urlkey : $hostPart;

        $labels = array_filter(explode(',', $hostPart), fn (string $label) => $label !== '');

        return implode('.', array_reverse($labels)) ?: $urlkey;
    }

    private function pathFromUrlkey(string $urlkey): string
    {
        $position = strpos($urlkey, ')');

        return $position === false ? '' : substr($urlkey, $position + 1);
    }
}