Function processGcsFile

server/src/routes/catalog-api.ts:297–327

(
        prefix: string,
        processor: (lines: string[]) => Promise<number>
      )

      // Iterate BigQuery EXPORT DATA shards (prefix-000000000000.csv, -000000000001.csv, ...)
      async function processGcsFile(
        prefix: string,
        processor: (lines: string[]) => Promise<number>
      ): Promise<{ rows: number; time_ms: number }> {
        const start = Date.now();
        let totalRows = 0;
        let shard = 0;

        while (true) {
          const shardId = String(shard).padStart(12, '0');
          const url = `${GCS_BASE}/${prefix}-${shardId}.csv`;
          logger.info(`Fetching ${url}`);

          const response = await fetch(url);
          if (response.status === 404) break;
          if (!response.ok) {
            throw new Error(`GCS fetch failed for ${url}: ${response.status} ${response.statusText}`);
          }

          totalRows += await processGcsShard(response, url, prefix, shard, processor);
          progress({ step: prefix, shard, rows: totalRows, status: 'shard_done' });
          logger.info(`  ${prefix}: shard ${shard} done (${totalRows.toLocaleString()} rows so far)`);
          shard++;
        }

        if (shard === 0) {
          throw new Error(`No shards found for ${prefix} at ${GCS_BASE}`);
        }

        return { rows: totalRows, time_ms: Date.now() - start };
      }

      // 1. Ad infra classifications (bulk insert)
      results.ad_infra = await processGcsFile('ad-infra', async (lines) => {