CaiWuXiTong/modules/EcommerceFinance/Repositories/ThirdPartyWarehouseStorageUpstreamRepository.php

<?php

namespace Modules\EcommerceFinance\Repositories;

use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;

class ThirdPartyWarehouseStorageUpstreamRepository
{
    public const SOURCE_KEY = 'jiansun';

    /** 三方仓页面仍固定读取该仓；共享流水镜像覆盖全部仓库。 */
    public const WAREHOUSE_ID = 15045364;

    public const START_DATE = '2026-08-01';

    private ?UpstreamReadOnlyQuery $query = null;

    public function __construct(private readonly UpstreamReadOnlyQueryFactory $queries)
    {
    }

    public function headerHighWatermark(): int
    {
        return (int) $this->query()->scalar(
            'SELECT COALESCE(MAX([Id]), 0) FROM [dbo].[jst_inoutWater]',
        );
    }

    /** @return list<array<string,mixed>> */
    public function headerRange(int $afterId, int $throughId): array
    {
        if ($throughId <= $afterId) {
            return [];
        }

        return $this->query()->select(<<<'SQL'
SELECT
    h.[Id] AS source_id,
    h.[io_id],
    h.[wms_co_id],
    h.[wh_id],
    h.[status],
    h.[type] AS io_type,
    CONVERT(varchar(27), h.[io_date], 126) AS occurred_at,
    CONVERT(varchar(27), h.[modified], 126) AS modified_at
FROM [dbo].[jst_inoutWater] h
WHERE h.[Id] > ?
  AND h.[Id] <= ?
  AND h.[io_date] >= ?
  AND NOT EXISTS (
      SELECT 1 FROM [dbo].[jst_inoutWater] newer
      WHERE newer.[io_id] = h.[io_id] AND newer.[Id] > h.[Id]
  )
ORDER BY h.[Id]
SQL, [$afterId, $throughId, self::START_DATE]);
    }

    /** @return list<array<string,mixed>> */
    public function modifiedHeadersBatch(
        string $cursorAt,
        int $cursorId,
        string $throughAt,
        int $limit,
    ): array {
        $limit = min(max($limit, 1), 5000);

        return $this->query()->select('SELECT TOP (' . $limit . ') '
            . <<<'SQL'
    [Id] AS source_id,
    [io_id],
    [wms_co_id],
    [wh_id],
    [status],
    [type] AS io_type,
    CONVERT(varchar(27), [io_date], 126) AS occurred_at,
    CONVERT(varchar(27), [modified], 126) AS modified_at
FROM [dbo].[jst_inoutWater]
WHERE [modified] >= ?
  AND [modified] < ?
  AND ([modified] > ? OR ([modified] = ? AND [Id] > ?))
ORDER BY [modified], [Id]
SQL, [$cursorAt, $throughAt, $cursorAt, $cursorAt, $cursorId]);
    }

    /**
     * Reload the current version for every changed document before replacing
     * its mirror rows. A document can have more than one historical header Id.
     *
     * @param list<int|string> $ioIds
     * @return list<array<string,mixed>>
     */
    public function latestHeadersForIoIds(array $ioIds, string $throughAt): array
    {
        $ioIds = array_values(array_unique(array_filter(
            array_map(static fn (mixed $id): int => (int) $id, $ioIds),
            static fn (int $id): bool => $id > 0,
        )));
        if ($ioIds === []) {
            return [];
        }

        $rows = [];
        foreach (array_chunk($ioIds, 500) as $chunk) {
            $placeholders = implode(',', array_fill(0, count($chunk), '?'));
            $batch = $this->query()->select(
                'SELECT h.[Id] AS source_id, h.[io_id], h.[wms_co_id], h.[wh_id], h.[status], '
                . 'h.[type] AS io_type, CONVERT(varchar(27), h.[io_date], 126) AS occurred_at, '
                . 'CONVERT(varchar(27), h.[modified], 126) AS modified_at '
                . 'FROM [dbo].[jst_inoutWater] h '
                . 'INNER JOIN (SELECT [io_id], MAX([Id]) AS latest_id '
                . 'FROM [dbo].[jst_inoutWater] '
                . "WHERE [io_id] IN ({$placeholders}) AND [modified] < ? GROUP BY [io_id]) latest "
                . 'ON latest.[latest_id] = h.[Id] '
                . 'WHERE h.[io_date] >= ? '
                . 'ORDER BY h.[Id]',
                [...$chunk, $throughAt, self::START_DATE],
            );
            array_push($rows, ...$batch);
        }

        usort($rows, static fn (array $left, array $right): int =>
            ((int) ($left['source_id'] ?? 0)) <=> ((int) ($right['source_id'] ?? 0))
        );

        return $rows;
    }

    /**
     * The upstream item table has an index on io_id. Fetching items only after
     * the warehouse headers have been selected avoids scanning unrelated rows.
     *
     * @param list<int|string> $ioIds
     * @return list<array<string,mixed>>
     */
    public function itemsForIoIds(array $ioIds): array
    {
        $ioIds = array_values(array_unique(array_filter(
            array_map(static fn (mixed $id): int => (int) $id, $ioIds),
            static fn (int $id): bool => $id > 0,
        )));

        if ($ioIds === []) {
            return [];
        }

        $rows = [];
        foreach (array_chunk($ioIds, 500) as $chunk) {
            $placeholders = implode(',', array_fill(0, count($chunk), '?'));
            $batch = $this->query()->select(
                'SELECT [Id] AS source_id, [io_id], [sku_id] AS sku_code, [qty], [qty_type] '
                . 'FROM [dbo].[jst_inoutWaterItem] '
                . "WHERE [io_id] IN ({$placeholders}) ORDER BY [Id]",
                $chunk,
            );
            array_push($rows, ...$batch);
        }

        usort($rows, static fn (array $left, array $right): int =>
            ((int) ($left['source_id'] ?? 0)) <=> ((int) ($right['source_id'] ?? 0))
        );

        return $rows;
    }

    public function inventoryHighWatermark(): int
    {
        return (int) $this->query()->scalar(
            'SELECT COALESCE(MAX([Id]), 0) FROM [dbo].[jst_inventoryInfo]',
        );
    }

    /** @return list<array<string,mixed>> */
    public function inventoryBatch(int $afterId, int $highWatermark, int $limit): array
    {
        if ($highWatermark <= $afterId) {
            return [];
        }

        $limit = min(max($limit, 1), 5000);

        return $this->query()->select('SELECT TOP (' . $limit . ') '
            . <<<'SQL'
    [Id] AS source_id,
    [wms_co_id],
    [sku_id] AS sku_code,
    [qty],
    CONVERT(varchar(27), [modified], 126) AS source_modified_at
FROM [dbo].[jst_inventoryInfo]
WHERE [Id] > ?
  AND [Id] <= ?
ORDER BY [Id]
SQL, [$afterId, $highWatermark]);
    }

    private function query(): UpstreamReadOnlyQuery
    {
        return $this->query ??= $this->queries->forSource(self::SOURCE_KEY);
    }
}