$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'), ]; } }