CaiWuXiTong/modules/EcommerceFinance/Repositories/ThirdPartyWarehouseStorageMirrorRepository.php
WuYongJie 374df29d2b 修复生产报表负载根因 (#257)
消除报表缓存穿透与后台千万级全表计数,降低同步频率并调整共享数据库资源。
2026-09-23 20:15:59 +08:00

1765 lines
75 KiB
PHP
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

<?php
namespace Modules\EcommerceFinance\Repositories;
use Carbon\CarbonInterface;
use Generator;
use Illuminate\Contracts\Cache\LockTimeoutException;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Database\Query\Builder;
use Illuminate\Support\Carbon;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Schema;
use RuntimeException;
use Throwable;
/**
* 三方仓进销存的共享业务镜像只读查询。
*
* 已导入的历史月份直接读取权威日明细;后续月份从每个 SKU 的固定
* 月末锚点按已确认流水前推。未导入锚点时保留实时库存兼容逻辑。
*/
class ThirdPartyWarehouseStorageMirrorRepository
{
private const REPORT_QUERY_TIMEOUT_SECONDS = 30;
private const REPORT_CACHE_SECONDS = 93600;
private const REPORT_CACHE_LOCK_SECONDS = 45;
private const REPORT_CACHE_WAIT_SECONDS = 35;
public const DATASET_KEY = 'jiansun.third-party-warehouse-storage';
public const WAREHOUSE_ID = 15045364;
private const STATE_TABLE = 'third_party_warehouse_sync_states';
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';
private const AUTHORITATIVE_STATE_TABLE = 'third_party_warehouse_authoritative_anchor_states';
private const AUTHORITATIVE_ANCHOR_TABLE = 'third_party_warehouse_authoritative_anchors';
private const AUTHORITATIVE_ROW_TABLE = 'third_party_warehouse_authoritative_rows';
private const AUTHORITATIVE_DATASET_KEY = 'jiansun.third-party-warehouse-storage.authoritative-anchor';
private const INVENTORY_REPORT_AUTHORITATIVE_STATE_TABLE = 'inventory_report_authoritative_states';
private const INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE = 'inventory_report_authoritative_rows';
private const INVENTORY_REPORT_AUTHORITATIVE_DATASET_KEY = 'jiansun.inventory-report.authoritative-month';
private const PRODUCT_TABLE = 'ops_jst_sku';
private const WAREHOUSE_TABLE = 'ops_jst_warehouse_info';
private const GENERIC_STATE_TABLE = 'business_mirror_table_states';
private const PRODUCT_DATASET_KEY = 'operations_replica.rpt.jst_sku';
private const WAREHOUSE_DATASET_KEY = 'operations_replica.rpt.jst_warehouseInfo';
public const START_DATE = '2026-08-01';
/** Historical monthly report range can start one month before the flow page. */
public const INVENTORY_REPORT_START_DATE = '2026-07-01';
/** @var array<string,string> */
private const MOVEMENT_COLUMNS = [
'processing_in' => '加工进仓',
'processing_out' => '加工出仓',
'purchase_in' => '采购进仓',
'purchase_return' => '采购退货',
'other_in' => '其它进仓',
'other_return' => '其它退货',
'sales_out' => '销售出仓',
'sales_return' => '销售退货',
'stocktake' => '盘点',
'transfer_out' => '调拨出',
'transfer_in' => '调拨入',
'other_out' => '其它出库',
];
/** @var array<int,string> */
private const SUBWAREHOUSE_SUFFIXES = [
1 => '主仓',
2 => '销退仓',
3 => '进货仓',
4 => '次品仓',
6 => '自定义仓1',
7 => '自定义仓2',
8 => '自定义仓3',
];
/**
* 聚水潭“按商品、按仓库、按月”导出中使用的实际子仓名。
* 其余仓库按“仓储方 + 子仓后缀”生成,不再只显示通用类型名。
*
* @var array<string,array<int,string>>
*/
private const SPECIAL_SUBWAREHOUSE_NAMES = [
'常州简尚家居有限公司' => [
1 => '配件单仓',
],
'简尚-横林二仓' => [
6 => '客退临时上架仓',
7 => '数量校验仓',
],
'简尚-新苏北仓' => [
6 => '新苏北客退上架仓',
],
'简尚-睢宁仓' => [
6 => '睢宁客退上架仓',
],
'简尚-河南洛宁仓' => [
6 => '河南洛宁客退上架仓',
],
'简尚-山东青岗集仓' => [
7 => '青岗集客退上架仓',
],
];
/** @var list<string> */
private const OUTPUT_COLUMNS = [
'owner', 'date', 'sku_code', 'image_url', 'product_name', 'short_name',
'supplier', 'category', 'virtual_category', 'supplier_style_no',
'color_spec', 'brand', 'unit', 'opening_qty', 'closing_qty',
'processing_in', 'processing_out', 'purchase_in', 'purchase_return',
'other_in', 'other_return', 'sales_out', 'sales_return', 'stocktake',
'transfer_out', 'transfer_in', 'opening_aux', 'other_out',
];
private readonly ConnectionInterface $connection;
public function __construct(DatabaseManager $databases)
{
$this->connection = $databases->connection(
(string) config('upstream_mirrors.connection', 'business_mirror'),
);
}
public function available(): bool
{
try {
$schema = Schema::connection($this->connection->getName());
foreach ([
self::STATE_TABLE,
self::HEADER_TABLE,
self::ITEM_TABLE,
self::INVENTORY_TABLE,
self::PRODUCT_TABLE,
self::GENERIC_STATE_TABLE,
] as $table) {
if (! $schema->hasTable($table)) {
return false;
}
}
return $schema->hasColumn(self::ITEM_TABLE, 'qty_type')
&& $schema->hasColumn(self::HEADER_TABLE, 'source_modified_at')
&& $schema->hasColumn(self::ITEM_TABLE, 'source_modified_at')
&& $schema->hasColumn(self::INVENTORY_TABLE, 'source_modified_at')
&& $schema->hasColumn(self::INVENTORY_TABLE, 'captured_at');
} catch (Throwable) {
return false;
}
}
public function ready(): bool
{
$state = $this->state();
return is_array($state)
&& ($state['initial_completed_at'] ?? null) !== null
&& $this->generation($state, 'active_header_generation') !== null
&& $this->generation($state, 'active_item_generation') !== null
&& $this->generation($state, 'active_inventory_generation') !== null
&& $this->dependenciesReady();
}
public function allWarehousesReady(): bool
{
$state = $this->state();
return $this->ready()
&& (int) ($state['scope_version'] ?? 1) >= 2
&& $this->warehouseDependencyReady();
}
/** @return array<string,mixed>|null */
public function state(): ?array
{
if (! $this->available()) {
return null;
}
try {
$row = $this->connection->table(self::STATE_TABLE)
->where('dataset_key', self::DATASET_KEY)
->first();
return $row === null ? null : (array) $row;
} catch (Throwable) {
return null;
}
}
/**
* @param array<string,mixed>|null $state
* @return array{code:string,last_succeeded_at:?string,lag_minutes:?int}
*/
public function syncStatus(?array $state = null, ?CarbonInterface $at = null): array
{
$state ??= $this->state();
if (! is_array($state)) {
return ['code' => 'unavailable', 'last_succeeded_at' => null, 'lag_minutes' => null];
}
$lastSucceededAt = $this->timestamp($state['last_succeeded_at'] ?? null);
$lastSucceeded = $lastSucceededAt === null
? null
: Carbon::createFromTimestamp($lastSucceededAt)->toDateTimeString();
$lagMinutes = $lastSucceededAt === null
? null
: (int) ceil(max(0, ($at ?? now())->getTimestamp() - $lastSucceededAt) / 60);
$status = (string) ($state['status'] ?? 'pending');
if (($state['initial_completed_at'] ?? null) === null) {
$code = $status === 'failed'
? 'initial_failed'
: ($status === 'running' ? 'initializing' : 'pending');
} elseif (! $this->dependenciesReady()) {
$code = 'dependency_pending';
} elseif ($status === 'failed') {
$code = 'failed';
} elseif ($status === 'running') {
$code = ($state['run_mode'] ?? null) === 'full' ? 'reconciling' : 'syncing';
} elseif ($lagMinutes === null
|| $lagMinutes > max((int) config(
'upstream_mirrors.third_party_warehouse_storage.stale_after_minutes',
1560,
), 1)) {
$code = 'delayed';
} else {
$code = 'healthy';
}
return [
'code' => $code,
'last_succeeded_at' => $lastSucceeded,
'lag_minutes' => $lagMinutes,
];
}
/**
* @param array<string,mixed> $filters
* @return array{data:list<array<string,mixed>>,total:int,summary:array<string,int|float>,state:array<string,mixed>,sync:array<string,mixed>}
*/
public function paginate(array $filters, int $page, int $perPage): array
{
$state = $this->requireReadyState();
$page = max($page, 1);
$perPage = min(max($perPage, 1), 100);
$query = $this->filteredReportQuery($filters, $state);
$summary = $this->summarizeQuery($this->filteredReportQuery(
$filters,
$state,
$this->hasProductFilters($filters),
));
$rows = $query->orderBy('report_date')->orderBy('sku_code')
->forPage($page, $perPage)
->get()
->map(fn (object $row): array => $this->mapRow($row))
->all();
return [
'data' => $rows,
'total' => $summary['count'],
'summary' => $summary,
'state' => $state,
'sync' => $this->syncStatus($state),
];
}
/**
* 商品价格管理的进销存报表与三方仓共用同一套已确认流水、
* 库存锚点和业务类型口径,只在这里将日明细再汇总为“商品 + 仓库 + 月”。
*
* @param array<string,mixed> $filters
* @return array{data:list<object>,total:int,state:array<string,mixed>,sync:array<string,mixed>}
*/
public function paginateMonthlyInventoryReport(
array $filters,
int $page,
int $perPage,
bool $includeTotal = true,
?array $state = null,
): array
{
$state ??= $this->requireReadyState();
if ((int) ($state['scope_version'] ?? 1) < 2) {
throw new RuntimeException('inventory_report_all_warehouses_not_ready');
}
$page = max($page, 1);
$perPage = min(max($perPage, 1), 5000);
$period = $this->inventoryReportPeriod((string) ($filters['month'] ?? ''));
$authoritative = $this->monthlyInventoryAuthoritativeQuery($period, $filters);
if ($authoritative !== null) {
$total = $includeTotal ? $this->boundedCount(clone $authoritative) : 0;
$authoritativePage = (clone $authoritative)
->orderBy('warehouse_provider')
->orderBy('warehouse')
->orderBy('product_code')
->forPage($page, $perPage);
$rows = $this->boundedGet($authoritativePage)
->map(static fn (object $row): object => (object) [
'owner' => $row->owner,
'month' => $row->report_month,
'warehouse_provider' => $row->warehouse_provider,
'warehouse' => $row->warehouse,
'product_code' => $row->product_code,
'category' => $row->category,
'opening_quantity' => $row->opening_quantity,
'closing_quantity' => $row->closing_quantity,
'processing_in' => $row->processing_in,
'processing_out' => $row->processing_out,
'purchase_in' => $row->purchase_in,
'purchase_return' => $row->purchase_return,
'other_in' => $row->other_in,
'other_return' => $row->other_return,
'sales_out' => $row->sales_out,
'sales_return' => $row->sales_return,
'stocktake' => $row->stocktake,
'transfer_out' => $row->transfer_out,
'transfer_in' => $row->transfer_in,
'opening' => $row->opening,
'other_out' => $row->other_out,
])
->all();
return [
'data' => $rows,
'total' => $total,
'state' => $state,
'sync' => $this->syncStatus($state),
];
}
$headerGeneration = $this->generation($state, 'active_header_generation');
$itemGeneration = $this->generation($state, 'active_item_generation');
$inventoryGeneration = $this->generation($state, 'active_inventory_generation');
if ($headerGeneration === null || $itemGeneration === null || $inventoryGeneration === null) {
throw new RuntimeException('inventory_report_mirror_not_ready');
}
$authority = $this->previousMonthlyInventoryAuthority($period);
$resolveRollForward = fn (): ?array => $authority === null
? null
: $this->monthlyInventoryRollForward(
$period,
$filters,
$authority,
$headerGeneration,
$itemGeneration,
$inventoryGeneration,
$page,
$perPage,
$includeTotal,
$state,
);
$rolledForward = $includeTotal && $perPage <= 100 && $authority !== null
? $this->rememberMonthlyInventoryReport(
$this->monthlyInventoryReportCacheKey(
$period,
$filters,
$authority,
$headerGeneration,
$itemGeneration,
$inventoryGeneration,
$page,
$perPage,
),
$resolveRollForward,
)
: $resolveRollForward();
if ($rolledForward !== null) {
$rolledForward['state'] = $state;
$rolledForward['sync'] = $this->syncStatus($state);
return $rolledForward;
}
$anchorAt = $this->stateAnchor($state);
$rangeStart = min($period['start'], $anchorAt);
$rangeEnd = max($period['end'], $anchorAt);
$latestProducts = $this->connection->table(self::PRODUCT_TABLE)
->where('sku_type', 'normal')->whereNotNull('sku_id')->where('sku_id', '!=', '')
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
$latestWarehouses = $this->connection->table(self::WAREHOUSE_TABLE)
->whereNotNull('wms_co_id')->groupBy('wms_co_id')
->select('wms_co_id')->selectRaw('MAX(Id) AS Id');
$inventoryDimensions = $this->connection->table(self::SUBWAREHOUSE_INVENTORY_TABLE)
->where('generation', $inventoryGeneration)
->where('qty', '!=', 0)
->whereNotNull('sku_code')->where('sku_code', '!=', '')
->select(['wms_co_id', 'wh_id', 'sku_code as inventory_sku_code', 'captured_at'])
->selectRaw('qty AS snapshot_qty');
$candidates = $this->connection->query()->fromSub($inventoryDimensions, 'pairs')
->joinSub($latestProducts, 'latest_product', 'latest_product.sku_id', '=', 'pairs.inventory_sku_code')
->join(self::PRODUCT_TABLE . ' as s', function ($join): void {
$join->on('s.sku_id', '=', 'latest_product.sku_id')
->on('s.autoid', '=', 'latest_product.autoid');
})
->select([
'pairs.wms_co_id', 'pairs.wh_id', 'pairs.inventory_sku_code',
's.sku_id as sku_code', 's.category',
]);
$productCode = trim((string) ($filters['product_code'] ?? ''));
if ($productCode !== '') {
$candidates->where('s.sku_id', $productCode);
}
$warehouseFilter = trim((string) ($filters['warehouse_provider'] ?? ''));
if ($warehouseFilter !== '') {
$warehouseIds = $this->connection->query()->fromSub(
clone $latestWarehouses,
'filter_latest_warehouse',
)->join(self::WAREHOUSE_TABLE . ' as filter_warehouse', function ($join): void {
$join->on('filter_warehouse.wms_co_id', '=', 'filter_latest_warehouse.wms_co_id')
->on('filter_warehouse.Id', '=', 'filter_latest_warehouse.Id');
});
$this->containsFilter($warehouseIds, 'filter_warehouse.name', $warehouseFilter);
$candidates->whereIn('pairs.wms_co_id', $warehouseIds->select('filter_warehouse.wms_co_id'));
}
$categoryFilter = trim((string) ($filters['category'] ?? ''));
if ($categoryFilter !== '') {
$this->containsFilter($candidates, 's.category', $categoryFilter);
}
$total = $includeTotal ? $this->boundedCount(clone $candidates) : 0;
$pagePairs = (clone $candidates)
->orderBy('pairs.wms_co_id')->orderBy('pairs.wh_id')->orderBy('s.sku_id')
->forPage($page, $perPage);
$pairRows = $this->boundedGet(clone $pagePairs);
if ($pairRows->isEmpty()) {
return [
'data' => [],
'total' => $total,
'state' => $state,
'sync' => $this->syncStatus($state),
];
}
$dimensionedPagePairs = $this->connection->query()->fromSub($pagePairs, 'pairs')
->leftJoinSub($latestWarehouses, 'latest_warehouse', 'latest_warehouse.wms_co_id', '=', 'pairs.wms_co_id')
->leftJoin(self::WAREHOUSE_TABLE . ' as w', function ($join): void {
$join->on('w.wms_co_id', '=', 'latest_warehouse.wms_co_id')
->on('w.Id', '=', 'latest_warehouse.Id');
})
->select([
'pairs.wms_co_id', 'pairs.wh_id', 'pairs.inventory_sku_code',
'pairs.sku_code', 'pairs.category', 'w.name as warehouse_name',
]);
$anchors = $this->connection->query()->fromSub($dimensionedPagePairs, 'pairs')
->leftJoinSub(clone $inventoryDimensions, 'inventory', function ($join): void {
$join->on('inventory.wms_co_id', '=', 'pairs.wms_co_id')
->on('inventory.inventory_sku_code', '=', 'pairs.inventory_sku_code')
->on('inventory.wh_id', '=', 'pairs.wh_id');
})
->select([
'pairs.wms_co_id', 'pairs.wh_id', 'pairs.inventory_sku_code',
'pairs.sku_code', 'pairs.category', 'pairs.warehouse_name',
])
->selectRaw('COALESCE(inventory.snapshot_qty, 0) AS snapshot_qty')
->selectRaw('COALESCE(inventory.captured_at, ?) AS snapshot_at', [$anchorAt]);
$movements = $this->confirmedMovements($headerGeneration, $itemGeneration, null)
->where('h.occurred_at', '>=', $rangeStart)
->where('h.occurred_at', '<', $rangeEnd)
->whereIn('h.wms_co_id', $pairRows->pluck('wms_co_id')->unique()->values()->all())
->whereIn('i.sku_code', $pairRows->pluck('inventory_sku_code')->unique()->values()->all())
->select([
'h.wms_co_id', 'h.io_type', 'h.occurred_at', 'i.sku_code as inventory_sku_code',
'i.qty as movement_qty', 'i.qty_type as movement_qty_type',
])
->selectRaw('COALESCE(NULLIF(h.wh_id, 0), 1) AS wh_id');
$signed = '(COALESCE(mv.movement_qty, 0) * COALESCE(mv.movement_qty_type, 0))';
$balanceAt = static fn (string $target): array => [
'a.snapshot_qty + COALESCE(SUM(CASE '
. 'WHEN a.snapshot_at < ? AND mv.occurred_at >= a.snapshot_at AND mv.occurred_at < ? THEN ' . $signed . ' '
. 'WHEN a.snapshot_at >= ? AND mv.occurred_at >= ? AND mv.occurred_at < a.snapshot_at THEN -' . $signed . ' '
. 'ELSE 0 END), 0)',
[$target, $target, $target, $target],
];
[$openingSql, $openingBindings] = $balanceAt($period['start']);
[$closingSql, $closingBindings] = $balanceAt($period['end']);
$monthly = $this->connection->query()->fromSub($anchors, 'a')
->leftJoinSub($movements, 'mv', function ($join): void {
$join->on('mv.wms_co_id', '=', 'a.wms_co_id')
->on('mv.wh_id', '=', 'a.wh_id')
->on('mv.inventory_sku_code', '=', 'a.inventory_sku_code');
})
->groupBy([
'a.wms_co_id', 'a.wh_id', 'a.inventory_sku_code', 'a.sku_code',
'a.category', 'a.warehouse_name',
'a.snapshot_qty', 'a.snapshot_at',
])
->select(['a.wms_co_id', 'a.wh_id', 'a.sku_code', 'a.category', 'a.warehouse_name'])
->selectRaw($openingSql . ' AS opening_qty', $openingBindings)
->selectRaw($closingSql . ' AS closing_qty', $closingBindings);
foreach (self::MOVEMENT_COLUMNS as $column => $type) {
$monthly->selectRaw(
'COALESCE(SUM(CASE WHEN mv.occurred_at >= ? AND mv.occurred_at < ? '
. 'AND mv.io_type = ? THEN ' . $signed . ' ELSE 0 END), 0) AS ' . $column,
[$period['start'], $period['end'], $type],
);
}
$monthly->selectRaw('0 AS opening_aux');
$monthlyRows = $this->connection->query()->fromSub($monthly, 'm')
->select('m.*')->orderBy('m.warehouse_name')->orderBy('m.wh_id')->orderBy('m.sku_code');
$rows = $this->boundedGet($monthlyRows)
->map(function (object $row) use ($period): object {
$warehouse = trim((string) ($row->warehouse_name ?? ''));
$subwarehouseId = (int) ($row->wh_id ?? 0);
$subwarehouse = $this->subwarehouseName($warehouse, $subwarehouseId);
return (object) [
'owner' => (string) config(
'upstream_mirrors.third_party_warehouse_storage.owner_name',
'常州简尚家居有限公司',
),
'month' => Carbon::parse($period['start'])->toDateString(),
'warehouse_provider' => $warehouse,
'warehouse' => $subwarehouse,
'product_code' => (string) ($row->sku_code ?? ''),
'category' => (string) ($row->category ?? ''),
'opening_quantity' => $row->opening_qty,
'closing_quantity' => $row->closing_qty,
'processing_in' => $row->processing_in,
'processing_out' => $row->processing_out,
'purchase_in' => $row->purchase_in,
'purchase_return' => $row->purchase_return,
'other_in' => $row->other_in,
'other_return' => $row->other_return,
'sales_out' => $row->sales_out,
'sales_return' => $row->sales_return,
'stocktake' => $row->stocktake,
'transfer_out' => $row->transfer_out,
'transfer_in' => $row->transfer_in,
'opening' => $row->opening_aux,
'other_out' => $row->other_out,
];
})->all();
return [
'data' => $rows,
'total' => $total,
'state' => $state,
'sync' => $this->syncStatus($state),
];
}
private function subwarehouseName(string $warehouseProvider, int $subwarehouseId): string
{
$specialName = self::SPECIAL_SUBWAREHOUSE_NAMES[$warehouseProvider][$subwarehouseId] ?? null;
if ($specialName !== null) {
return $specialName;
}
$suffix = self::SUBWAREHOUSE_SUFFIXES[$subwarehouseId] ?? '其他子仓(' . $subwarehouseId . ')';
return $warehouseProvider === '' ? $suffix : $warehouseProvider . $suffix;
}
/**
* @param array<string,mixed> $filters
* @return Generator<int,object>
*/
public function exportMonthlyInventoryReportRows(array $filters, int $batchSize = 2000): Generator
{
$batchSize = min(max($batchSize, 1), 5000);
$page = 1;
$exported = 0;
$total = null;
$state = $this->requireReadyState();
while (true) {
$result = $this->paginateMonthlyInventoryReport(
$filters,
$page,
$batchSize,
$total === null,
$state,
);
if ($total === null) {
$total = (int) ($result['total'] ?? 0);
}
$rows = $result['data'] ?? [];
if ($rows === []) {
return;
}
foreach ($rows as $row) {
yield $row;
}
$exported += count($rows);
if ($exported >= $total || count($rows) < $batchSize) {
return;
}
$page++;
}
}
/**
* 用户确认的聚水潭月报保留了“仓储方 + 子仓 + 商品”维度。
* 命中同月权威数据时整月直接读取,不再用当前库存倒推历史子仓余额。
*
* @param array{start:string,end:string} $period
* @param array<string,mixed> $filters
*/
private function monthlyInventoryAuthoritativeQuery(array $period, array $filters): ?Builder
{
try {
$schema = Schema::connection($this->connection->getName());
if (! $schema->hasTable(self::INVENTORY_REPORT_AUTHORITATIVE_STATE_TABLE)
|| ! $schema->hasTable(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE)) {
return null;
}
$month = Carbon::parse($period['start'])->startOfMonth()->toDateString();
$state = $this->connection->table(self::INVENTORY_REPORT_AUTHORITATIVE_STATE_TABLE)
->where('dataset_key', self::INVENTORY_REPORT_AUTHORITATIVE_DATASET_KEY)
->whereNotNull('active_generation')
->where('source_month', $month)
->first();
if ($state === null || trim((string) $state->active_generation) === '') {
return null;
}
$query = $this->connection->table(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE)
->where('generation', (string) $state->active_generation)
->where('report_month', $month)
->select([
'owner', 'report_month', 'warehouse_provider', 'warehouse',
'product_code', 'category', 'opening_quantity', 'closing_quantity',
'processing_in', 'processing_out', 'purchase_in', 'purchase_return',
'other_in', 'other_return', 'sales_out', 'sales_return', 'stocktake',
'transfer_out', 'transfer_in', 'opening', 'other_out',
]);
$productCode = trim((string) ($filters['product_code'] ?? ''));
if ($productCode !== '') {
$query->where('product_code', $productCode);
}
$this->containsFilter($query, 'warehouse_provider', $filters['warehouse_provider'] ?? '');
$this->containsFilter($query, 'category', $filters['category'] ?? '');
return $query;
} catch (Throwable) {
return null;
}
}
/**
* Every later month starts from the latest authoritative month closing quantity and
* rolls confirmed movements forward to the requested month. Current inventory is
* deliberately ignored because it cannot reproduce historical opening balances.
*
* @param array{start:string,end:string} $period
* @param array<string,mixed> $filters
* @param array{generation:string,source_month:string} $authority
* @param array<string,mixed> $state
* @return array{data:list<object>,total:int,state:array<string,mixed>,sync:array<string,mixed>}|null
*/
private function monthlyInventoryRollForward(
array $period,
array $filters,
array $authority,
string $headerGeneration,
string $itemGeneration,
string $inventoryGeneration,
int $page,
int $perPage,
bool $includeTotal,
array $state,
): ?array {
try {
$latestProducts = $this->connection->table(self::PRODUCT_TABLE)
->where('sku_type', 'normal')->whereNotNull('sku_id')->where('sku_id', '!=', '')
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
$latestWarehouses = $this->connection->table(self::WAREHOUSE_TABLE)
->whereNotNull('wms_co_id')->groupBy('wms_co_id')
->select('wms_co_id')->selectRaw('MAX(Id) AS Id');
$authorityRows = $this->connection
->table(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE . ' as authority')
->where('authority.generation', $authority['generation'])
->where('authority.report_month', $authority['source_month']);
$productCode = trim((string) ($filters['product_code'] ?? ''));
if ($productCode !== '') {
$authorityRows->where('authority.product_code', $productCode);
}
$this->containsFilter($authorityRows, 'authority.warehouse_provider', $filters['warehouse_provider'] ?? '');
$this->containsFilter($authorityRows, 'authority.category', $filters['category'] ?? '');
[$authorityWarehouseIdSql, $authorityWarehouseIdBindings] = $this->subwarehouseIdSql(
'authority.warehouse_provider',
'authority.warehouse',
);
$authorityCandidates = (clone $authorityRows)
->leftJoin(self::WAREHOUSE_TABLE . ' as authority_warehouse', function ($join): void {
$join->on('authority_warehouse.name', '=', 'authority.warehouse_provider');
})
->leftJoinSub($latestWarehouses, 'authority_latest_warehouse', function ($join): void {
$join->on('authority_latest_warehouse.wms_co_id', '=', 'authority_warehouse.wms_co_id')
->on('authority_latest_warehouse.Id', '=', 'authority_warehouse.Id');
})->selectRaw(
'authority.owner AS owner, authority.warehouse_provider AS warehouse_provider, '
. 'authority.warehouse AS source_warehouse, authority_warehouse.wms_co_id AS wms_co_id, '
. $authorityWarehouseIdSql . ' AS wh_id, authority.product_code AS sku_code, '
. 'authority.category AS category, authority.closing_quantity AS opening_qty',
$authorityWarehouseIdBindings,
);
$inventoryDimensions = $this->connection->table(self::SUBWAREHOUSE_INVENTORY_TABLE . ' as inventory')
->where('inventory.generation', $inventoryGeneration)
->where('inventory.qty', '!=', 0)
->whereNotNull('inventory.sku_code')->where('inventory.sku_code', '!=', '')
->joinSub($latestProducts, 'latest_product', 'latest_product.sku_id', '=', 'inventory.sku_code')
->join(self::PRODUCT_TABLE . ' as s', function ($join): void {
$join->on('s.sku_id', '=', 'latest_product.sku_id')
->on('s.autoid', '=', 'latest_product.autoid');
})
->joinSub($latestWarehouses, 'latest_warehouse', 'latest_warehouse.wms_co_id', '=', 'inventory.wms_co_id')
->join(self::WAREHOUSE_TABLE . ' as w', function ($join): void {
$join->on('w.wms_co_id', '=', 'latest_warehouse.wms_co_id')
->on('w.Id', '=', 'latest_warehouse.Id');
});
if ($productCode !== '') {
$inventoryDimensions->where('s.sku_id', $productCode);
}
$this->containsFilter($inventoryDimensions, 'w.name', $filters['warehouse_provider'] ?? '');
$this->containsFilter($inventoryDimensions, 's.category', $filters['category'] ?? '');
$owner = (string) config(
'upstream_mirrors.third_party_warehouse_storage.owner_name',
'常州简尚家居有限公司',
);
[$inventoryWarehouseSql, $inventoryWarehouseBindings] = $this->subwarehouseNameSql(
'w.name',
'inventory.wh_id',
);
$inventoryDimensions->whereNotExists(function (Builder $prior) use (
$authority,
$inventoryWarehouseSql,
$inventoryWarehouseBindings,
): void {
$prior->selectRaw('1')
->from(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE . ' as prior')
->where('prior.generation', $authority['generation'])
->where('prior.report_month', $authority['source_month'])
->whereColumn('prior.product_code', 'inventory.sku_code')
->whereColumn('prior.warehouse_provider', 'w.name')
->whereRaw('prior.warehouse = ' . $inventoryWarehouseSql, $inventoryWarehouseBindings);
});
$inventoryDimensions->distinct()
->selectRaw(
'? AS owner, w.name AS warehouse_provider, '
. $inventoryWarehouseSql . ' AS source_warehouse, '
. 'inventory.wms_co_id AS wms_co_id, inventory.wh_id AS wh_id, '
. 'inventory.sku_code AS sku_code, s.category AS category, 0 AS opening_qty',
array_merge([$owner], $inventoryWarehouseBindings),
);
$candidateUnion = $authorityCandidates->unionAll($inventoryDimensions);
$candidates = $this->connection->query()->fromSub($candidateUnion, 'candidate_rows');
$pageCandidates = (clone $candidates)
->orderBy('warehouse_provider')->orderBy('wh_id')->orderBy('sku_code')
->forPage($page, $perPage);
if ($includeTotal) {
$pageCandidates->select('candidate_rows.*')->selectRaw('COUNT(*) OVER() AS report_total');
}
$pairRows = $this->boundedGet(clone $pageCandidates);
$total = $includeTotal ? (int) ($pairRows->first()->report_total ?? 0) : 0;
if ($pairRows->isEmpty()) {
if ($includeTotal && $page > 1) {
$total = $this->boundedCount(clone $authorityRows)
+ $this->boundedCount(clone $inventoryDimensions);
}
return [
'data' => [],
'total' => $total,
'state' => $state,
'sync' => $this->syncStatus($state),
];
}
$authorityEnd = Carbon::parse($authority['source_month'])
->addMonth()->startOfMonth()->format('Y-m-d 00:00:00');
$movementPairs = $pairRows->filter(
static fn (object $row): bool => $row->wms_co_id !== null && trim((string) $row->sku_code) !== '',
)->unique(static fn (object $row): string => implode(':', [
(string) $row->wms_co_id,
(string) $row->wh_id,
(string) $row->sku_code,
]))->values();
$movementRows = collect();
if ($movementPairs->isNotEmpty()) {
$movements = $this->confirmedMovements($headerGeneration, $itemGeneration, null)
->where('h.occurred_at', '>=', $authorityEnd)
->where('h.occurred_at', '<', $period['end'])
->where(function (Builder $pairs) use ($movementPairs): void {
foreach ($movementPairs as $pair) {
$pairs->orWhere(function (Builder $candidate) use ($pair): void {
$candidate->where('h.wms_co_id', $pair->wms_co_id)
->whereRaw('COALESCE(NULLIF(h.wh_id, 0), 1) = ?', [(int) $pair->wh_id])
->where('i.sku_code', $pair->sku_code);
});
}
})
->groupBy(['h.wms_co_id', 'i.sku_code'])
->groupByRaw('COALESCE(NULLIF(h.wh_id, 0), 1)')
->select(['h.wms_co_id', 'i.sku_code'])
->selectRaw('COALESCE(NULLIF(h.wh_id, 0), 1) AS wh_id');
$signed = '(COALESCE(i.qty, 0) * COALESCE(i.qty_type, 0))';
$movements->selectRaw(
'COALESCE(SUM(CASE WHEN h.occurred_at >= ? AND h.occurred_at < ? '
. 'THEN ' . $signed . ' ELSE 0 END), 0) AS rolled_opening_delta',
[$authorityEnd, $period['start']],
);
foreach (self::MOVEMENT_COLUMNS as $column => $type) {
$expression = 'COALESCE(SUM(CASE WHEN h.occurred_at >= ? AND h.occurred_at < ? '
. 'AND h.io_type = ? THEN ' . $signed . ' ELSE 0 END), 0)';
$movements->selectRaw($expression . ' AS ' . $column, [$period['start'], $period['end'], $type]);
}
$movementRows = $this->boundedGet($movements)->keyBy(
static fn (object $row): string => implode(':', [
(string) $row->wms_co_id,
(string) $row->wh_id,
(string) $row->sku_code,
]),
);
}
$rows = $pairRows->map(function (object $row) use ($movementRows, $period): object {
$key = implode(':', [(string) $row->wms_co_id, (string) $row->wh_id, (string) $row->sku_code]);
$movement = $movementRows->get($key);
$sourceWarehouse = trim((string) ($row->source_warehouse ?? ''));
$warehouse = $sourceWarehouse !== ''
? $sourceWarehouse
: $this->subwarehouseName(
trim((string) $row->warehouse_provider),
(int) $row->wh_id,
);
$rolledOpening = (float) $row->opening_qty + (float) ($movement->rolled_opening_delta ?? 0);
$closingQuantity = $rolledOpening;
foreach (array_keys(self::MOVEMENT_COLUMNS) as $column) {
$closingQuantity += (float) ($movement->{$column} ?? 0);
}
return (object) [
'owner' => $row->owner,
'month' => Carbon::parse($period['start'])->toDateString(),
'warehouse_provider' => $row->warehouse_provider,
'warehouse' => $warehouse,
'product_code' => $row->sku_code,
'category' => $row->category,
'opening_quantity' => $rolledOpening,
'closing_quantity' => $closingQuantity,
'processing_in' => $movement->processing_in ?? 0,
'processing_out' => $movement->processing_out ?? 0,
'purchase_in' => $movement->purchase_in ?? 0,
'purchase_return' => $movement->purchase_return ?? 0,
'other_in' => $movement->other_in ?? 0,
'other_return' => $movement->other_return ?? 0,
'sales_out' => $movement->sales_out ?? 0,
'sales_return' => $movement->sales_return ?? 0,
'stocktake' => $movement->stocktake ?? 0,
'transfer_out' => $movement->transfer_out ?? 0,
'transfer_in' => $movement->transfer_in ?? 0,
'opening' => 0,
'other_out' => $movement->other_out ?? 0,
];
})->all();
return [
'data' => $rows,
'total' => $total,
'state' => $state,
'sync' => $this->syncStatus($state),
];
} catch (Throwable $exception) {
if ($this->isStatementTimeout($exception)) {
throw $exception;
}
report($exception);
return null;
}
}
/**
* @param array{start:string,end:string} $period
* @param array<string,mixed> $filters
* @param array{generation:string,source_month:string} $authority
*/
private function monthlyInventoryReportCacheKey(
array $period,
array $filters,
array $authority,
string $headerGeneration,
string $itemGeneration,
string $inventoryGeneration,
int $page,
int $perPage,
): string {
$normalizedFilters = [
'product_code' => trim((string) ($filters['product_code'] ?? '')),
'warehouse_provider' => trim((string) ($filters['warehouse_provider'] ?? '')),
'category' => trim((string) ($filters['category'] ?? '')),
];
return 'inventory-report:monthly-page:v3:' . hash('sha256', (string) json_encode([
'period' => $period,
'filters' => $normalizedFilters,
'authority' => $authority,
'header_generation' => $headerGeneration,
'item_generation' => $itemGeneration,
'inventory_generation' => $inventoryGeneration,
'page' => $page,
'per_page' => $perPage,
], JSON_UNESCAPED_UNICODE));
}
/** @param callable():?array $resolver */
private function rememberMonthlyInventoryReport(string $key, callable $resolver): ?array
{
$cached = Cache::get($key);
if (is_array($cached)) {
return $cached;
}
try {
return Cache::lock($key . ':lock', self::REPORT_CACHE_LOCK_SECONDS)
->block(self::REPORT_CACHE_WAIT_SECONDS, function () use ($key, $resolver): ?array {
$cached = Cache::get($key);
if (is_array($cached)) {
return $cached;
}
$resolved = $resolver();
if (is_array($resolved)) {
Cache::put($key, $resolved, self::REPORT_CACHE_SECONDS);
}
return $resolved;
});
} catch (LockTimeoutException $exception) {
$cached = Cache::get($key);
if (is_array($cached)) {
return $cached;
}
throw new RuntimeException('inventory_report_generation_busy', 0, $exception);
}
}
/** @param array{start:string,end:string} $period */
private function previousMonthlyInventoryAuthority(array $period): ?array
{
$state = $this->connection->table(self::INVENTORY_REPORT_AUTHORITATIVE_STATE_TABLE)
->where('dataset_key', self::INVENTORY_REPORT_AUTHORITATIVE_DATASET_KEY)
->whereNotNull('active_generation')
->whereDate('source_month', '<', Carbon::parse($period['start'])->startOfMonth())
->orderByDesc('source_month')
->first();
if ($state === null || trim((string) $state->active_generation) === '') {
return null;
}
return [
'generation' => (string) $state->active_generation,
'source_month' => Carbon::parse($state->source_month)->startOfMonth()->toDateString(),
];
}
/** @return array{0:string,1:list<mixed>} */
private function subwarehouseIdSql(string $providerColumn, string $warehouseColumn): array
{
$cases = [];
$bindings = [];
foreach (self::SPECIAL_SUBWAREHOUSE_NAMES as $provider => $names) {
foreach ($names as $id => $name) {
$cases[] = "WHEN {$providerColumn} = ? AND {$warehouseColumn} = ? THEN {$id}";
array_push($bindings, $provider, $name);
}
}
foreach (self::SUBWAREHOUSE_SUFFIXES as $id => $suffix) {
$cases[] = 'WHEN ' . $warehouseColumn . ' = '
. $this->concatSql([$providerColumn, '?']) . " THEN {$id}";
$bindings[] = $suffix;
}
return ['CASE ' . implode(' ', $cases) . ' ELSE -1 END', $bindings];
}
/** @return array{0:string,1:list<mixed>} */
private function subwarehouseNameSql(string $providerColumn, string $subwarehouseColumn): array
{
$cases = [];
$bindings = [];
foreach (self::SPECIAL_SUBWAREHOUSE_NAMES as $provider => $names) {
foreach ($names as $id => $name) {
$cases[] = "WHEN {$providerColumn} = ? AND {$subwarehouseColumn} = {$id} THEN ?";
array_push($bindings, $provider, $name);
}
}
foreach (self::SUBWAREHOUSE_SUFFIXES as $id => $suffix) {
$cases[] = "WHEN {$subwarehouseColumn} = {$id} THEN " . $this->concatSql([$providerColumn, '?']);
$bindings[] = $suffix;
}
$fallback = $this->concatSql([$providerColumn, "'其他子仓('", $subwarehouseColumn, "')'"]);
return ['CASE ' . implode(' ', $cases) . ' ELSE ' . $fallback . ' END', $bindings];
}
/** @param list<string> $parts */
private function concatSql(array $parts): string
{
return $this->connection->getDriverName() === 'sqlite'
? '(' . implode(' || ', $parts) . ')'
: 'CONCAT(' . implode(', ', $parts) . ')';
}
/**
* @param array<string,mixed> $filters
* @return array{count:int,opening_qty:float,closing_qty:float}
*/
public function summary(array $filters): array
{
$state = $this->requireReadyState();
return $this->summarizeQuery($this->filteredReportQuery(
$filters,
$state,
$this->hasProductFilters($filters),
));
}
/**
* @param array<string,mixed> $filters
* @return array{suppliers:list<string>,categories:list<string>}
*/
public function filterOptions(array $filters = []): array
{
$state = $this->requireReadyState();
$query = $this->filteredReportQuery($filters, $state);
$suppliers = (clone $query)->whereNotNull('supplier')->where('supplier', '!=', '')
->distinct()->orderBy('supplier')->pluck('supplier')
->map(static fn (mixed $value): string => (string) $value)->all();
$categories = (clone $query)->whereNotNull('category')->where('category', '!=', '')
->distinct()->orderBy('category')->pluck('category')
->map(static fn (mixed $value): string => (string) $value)->all();
return ['suppliers' => $suppliers, 'categories' => $categories];
}
/**
* @param array<string,mixed> $filters
* @return Generator<int,array<string,mixed>>
*/
public function exportRows(
array $filters,
int $batchSize = 500,
bool $includeProductDetails = true,
): Generator
{
$state = $this->requireReadyState();
$batchSize = min(max($batchSize, 1), 5000);
$lastDate = null;
$lastSku = null;
while (true) {
// Settlement only needs quantities and SKU/date fields. Skipping the
// large product catalogue join keeps the page calculation bounded
// while the normal CSV export retains product details.
$query = $this->filteredReportQuery($filters, $state, $includeProductDetails);
if ($lastDate !== null && $lastSku !== null) {
$query->where(function (Builder $cursor) use ($lastDate, $lastSku): void {
$cursor->where('report_date', '>', $lastDate)
->orWhere(function (Builder $sameDate) use ($lastDate, $lastSku): void {
$sameDate->where('report_date', $lastDate)
->where('sku_code', '>', $lastSku);
});
});
}
$rows = $query->orderBy('report_date')->orderBy('sku_code')->limit($batchSize)->get();
if ($rows->isEmpty()) {
return;
}
foreach ($rows as $row) {
yield $this->mapRow($row);
}
$lastDate = (string) $rows->last()->report_date;
$lastSku = (string) $rows->last()->sku_code;
if ($rows->count() < $batchSize) {
return;
}
}
}
/** @param array<string,mixed> $state */
private function filteredReportQuery(
array $filters,
array $state,
bool $includeProductDetails = true,
): Builder
{
$query = $this->connection->query()->fromSub(
$this->reportQuery($filters, $state, $includeProductDetails),
'report',
);
$this->containsFilter($query, 'sku_code', $filters['sku_code'] ?? '');
$this->containsFilter($query, 'product_name', $filters['product_name'] ?? '');
$this->containsFilter($query, 'supplier', $filters['supplier'] ?? '');
$this->containsFilter($query, 'category', $filters['category'] ?? '');
$nonZero = ['opening_qty', 'closing_qty', ...array_keys(self::MOVEMENT_COLUMNS)];
$query->where(function (Builder $values) use ($nonZero): void {
foreach ($nonZero as $index => $column) {
$method = $index === 0 ? 'where' : 'orWhere';
$values->{$method}($column, '<>', 0);
}
});
return $query;
}
/** @param array<string,mixed> $state */
private function reportQuery(
array $filters,
array $state,
bool $includeProductDetails,
): Builder
{
$period = $this->period($filters);
$headerGeneration = $this->generation($state, 'active_header_generation');
$itemGeneration = $this->generation($state, 'active_item_generation');
$inventoryGeneration = $this->generation($state, 'active_inventory_generation');
$authoritativeAnchor = $this->authoritativeAnchor();
$anchorAt = $authoritativeAnchor['anchor_at'] ?? $this->stateAnchor($state);
if ($headerGeneration === null || $itemGeneration === null || $inventoryGeneration === null) {
throw new RuntimeException('third_party_warehouse_mirror_not_ready');
}
$authoritativeRows = $this->authoritativeReportQuery(
$period,
$authoritativeAnchor,
$includeProductDetails,
);
if ($authoritativeRows !== null) {
return $authoritativeRows;
}
$skuRangeStart = min($period['start'], $anchorAt);
$skuRangeEnd = max($period['end'], $anchorAt);
$movementSkus = $this->confirmedMovements($headerGeneration, $itemGeneration)
->where('h.occurred_at', '>=', $skuRangeStart)
->where('h.occurred_at', '<', $skuRangeEnd)
->select('i.sku_code');
if ($authoritativeAnchor !== null) {
$skuUnion = $this->connection->table(self::AUTHORITATIVE_ANCHOR_TABLE . ' as inventory_skus')
->where('inventory_skus.generation', $authoritativeAnchor['generation'])
->whereNotNull('inventory_skus.sku_code')
->where('inventory_skus.sku_code', '!=', '')
->select('inventory_skus.sku_code')
->union($movementSkus);
$anchors = $this->connection->query()->fromSub($skuUnion, 'sku_set')
->leftJoin(self::AUTHORITATIVE_ANCHOR_TABLE . ' as inventory', function ($join) use ($authoritativeAnchor): void {
$join->on('inventory.sku_code', '=', 'sku_set.sku_code')
->where('inventory.generation', '=', $authoritativeAnchor['generation']);
})
->select('sku_set.sku_code')
->selectRaw('COALESCE(inventory.qty, 0) AS snapshot_qty')
->selectRaw('COALESCE(inventory.anchor_at, ?) AS snapshot_at', [$anchorAt]);
} else {
$skuUnion = $this->connection->table(self::INVENTORY_TABLE . ' as inventory_skus')
->where('inventory_skus.generation', $inventoryGeneration)
->where('inventory_skus.wms_co_id', self::WAREHOUSE_ID)
->whereNotNull('inventory_skus.sku_code')
->where('inventory_skus.sku_code', '!=', '')
->select('inventory_skus.sku_code')
->union($movementSkus);
$anchors = $this->connection->query()->fromSub($skuUnion, 'sku_set')
->leftJoin(self::INVENTORY_TABLE . ' as inventory', function ($join) use ($inventoryGeneration): void {
$join->on('inventory.sku_code', '=', 'sku_set.sku_code')
->where('inventory.generation', '=', $inventoryGeneration)
->where('inventory.wms_co_id', '=', self::WAREHOUSE_ID);
})
->select('sku_set.sku_code')
->selectRaw('COALESCE(inventory.qty, 0) AS snapshot_qty')
->selectRaw('COALESCE(inventory.captured_at, ?) AS snapshot_at', [$anchorAt]);
}
$dateSkuAnchors = $this->connection->query()
->fromSub($this->dateSeries($period['start'], $period['end']), 'd')
->crossJoinSub($anchors, 'a')
->select([
'd.report_date',
'd.day_start',
'd.day_end',
'a.sku_code',
'a.snapshot_qty',
'a.snapshot_at',
]);
$signedQuantity = '(COALESCE(movement_qty, 0) * COALESCE(movement_qty_type, 0))';
$forwardAnchorOperator = $authoritativeAnchor === null ? '>' : '>=';
$closingExpression = 'da.snapshot_qty + COALESCE(SUM(CASE '
. 'WHEN da.snapshot_at < da.day_end AND mv.occurred_at ' . $forwardAnchorOperator . ' da.snapshot_at '
. 'AND mv.occurred_at < da.day_end THEN ' . $signedQuantity . ' '
. 'WHEN da.snapshot_at >= da.day_end AND mv.occurred_at >= da.day_end '
. 'AND mv.occurred_at < da.snapshot_at THEN -' . $signedQuantity . ' '
. 'ELSE 0 END), 0)';
$allMovements = $this->confirmedMovements($headerGeneration, $itemGeneration)
->where('h.occurred_at', '>=', $skuRangeStart)
->where('h.occurred_at', '<', $skuRangeEnd)
->select([
'i.sku_code',
'i.qty as movement_qty',
'i.qty_type as movement_qty_type',
'h.occurred_at',
]);
$balances = $this->connection->query()->fromSub($dateSkuAnchors, 'da')
->leftJoinSub($allMovements, 'mv', 'mv.sku_code', '=', 'da.sku_code')
->groupBy(
'da.report_date',
'da.day_start',
'da.day_end',
'da.sku_code',
'da.snapshot_qty',
'da.snapshot_at',
)
->select(['da.report_date', 'da.sku_code'])
->selectRaw($closingExpression . ' AS closing_qty');
$periodMovements = $this->confirmedMovements($headerGeneration, $itemGeneration)
->where('h.occurred_at', '>=', $period['start'])
->where('h.occurred_at', '<', $period['end'])
->groupBy('i.sku_code')
->groupByRaw('DATE(h.occurred_at)')
->select('i.sku_code')
->selectRaw('DATE(h.occurred_at) AS report_date');
foreach (self::MOVEMENT_COLUMNS as $column => $type) {
$periodMovements->selectRaw(
'COALESCE(SUM(CASE WHEN h.io_type = ? THEN '
. '(COALESCE(i.qty, 0) * COALESCE(i.qty_type, 0)) ELSE 0 END), 0) AS ' . $column,
[$type],
);
}
$report = $this->connection->query()->fromSub($balances, 'b')
->leftJoinSub($periodMovements, 'm', function ($join): void {
$join->on('m.sku_code', '=', 'b.sku_code')
->on('m.report_date', '=', 'b.report_date');
});
if ($includeProductDetails) {
$latestProducts = $this->connection->table(self::PRODUCT_TABLE)
->where('sku_type', 'normal')
->whereNotNull('sku_id')->where('sku_id', '!=', '')
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
$report->leftJoinSub(
$latestProducts,
'latest_product',
'latest_product.sku_id',
'=',
'b.sku_code',
)->leftJoin(self::PRODUCT_TABLE . ' as s', function ($join): void {
$join->on('s.sku_id', '=', 'latest_product.sku_id')
->on('s.autoid', '=', 'latest_product.autoid')
->where('s.sku_type', '=', 'normal');
});
}
$movementSelects = [];
foreach (array_keys(self::MOVEMENT_COLUMNS) as $column) {
$movementSelects[$column] = 'COALESCE(m.' . $column . ', 0)';
}
$netMovement = implode(' + ', array_values($movementSelects));
$openingQuantity = '(b.closing_qty - (' . $netMovement . '))';
$owner = (string) config(
'upstream_mirrors.third_party_warehouse_storage.owner_name',
'常州简尚家居有限公司',
);
$report->selectRaw('? AS owner', [$owner])
->addSelect('b.report_date')
->addSelect('b.sku_code');
if ($includeProductDetails) {
$report->addSelect([
$this->connection->raw('COALESCE(s.pic_big, s.pic) AS image_url'),
$this->connection->raw('s.name AS product_name'),
$this->connection->raw('COALESCE(s.short_name, s.name) AS short_name'),
$this->connection->raw('s.supplier_name AS supplier'),
$this->connection->raw('s.category AS category'),
$this->connection->raw('s.vc_name AS virtual_category'),
$this->connection->raw('s.supplier_sku_id AS supplier_style_no'),
$this->connection->raw('COALESCE(s.properties_value, s.color) AS color_spec'),
$this->connection->raw('s.brand AS brand'),
$this->connection->raw('s.unit AS unit'),
]);
}
$report->selectRaw($openingQuantity . ' AS opening_qty')
->selectRaw('b.closing_qty AS closing_qty');
foreach ($movementSelects as $column => $expression) {
$report->selectRaw($expression . ' AS ' . $column);
}
$report->selectRaw('0 AS opening_aux');
return $report;
}
private function dateSeries(string $start, string $end): Builder
{
$cursor = Carbon::parse($start)->startOfDay();
$through = Carbon::parse($end)->startOfDay();
$dates = null;
while ($cursor < $through) {
$dayStart = $cursor->copy();
$dayEnd = $cursor->copy()->addDay();
$row = $this->connection->query()->selectRaw(
'? AS report_date, ? AS day_start, ? AS day_end',
[$dayStart->toDateString(), $dayStart->toDateTimeString(), $dayEnd->toDateTimeString()],
);
$dates = $dates === null ? $row : $dates->unionAll($row);
$cursor->addDay();
}
if (! $dates instanceof Builder) {
throw new RuntimeException('third_party_warehouse_period_invalid');
}
return $dates;
}
private function confirmedMovements(
string $headerGeneration,
string $itemGeneration,
?int $warehouseId = self::WAREHOUSE_ID,
): Builder
{
$query = $this->connection->table(self::ITEM_TABLE . ' as i')
->join(self::HEADER_TABLE . ' as h', function ($join) use ($headerGeneration): void {
$join->on('h.io_id', '=', 'i.io_id')
->where('h.generation', '=', $headerGeneration);
})
->where('i.generation', $itemGeneration)
->where('h.status', 'Confirmed')
->whereIn('h.io_type', array_values(self::MOVEMENT_COLUMNS))
->whereNotExists(function (Builder $newer) use ($headerGeneration): void {
$newer->selectRaw('1')->from(self::HEADER_TABLE . ' as newer')
->where('newer.generation', $headerGeneration)
->whereColumn('newer.io_id', 'h.io_id')
->whereColumn('newer.id', '>', 'h.id');
})
->whereNotNull('i.sku_code')->where('i.sku_code', '!=', '');
if ($warehouseId !== null) {
$query->where('h.wms_co_id', $warehouseId);
}
return $query;
}
/**
* @param array<string,mixed> $filters
* @return array{dimension:string,value:string,start:string,end:string,days:int}
*/
private function period(array $filters): array
{
$dimension = ($filters['dimension'] ?? 'month') === 'day' ? 'day' : 'month';
$candidate = trim((string) ($filters['period'] ?? ''));
if ($dimension === 'day') {
if (preg_match('/^\d{4}-\d{2}-\d{2}$/', $candidate) !== 1) {
$candidate = trim((string) ($filters['date'] ?? ''));
}
try {
$start = preg_match('/^\d{4}-\d{2}-\d{2}$/', $candidate) === 1
? Carbon::createFromFormat('!Y-m-d', $candidate)
: now()->startOfDay();
} catch (Throwable) {
$start = now()->startOfDay();
}
if ($start->lt(Carbon::parse(self::START_DATE)->startOfDay())) {
$start = Carbon::parse(self::START_DATE)->startOfDay();
}
return [
'dimension' => 'day',
'value' => $start->format('Y-m-d'),
'start' => $start->format('Y-m-d 00:00:00'),
'end' => $start->copy()->addDay()->format('Y-m-d 00:00:00'),
'days' => 1,
];
}
if (preg_match('/^\d{4}-\d{2}$/', $candidate) !== 1) {
$candidate = trim((string) ($filters['month'] ?? ''));
}
try {
$start = preg_match('/^\d{4}-\d{2}$/', $candidate) === 1
? Carbon::createFromFormat('!Y-m', $candidate)->startOfMonth()
: now()->startOfMonth();
} catch (Throwable) {
$start = now()->startOfMonth();
}
$minimum = Carbon::parse(self::START_DATE)->startOfMonth();
if ($start->lt($minimum)) {
$start = $minimum;
}
return [
'dimension' => 'month',
'value' => $start->format('Y-m'),
'start' => $start->format('Y-m-d 00:00:00'),
'end' => $start->copy()->addMonth()->format('Y-m-d 00:00:00'),
'days' => $start->daysInMonth,
];
}
/** @return array{dimension:string,value:string,start:string,end:string,days:int} */
private function inventoryReportPeriod(string $month): array
{
try {
$start = preg_match('/^\d{4}-\d{2}$/', trim($month)) === 1
? Carbon::createFromFormat('!Y-m', trim($month))->startOfMonth()
: now()->startOfMonth();
} catch (Throwable) {
$start = now()->startOfMonth();
}
$minimum = Carbon::parse(self::INVENTORY_REPORT_START_DATE)->startOfMonth();
if ($start->lt($minimum)) {
$start = $minimum;
}
return [
'dimension' => 'month',
'value' => $start->format('Y-m'),
'start' => $start->format('Y-m-d 00:00:00'),
'end' => $start->copy()->addMonth()->format('Y-m-d 00:00:00'),
'days' => $start->daysInMonth,
];
}
/** @param array<string,mixed> $state */
private function stateAnchor(array $state): string
{
$timestamp = $this->timestamp($state['last_succeeded_at'] ?? null);
return $timestamp === null
? now()->toDateTimeString()
: Carbon::createFromTimestamp($timestamp)->toDateTimeString();
}
/**
* @param array{generation:string,anchor_at:string,source_period_start:?string,source_period_end:?string}|null $authoritativeAnchor
*/
private function authoritativeReportQuery(
array $period,
?array $authoritativeAnchor,
bool $includeProductDetails,
): ?Builder {
if ($authoritativeAnchor === null) {
return null;
}
try {
$schema = Schema::connection($this->connection->getName());
if (! $schema->hasTable(self::AUTHORITATIVE_ROW_TABLE)) {
return null;
}
$sourceStart = $authoritativeAnchor['source_period_start'] ?? null;
$sourceEnd = $authoritativeAnchor['source_period_end'] ?? null;
if (! is_string($sourceStart) || ! is_string($sourceEnd)
|| $sourceStart === '' || $sourceEnd === '') {
return null;
}
$sourceStartAt = Carbon::parse($sourceStart)->startOfDay();
$sourceEndAt = Carbon::parse($sourceEnd)->addDay()->startOfDay();
$periodStartAt = Carbon::parse($period['start']);
$periodEndAt = Carbon::parse($period['end']);
if ($periodStartAt->lt($sourceStartAt) || $periodEndAt->gt($sourceEndAt)) {
return null;
}
$rows = $this->connection->table(self::AUTHORITATIVE_ROW_TABLE . ' as b')
->where('b.generation', $authoritativeAnchor['generation'])
->where('b.report_date', '>=', $periodStartAt->toDateString())
->where('b.report_date', '<', $periodEndAt->toDateString());
$owner = (string) config(
'upstream_mirrors.third_party_warehouse_storage.owner_name',
'常州简尚家居有限公司',
);
$rows->selectRaw('? AS owner', [$owner])->addSelect(['b.report_date', 'b.sku_code']);
if ($includeProductDetails) {
$latestProducts = $this->connection->table(self::PRODUCT_TABLE)
->where('sku_type', 'normal')
->whereNotNull('sku_id')->where('sku_id', '!=', '')
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
$rows->leftJoinSub(
$latestProducts,
'latest_product',
'latest_product.sku_id',
'=',
'b.sku_code',
)->leftJoin(self::PRODUCT_TABLE . ' as s', function ($join): void {
$join->on('s.sku_id', '=', 'latest_product.sku_id')
->on('s.autoid', '=', 'latest_product.autoid')
->where('s.sku_type', '=', 'normal');
})->addSelect([
$this->connection->raw('COALESCE(s.pic_big, s.pic) AS image_url'),
$this->connection->raw('s.name AS product_name'),
$this->connection->raw('COALESCE(s.short_name, s.name) AS short_name'),
$this->connection->raw('s.supplier_name AS supplier'),
$this->connection->raw('s.category AS category'),
$this->connection->raw('s.vc_name AS virtual_category'),
$this->connection->raw('s.supplier_sku_id AS supplier_style_no'),
$this->connection->raw('COALESCE(s.properties_value, s.color) AS color_spec'),
$this->connection->raw('s.brand AS brand'),
$this->connection->raw('s.unit AS unit'),
]);
}
$rows->addSelect([
'b.opening_qty', 'b.closing_qty', 'b.processing_in', 'b.processing_out',
'b.purchase_in', 'b.purchase_return', 'b.other_in', 'b.other_return',
'b.sales_out', 'b.sales_return', 'b.stocktake', 'b.transfer_out',
'b.transfer_in', 'b.opening_aux', 'b.other_out',
]);
return $rows;
} catch (Throwable) {
return null;
}
}
/** @return array{generation:string,anchor_at:string,source_period_start:?string,source_period_end:?string}|null */
private function authoritativeAnchor(): ?array
{
try {
$schema = Schema::connection($this->connection->getName());
if (! $schema->hasTable(self::AUTHORITATIVE_STATE_TABLE)
|| ! $schema->hasTable(self::AUTHORITATIVE_ANCHOR_TABLE)) {
return null;
}
$state = $this->connection->table(self::AUTHORITATIVE_STATE_TABLE)
->where('dataset_key', self::AUTHORITATIVE_DATASET_KEY)
->whereNotNull('active_generation')
->whereNotNull('anchor_at')
->first();
if ($state === null) {
return null;
}
$generation = trim((string) $state->active_generation);
$anchorAt = $this->timestamp($state->anchor_at ?? null);
if ($generation === '' || $anchorAt === null) {
return null;
}
return [
'generation' => $generation,
'anchor_at' => Carbon::createFromTimestamp($anchorAt)->toDateTimeString(),
'source_period_start' => isset($state->source_period_start)
? (string) $state->source_period_start : null,
'source_period_end' => isset($state->source_period_end)
? (string) $state->source_period_end : null,
];
} catch (Throwable) {
return null;
}
}
/** @return array<string,mixed> */
private function requireReadyState(): array
{
$state = $this->state();
if (! is_array($state)
|| ($state['initial_completed_at'] ?? null) === null
|| $this->generation($state, 'active_header_generation') === null
|| $this->generation($state, 'active_item_generation') === null
|| $this->generation($state, 'active_inventory_generation') === null
|| ! $this->dependenciesReady()) {
throw new RuntimeException('third_party_warehouse_mirror_not_ready');
}
return $state;
}
private function dependenciesReady(): bool
{
try {
$required = [self::PRODUCT_DATASET_KEY];
$ready = $this->connection->table(self::GENERIC_STATE_TABLE)
->whereIn('dataset_key', $required)
->whereNotNull('initial_completed_at')
->pluck('dataset_key')
->map(static fn (mixed $dataset): string => (string) $dataset)
->all();
return array_diff($required, $ready) === [];
} catch (Throwable) {
return false;
}
}
private function warehouseDependencyReady(): bool
{
try {
if (! $this->connection->getSchemaBuilder()->hasTable(self::WAREHOUSE_TABLE)) {
return false;
}
return $this->connection->table(self::GENERIC_STATE_TABLE)
->where('dataset_key', self::WAREHOUSE_DATASET_KEY)
->whereNotNull('initial_completed_at')
->whereIn('status', ['succeeded', 'running', 'failed'])
->exists();
} catch (Throwable) {
return false;
}
}
/** @param array<string,mixed> $state */
private function generation(array $state, string $key): ?string
{
$value = trim((string) ($state[$key] ?? ''));
return $value === '' ? null : $value;
}
private function timestamp(mixed $value): ?int
{
if ($value instanceof CarbonInterface) {
return $value->getTimestamp();
}
if (! is_string($value) || trim($value) === '') {
return null;
}
try {
return Carbon::parse($value)->getTimestamp();
} catch (Throwable) {
return null;
}
}
/**
* @return array{count:int,opening_qty:float,closing_qty:float}
*/
private function summarizeQuery(Builder $query): array
{
$row = $query->selectRaw(
'COUNT(*) AS row_count, '
. 'COALESCE(SUM(opening_qty), 0) AS opening_qty, '
. 'COALESCE(SUM(closing_qty), 0) AS closing_qty'
)->first();
return [
'count' => (int) ($row->row_count ?? 0),
'opening_qty' => round((float) ($row->opening_qty ?? 0), 4),
'closing_qty' => round((float) ($row->closing_qty ?? 0), 4),
];
}
/** @return array<string,mixed> */
private function mapRow(object $row): array
{
$mapped = [];
foreach (self::OUTPUT_COLUMNS as $column) {
$mapped[$column] = $column === 'date'
? ($row->report_date ?? null)
: ($row->{$column} ?? null);
}
foreach (['opening_qty', 'closing_qty', ...array_keys(self::MOVEMENT_COLUMNS), 'opening_aux'] as $column) {
$mapped[$column] = round((float) ($mapped[$column] ?? 0), 4);
}
return $mapped;
}
private function containsFilter(Builder $query, string $column, mixed $value): void
{
$value = trim((string) $value);
if ($value !== '') {
$query->whereRaw($column . " LIKE ? ESCAPE '!'", [$this->contains($value)]);
}
}
/** @param array<string,mixed> $filters */
private function hasProductFilters(array $filters): bool
{
foreach (['product_name', 'supplier', 'category'] as $filter) {
if (trim((string) ($filters[$filter] ?? '')) !== '') {
return true;
}
}
return false;
}
private function contains(string $value): string
{
return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], $value) . '%';
}
/** @return Collection<int,object> */
private function boundedGet(Builder $query): Collection
{
if ($this->connection->getDriverName() !== 'mysql') {
return $query->get();
}
return collect($this->connection->select(
'SET STATEMENT max_statement_time=' . self::REPORT_QUERY_TIMEOUT_SECONDS . ' FOR ' . $query->toSql(),
$query->getBindings(),
));
}
private function boundedCount(Builder $query): int
{
$count = $this->connection->query()
->fromSub($query, 'bounded_count')
->selectRaw('COUNT(*) AS aggregate');
return (int) ($this->boundedGet($count)->first()->aggregate ?? 0);
}
private function isStatementTimeout(Throwable $exception): bool
{
$message = strtolower($exception->getMessage());
return str_contains($message, 'max_statement_time')
|| str_contains($message, 'query execution was interrupted');
}
}