1337 lines
48 KiB
PHP
1337 lines
48 KiB
PHP
<?php
|
|
|
|
namespace Modules\EcommerceFinance\Services;
|
|
|
|
use Illuminate\Cache\CacheManager;
|
|
use Illuminate\Contracts\Cache\Repository;
|
|
use Illuminate\Database\ConnectionInterface;
|
|
use Illuminate\Database\DatabaseManager;
|
|
use Illuminate\Support\Carbon;
|
|
use Illuminate\Support\Facades\Log;
|
|
use Illuminate\Support\Facades\Schema;
|
|
use Illuminate\Support\Str;
|
|
use Modules\EcommerceFinance\Repositories\ThirdPartyWarehouseStorageUpstreamRepository;
|
|
use RuntimeException;
|
|
use Throwable;
|
|
|
|
class ThirdPartyWarehouseStorageSynchronizer
|
|
{
|
|
public const DATASET_KEY = 'jiansun.third-party-warehouse-storage';
|
|
|
|
public const LOCK_KEY = 'business-mirror:sync:jiansun.third-party-warehouse-storage';
|
|
|
|
private const STATE_TABLE = 'third_party_warehouse_sync_states';
|
|
|
|
private const RELEASE_TABLE = 'third_party_warehouse_generation_releases';
|
|
|
|
private const HEADER_TABLE = 'third_party_warehouse_inout_headers';
|
|
|
|
private const ITEM_TABLE = 'third_party_warehouse_inout_items';
|
|
|
|
private const INVENTORY_TABLE = 'third_party_warehouse_inventory_snapshots';
|
|
|
|
private const SUBWAREHOUSE_INVENTORY_TABLE = 'third_party_warehouse_inventory_subwarehouse_snapshots';
|
|
|
|
/** @var array<int,string> */
|
|
private const SUBWAREHOUSE_QUANTITY_FIELDS = [
|
|
1 => 'qty', 2 => 'return_qty', 3 => 'in_qty', 4 => 'defective_qty',
|
|
6 => 'customize_qty_1', 7 => 'customize_qty_2', 8 => 'customize_qty_3',
|
|
];
|
|
|
|
private readonly ConnectionInterface $connection;
|
|
|
|
private readonly Repository $cache;
|
|
|
|
public function __construct(
|
|
private readonly ThirdPartyWarehouseStorageUpstreamRepository $upstream,
|
|
DatabaseManager $databases,
|
|
CacheManager $cacheManager,
|
|
) {
|
|
$this->connection = $databases->connection(
|
|
(string) config('upstream_mirrors.connection', 'business_mirror'),
|
|
);
|
|
$this->cache = $cacheManager->store(
|
|
(string) config('upstreams.snapshots.cache_store', 'upstream_shared'),
|
|
);
|
|
}
|
|
|
|
/**
|
|
* @return array{status:string,mode:string,phase:string,batches:int,headers:int,items:int,inventory:int}
|
|
*/
|
|
public function sync(bool $full = false, ?int $maxBatches = null, ?int $scanSpan = null): array
|
|
{
|
|
$lock = $this->cache->lock(self::LOCK_KEY, max((int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.sync_lock_seconds',
|
|
21600,
|
|
), 300));
|
|
|
|
if (! $lock->get()) {
|
|
return $this->result(
|
|
'already_running',
|
|
$full ? 'full' : 'incremental',
|
|
'locked',
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
);
|
|
}
|
|
|
|
try {
|
|
return $this->syncUnlocked($full, $maxBatches, $scanSpan);
|
|
} finally {
|
|
$lock->release();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @return array{status:string,mode:string,phase:string,batches:int,headers:int,items:int,inventory:int}
|
|
*/
|
|
private function syncUnlocked(bool $full, ?int $maxBatches, ?int $scanSpan): array
|
|
{
|
|
$this->assertSchemaReady();
|
|
|
|
$state = $this->state();
|
|
$resumeFull = $this->resumable($state, 'full');
|
|
if (! $full && ! $this->ready($state) && ! $resumeFull) {
|
|
return $this->result('requires_full_sync', 'incremental', 'headers', 0, 0, 0, 0);
|
|
}
|
|
|
|
$mode = $full || $resumeFull ? 'full' : 'incremental';
|
|
$resume = $this->resumable($state, $mode);
|
|
$scanSpan = min(max(
|
|
$scanSpan ?? (int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.header_scan_size',
|
|
(int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.scan_span',
|
|
20000,
|
|
),
|
|
),
|
|
100,
|
|
), 100000);
|
|
$maxBatches = max(
|
|
$maxBatches ?? (int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.max_header_batches_per_run',
|
|
(int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.max_batches_per_run',
|
|
5,
|
|
),
|
|
),
|
|
0,
|
|
);
|
|
$inventoryBatchSize = min(max(
|
|
(int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.inventory_batch_size',
|
|
2000,
|
|
),
|
|
1,
|
|
), 5000);
|
|
$incrementalBatchSize = min(max(
|
|
(int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.incremental_batch_size',
|
|
2000,
|
|
),
|
|
1,
|
|
), 5000);
|
|
|
|
$batches = 0;
|
|
$headerRows = 0;
|
|
$itemRows = 0;
|
|
$inventoryRows = 0;
|
|
$runToken = null;
|
|
|
|
try {
|
|
$state = $resume ? $this->resumeRun($state) : $this->startRun($state, $mode);
|
|
$runToken = $this->requiredRunToken($state);
|
|
$headerGeneration = (string) $state['pending_header_generation'];
|
|
$itemGeneration = (string) $state['pending_item_generation'];
|
|
$phase = (string) ($state['sync_phase'] ?? ($mode === 'full' ? 'headers' : 'inventory'));
|
|
|
|
if ($phase === 'headers') {
|
|
if ($mode !== 'full') {
|
|
throw new RuntimeException('third_party_warehouse_incremental_header_phase_invalid');
|
|
}
|
|
|
|
$headerHighWatermark = $state['header_high_watermark'] ?? null;
|
|
if ($headerHighWatermark === null) {
|
|
$headerHighWatermark = $this->upstream->headerHighWatermark();
|
|
$this->updateState(['header_high_watermark' => $headerHighWatermark], $runToken);
|
|
}
|
|
|
|
$headerCursor = (int) ($state['header_cursor'] ?? 0);
|
|
while ($headerCursor < (int) $headerHighWatermark) {
|
|
$throughId = min($headerCursor + $scanSpan, (int) $headerHighWatermark);
|
|
$headers = $this->upstream->headerRange($headerCursor, $throughId);
|
|
$items = $this->upstream->itemsForIoIds(array_map(
|
|
static fn (array $row): mixed => $row['io_id'] ?? null,
|
|
$headers,
|
|
));
|
|
|
|
$this->storeHeaderRange(
|
|
$headers,
|
|
$items,
|
|
$headerGeneration,
|
|
$itemGeneration,
|
|
$headerCursor,
|
|
$throughId,
|
|
$runToken,
|
|
);
|
|
|
|
$headerCursor = $throughId;
|
|
$headerRows += count($headers);
|
|
$itemRows += count($items);
|
|
$batches++;
|
|
|
|
if ($headerCursor < (int) $headerHighWatermark
|
|
&& $this->batchLimitReached($batches, $maxBatches)) {
|
|
return $this->result(
|
|
'running', $mode, 'headers', $batches, $headerRows, $itemRows, $inventoryRows,
|
|
);
|
|
}
|
|
}
|
|
$state = $this->startInventorySnapshot($runToken);
|
|
$phase = 'inventory';
|
|
}
|
|
|
|
if ($phase === 'inventory' && ($state['pending_inventory_generation'] ?? null) === null) {
|
|
$state = $this->startInventorySnapshot($runToken);
|
|
}
|
|
|
|
if ($phase === 'inventory' && $this->batchLimitReached($batches, $maxBatches)) {
|
|
return $this->result(
|
|
'running', $mode, 'inventory', $batches, $headerRows, $itemRows, $inventoryRows,
|
|
);
|
|
}
|
|
|
|
if ($phase === 'inventory') {
|
|
$inventoryGeneration = (string) $state['pending_inventory_generation'];
|
|
$inventoryCursor = (int) ($state['inventory_cursor'] ?? 0);
|
|
$inventoryHighWatermark = (int) ($state['inventory_high_watermark'] ?? 0);
|
|
|
|
while ($inventoryCursor < $inventoryHighWatermark) {
|
|
$rows = $this->upstream->inventoryBatch(
|
|
$inventoryCursor,
|
|
$inventoryHighWatermark,
|
|
$inventoryBatchSize,
|
|
);
|
|
$nextCursor = $this->inventoryNextCursor(
|
|
$rows,
|
|
$inventoryCursor,
|
|
$inventoryHighWatermark,
|
|
$inventoryBatchSize,
|
|
);
|
|
$this->storeInventoryRange($rows, $inventoryGeneration, $nextCursor, $runToken);
|
|
|
|
$inventoryCursor = $nextCursor;
|
|
$inventoryRows += count($rows);
|
|
$batches++;
|
|
|
|
if ($inventoryCursor < $inventoryHighWatermark
|
|
&& $this->batchLimitReached($batches, $maxBatches)) {
|
|
return $this->result(
|
|
'running', $mode, 'inventory', $batches, $headerRows, $itemRows, $inventoryRows,
|
|
);
|
|
}
|
|
}
|
|
|
|
$state = $this->startHeaderCatchup($runToken);
|
|
$phase = 'catchup';
|
|
}
|
|
|
|
if ($phase !== 'catchup') {
|
|
throw new RuntimeException('third_party_warehouse_sync_phase_invalid');
|
|
}
|
|
|
|
if ($this->batchLimitReached($batches, $maxBatches)) {
|
|
return $this->result(
|
|
'running', $mode, 'catchup', $batches, $headerRows, $itemRows, $inventoryRows,
|
|
);
|
|
}
|
|
|
|
$cursorAt = $this->dateTime($state['header_modified_cursor_at'] ?? null);
|
|
$throughAt = $this->dateTime($state['header_window_ends_at'] ?? null);
|
|
$cursorId = (int) ($state['header_modified_cursor_id'] ?? 0);
|
|
if ($cursorAt === null || $throughAt === null) {
|
|
throw new RuntimeException('third_party_warehouse_catchup_window_missing');
|
|
}
|
|
|
|
while (true) {
|
|
$changedHeaders = $this->upstream->modifiedHeadersBatch(
|
|
$cursorAt,
|
|
$cursorId,
|
|
$throughAt,
|
|
$incrementalBatchSize,
|
|
);
|
|
if ($changedHeaders === []) {
|
|
break;
|
|
}
|
|
|
|
$changedIoIds = array_values(array_unique(array_filter(array_map(
|
|
static fn (array $row): int => (int) ($row['io_id'] ?? 0),
|
|
$changedHeaders,
|
|
))));
|
|
$latestHeaders = $this->upstream->latestHeadersForIoIds($changedIoIds, $throughAt);
|
|
$targetHeaders = $latestHeaders;
|
|
$items = $this->upstream->itemsForIoIds(array_map(
|
|
static fn (array $row): mixed => $row['io_id'] ?? null,
|
|
$targetHeaders,
|
|
));
|
|
$last = $changedHeaders[array_key_last($changedHeaders)];
|
|
$nextCursorAt = $this->dateTime($last['modified_at'] ?? null);
|
|
$nextCursorId = (int) ($last['source_id'] ?? 0);
|
|
if ($nextCursorAt === null
|
|
|| $nextCursorAt < $cursorAt
|
|
|| ($nextCursorAt === $cursorAt && $nextCursorId <= $cursorId)) {
|
|
throw new RuntimeException('third_party_warehouse_modified_cursor_invalid');
|
|
}
|
|
|
|
$this->storeModifiedHeaderBatch(
|
|
$changedIoIds,
|
|
$targetHeaders,
|
|
$items,
|
|
$headerGeneration,
|
|
$itemGeneration,
|
|
$nextCursorAt,
|
|
$nextCursorId,
|
|
$runToken,
|
|
);
|
|
|
|
$cursorAt = $nextCursorAt;
|
|
$cursorId = $nextCursorId;
|
|
$headerRows += count($targetHeaders);
|
|
$itemRows += count($items);
|
|
$batches++;
|
|
|
|
if (count($changedHeaders) < $incrementalBatchSize) {
|
|
break;
|
|
}
|
|
if ($this->batchLimitReached($batches, $maxBatches)) {
|
|
return $this->result(
|
|
'running', $mode, 'catchup', $batches, $headerRows, $itemRows, $inventoryRows,
|
|
);
|
|
}
|
|
}
|
|
|
|
$inventoryGeneration = (string) $state['pending_inventory_generation'];
|
|
$counts = $this->complete(
|
|
$mode,
|
|
$headerGeneration,
|
|
$itemGeneration,
|
|
$inventoryGeneration,
|
|
$runToken,
|
|
);
|
|
|
|
return $this->result(
|
|
'succeeded',
|
|
$mode,
|
|
'complete',
|
|
$batches,
|
|
$counts['headers'],
|
|
$counts['items'],
|
|
$counts['inventory'],
|
|
);
|
|
} catch (Throwable $exception) {
|
|
if ($exception instanceof RuntimeException
|
|
&& $exception->getMessage() === 'third_party_warehouse_sync_run_superseded') {
|
|
return $this->result(
|
|
'already_running', $mode, 'superseded', $batches, $headerRows, $itemRows, $inventoryRows,
|
|
);
|
|
}
|
|
|
|
$this->fail($exception::class, $runToken);
|
|
|
|
throw $exception;
|
|
}
|
|
}
|
|
|
|
/** @param array<string,mixed> $state */
|
|
private function startRun(array $state, string $mode): array
|
|
{
|
|
$now = now();
|
|
$runToken = (string) Str::uuid();
|
|
$expectedRunToken = $this->nullableRunToken($state);
|
|
$overlapHours = max((int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.incremental_overlap_hours',
|
|
48,
|
|
), 0);
|
|
try {
|
|
$sourceBoundary = $mode === 'full'
|
|
? $now
|
|
: ($state['header_modified_cursor_at'] ?? $state['last_succeeded_at'] ?? $now);
|
|
$windowStart = Carbon::parse((string) $sourceBoundary)
|
|
->subHours($overlapHours);
|
|
} catch (Throwable) {
|
|
$windowStart = $now->copy()->subHours($overlapHours);
|
|
}
|
|
$values = [
|
|
'status' => 'running',
|
|
'run_mode' => $mode,
|
|
'sync_phase' => $mode === 'full' ? 'headers' : 'inventory',
|
|
'run_token' => $runToken,
|
|
'header_cursor' => $mode === 'full' ? 0 : (int) ($state['header_cursor'] ?? 0),
|
|
'header_high_watermark' => null,
|
|
'header_window_started_at' => $windowStart->format('Y-m-d H:i:s.v'),
|
|
'header_window_ends_at' => null,
|
|
'header_modified_cursor_at' => $windowStart->format('Y-m-d H:i:s.v'),
|
|
'header_modified_cursor_id' => 0,
|
|
'inventory_cursor' => 0,
|
|
'inventory_high_watermark' => null,
|
|
'pending_header_generation' => (string) Str::uuid(),
|
|
'pending_item_generation' => (string) Str::uuid(),
|
|
'pending_inventory_generation' => null,
|
|
'last_attempted_at' => $now,
|
|
'last_error_code' => null,
|
|
'updated_at' => $now,
|
|
];
|
|
|
|
$this->connection->transaction(function () use (
|
|
$state,
|
|
$mode,
|
|
$values,
|
|
$expectedRunToken,
|
|
): void {
|
|
$this->assertExpectedRunToken($expectedRunToken);
|
|
$this->updateStateForToken($values, $expectedRunToken);
|
|
|
|
if ($mode === 'incremental') {
|
|
$this->cloneActiveGeneration(
|
|
self::HEADER_TABLE,
|
|
(string) ($state['active_header_generation'] ?? ''),
|
|
(string) $values['pending_header_generation'],
|
|
['source_key', 'id', 'io_id', 'wms_co_id', 'wh_id', 'status', 'io_type', 'occurred_at', 'source_modified_at', 'row_hash', 'synced_at'],
|
|
);
|
|
$this->cloneActiveGeneration(
|
|
self::ITEM_TABLE,
|
|
(string) ($state['active_item_generation'] ?? ''),
|
|
(string) $values['pending_item_generation'],
|
|
['source_key', 'id', 'io_id', 'sku_code', 'qty', 'qty_type', 'source_modified_at', 'row_hash', 'synced_at'],
|
|
);
|
|
}
|
|
});
|
|
|
|
return $this->stateForRun($runToken);
|
|
}
|
|
|
|
/** @param array<string,mixed> $state */
|
|
private function resumeRun(array $state): array
|
|
{
|
|
$expectedRunToken = $this->nullableRunToken($state);
|
|
$runToken = (string) Str::uuid();
|
|
$this->connection->transaction(function () use ($expectedRunToken, $runToken): void {
|
|
$this->assertExpectedRunToken($expectedRunToken);
|
|
$this->updateStateForToken([
|
|
'status' => 'running',
|
|
'run_token' => $runToken,
|
|
'last_attempted_at' => now(),
|
|
'last_error_code' => null,
|
|
'updated_at' => now(),
|
|
], $expectedRunToken);
|
|
});
|
|
|
|
return $this->stateForRun($runToken);
|
|
}
|
|
|
|
/** @return array<string,mixed> */
|
|
private function startInventorySnapshot(string $runToken): array
|
|
{
|
|
$highWatermark = $this->upstream->inventoryHighWatermark();
|
|
$this->updateState([
|
|
'pending_inventory_generation' => (string) Str::uuid(),
|
|
'sync_phase' => 'inventory',
|
|
'inventory_cursor' => 0,
|
|
'inventory_high_watermark' => $highWatermark,
|
|
], $runToken);
|
|
|
|
return $this->stateForRun($runToken);
|
|
}
|
|
|
|
/** @return array<string,mixed> */
|
|
private function startHeaderCatchup(string $runToken): array
|
|
{
|
|
$state = $this->stateForRun($runToken);
|
|
$windowStart = $this->dateTime($state['header_window_started_at'] ?? null);
|
|
if ($windowStart === null) {
|
|
throw new RuntimeException('third_party_warehouse_catchup_start_missing');
|
|
}
|
|
|
|
$this->updateState([
|
|
'sync_phase' => 'catchup',
|
|
'header_window_ends_at' => now()->format('Y-m-d H:i:s.v'),
|
|
'header_modified_cursor_at' => $windowStart,
|
|
'header_modified_cursor_id' => 0,
|
|
], $runToken);
|
|
|
|
return $this->stateForRun($runToken);
|
|
}
|
|
|
|
/**
|
|
* @param list<array<string,mixed>> $headers
|
|
* @param list<array<string,mixed>> $items
|
|
*/
|
|
private function storeHeaderRange(
|
|
array $headers,
|
|
array $items,
|
|
string $headerGeneration,
|
|
string $itemGeneration,
|
|
int $afterId,
|
|
int $throughId,
|
|
string $runToken,
|
|
): void {
|
|
$now = now();
|
|
$mappedHeaders = array_values(array_filter(array_map(
|
|
fn (array $row): ?array => $this->mapHeader($row, $headerGeneration, $now),
|
|
$headers,
|
|
)));
|
|
$modifiedByIoId = [];
|
|
foreach ($mappedHeaders as $header) {
|
|
$modifiedByIoId[(string) $header['io_id']] = $header['source_modified_at'];
|
|
}
|
|
$mappedItems = array_values(array_filter(array_map(
|
|
fn (array $row): ?array => $this->mapItem(
|
|
$row,
|
|
$itemGeneration,
|
|
$modifiedByIoId[(string) ($row['io_id'] ?? '')] ?? null,
|
|
$now,
|
|
),
|
|
$items,
|
|
)));
|
|
|
|
$this->connection->transaction(function () use (
|
|
$mappedHeaders,
|
|
$mappedItems,
|
|
$headerGeneration,
|
|
$itemGeneration,
|
|
$afterId,
|
|
$throughId,
|
|
$now,
|
|
$runToken,
|
|
): void {
|
|
$this->assertCurrentRunToken($runToken);
|
|
$ioIds = array_values(array_unique(array_filter(array_column($mappedHeaders, 'io_id'))));
|
|
$oldIoIds = $this->connection->table(self::HEADER_TABLE)
|
|
->where('generation', $headerGeneration)
|
|
->where('id', '>', $afterId)
|
|
->where('id', '<=', $throughId)
|
|
->pluck('io_id')
|
|
->map(static fn (mixed $id): string => (string) $id)
|
|
->all();
|
|
$ioIds = array_values(array_unique(array_merge($ioIds, $oldIoIds)));
|
|
|
|
if ($ioIds !== []) {
|
|
$this->connection->table(self::ITEM_TABLE)
|
|
->where('generation', $itemGeneration)
|
|
->whereIn('io_id', $ioIds)
|
|
->delete();
|
|
}
|
|
$this->connection->table(self::HEADER_TABLE)
|
|
->where('generation', $headerGeneration)
|
|
->where('id', '>', $afterId)
|
|
->where('id', '<=', $throughId)
|
|
->delete();
|
|
|
|
$this->upsertRows(self::HEADER_TABLE, $mappedHeaders, ['generation', 'id']);
|
|
$this->upsertRows(self::ITEM_TABLE, $mappedItems, ['generation', 'id']);
|
|
|
|
$this->updateStateForToken([
|
|
'header_cursor' => $throughId,
|
|
'updated_at' => $now,
|
|
], $runToken);
|
|
});
|
|
}
|
|
|
|
/**
|
|
* @param list<int> $changedIoIds
|
|
* @param list<array<string,mixed>> $targetHeaders
|
|
* @param list<array<string,mixed>> $items
|
|
*/
|
|
private function storeModifiedHeaderBatch(
|
|
array $changedIoIds,
|
|
array $targetHeaders,
|
|
array $items,
|
|
string $headerGeneration,
|
|
string $itemGeneration,
|
|
string $cursorAt,
|
|
int $cursorId,
|
|
string $runToken,
|
|
): void {
|
|
$now = now();
|
|
$mappedHeaders = array_values(array_filter(array_map(
|
|
fn (array $row): ?array => $this->mapHeader($row, $headerGeneration, $now),
|
|
$targetHeaders,
|
|
)));
|
|
$modifiedByIoId = [];
|
|
foreach ($mappedHeaders as $header) {
|
|
$modifiedByIoId[(string) $header['io_id']] = $header['source_modified_at'];
|
|
}
|
|
$mappedItems = array_values(array_filter(array_map(
|
|
fn (array $row): ?array => $this->mapItem(
|
|
$row,
|
|
$itemGeneration,
|
|
$modifiedByIoId[(string) ($row['io_id'] ?? '')] ?? null,
|
|
$now,
|
|
),
|
|
$items,
|
|
)));
|
|
$this->connection->transaction(function () use (
|
|
$mappedHeaders,
|
|
$mappedItems,
|
|
$changedIoIds,
|
|
$headerGeneration,
|
|
$itemGeneration,
|
|
$cursorAt,
|
|
$cursorId,
|
|
$now,
|
|
$runToken,
|
|
): void {
|
|
$this->assertCurrentRunToken($runToken);
|
|
if ($changedIoIds !== []) {
|
|
$this->connection->table(self::ITEM_TABLE)
|
|
->where('generation', $itemGeneration)
|
|
->whereIn('io_id', $changedIoIds)
|
|
->delete();
|
|
$this->connection->table(self::HEADER_TABLE)
|
|
->where('generation', $headerGeneration)
|
|
->whereIn('io_id', $changedIoIds)
|
|
->delete();
|
|
}
|
|
|
|
$this->upsertRows(self::HEADER_TABLE, $mappedHeaders, ['generation', 'id']);
|
|
$this->upsertRows(self::ITEM_TABLE, $mappedItems, ['generation', 'id']);
|
|
$this->updateStateForToken([
|
|
'header_modified_cursor_at' => $cursorAt,
|
|
'header_modified_cursor_id' => $cursorId,
|
|
'updated_at' => $now,
|
|
], $runToken);
|
|
});
|
|
}
|
|
|
|
/** @param list<array<string,mixed>> $rows */
|
|
private function storeInventoryRange(
|
|
array $rows,
|
|
string $generation,
|
|
int $cursor,
|
|
string $runToken,
|
|
): void
|
|
{
|
|
$now = now();
|
|
$mapped = array_values(array_filter(array_map(
|
|
fn (array $row): ?array => $this->mapInventory($row, $generation, $now),
|
|
$rows,
|
|
)));
|
|
$subwarehouseMapped = [];
|
|
foreach ($rows as $row) {
|
|
array_push($subwarehouseMapped, ...$this->mapSubwarehouseInventory($row, $generation, $now));
|
|
}
|
|
|
|
$this->connection->transaction(function () use (
|
|
$mapped,
|
|
$subwarehouseMapped,
|
|
$generation,
|
|
$cursor,
|
|
$now,
|
|
$runToken,
|
|
): void {
|
|
$this->assertCurrentRunToken($runToken);
|
|
$this->upsertRows(self::INVENTORY_TABLE, $mapped, ['generation', 'id']);
|
|
$this->upsertRows(
|
|
self::SUBWAREHOUSE_INVENTORY_TABLE,
|
|
$subwarehouseMapped,
|
|
['generation', 'id', 'wh_id'],
|
|
);
|
|
$this->updateStateForToken([
|
|
'inventory_cursor' => $cursor,
|
|
'updated_at' => $now,
|
|
], $runToken);
|
|
});
|
|
}
|
|
|
|
/** @return array{headers:int,items:int,inventory:int} */
|
|
private function complete(
|
|
string $mode,
|
|
string $headerGeneration,
|
|
string $itemGeneration,
|
|
string $inventoryGeneration,
|
|
string $runToken,
|
|
): array {
|
|
$counts = [
|
|
'headers' => (int) $this->connection->table(self::HEADER_TABLE)
|
|
->where('generation', $headerGeneration)->count(),
|
|
'items' => (int) $this->connection->table(self::ITEM_TABLE)
|
|
->where('generation', $itemGeneration)->count(),
|
|
'inventory' => (int) $this->connection->table(self::INVENTORY_TABLE)
|
|
->where('generation', $inventoryGeneration)->count(),
|
|
];
|
|
|
|
if ($counts['inventory'] === 0) {
|
|
throw new RuntimeException('third_party_warehouse_inventory_unexpected_empty_source');
|
|
}
|
|
if ($mode === 'full' && $counts['headers'] === 0) {
|
|
throw new RuntimeException('third_party_warehouse_headers_unexpected_empty_source');
|
|
}
|
|
|
|
$now = now();
|
|
$this->connection->transaction(function () use (
|
|
$mode,
|
|
$headerGeneration,
|
|
$itemGeneration,
|
|
$inventoryGeneration,
|
|
$counts,
|
|
$now,
|
|
$runToken,
|
|
): void {
|
|
$state = $this->assertCurrentRunToken($runToken);
|
|
$values = [
|
|
'status' => 'succeeded',
|
|
'run_mode' => null,
|
|
'sync_phase' => null,
|
|
'run_token' => null,
|
|
'active_header_generation' => $headerGeneration,
|
|
'active_item_generation' => $itemGeneration,
|
|
'active_inventory_generation' => $inventoryGeneration,
|
|
'pending_header_generation' => null,
|
|
'pending_item_generation' => null,
|
|
'pending_inventory_generation' => null,
|
|
'header_window_started_at' => null,
|
|
'header_window_ends_at' => null,
|
|
'header_row_count' => $counts['headers'],
|
|
'item_row_count' => $counts['items'],
|
|
'inventory_row_count' => $counts['inventory'],
|
|
'last_succeeded_at' => $now,
|
|
'last_error_code' => null,
|
|
'updated_at' => $now,
|
|
];
|
|
$values['header_modified_cursor_at'] = $this->dateTime(
|
|
$state['header_window_ends_at'] ?? $now,
|
|
);
|
|
$values['header_modified_cursor_id'] = 0;
|
|
if ($mode === 'full') {
|
|
$values['scope_version'] = 2;
|
|
$values['initial_completed_at'] = $now;
|
|
$values['last_full_succeeded_at'] = $now;
|
|
}
|
|
|
|
$this->publishGenerationRelease(
|
|
$state,
|
|
$headerGeneration,
|
|
$itemGeneration,
|
|
$inventoryGeneration,
|
|
$now,
|
|
);
|
|
$this->updateStateForToken($values, $runToken);
|
|
});
|
|
|
|
try {
|
|
$this->cleanupRetiredGenerations();
|
|
} catch (Throwable $exception) {
|
|
Log::warning('三方仓仓储数据旧代次清理失败，已保留数据等待下次重试。', [
|
|
'dataset' => self::DATASET_KEY,
|
|
'exception' => $exception::class,
|
|
]);
|
|
}
|
|
|
|
return $counts;
|
|
}
|
|
|
|
private function fail(string $errorCode, ?string $runToken): void
|
|
{
|
|
if ($runToken === null
|
|
|| ! Schema::connection($this->connection->getName())->hasTable(self::STATE_TABLE)) {
|
|
return;
|
|
}
|
|
|
|
$this->connection->table(self::STATE_TABLE)
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->where('run_token', $runToken)
|
|
->update([
|
|
'status' => 'failed',
|
|
'last_error_code' => mb_substr($errorCode, 0, 120),
|
|
'updated_at' => now(),
|
|
]);
|
|
}
|
|
|
|
/** @return array<string,mixed> */
|
|
private function state(): array
|
|
{
|
|
$now = now();
|
|
$this->connection->table(self::STATE_TABLE)->insertOrIgnore([
|
|
'dataset_key' => self::DATASET_KEY,
|
|
'status' => 'pending',
|
|
'header_row_count' => 0,
|
|
'item_row_count' => 0,
|
|
'inventory_row_count' => 0,
|
|
'created_at' => $now,
|
|
'updated_at' => $now,
|
|
]);
|
|
|
|
return (array) $this->connection->table(self::STATE_TABLE)
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->first();
|
|
}
|
|
|
|
/** @param array<string,mixed> $state */
|
|
private function ready(array $state): bool
|
|
{
|
|
return ($state['initial_completed_at'] ?? null) !== null
|
|
&& trim((string) ($state['active_header_generation'] ?? '')) !== ''
|
|
&& trim((string) ($state['active_item_generation'] ?? '')) !== ''
|
|
&& trim((string) ($state['active_inventory_generation'] ?? '')) !== '';
|
|
}
|
|
|
|
/** @param array<string,mixed> $state */
|
|
private function resumable(array $state, string $mode): bool
|
|
{
|
|
if (($state['run_mode'] ?? null) !== $mode
|
|
|| ! in_array($state['status'] ?? null, ['running', 'failed'], true)) {
|
|
return false;
|
|
}
|
|
|
|
if ($mode === 'full') {
|
|
return trim((string) ($state['pending_header_generation'] ?? '')) !== ''
|
|
&& trim((string) ($state['pending_item_generation'] ?? '')) !== '';
|
|
}
|
|
|
|
return trim((string) ($state['active_header_generation'] ?? '')) !== ''
|
|
&& trim((string) ($state['active_item_generation'] ?? '')) !== ''
|
|
&& trim((string) ($state['pending_header_generation'] ?? '')) !== ''
|
|
&& trim((string) ($state['pending_item_generation'] ?? '')) !== '';
|
|
}
|
|
|
|
/** @param array<string,mixed> $values */
|
|
private function updateState(array $values, string $runToken): void
|
|
{
|
|
$this->connection->transaction(function () use ($values, $runToken): void {
|
|
$this->assertCurrentRunToken($runToken);
|
|
$this->updateStateForToken($values + ['updated_at' => now()], $runToken);
|
|
});
|
|
}
|
|
|
|
/** @param array<string,mixed> $values */
|
|
private function updateStateForToken(array $values, ?string $expectedRunToken): void
|
|
{
|
|
$query = $this->connection->table(self::STATE_TABLE)
|
|
->where('dataset_key', self::DATASET_KEY);
|
|
$expectedRunToken === null
|
|
? $query->whereNull('run_token')
|
|
: $query->where('run_token', $expectedRunToken);
|
|
|
|
if ($query->update($values) !== 1) {
|
|
throw new RuntimeException('third_party_warehouse_sync_run_superseded');
|
|
}
|
|
}
|
|
|
|
/** @return array<string,mixed> */
|
|
private function assertCurrentRunToken(string $runToken): array
|
|
{
|
|
return $this->assertExpectedRunToken($runToken);
|
|
}
|
|
|
|
/** @return array<string,mixed> */
|
|
private function assertExpectedRunToken(?string $expectedRunToken): array
|
|
{
|
|
$state = $this->connection->table(self::STATE_TABLE)
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->lockForUpdate()
|
|
->first();
|
|
if ($state === null
|
|
|| $this->nullableRunToken((array) $state) !== $expectedRunToken) {
|
|
throw new RuntimeException('third_party_warehouse_sync_run_superseded');
|
|
}
|
|
|
|
return (array) $state;
|
|
}
|
|
|
|
/** @param array<string,mixed> $state */
|
|
private function nullableRunToken(array $state): ?string
|
|
{
|
|
$runToken = trim((string) ($state['run_token'] ?? ''));
|
|
|
|
return $runToken === '' ? null : $runToken;
|
|
}
|
|
|
|
/** @param array<string,mixed> $state */
|
|
private function requiredRunToken(array $state): string
|
|
{
|
|
$runToken = $this->nullableRunToken($state);
|
|
if ($runToken === null) {
|
|
throw new RuntimeException('third_party_warehouse_sync_run_token_missing');
|
|
}
|
|
|
|
return $runToken;
|
|
}
|
|
|
|
/** @return array<string,mixed> */
|
|
private function stateForRun(string $runToken): array
|
|
{
|
|
$state = $this->state();
|
|
if ($this->nullableRunToken($state) !== $runToken) {
|
|
throw new RuntimeException('third_party_warehouse_sync_run_superseded');
|
|
}
|
|
|
|
return $state;
|
|
}
|
|
|
|
/**
|
|
* @param array<string,mixed> $state
|
|
*/
|
|
private function publishGenerationRelease(
|
|
array $state,
|
|
string $headerGeneration,
|
|
string $itemGeneration,
|
|
string $inventoryGeneration,
|
|
mixed $now,
|
|
): void {
|
|
$this->connection->table(self::RELEASE_TABLE)
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->where('status', 'active')
|
|
->update([
|
|
'status' => 'retired',
|
|
'retired_at' => $now,
|
|
'updated_at' => $now,
|
|
]);
|
|
|
|
$previous = [
|
|
trim((string) ($state['active_header_generation'] ?? '')),
|
|
trim((string) ($state['active_item_generation'] ?? '')),
|
|
trim((string) ($state['active_inventory_generation'] ?? '')),
|
|
];
|
|
$current = [$headerGeneration, $itemGeneration, $inventoryGeneration];
|
|
if (! in_array('', $previous, true) && $previous !== $current) {
|
|
$this->recordGenerationRelease(
|
|
$previous[0],
|
|
$previous[1],
|
|
$previous[2],
|
|
'retired',
|
|
$state['last_succeeded_at'] ?? $now,
|
|
$now,
|
|
);
|
|
}
|
|
|
|
$this->recordGenerationRelease(
|
|
$headerGeneration,
|
|
$itemGeneration,
|
|
$inventoryGeneration,
|
|
'active',
|
|
$now,
|
|
null,
|
|
);
|
|
}
|
|
|
|
private function recordGenerationRelease(
|
|
string $headerGeneration,
|
|
string $itemGeneration,
|
|
string $inventoryGeneration,
|
|
string $status,
|
|
mixed $publishedAt,
|
|
mixed $retiredAt,
|
|
): void {
|
|
$key = [
|
|
'dataset_key' => self::DATASET_KEY,
|
|
'header_generation' => $headerGeneration,
|
|
'item_generation' => $itemGeneration,
|
|
'inventory_generation' => $inventoryGeneration,
|
|
];
|
|
$existing = $this->connection->table(self::RELEASE_TABLE)->where($key)->first();
|
|
$values = [
|
|
'status' => $status,
|
|
'published_at' => $publishedAt,
|
|
'retired_at' => $retiredAt,
|
|
'updated_at' => now(),
|
|
];
|
|
|
|
if ($existing !== null) {
|
|
$this->connection->table(self::RELEASE_TABLE)
|
|
->where('release_id', (string) $existing->release_id)
|
|
->update($values);
|
|
|
|
return;
|
|
}
|
|
|
|
$this->connection->table(self::RELEASE_TABLE)->insert($key + $values + [
|
|
'release_id' => (string) Str::uuid(),
|
|
'created_at' => now(),
|
|
]);
|
|
}
|
|
|
|
private function cleanupRetiredGenerations(): void
|
|
{
|
|
$retentionHours = max((int) config(
|
|
'upstream_mirrors.third_party_warehouse_storage.generation_retention_hours',
|
|
48,
|
|
), 1);
|
|
$cutoff = now()->subHours($retentionHours);
|
|
$releaseIds = $this->connection->table(self::RELEASE_TABLE)
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->where('status', 'retired')
|
|
->whereNotNull('retired_at')
|
|
->where('retired_at', '<=', $cutoff)
|
|
->orderBy('retired_at')
|
|
->limit(50)
|
|
->pluck('release_id')
|
|
->map(static fn (mixed $releaseId): string => (string) $releaseId)
|
|
->all();
|
|
|
|
foreach ($releaseIds as $releaseId) {
|
|
$this->cleanupRetiredGeneration($releaseId, $cutoff);
|
|
}
|
|
}
|
|
|
|
private function cleanupRetiredGeneration(string $releaseId, mixed $cutoff): void
|
|
{
|
|
$this->connection->transaction(function () use ($releaseId, $cutoff): void {
|
|
$state = (array) $this->connection->table(self::STATE_TABLE)
|
|
->where('dataset_key', self::DATASET_KEY)
|
|
->lockForUpdate()
|
|
->first();
|
|
$release = $this->connection->table(self::RELEASE_TABLE)
|
|
->where('release_id', $releaseId)
|
|
->lockForUpdate()
|
|
->first();
|
|
if ($release === null
|
|
|| (string) $release->status !== 'retired'
|
|
|| $release->retired_at === null
|
|
|| Carbon::parse((string) $release->retired_at)->gt($cutoff)) {
|
|
return;
|
|
}
|
|
|
|
$protected = array_filter([
|
|
$state['active_header_generation'] ?? null,
|
|
$state['active_item_generation'] ?? null,
|
|
$state['active_inventory_generation'] ?? null,
|
|
$state['pending_header_generation'] ?? null,
|
|
$state['pending_item_generation'] ?? null,
|
|
$state['pending_inventory_generation'] ?? null,
|
|
]);
|
|
foreach ([
|
|
[self::HEADER_TABLE, 'header_generation', (string) $release->header_generation],
|
|
[self::ITEM_TABLE, 'item_generation', (string) $release->item_generation],
|
|
[self::INVENTORY_TABLE, 'inventory_generation', (string) $release->inventory_generation],
|
|
[self::SUBWAREHOUSE_INVENTORY_TABLE, 'inventory_generation', (string) $release->inventory_generation],
|
|
] as [$table, $column, $generation]) {
|
|
if ($generation === '' || in_array($generation, $protected, true)) {
|
|
continue;
|
|
}
|
|
|
|
$stillReferenced = $this->connection->table(self::RELEASE_TABLE)
|
|
->where('release_id', '!=', $releaseId)
|
|
->where($column, $generation)
|
|
->where(function ($query) use ($cutoff): void {
|
|
$query->where('status', 'active')
|
|
->orWhereNull('retired_at')
|
|
->orWhere('retired_at', '>', $cutoff);
|
|
})
|
|
->exists();
|
|
if (! $stillReferenced) {
|
|
$this->connection->table($table)->where('generation', $generation)->delete();
|
|
}
|
|
}
|
|
|
|
$this->connection->table(self::RELEASE_TABLE)
|
|
->where('release_id', $releaseId)
|
|
->delete();
|
|
});
|
|
}
|
|
|
|
/** @return array<string,mixed>|null */
|
|
private function mapHeader(array $row, string $generation, mixed $now): ?array
|
|
{
|
|
$id = (int) ($row['source_id'] ?? 0);
|
|
if ($id <= 0) {
|
|
return null;
|
|
}
|
|
|
|
$ioId = (int) ($row['io_id'] ?? 0);
|
|
$warehouseId = (int) ($row['wms_co_id'] ?? 0);
|
|
if ($ioId <= 0 || $warehouseId <= 0) {
|
|
return null;
|
|
}
|
|
|
|
$values = [
|
|
'generation' => $generation,
|
|
'source_key' => ThirdPartyWarehouseStorageUpstreamRepository::SOURCE_KEY,
|
|
'id' => $id,
|
|
'io_id' => $ioId,
|
|
'wms_co_id' => $warehouseId,
|
|
'wh_id' => max((int) ($row['wh_id'] ?? 0), 0),
|
|
'status' => $this->text($row['status'] ?? null, 32),
|
|
'io_type' => $this->text($row['io_type'] ?? null, 64),
|
|
'occurred_at' => $this->dateTime($row['occurred_at'] ?? null),
|
|
'source_modified_at' => $this->dateTime($row['modified_at'] ?? null),
|
|
];
|
|
|
|
return $values + [
|
|
'row_hash' => $this->rowHash($values),
|
|
'synced_at' => $now,
|
|
];
|
|
}
|
|
|
|
/** @return array<string,mixed>|null */
|
|
private function mapItem(array $row, string $generation, ?string $modifiedAt, mixed $now): ?array
|
|
{
|
|
$id = (int) ($row['source_id'] ?? 0);
|
|
if ($id <= 0) {
|
|
return null;
|
|
}
|
|
|
|
$ioId = (int) ($row['io_id'] ?? 0);
|
|
$skuCode = $this->text($row['sku_code'] ?? null, 191);
|
|
$qtyType = (int) ($row['qty_type'] ?? 0);
|
|
if ($ioId <= 0 || $skuCode === null || ! in_array($qtyType, [-1, 1], true)) {
|
|
return null;
|
|
}
|
|
|
|
$values = [
|
|
'generation' => $generation,
|
|
'source_key' => ThirdPartyWarehouseStorageUpstreamRepository::SOURCE_KEY,
|
|
'id' => $id,
|
|
'io_id' => $ioId,
|
|
'sku_code' => $skuCode,
|
|
'qty' => (float) ($row['qty'] ?? 0),
|
|
'qty_type' => $qtyType,
|
|
// jst_inoutWaterItem has no update timestamp; the owning header's
|
|
// modified time is the auditable refresh marker for all its items.
|
|
'source_modified_at' => $modifiedAt,
|
|
];
|
|
|
|
return $values + [
|
|
'row_hash' => $this->rowHash($values),
|
|
'synced_at' => $now,
|
|
];
|
|
}
|
|
|
|
/** @return array<string,mixed>|null */
|
|
private function mapInventory(array $row, string $generation, mixed $now): ?array
|
|
{
|
|
$id = (int) ($row['source_id'] ?? 0);
|
|
if ($id <= 0) {
|
|
return null;
|
|
}
|
|
|
|
$warehouseId = (int) ($row['wms_co_id'] ?? 0);
|
|
$skuCode = $this->text($row['sku_code'] ?? null, 191);
|
|
if ($warehouseId <= 0 || $skuCode === null) {
|
|
return null;
|
|
}
|
|
|
|
$values = [
|
|
'generation' => $generation,
|
|
'source_key' => ThirdPartyWarehouseStorageUpstreamRepository::SOURCE_KEY,
|
|
'id' => $id,
|
|
'wms_co_id' => $warehouseId,
|
|
'sku_code' => $skuCode,
|
|
'qty' => (float) ($row['qty'] ?? 0),
|
|
'source_modified_at' => $this->dateTime($row['source_modified_at'] ?? null),
|
|
'captured_at' => $now->format('Y-m-d H:i:s.u'),
|
|
];
|
|
|
|
return $values + [
|
|
'row_hash' => $this->rowHash($values),
|
|
'synced_at' => $now,
|
|
];
|
|
}
|
|
|
|
/** @return list<array<string,mixed>> */
|
|
private function mapSubwarehouseInventory(array $row, string $generation, mixed $now): array
|
|
{
|
|
$base = $this->mapInventory($row, $generation, $now);
|
|
if ($base === null) {
|
|
return [];
|
|
}
|
|
|
|
$rows = [];
|
|
foreach (self::SUBWAREHOUSE_QUANTITY_FIELDS as $warehouseId => $field) {
|
|
$quantity = (float) ($row[$field] ?? 0);
|
|
if ($quantity == 0.0) {
|
|
continue;
|
|
}
|
|
$values = [
|
|
'generation' => $generation,
|
|
'source_key' => $base['source_key'],
|
|
'id' => $base['id'],
|
|
'wms_co_id' => $base['wms_co_id'],
|
|
'wh_id' => $warehouseId,
|
|
'sku_code' => $base['sku_code'],
|
|
'qty' => $quantity,
|
|
'source_modified_at' => $base['source_modified_at'],
|
|
'captured_at' => $base['captured_at'],
|
|
];
|
|
$rows[] = $values + [
|
|
'row_hash' => $this->rowHash($values),
|
|
'synced_at' => $now,
|
|
];
|
|
}
|
|
|
|
return $rows;
|
|
}
|
|
|
|
/**
|
|
* @param list<array<string,mixed>> $rows
|
|
* @param list<string> $uniqueBy
|
|
*/
|
|
private function upsertRows(string $table, array $rows, array $uniqueBy): void
|
|
{
|
|
if ($rows === []) {
|
|
return;
|
|
}
|
|
|
|
$updates = array_values(array_diff(array_keys($rows[0]), $uniqueBy));
|
|
$columnsPerRow = max(count($rows[0]), 1);
|
|
$chunkSize = max(1, min(1000, intdiv(50000, $columnsPerRow)));
|
|
foreach (array_chunk($rows, $chunkSize) as $chunk) {
|
|
$this->connection->table($table)->upsert($chunk, $uniqueBy, $updates);
|
|
}
|
|
}
|
|
|
|
/** @param list<string> $columns */
|
|
private function cloneActiveGeneration(
|
|
string $table,
|
|
string $activeGeneration,
|
|
string $pendingGeneration,
|
|
array $columns,
|
|
): void {
|
|
if ($activeGeneration === '' || $pendingGeneration === '') {
|
|
throw new RuntimeException('third_party_warehouse_active_generation_missing');
|
|
}
|
|
|
|
$source = $this->connection->table($table)
|
|
->where('generation', $activeGeneration)
|
|
->selectRaw('? AS generation', [$pendingGeneration])
|
|
->addSelect($columns);
|
|
|
|
$this->connection->table($table)->insertUsing(['generation', ...$columns], $source);
|
|
}
|
|
|
|
/** @param array<string,mixed> $values */
|
|
private function rowHash(array $values): string
|
|
{
|
|
unset($values['generation']);
|
|
|
|
return hash('sha256', json_encode(
|
|
$values,
|
|
JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION,
|
|
));
|
|
}
|
|
|
|
private function text(mixed $value, int $maxLength): ?string
|
|
{
|
|
$value = trim((string) ($value ?? ''));
|
|
|
|
return $value === '' ? null : mb_substr($value, 0, $maxLength);
|
|
}
|
|
|
|
private function dateTime(mixed $value): ?string
|
|
{
|
|
$value = trim((string) ($value ?? ''));
|
|
|
|
return $value === '' ? null : Carbon::parse($value)->format('Y-m-d H:i:s.v');
|
|
}
|
|
|
|
private function batchLimitReached(int $batches, int $maxBatches): bool
|
|
{
|
|
return $maxBatches > 0 && $batches >= $maxBatches;
|
|
}
|
|
|
|
/** @param list<array<string,mixed>> $rows */
|
|
private function inventoryNextCursor(
|
|
array $rows,
|
|
int $cursor,
|
|
int $highWatermark,
|
|
int $batchSize,
|
|
): int {
|
|
if ($rows === []) {
|
|
return $highWatermark;
|
|
}
|
|
|
|
$nextCursor = max(array_map(
|
|
static fn (array $row): int => (int) ($row['source_id'] ?? 0),
|
|
$rows,
|
|
));
|
|
if ($nextCursor <= $cursor || $nextCursor > $highWatermark) {
|
|
throw new RuntimeException('third_party_warehouse_inventory_cursor_invalid');
|
|
}
|
|
|
|
return count($rows) < $batchSize ? $highWatermark : $nextCursor;
|
|
}
|
|
|
|
/**
|
|
* @return array{status:string,mode:string,phase:string,batches:int,headers:int,items:int,inventory:int}
|
|
*/
|
|
private function result(
|
|
string $status,
|
|
string $mode,
|
|
string $phase,
|
|
int $batches,
|
|
int $headers,
|
|
int $items,
|
|
int $inventory,
|
|
): array {
|
|
return compact('status', 'mode', 'phase', 'batches', 'headers', 'items', 'inventory');
|
|
}
|
|
|
|
private function assertSchemaReady(): void
|
|
{
|
|
$schema = Schema::connection($this->connection->getName());
|
|
foreach ([
|
|
self::STATE_TABLE,
|
|
self::RELEASE_TABLE,
|
|
self::HEADER_TABLE,
|
|
self::ITEM_TABLE,
|
|
self::INVENTORY_TABLE,
|
|
self::SUBWAREHOUSE_INVENTORY_TABLE,
|
|
] as $table) {
|
|
if (! $schema->hasTable($table)) {
|
|
throw new RuntimeException('third_party_warehouse_mirror_schema_missing:' . $table);
|
|
}
|
|
}
|
|
|
|
if (! $schema->hasColumn(self::ITEM_TABLE, 'qty_type')) {
|
|
throw new RuntimeException('third_party_warehouse_mirror_item_missing:qty_type');
|
|
}
|
|
foreach ([self::HEADER_TABLE, self::ITEM_TABLE] as $table) {
|
|
if (! $schema->hasColumn($table, 'source_key')) {
|
|
throw new RuntimeException('third_party_warehouse_mirror_source_key_missing:' . $table);
|
|
}
|
|
if (! $schema->hasColumn($table, 'source_modified_at')) {
|
|
throw new RuntimeException('third_party_warehouse_mirror_modified_at_missing:' . $table);
|
|
}
|
|
}
|
|
foreach (['source_modified_at', 'captured_at'] as $column) {
|
|
if (! $schema->hasColumn(self::INVENTORY_TABLE, $column)) {
|
|
throw new RuntimeException('third_party_warehouse_mirror_inventory_missing:' . $column);
|
|
}
|
|
}
|
|
if (! $schema->hasColumn(self::INVENTORY_TABLE, 'source_key')) {
|
|
throw new RuntimeException('third_party_warehouse_mirror_source_key_missing:' . self::INVENTORY_TABLE);
|
|
}
|
|
|
|
foreach ([
|
|
'run_mode',
|
|
'sync_phase',
|
|
'run_token',
|
|
'active_header_generation',
|
|
'active_item_generation',
|
|
'active_inventory_generation',
|
|
'pending_header_generation',
|
|
'pending_item_generation',
|
|
'pending_inventory_generation',
|
|
'header_cursor',
|
|
'header_high_watermark',
|
|
'header_window_started_at',
|
|
'header_window_ends_at',
|
|
'header_modified_cursor_at',
|
|
'header_modified_cursor_id',
|
|
'inventory_cursor',
|
|
'inventory_high_watermark',
|
|
'header_row_count',
|
|
'item_row_count',
|
|
'inventory_row_count',
|
|
'initial_completed_at',
|
|
'last_full_succeeded_at',
|
|
'scope_version',
|
|
] as $column) {
|
|
if (! $schema->hasColumn(self::STATE_TABLE, $column)) {
|
|
throw new RuntimeException('third_party_warehouse_mirror_state_missing:' . $column);
|
|
}
|
|
}
|
|
}
|
|
}
|