Migration & Datapool created with job and scheduler backend->frontend(public) schema
This commit is contained in:
@@ -0,0 +1,228 @@
|
||||
<?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,
|
||||
'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',
|
||||
'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'),
|
||||
];
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user