webstatement/app/Jobs/ProcessAccountDataJob.php

<?php

namespace Modules\Webstatement\Jobs;

use Exception;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Storage;
use Modules\Webstatement\Models\Account;
use Modules\Webstatement\Models\AccountBalance;

class ProcessAccountDataJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    private const CSV_DELIMITER      = '~';
    private const MAX_EXECUTION_TIME = 86400; // 24 hours in seconds
    private const FILENAME           = 'ST.ACCOUNT.csv';
    private const DISK_NAME          = 'staging';
    private const CHUNK_SIZE         = 1000; // Process data in chunks to reduce memory usage

    private string $period = '';
    private int    $processedCount = 0;
    private int    $errorCount     = 0;

    private $balanceData = [];
    private $accountBatch = [];
    private $balanceBatch = [];

    /**
     * Create a new job instance.
     */
    public function __construct(string $period = '')
    {
        $this->period = $period;
    }

    /**
     * Execute the job.
     */
    public function handle()
    : void
    {
        try {
            $this->initializeJob();

            if ($this->period === '') {
                Log::warning('No period provided for account data processing');
                return;
            }

            $this->processPeriod();
            $this->logJobCompletion();
        } catch (Exception $e) {
            Log::error('Error in ProcessAccountDataJob: ' . $e->getMessage());
            throw $e;
        }
    }

    private function initializeJob()
    : void
    {
        set_time_limit(self::MAX_EXECUTION_TIME);
        $this->processedCount = 0;
        $this->errorCount     = 0;
        $this->accountBatch   = [];
        $this->balanceBatch   = [];
    }

    private function processPeriod()
    : void
    {
        $disk     = Storage::disk(self::DISK_NAME);
        $filename = "{$this->period}." . self::FILENAME;
        $filePath = "{$this->period}/$filename";

        if (!$this->validateFile($disk, $filePath)) {
            return;
        }

        $tempFilePath = $this->createTemporaryFile($disk, $filePath, $filename);
        $this->processFile($tempFilePath, $filePath);
        $this->cleanup($tempFilePath);
    }

    private function validateFile($disk, string $filePath)
    : bool
    {
        Log::info("Processing account file: $filePath");

        if (!$disk->exists($filePath)) {
            Log::warning("File not found: $filePath");
            return false;
        }

        return true;
    }

    private function createTemporaryFile($disk, string $filePath, string $filename)
    : string
    {
        $tempFilePath = storage_path("app/temp_$filename");
        file_put_contents($tempFilePath, $disk->get($filePath));
        return $tempFilePath;
    }

    private function processFile(string $tempFilePath, string $filePath)
    : void
    {
        $handle = fopen($tempFilePath, "r");
        if ($handle === false) {
            Log::error("Unable to open file: $filePath");
            return;
        }

        $headers = (new Account())->getFillable();
        Log::info('Headers: ' . implode(", ", $headers));
        $rowCount = 0;
        $chunkCount = 0;

        while (($row = fgetcsv($handle, 0, self::CSV_DELIMITER)) !== false) {
            $rowCount++;
            $this->processRow($row, $headers, $rowCount, $filePath);

            // Process in chunks to avoid memory issues
            if (count($this->accountBatch) >= self::CHUNK_SIZE) {
                $this->saveBatch();
                $chunkCount++;
                Log::info("Processed chunk $chunkCount ({$this->processedCount} records so far)");
            }
        }

        // Process any remaining records
        if (!empty($this->accountBatch) || !empty($this->balanceBatch)) {
            $this->saveBatch();
        }

        fclose($handle);
        Log::info("Completed processing $filePath. Processed {$this->processedCount} records with {$this->errorCount} errors.");
    }

    private function processRow(array $row, array $headers, int $rowCount, string $filePath)
    : void
    {
        if (count($headers) !== count($row)) {
            Log::warning("Row $rowCount in $filePath has incorrect column count. Expected: " .
                count($headers) . ", Got: " . count($row));
            return;
        }

        $data = array_combine($headers, $row);
        $this->normalizeData($data);
        $this->addToBatch($data, $rowCount, $filePath);
    }

    private function normalizeData(array &$data)
    : void
    {
        // Check if start_year_bal is empty and set it to 0 if so
        if (empty($data['start_year_bal']) || $data['start_year_bal'] == "" || $data['start_year_bal'] == null) {
            $data['start_year_bal'] = 0;
        }

        if (empty($data['closure_date']) || $data['closure_date'] == "" || $data['closure_date'] == null) {
            $data['closure_date'] = null;
        }

        // Store balance data separately before removing from Account data
        $this->balanceData = [
            'open_actual_bal'  => empty($data['open_actual_bal']) ? 0 : $data['open_actual_bal'],
            'open_cleared_bal' => empty($data['open_cleared_bal']) ? 0 : $data['open_cleared_bal'],
        ];

        // Remove balance fields from Account data
        unset($data['open_actual_bal']);
        unset($data['open_cleared_bal']);
    }

    /**
     * Add record to batch instead of saving immediately
     */
    private function addToBatch(array $data, int $rowCount, string $filePath)
    : void
    {
        try {
            if ($data['account_number'] !== 'account_number') {
                // Add timestamp fields
                $now = now();
                $data['created_at'] = $now;
                $data['updated_at'] = $now;

                // Add to account batch
                $this->accountBatch[] = $data;

                // Add to balance batch
                if (isset($this->balanceData['open_actual_bal']) || isset($this->balanceData['open_cleared_bal'])) {
                    $this->balanceBatch[] = [
                        'account_number'  => $data['account_number'],
                        'period'          => $this->period,
                        'actual_balance'  => $this->balanceData['open_actual_bal'],
                        'cleared_balance' => $this->balanceData['open_cleared_bal'],
                        'created_at'      => $now,
                        'updated_at'      => $now
                    ];
                }

                $this->processedCount++;
            }
        } catch (Exception $e) {
            $this->errorCount++;
            Log::error("Error processing Account at row $rowCount in $filePath: " . $e->getMessage());
        }
    }

    /**
     * Save batched records to the database
     */
    private function saveBatch()
    : void
    {
        try {
            if (!empty($this->accountBatch)) {
                // Bulk insert/update accounts
                Account::upsert(
                    $this->accountBatch,
                    ['account_number'], // Unique key
                    array_diff((new Account())->getFillable(), ['account_number']) // Update columns
                );

                // Reset account batch after processing
                $this->accountBatch = [];
            }

            if (!empty($this->balanceBatch)) {
                // Bulk insert/update account balances
                AccountBalance::upsert(
                    $this->balanceBatch,
                    ['account_number', 'period'], // Composite unique key
                    ['actual_balance', 'cleared_balance', 'updated_at'] // Update columns
                );

                // Reset balance batch after processing
                $this->balanceBatch = [];
            }
        } catch (Exception $e) {
            Log::error("Error in saveBatch: " . $e->getMessage());
            $this->errorCount += count($this->accountBatch);
        }
    }

    private function cleanup(string $tempFilePath)
    : void
    {
        if (file_exists($tempFilePath)) {
            unlink($tempFilePath);
        }
    }

    private function logJobCompletion()
    : void
    {
        Log::info("Account data processing completed. " .
            "Total processed: {$this->processedCount}, Total errors: {$this->errorCount}");
    }
}