The 100 Lines Challenge: High-Performance Data Ingestion Pipeline in PHP

<?php

declare(strict_types=1);

namespace App\Services\Ingestion;

use Illuminate\Support\LazyCollection;
use Illuminate\Support\Facades\DB;
use Generator;
use RuntimeException;

final class StreamCsvIngestionService
{
    public function import(string $filePath, int $chunkSize = 1000): int
    {
        if (!file_exists($filePath) || !is_readable($filePath)) {
            throw new RuntimeException("File unreadable: {$filePath}");
        }

        $totalProcessed = 0;
        LazyCollection::make(function () use ($filePath): Generator {
            $handle = fopen($filePath, 'rb');
            $header = fgetcsv($handle, 0, ',');
            while (($row = fgetcsv($handle, 0, ',')) !== false) {
                if (count($row) === count($header)) {
                    yield array_combine($header, $row);
                }
            }
            fclose($handle);
        })
        ->chunk($chunkSize)
        ->each(function ($chunk) use (&$totalProcessed): void {
            $records = $chunk->map(fn (array $row): array => [
                'account_number' => trim((string)$row['account_no']),
                'transaction_ref'=> trim((string)$row['txn_ref']),
                'amount'         => (float)$row['amount'],
                'currency'       => strtoupper(trim((string)$row['currency'])),
                'status'         => 'processed',
                'synced_at'      => now(),
                'created_at'     => now(),
                'updated_at'     => now(),
            ])->toArray();

            DB::table('financial_transactions')->upsert(
                $records,
                ['transaction_ref'],
                ['amount', 'status', 'synced_at', 'updated_at']
            );
            $totalProcessed += count($records);
        });

        return $totalProcessed;
    }
}