AFC-Demo/app/Jobs/SyncBackendDataPool.php

<?php

declare(strict_types=1);

namespace App\Jobs;

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\DB;
use Illuminate\Support\Facades\Log;

class SyncBackendDataPool implements ShouldQueue
{
    use Dispatchable;
    use InteractsWithQueue;
    use Queueable;
    use SerializesModels;

    public int $timeout = 3600; // 1 hour timeout

    public int $tries = 3;

    /**
     * Create a new job instance.
     */
    public function __construct(
        public bool $fullSync = true,
        public ?int $batchSize = 1000,
    ) {
    }

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        $startTime = now();

        Log::info('SyncBackendDataPool: Starting sync', [
            'full_sync' => $this->fullSync,
            'batch_size' => $this->batchSize,
        ]);

        try {
            if ($this->fullSync) {
                $this->performFullSync();
            } else {
                $this->performIncrementalSync();
            }

            $duration = now()->diffInSeconds($startTime);

            Log::info('SyncBackendDataPool: Sync completed successfully', [
                'duration_seconds' => $duration,
            ]);
        } catch (\Exception $e) {
            Log::error('SyncBackendDataPool: Sync failed', [
                'error' => $e->getMessage(),
                'trace' => $e->getTraceAsString(),
            ]);

            throw $e;
        }
    }

    /**
     * Perform full synchronization - truncates and rebuilds the data pool
     */
    private function performFullSync(): void
    {
        DB::transaction(function () {
            // Truncate existing data pool
            DB::table('backend_data_pool')->truncate();

            Log::info('SyncBackendDataPool: Truncated backend_data_pool');

            // Insert all data in batches
            $this->insertDataInBatches();
        });
    }

    /**
     * Perform incremental synchronization - only syncs new/updated records
     */
    private function performIncrementalSync(): void
    {
        DB::transaction(function () {
            // Get the last synced timestamp
            $lastSyncedAt = DB::table('backend_data_pool')
                ->max('synced_at');

            Log::info('SyncBackendDataPool: Incremental sync', [
                'last_synced_at' => $lastSyncedAt,
            ]);

            // Insert new records
            $this->insertDataInBatches($lastSyncedAt);
        });
    }

    /**
     * Insert data from backend schema into public.backend_data_pool in batches
     */
    private function insertDataInBatches(?string $lastSyncedAt = null): void
    {
        $totalInserted = 0;
        $offset = 0;

        while (true) {
            // Build the query
            $query = $this->buildSyncQuery($lastSyncedAt)
                ->limit($this->batchSize)
                ->offset($offset);

            $records = $query->get();

            if ($records->isEmpty()) {
                break;
            }

            // Transform records for insertion
            $dataToInsert = $records->map(function ($record) {
                return [
                    'transaction_id' => $record->transaction_id,
                    'corporate_entity' => $record->corporate_entity,
                    'corporate_counterparty' => $record->corporate_counterparty,
                    'tx_date' => $record->tx_date,
                    'tx_amount' => $record->tx_amount,
                    'tx_currency' => $record->tx_currency,
                    'tx_purpose' => $record->tx_purpose,
                    'tx_country_outgoing' => $record->tx_country_outgoing,
                    'tx_country_incoming' => $record->tx_country_incoming,
                    'source_file' => $record->source_file,
                    'raw_payload' => $record->raw_payload,
                    'status' => $record->status,
                    'created_at' => $record->created_at,
                    'last_modified_at' => $record->last_modified_at,
                    'risk_score' => $record->risk_score,
                    'risk_level' => $record->risk_level,
                    'risk_details_json' => $record->risk_details_json,
                    'prompt_id' => $record->prompt_id,
                    'output_key' => $record->output_key,
                    'content' => $record->content,
                    'run_id' => $record->run_id,
                    'synced_at' => now(),
                ];
            })->toArray();

            // Insert batch
            DB::table('backend_data_pool')->insert($dataToInsert);

            $totalInserted += count($dataToInsert);

            Log::info('SyncBackendDataPool: Batch inserted', [
                'batch_size' => count($dataToInsert),
                'total_inserted' => $totalInserted,
            ]);

            $offset += $this->batchSize;
        }

        Log::info('SyncBackendDataPool: All batches inserted', [
            'total_records' => $totalInserted,
        ]);
    }

    /**
     * Build the sync query that replicates the SQL:
     * SELECT * FROM backend.transactions t
     * LEFT JOIN backend.transaction_outputs tout ON t.id = tout.transaction_id
     * WHERE t.status = 'done'
     * ORDER BY transaction_id, prompt_id
     */
    private function buildSyncQuery(?string $lastSyncedAt = null): \Illuminate\Database\Query\Builder
    {
        $query = DB::connection('pgsql')
            ->table('backend.transactions as t')
            ->leftJoin('backend.transaction_outputs as tout', 't.id', '=', 'tout.transaction_id')
            ->select([
                't.id as transaction_id',
                't.corporate_entity',
                't.corporate_counterparty',
                't.tx_date',
                't.tx_amount',
                't.tx_currency',
                't.tx_purpose',
                't.tx_country_outgoing',
                't.tx_country_incoming',
                't.source_file',
                't.raw_payload',
                't.status',
                't.created_at',
                't.last_modified_at',
                't.risk_score',
                't.risk_level',
                't.risk_details_json',
                'tout.prompt_id',
                'tout.output_key',
                'tout.content',
                'tout.run_id',
            ])
            ->where('t.status', '=', 'done')
            ->orderBy('t.id')
            ->orderBy('tout.prompt_id');

        // For incremental sync, only get records modified after last sync
        if ($lastSyncedAt !== null) {
            $query->where('t.last_modified_at', '>', $lastSyncedAt);
        }

        return $query;
    }

    /**
     * Get statistics about the sync
     */
    public function getStats(): array
    {
        return [
            'backend_transactions_done' => DB::connection('pgsql')
                ->table('backend.transactions')
                ->where('status', 'done')
                ->count(),
            'backend_transaction_outputs' => DB::connection('pgsql')
                ->table('backend.transaction_outputs')
                ->count(),
            'backend_data_pool_records' => DB::table('backend_data_pool')->count(),
            'last_synced_at' => DB::table('backend_data_pool')->max('synced_at'),
        ];
    }
}