421 lines
19 KiB
PHP
421 lines
19 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Repositories;
|
||
|
||
use Illuminate\Database\ConnectionInterface;
|
||
use Illuminate\Database\Query\Builder;
|
||
use Illuminate\Support\Carbon;
|
||
use Illuminate\Support\Facades\Cache;
|
||
use Illuminate\Support\Facades\Schema;
|
||
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
|
||
use RuntimeException;
|
||
use Throwable;
|
||
|
||
/** 快递应付运费对账的共享业务镜像只读访问。 */
|
||
class CourierFreightReconciliationMirrorRepository
|
||
{
|
||
/** 保留旧契约，供现有结算同步和调用方识别。 */
|
||
public const DATASET = 'jiansun.dbo.fys_express_courier_settlement';
|
||
public const DETAILS_DATASET = 'jiansun.dbo.jst_order_express_details';
|
||
public const EXTENSION_DATASET = 'jiansun.dbo.Jst_SendData_View_extension';
|
||
public const OUTBOUND_DATASET = 'jiansun.dbo.jst_salesOutboundInfo';
|
||
|
||
/** @var list<string> */
|
||
private const REQUIRED_DATASETS = [
|
||
self::DETAILS_DATASET,
|
||
self::EXTENSION_DATASET,
|
||
self::OUTBOUND_DATASET,
|
||
];
|
||
|
||
/** 仅暴露附件中有演示值、且已经核对来源的字段。 */
|
||
private const COLUMNS = [
|
||
'detail_id', 'waybill', 'product_id', 'send_time', 'transaction_order_no',
|
||
'province', 'city', 'district', 'barcode', 'express_name', 'express_name_2', 'quantity',
|
||
'product_name', 'platform', 'shop_name', 'warehouse', 'volume', 'supplier',
|
||
'length', 'width', 'height', 'freight', 'reconciliation_weight',
|
||
'reconciliation_volume', 'month', 'status',
|
||
];
|
||
|
||
private readonly ConnectionInterface $connection;
|
||
|
||
public function __construct(
|
||
private readonly BusinessMirrorDatasetRegistry $datasets,
|
||
private readonly GenericBusinessMirrorRepository $mirror,
|
||
) {
|
||
$this->connection = $mirror->connection();
|
||
}
|
||
|
||
public function available(): bool
|
||
{
|
||
try {
|
||
$schema = Schema::connection($this->connection->getName());
|
||
if (! $schema->hasTable('business_mirror_table_states')) {
|
||
return false;
|
||
}
|
||
|
||
foreach (self::REQUIRED_DATASETS as $key) {
|
||
if (! $schema->hasTable($this->datasets->get($key)->targetTable())) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
return true;
|
||
} catch (Throwable) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
public function ready(): bool
|
||
{
|
||
try {
|
||
foreach (self::REQUIRED_DATASETS as $key) {
|
||
if (! $this->mirror->ready($this->datasets->get($key))) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
return true;
|
||
} catch (Throwable) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/** @return array<string,mixed>|null */
|
||
public function state(): ?array
|
||
{
|
||
try {
|
||
$states = [];
|
||
foreach (self::REQUIRED_DATASETS as $key) {
|
||
$state = $this->mirror->state($this->datasets->get($key));
|
||
if (is_array($state)) {
|
||
$states[$key] = $state;
|
||
}
|
||
}
|
||
if ($states === []) {
|
||
return null;
|
||
}
|
||
|
||
$first = $states[self::DETAILS_DATASET] ?? reset($states);
|
||
$initial = array_values(array_filter(array_map(
|
||
static fn (array $state): ?string => $state['initial_completed_at'] ?? null,
|
||
$states,
|
||
)));
|
||
$succeeded = array_values(array_filter(array_map(
|
||
static fn (array $state): ?string => $state['last_succeeded_at'] ?? null,
|
||
$states,
|
||
)));
|
||
$statuses = array_map(static fn (array $state): string => (string) ($state['status'] ?? 'pending'), $states);
|
||
|
||
$first['initial_completed_at'] = count($initial) === count(self::REQUIRED_DATASETS) ? min($initial) : null;
|
||
$first['last_succeeded_at'] = count($succeeded) === count(self::REQUIRED_DATASETS) ? min($succeeded) : null;
|
||
$first['status'] = in_array('failed', $statuses, true)
|
||
? 'failed'
|
||
: (in_array('running', $statuses, true) ? 'running' : (count(array_filter($statuses, static fn (string $value): bool => $value === 'succeeded')) === count(self::REQUIRED_DATASETS) ? 'succeeded' : 'pending'));
|
||
$first['datasets'] = $states;
|
||
|
||
return $first;
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/** @return array{code:string,last_succeeded_at:?string,lag_minutes:?int} */
|
||
public function syncStatus(?array $state = null): array
|
||
{
|
||
$state ??= $this->state();
|
||
if (! is_array($state)) {
|
||
return ['code' => 'unavailable', 'last_succeeded_at' => null, 'lag_minutes' => null];
|
||
}
|
||
|
||
$lastSucceeded = $state['last_succeeded_at'] ?? null;
|
||
$lagMinutes = null;
|
||
if (is_string($lastSucceeded) && $lastSucceeded !== '') {
|
||
try {
|
||
$lagMinutes = (int) ceil(max(0, now()->getTimestamp() - Carbon::parse($lastSucceeded)->getTimestamp()) / 60);
|
||
} catch (Throwable) {
|
||
$lagMinutes = null;
|
||
}
|
||
}
|
||
|
||
$status = (string) ($state['status'] ?? 'pending');
|
||
if (($state['initial_completed_at'] ?? null) === null) {
|
||
$code = $status === 'failed' ? 'initial_failed' : ($status === 'running' ? 'initializing' : 'pending');
|
||
} elseif ($status === 'failed') {
|
||
$code = 'failed';
|
||
} elseif ($status === 'running') {
|
||
$code = 'syncing';
|
||
} elseif ($lagMinutes === null || $lagMinutes > max((int) config('upstream_mirrors.courier_information.stale_after_minutes', 60), 1)) {
|
||
$code = 'stale';
|
||
} else {
|
||
$code = 'fresh';
|
||
}
|
||
|
||
return ['code' => $code, 'last_succeeded_at' => $lastSucceeded, 'lag_minutes' => $lagMinutes];
|
||
}
|
||
|
||
/**
|
||
* @param array<string,string> $filters
|
||
* @return array{data:list<array<string,mixed>>,total:?int,has_more:bool,summary:?array,state:array<string,mixed>|null,sync:array<string,mixed>}
|
||
*/
|
||
public function paginate(array $filters, int $page, int $perPage): array
|
||
{
|
||
if (! $this->ready()) {
|
||
throw new RuntimeException('courier_freight_reconciliation_mirror_not_ready');
|
||
}
|
||
|
||
$page = max($page, 1);
|
||
$perPage = min(max($perPage, 1), 100);
|
||
$summary = Cache::get($this->summaryCacheKey($filters));
|
||
// Select page identifiers before loading the wide optional details. A
|
||
// contains-waybill filter can then scan the compact covering index.
|
||
$candidates = $this->filteredQuery($filters, trim((string) ($filters['express_name'] ?? '')) !== '')
|
||
->select('d.id')
|
||
->orderByDesc('d.id')
|
||
->offset(($page - 1) * $perPage)->limit($perPage + 1);
|
||
$rows = $this->detailQuery($candidates)->select($this->selectColumns())
|
||
->orderByDesc('d.id')
|
||
->get()
|
||
->map(fn (object $row): array => $this->mapRow($row))
|
||
->all();
|
||
$state = $this->state();
|
||
|
||
return [
|
||
'data' => array_slice($rows, 0, $perPage),
|
||
'total' => $summary['count'] ?? null,
|
||
'has_more' => count($rows) > $perPage,
|
||
'summary' => $summary,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
|
||
/** Send a small pending response before starting the bounded calculation. */
|
||
public function requestSummary(array $filters): ?array
|
||
{
|
||
$key = $this->summaryCacheKey($filters);
|
||
if (is_array($cached = Cache::get($key))) return $cached;
|
||
if (Cache::has($key . ':failed')) throw new RuntimeException('courier_summary_recently_failed');
|
||
app()->terminating(function () use ($filters, $key): void {
|
||
try {
|
||
$this->summarize($filters);
|
||
} catch (Throwable $exception) {
|
||
Cache::put($key . ':failed', true, 60);
|
||
report($exception);
|
||
}
|
||
});
|
||
return null;
|
||
}
|
||
|
||
/** The separate summary request preserves full-filter totals without delaying rows. */
|
||
public function summarize(array $filters): ?array
|
||
{
|
||
if (! $this->ready()) throw new RuntimeException('courier_freight_reconciliation_mirror_not_ready');
|
||
$key = $this->summaryCacheKey($filters);
|
||
if (is_array($cached = Cache::get($key))) return $cached;
|
||
// One expensive aggregate at a time, including requests for different filters.
|
||
$lock = Cache::lock('courier-freight-summary:calculation', 360);
|
||
if (! $lock->get()) return null;
|
||
$previousSettings = [];
|
||
try {
|
||
if (is_array($cached = Cache::get($key))) return $cached;
|
||
$unfiltered = array_filter($filters, static fn ($value): bool => trim((string) $value) !== '') === [];
|
||
if ($this->connection->getDriverName() === 'mysql') {
|
||
$settings = ['max_statement_time' => 300];
|
||
if ($unfiltered) {
|
||
$settings += [
|
||
'optimizer_switch' => 'derived_merge=off,derived_with_keys=off',
|
||
'join_cache_level' => 8,
|
||
// 全量汇总要把 300 万行左右的明细和 400 万行的最新关联表做哈希
|
||
// 关联；16MB 的哈希表装不下会退化成磁盘多趟扫描（实测超过
|
||
// max_statement_time 也跑不完）。128MB 缓冲约 110 秒出结果。
|
||
'join_buffer_size' => 134217728,
|
||
'join_buffer_space_limit' => 402653184,
|
||
];
|
||
}
|
||
foreach ($settings as $name => $value) {
|
||
$previousSettings[$name] = $this->connection->selectOne('SELECT @@SESSION.' . $name . ' AS value')->value;
|
||
$this->setSummarySessionSetting($name, $value);
|
||
}
|
||
}
|
||
$asOf = now()->toDateTimeString();
|
||
$row = $this->summaryQuery($filters)->selectRaw(
|
||
'COUNT(d.id) AS aggregate_count, COALESCE(SUM(o.freight), 0) AS freight',
|
||
)->first();
|
||
$summary = ['count' => (int) $row->aggregate_count, 'freight' => (float) $row->freight, 'computed_at' => $asOf];
|
||
// 全量（无筛选）汇总要跑两分钟左右，缓存久一点，避免频繁重算；
|
||
// 带筛选的汇总走索引，很快，保持原来的 15 分钟。
|
||
Cache::put($key, $summary, $unfiltered ? 3600 : 900);
|
||
return $summary;
|
||
} finally {
|
||
try {
|
||
foreach (array_reverse($previousSettings, true) as $name => $value) {
|
||
$this->setSummarySessionSetting($name, $value);
|
||
}
|
||
} finally {
|
||
$lock->release();
|
||
}
|
||
}
|
||
}
|
||
|
||
private function setSummarySessionSetting(string $name, mixed $value): void
|
||
{
|
||
if ($name === 'optimizer_switch') {
|
||
$this->connection->statement('SET SESSION optimizer_switch=?', [$value]);
|
||
return;
|
||
}
|
||
// MariaDB rejects quoted numeric values for these session variables.
|
||
// Names come only from the fixed settings map above.
|
||
$this->connection->statement('SET SESSION ' . $name . '=' . (float) $value);
|
||
}
|
||
|
||
private function summaryCacheKey(array $filters): string
|
||
{
|
||
$filters = array_filter(array_map(static fn ($value): string => trim((string) $value), $filters), static fn (string $value): bool => $value !== '');
|
||
ksort($filters);
|
||
return 'courier-freight-summary:v3:' . hash('sha256', json_encode($filters));
|
||
}
|
||
|
||
/** Materialize latest associations once; use compact collation-aware hash joins for the full scan. */
|
||
private function summaryQuery(array $filters = []): Builder
|
||
{
|
||
$activeFilters = array_filter($filters, static fn ($value): bool => trim((string) $value) !== '');
|
||
if ($activeFilters !== []) {
|
||
$ids = $this->filteredQuery($filters, ($filters['express_name'] ?? '') !== '')->select('d.id')->distinct();
|
||
return $this->detailQuery($ids);
|
||
}
|
||
$columns = ['so_id', 'io_id'];
|
||
$lookup = fn (string $column): string => $this->connection->getDriverName() === 'mysql'
|
||
? 'CRC32(WEIGHT_STRING(RTRIM(' . $column . ')))'
|
||
: 'SUBSTR(' . $column . ', 1, 64)';
|
||
$extension = $this->latestSummaryProjection(self::EXTENSION_DATASET, 'so_id', $columns)
|
||
->selectRaw($lookup('candidate.so_id') . ' AS lookup_key');
|
||
$outbound = $this->latestSummaryProjection(self::OUTBOUND_DATASET, 'io_id', ['io_id', 'freight']);
|
||
$query = $this->connection->table($this->datasets->get(self::DETAILS_DATASET)->targetTable() . ' as d')
|
||
->leftJoinSub($extension, 'x', static function ($join) use ($lookup): void {
|
||
$join->whereRaw('x.lookup_key = ' . $lookup('d.so_id'))
|
||
// Keep the full comparison: a hash collision must never
|
||
// combine different orders or multiply freight amounts.
|
||
->whereRaw("COALESCE(x.so_id, '') = COALESCE(d.so_id, '')");
|
||
})
|
||
->leftJoinSub($outbound, 'o', 'o.io_id', '=', 'x.io_id');
|
||
return $this->filteredQuery([], true, $query);
|
||
}
|
||
|
||
private function latestSummaryProjection(string $dataset, string $relation, array $columns): Builder
|
||
{
|
||
$table = $this->datasets->get($dataset)->targetTable();
|
||
$ids = $this->connection->table($table)->select($relation)->selectRaw('MAX(Id) AS latest_id')->groupBy($relation)->orderByRaw('NULL');
|
||
return $this->connection->query()->fromSub($ids, 'latest_ids')
|
||
->join($table . ' as candidate', static function ($join) use ($relation): void {
|
||
$join->on('latest_ids.' . $relation, '=', 'candidate.' . $relation)->on('latest_ids.latest_id', '=', 'candidate.Id');
|
||
})->select(array_map(static fn (string $column): string => 'candidate.' . $column, $columns));
|
||
}
|
||
|
||
private function filteredQuery(array $filters, bool $includeRelations = true, ?Builder $base = null): Builder
|
||
{
|
||
$query = ($base ?? ($includeRelations ? $this->detailQuery() : $this->connection->table($this->datasets->get(self::DETAILS_DATASET)->targetTable() . ' as d')))
|
||
->whereNotNull('d.logistic_id')->where('d.logistic_id', '!=', '');
|
||
$this->containsFilter($query, 'd.logistic_id', $filters['express_no'] ?? '');
|
||
|
||
if (($filters['express_name'] ?? '') !== '') {
|
||
$needle = $this->contains((string) $filters['express_name']);
|
||
$query->where(function (Builder $name) use ($needle): void {
|
||
$name->whereRaw("d.logistic_company_name LIKE ? ESCAPE '!'", [$needle])
|
||
->orWhereRaw("x.expressName LIKE ? ESCAPE '!'", [$needle]);
|
||
});
|
||
}
|
||
if (($filters['month'] ?? '') !== '' && preg_match('/^\d{4}-\d{2}$/', (string) $filters['month']) === 1) {
|
||
$from = (string) $filters['month'] . '-01';
|
||
$query->where('d.send_time', '>=', $from . ' 00:00:00')
|
||
->where('d.send_time', '<', date('Y-m-d 00:00:00', strtotime($from . ' +1 month')));
|
||
}
|
||
if (($filters['status'] ?? '') !== '') {
|
||
$query->where('d.logistic_status_name', '=', $filters['status']);
|
||
}
|
||
if (($filters['modified_from'] ?? '') !== '') {
|
||
$query->where('d.updatedAt', '>=', $filters['modified_from'] . ' 00:00:00');
|
||
}
|
||
if (($filters['modified_to'] ?? '') !== '') {
|
||
$query->where('d.updatedAt', '<', date('Y-m-d 00:00:00', strtotime($filters['modified_to'] . ' +1 day')));
|
||
}
|
||
|
||
return $query;
|
||
}
|
||
|
||
private function detailQuery(?Builder $pageIds = null): Builder
|
||
{
|
||
$details = $this->datasets->get(self::DETAILS_DATASET)->targetTable();
|
||
$extension = $this->datasets->get(self::EXTENSION_DATASET)->targetTable();
|
||
$outbound = $this->datasets->get(self::OUTBOUND_DATASET)->targetTable();
|
||
|
||
$query = $pageIds === null ? $this->connection->table($details . ' as d')
|
||
: $this->connection->query()->fromSub($pageIds, 'page_ids')->join($details . ' as d', 'd.id', '=', 'page_ids.id');
|
||
return $query
|
||
->leftJoinSub($this->latestBy($extension, 'so_id', 'Id'), 'x', 'x.so_id', '=', 'd.so_id')
|
||
->leftJoinSub($this->latestBy($outbound, 'io_id', 'Id'), 'o', 'o.io_id', '=', 'x.io_id');
|
||
}
|
||
|
||
/** 固定取关联键下 Id 最大的一行，避免一对多重复页面行和运费汇总。 */
|
||
private function latestBy(string $table, string $relationColumn, string $idColumn): Builder
|
||
{
|
||
return $this->connection->table($table . ' as candidate')
|
||
->whereNotExists(function (Builder $query) use ($table, $relationColumn, $idColumn): void {
|
||
$query->selectRaw('1')
|
||
->from($table . ' as newer')
|
||
->whereColumn('newer.' . $relationColumn, '=', 'candidate.' . $relationColumn)
|
||
->whereColumn('newer.' . $idColumn, '>', 'candidate.' . $idColumn);
|
||
});
|
||
}
|
||
|
||
/** @return list<string> */
|
||
private function selectColumns(): array
|
||
{
|
||
return [
|
||
'd.id as detail_id', 'd.logistic_id as waybill', 'x.shop_i_id as product_id',
|
||
'd.send_time', $this->connection->raw('COALESCE(x.so_id, d.so_id) as transaction_order_no'), 'x.receiver_state as province',
|
||
'x.receiver_city as city', 'x.receiver_district as district',
|
||
'x.sku_ids as barcode',
|
||
'd.logistic_company_name as express_name', 'x.expressName as express_name_2',
|
||
'x.qty as quantity', 'x.sku_name as product_name', 'd.platform_name as platform',
|
||
$this->connection->raw('COALESCE(x.shop_name, d.shop_name) as shop_name'),
|
||
$this->connection->raw('COALESCE(x.wareHoursName, d.warehouse_name) as warehouse'), 'x.volume as volume',
|
||
'x.supplier_name as supplier', 'x.l as length', 'x.w as width', 'x.h as height',
|
||
'o.freight as freight', 'd.f_weight as reconciliation_weight',
|
||
'x.volume as reconciliation_volume', 'd.logistic_status_name as status',
|
||
];
|
||
}
|
||
|
||
/** @return array<string,mixed> */
|
||
private function mapRow(object $row): array
|
||
{
|
||
$mapped = [];
|
||
foreach (self::COLUMNS as $column) {
|
||
$mapped[$column] = $row->{$column} ?? null;
|
||
}
|
||
|
||
$sendTime = $mapped['send_time'];
|
||
$mapped['month'] = is_string($sendTime) ? substr($sendTime, 0, 7) : null;
|
||
foreach (['length', 'width', 'height', 'volume', 'freight', 'reconciliation_weight', 'reconciliation_volume'] as $number) {
|
||
$mapped[$number] = $mapped[$number] === null || $mapped[$number] === '' ? null : (float) $mapped[$number];
|
||
}
|
||
if ($mapped['quantity'] !== null && $mapped['quantity'] !== '') {
|
||
$mapped['quantity'] = (int) $mapped['quantity'];
|
||
}
|
||
|
||
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)]);
|
||
}
|
||
}
|
||
|
||
private function contains(string $value): string
|
||
{
|
||
return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], $value) . '%';
|
||
}
|
||
}
|