294 lines
10 KiB
PHP
294 lines
10 KiB
PHP
<?php
|
|
|
|
namespace Modules\EcommerceFinance\Repositories;
|
|
|
|
use Illuminate\Database\ConnectionInterface;
|
|
use Illuminate\Database\DatabaseManager;
|
|
use Illuminate\Support\Facades\Schema;
|
|
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
|
|
use Modules\EcommerceFinance\Services\WorkOrderWaybillLookup;
|
|
use RuntimeException;
|
|
|
|
class GenericBusinessMirrorRepository
|
|
{
|
|
private readonly ConnectionInterface $connection;
|
|
private ?bool $supportsExactCount = null;
|
|
|
|
public function __construct(DatabaseManager $database)
|
|
{
|
|
$this->connection = $database->connection(
|
|
(string) config('upstream_mirrors.connection', 'business_mirror')
|
|
);
|
|
}
|
|
|
|
public function connection(): ConnectionInterface
|
|
{
|
|
return $this->connection;
|
|
}
|
|
|
|
public function ready(BusinessMirrorDataset $dataset): bool
|
|
{
|
|
$schema = Schema::connection($this->connection->getName());
|
|
|
|
if (! $schema->hasTable('business_mirror_table_states')
|
|
|| ! $schema->hasTable($dataset->targetTable())) {
|
|
return false;
|
|
}
|
|
|
|
$state = $this->state($dataset);
|
|
|
|
return is_array($state) && ($state['initial_completed_at'] ?? null) !== null;
|
|
}
|
|
|
|
/** @return array<string,mixed>|null */
|
|
public function state(BusinessMirrorDataset $dataset): ?array
|
|
{
|
|
if (! Schema::connection($this->connection->getName())->hasTable('business_mirror_table_states')) {
|
|
return null;
|
|
}
|
|
|
|
$row = $this->connection->table('business_mirror_table_states')
|
|
->where('dataset_key', $dataset->key())
|
|
->first();
|
|
|
|
if ($row === null) {
|
|
return null;
|
|
}
|
|
|
|
$state = (array) $row;
|
|
$state['cursor'] = $this->decode($state['cursor'] ?? null);
|
|
$state['high_watermark'] = $this->decode($state['high_watermark'] ?? null);
|
|
|
|
return $state;
|
|
}
|
|
|
|
public function startRun(
|
|
BusinessMirrorDataset $dataset,
|
|
string $mode,
|
|
string $generation,
|
|
?string $windowStart,
|
|
?string $windowEnd,
|
|
bool $resume,
|
|
): array {
|
|
$now = now();
|
|
$values = [
|
|
'source_key' => $dataset->sourceKey(),
|
|
'source_object' => $dataset->sourceSchema() . '.' . $dataset->sourceObject(),
|
|
'target_table' => $dataset->targetTable(),
|
|
'status' => 'running',
|
|
'run_mode' => $mode,
|
|
'generation' => $generation,
|
|
'last_attempted_at' => $now,
|
|
'last_error_code' => null,
|
|
'updated_at' => $now,
|
|
];
|
|
|
|
if (! $resume) {
|
|
$values += [
|
|
'cursor' => null,
|
|
'window_started_at' => $windowStart,
|
|
'window_ends_at' => $windowEnd,
|
|
'last_upserted_rows' => 0,
|
|
];
|
|
}
|
|
|
|
$query = $this->connection->table('business_mirror_table_states')
|
|
->where('dataset_key', $dataset->key());
|
|
|
|
$existing = $query->first();
|
|
if ($this->supportsExactCount()) {
|
|
$values['mirror_row_count_is_exact'] = $mode === 'incremental'
|
|
&& $existing !== null && $existing->initial_completed_at !== null
|
|
&& ctype_digit((string) ($existing->mirror_row_count ?? ''))
|
|
&& ($existing->status === 'succeeded' || (bool) ($existing->mirror_row_count_is_exact ?? false));
|
|
}
|
|
if ($existing !== null) {
|
|
$query->update($values);
|
|
} else {
|
|
$this->connection->table('business_mirror_table_states')->insert(
|
|
['dataset_key' => $dataset->key(), 'created_at' => $now] + $values,
|
|
);
|
|
}
|
|
|
|
return $this->state($dataset) ?? [];
|
|
}
|
|
|
|
public function storeBatch(
|
|
BusinessMirrorDataset $dataset,
|
|
array $columns,
|
|
array $rows,
|
|
array $cursor,
|
|
string $generation,
|
|
int $cumulativeRows,
|
|
): void {
|
|
$now = now();
|
|
$columnNames = array_map(fn (array $column): string => (string) $column['column_name'], $columns);
|
|
$mapped = [];
|
|
|
|
foreach ($rows as $row) {
|
|
$values = [];
|
|
foreach ($columnNames as $columnName) {
|
|
$values[$columnName] = $this->value($row, $columnName);
|
|
}
|
|
|
|
$identity = array_map(
|
|
fn (string $column): string => $this->identityValue($this->value($row, $column)),
|
|
$dataset->keyColumns(),
|
|
);
|
|
|
|
$mapped[] = $values + [
|
|
'_mirror_key' => hash('sha256', implode("\0", $identity)),
|
|
'_mirror_generation' => $generation,
|
|
'_mirror_synced_at' => $now,
|
|
];
|
|
}
|
|
|
|
$this->connection->transaction(function () use (
|
|
$dataset, $columnNames, $mapped, $cursor, $cumulativeRows, $now
|
|
): void {
|
|
$exactCount = null;
|
|
if ($this->supportsExactCount()) {
|
|
$state = $this->connection->table('business_mirror_table_states')
|
|
->where('dataset_key', $dataset->key())->lockForUpdate()->first();
|
|
if ($state && $state->run_mode === 'incremental' && $state->mirror_row_count_is_exact) {
|
|
$keys = array_values(array_unique(array_column($mapped, '_mirror_key')));
|
|
$existing = $this->connection->table($dataset->targetTable())->whereIn('_mirror_key', $keys)->count();
|
|
$exactCount = (int) $state->mirror_row_count + count($keys) - $existing;
|
|
}
|
|
}
|
|
$updates = [...$columnNames, '_mirror_generation', '_mirror_synced_at'];
|
|
$chunkSize = max(1, intdiv(50000, max(count($updates) + 1, 1)));
|
|
|
|
foreach (array_chunk($mapped, $chunkSize) as $chunk) {
|
|
$this->connection->table($dataset->targetTable())->upsert($chunk, ['_mirror_key'], $updates);
|
|
}
|
|
if ($dataset->targetTable() === WorkOrderWaybillLookup::SOURCE) {
|
|
app(WorkOrderWaybillLookup::class)->storeBatch($mapped);
|
|
}
|
|
|
|
$this->connection->table('business_mirror_table_states')
|
|
->where('dataset_key', $dataset->key())
|
|
->update([
|
|
'cursor' => json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION),
|
|
'last_upserted_rows' => $cumulativeRows,
|
|
'updated_at' => $now,
|
|
] + ($exactCount === null ? [] : ['mirror_row_count' => $exactCount]));
|
|
});
|
|
}
|
|
|
|
public function complete(
|
|
BusinessMirrorDataset $dataset,
|
|
string $mode,
|
|
string $generation,
|
|
array $highWatermark,
|
|
int $rows,
|
|
): void {
|
|
$now = now();
|
|
|
|
$this->connection->transaction(function () use (
|
|
$dataset, $mode, $generation, $highWatermark, $rows, $now
|
|
): void {
|
|
if ($mode === 'full') {
|
|
$existingRows = $this->connection->table($dataset->targetTable())->count();
|
|
|
|
if ($rows === 0 && $existingRows > 0) {
|
|
throw new RuntimeException('business_mirror_unexpected_empty_source');
|
|
}
|
|
|
|
$this->connection->table($dataset->targetTable())
|
|
->where(function ($query) use ($generation): void {
|
|
$query->whereNull('_mirror_generation')
|
|
->orWhere('_mirror_generation', '!=', $generation);
|
|
})
|
|
->delete();
|
|
if ($dataset->targetTable() === WorkOrderWaybillLookup::SOURCE) {
|
|
app(WorkOrderWaybillLookup::class)->prune();
|
|
}
|
|
}
|
|
|
|
$countState = $this->supportsExactCount()
|
|
? $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset->key())->lockForUpdate()->first()
|
|
: null;
|
|
$rowCount = $mode === 'incremental' && $countState && $countState->mirror_row_count_is_exact
|
|
? (int) $countState->mirror_row_count
|
|
: $this->connection->table($dataset->targetTable())->count();
|
|
$values = [
|
|
'status' => 'succeeded',
|
|
'cursor' => null,
|
|
'high_watermark' => json_encode($highWatermark, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION),
|
|
'last_succeeded_at' => $now,
|
|
'last_upserted_rows' => $rows,
|
|
'mirror_row_count' => $rowCount,
|
|
'last_error_code' => null,
|
|
'updated_at' => $now,
|
|
];
|
|
|
|
if ($mode === 'full') {
|
|
$values['initial_completed_at'] = $now;
|
|
$values['last_full_succeeded_at'] = $now;
|
|
}
|
|
if ($this->supportsExactCount()) $values['mirror_row_count_is_exact'] = true;
|
|
|
|
$this->connection->table('business_mirror_table_states')
|
|
->where('dataset_key', $dataset->key())
|
|
->update($values);
|
|
});
|
|
}
|
|
|
|
public function fail(BusinessMirrorDataset $dataset, string $errorCode): void
|
|
{
|
|
$this->connection->table('business_mirror_table_states')
|
|
->where('dataset_key', $dataset->key())
|
|
->update([
|
|
'status' => 'failed',
|
|
'last_error_code' => mb_substr($errorCode, 0, 120),
|
|
'updated_at' => now(),
|
|
]);
|
|
}
|
|
|
|
private function supportsExactCount(): bool
|
|
{
|
|
return $this->supportsExactCount ??= Schema::connection($this->connection->getName())
|
|
->hasColumn('business_mirror_table_states', 'mirror_row_count_is_exact');
|
|
}
|
|
|
|
private function decode(mixed $value): ?array
|
|
{
|
|
if (is_array($value)) {
|
|
return $value;
|
|
}
|
|
|
|
if (! is_string($value) || $value === '') {
|
|
return null;
|
|
}
|
|
|
|
$decoded = json_decode($value, true);
|
|
|
|
return is_array($decoded) ? $decoded : null;
|
|
}
|
|
|
|
private function value(array $row, string $name): mixed
|
|
{
|
|
foreach ($row as $key => $value) {
|
|
if (strcasecmp((string) $key, $name) === 0) {
|
|
return $value;
|
|
}
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private function identityValue(mixed $value): string
|
|
{
|
|
if ($value === null) {
|
|
return '<NULL>';
|
|
}
|
|
|
|
if (is_bool($value)) {
|
|
return $value ? '1' : '0';
|
|
}
|
|
|
|
return (string) $value;
|
|
}
|
|
}
|