Add bulk CDX shard ingestion command as an API-independent fallback

crawlindex:bulk-fetch walks Common Crawl's CDX shard files directly on
data.commoncrawl.org (via cc-index.paths.gz), instead of the
index.commoncrawl.org query API — usable for bulk backfill/refresh or
as a fallback when that API is down. Shares the group/save pipeline
with crawlindex:fetch via a new SavesCommonCrawlCaptures trait.

Also fixes a real bug this surfaced under load: site_pages.urlkey was
a varchar(255), which is too narrow for real-world captured URLs
(long query strings on spam/malformed pages overflow it). Widened to
text and switched the uniqueness constraint to a sha1 hash column,
since MySQL can't fully index a TEXT column.

app/Services/CommonCrawl/CommonCrawlIndexService.php

@@ -17,6 +17,8 @@ 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
@@ -134,13 +136,14 @@ class CommonCrawlIndexService
         foreach ($pages as $page) {
             $attributes = $page;
             unset($attributes['collection'], $attributes['domain']);
+            $attributes['urlkey_hash'] = sha1($page['urlkey']);
 
             if ($withContent) {
                 $attributes = array_merge($attributes, $this->fetchContent($page, $contentCache));
             }
 
             $result = SitePage::updateOrCreate(
-                ['site_id' => $site->id, 'urlkey' => $page['urlkey']],
+                ['site_id' => $site->id, 'urlkey_hash' => $attributes['urlkey_hash']],
                 $attributes,
             );
 
@@ -155,6 +158,87 @@ class CommonCrawlIndexService
         ];
     }
 
+    /**
+     * 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);