webstatement/app/Jobs/ProcessTellerDataJob.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\Teller;

    class ProcessTellerDataJob 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.TELLER.csv';
        private const DISK_NAME          = 'sftpStatement';
        private const CHUNK_SIZE         = 1000; // Process data in chunks to reduce memory usage
        private const HEADER_MAP         = [
            'id'                => 'id_teller',
            'account_1'         => 'account_1',
            'currency_1'        => 'currency_1',
            'amount_local_1'    => 'amount_local_1',
            'value_date_1'      => 'value_date_1',
            'account_2'         => 'account_2',
            'new_cust_bal'      => 'new_cust_bal',
            'term_type'         => 'term_type',
            'term_id'           => 'term_id',
            'trans_reff'        => 'trans_reff',
            'card_no'           => 'card_no',
            'recipt_no'         => 'recipt_no',
            'transaction_code'  => 'transaction_code',
            'date_time'         => 'date_time',
            'record_status'     => 'record_status',
            'amount_local_2'    => 'amount_local_2',
            'co_code'           => 'co_code',
            'narrative_1'       => 'narrative_1',
            'wic_flag'          => 'wic_flag',
            'wic_cust_type'     => 'wic_cust_type',
            'wic_full_name'     => 'wic_full_name',
            'wic_alias_name'    => 'wic_alias_name',
            'wic_acct_no'       => 'wic_acct_no',
            'wic_id_type'       => 'wic_id_type',
            'wic_id_no'         => 'wic_id_no',
            'wic_npwp'          => 'wic_npwp',
            'wic_nationality'   => 'wic_nationality',
            'wic_ind_birthpl'   => 'wic_ind_birthpl',
            'wic_ind_birthdt'   => 'wic_ind_birthdt',
            'wic_address_id'    => 'wic_address_id',
            'wic_address_cur'   => 'wic_address_cur',
            'wic_city'          => 'wic_city',
            'wic_province'      => 'wic_province',
            'wic_post_code'     => 'wic_post_code',
            'wic_phone'         => 'wic_phone',
            'wic_gender'        => 'wic_gender',
            'wic_marital_sts'   => 'wic_marital_sts',
            'wic_occptn'        => 'wic_occptn',
            'wic_occptn_dur'    => 'wic_occptn_dur',
            'wic_income_avg'    => 'wic_income_avg',
            'wic_cor_name'      => 'wic_cor_name',
            'wic_cor_address'   => 'wic_cor_address',
            'wic_cor_phone'     => 'wic_cor_phone',
            'wic_cor_lgl_typ'   => 'wic_cor_lgl_typ',
            'wic_cor_lic_no'    => 'wic_cor_lic_no',
            'wic_cor_birthpl'   => 'wic_cor_birthpl',
            'wic_cor_birthdt'   => 'wic_cor_birthdt',
            'wic_cor_rel'       => 'wic_cor_rel',
            'wic_party_rel'     => 'wic_party_rel',
            'wic_amount'        => 'wic_amount',
            'wic_amount_type'   => 'wic_amount_type',
            'wic_amtbnk_name'   => 'wic_amtbnk_name',
            'wic_amtbnk_cunm'   => 'wic_amtbnk_cunm',
            'wic_fund_source'   => 'wic_fund_source',
            'wic_fund_use'      => 'wic_fund_use',
            'dr_cr_marker'      => 'dr_cr_marker',
            'charge_code'       => 'charge_code',
            'chrg_amt_local'    => 'chrg_amt_local',
            'charge_category'   => 'charge_category',
            'charge_account'    => 'charge_account',
            'amount_fcy_1'      => 'amount_fcy_1',
            'rate_1'            => 'rate_1',
            'deal_rate'         => 'deal_rate',
            'l_wic_id'          => 'l_wic_id',
            'account_1_co_code' => 'account_1_co_code',
            'account_2_co_code' => 'account_2_co_code',
            'l_charge_amt'      => 'l_charge_amt',
            'bl_cust_no'        => 'bl_cust_no',
            'stmt_no'           => 'stmt_no',
            'bil_customer'      => 'bil_customer',
            'value_date_2'      => 'value_date_2',
            'cheq_type'         => 'cheq_type',
            'cheque_number'     => 'cheque_number',
            'inputter'          => 'inputter',
            'authoriser'        => 'authoriser',
            'bil_product'       => 'bil_product',
            'fx_document'       => 'fx_document',
            'fx_purpose'        => 'fx_purpose',
            'narrative_2'       => 'narrative_2',
            'customer_2'        => 'customer_2',
            'l_sms_1'           => 'l_sms_1',
            'l_phone_1'         => 'l_phone_1',
            'kyc_incom_rng'     => 'kyc_incom_rng',
            'wic_rt'            => 'wic_rt',
            'wic_rw'            => 'wic_rw',
            'ktp_kelurahan'     => 'ktp_kelurahan',
            'ktp_kecamatan'     => 'ktp_kecamatan',
            'ktp_provinsi'      => 'ktp_provinsi',
            'wic_jenis_kelam'   => 'wic_jenis_kelam',
            'kyc_sumber_dana'   => 'kyc_sumber_dana',
            'l_manual_risk'     => 'l_manual_risk',
            'l_slip_no'         => 'l_slip_no',
            'dest_bank_id'      => 'dest_bank_id',
            'dr_narrative'      => 'dr_narrative',
            'inter_type'        => 'inter_type',
            'l_va_number'       => 'l_va_number',
            'inter_bank_id'     => 'inter_bank_id',
            'term_narr'         => 'term_narr',
            'currency_2'        => 'currency_2',
            'amount_fcy_2'      => 'amount_fcy_2',
            'rate_2'            => 'rate_2',
            'customer_1'        => 'customer_1',
            'last_version'      => 'last_version',
            'dealer_desk'      => 'dealer_desk',
        ];

        private string $period = '';
        private int    $processedCount = 0;
        private int    $errorCount     = 0;
        private array  $tellerBatch    = [];

        /**
         * 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 teller data processing');
                    return;
                }

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

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

        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 teller 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;
            }

            // Get the headers from the first row
            $headerRow = fgetcsv($handle, 0, self::CSV_DELIMITER);
            if (!$headerRow) {
                fclose($handle);
                return;
            }

            $rowCount = 0;
            $chunkCount = 0;

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

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

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

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

        private function processRow(array $headerRow, array $row, int $rowCount, string $filePath)
        : void
        {
            // Skip if row doesn't have enough columns
            if (count($headerRow) !== count($row)) {
                Log::warning("Row $rowCount in $filePath has incorrect column count. Expected: " .
                    count($headerRow) . ", Got: " . count($row));
                $this->errorCount++;
                return;
            }

            // Combine the header row with the data row
            $rawData = array_combine($headerRow, $row);

            // Map the raw data to our model fields
            $data = [];
            foreach (self::HEADER_MAP as $csvField => $modelField) {
                $data[$modelField] = $rawData[$csvField] ?? null;
            }

            // Skip header row if it was included in the data
            if ($data['id_teller'] === 'id') {
                return;
            }

            try {
                // Add timestamps
                $now = now();
                $data['created_at'] = $now;
                $data['updated_at'] = $now;

                // Add to batch for bulk processing
                $this->tellerBatch[] = $data;
                $this->processedCount++;
            } catch (Exception $e) {
                Log::error("Error processing Teller at row $rowCount in $filePath: " . $e->getMessage());
                $this->errorCount++;
            }
        }

        private function saveBatch(): void
        {
            try {
                if (!empty($this->tellerBatch)) {
                    // Bulk insert/update teller records
                    Teller::upsert(
                        $this->tellerBatch,
                        ['id_teller'], // Unique key
                        array_diff(array_values(self::HEADER_MAP), ['id_teller']) // Update columns
                    );

                    // Reset batch after processing
                    $this->tellerBatch = [];
                }
            } catch (Exception $e) {
                Log::error("Error in saveBatch: " . $e->getMessage());
                $this->errorCount += count($this->tellerBatch);
                // Reset batch even if there's an error to prevent reprocessing the same failed records
                $this->tellerBatch = [];
            }
        }

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

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