CaiWuXiTong/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php

<?php

namespace Tests\Unit\EcommerceFinance;

use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\DB;
use Modules\EcommerceFinance\Services\Jushuitan\ShippingReplicaSynchronizer;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
use Modules\EcommerceFinance\Repositories\ShippingOrderRepository;
use PDOException;
use RuntimeException;
use Tests\TestCase;

class ShippingReplicaSynchronizerTest extends TestCase
{
    public function test_windowed_express_scan_maps_order_metadata_without_joining_orders_per_page(): void
    {
        $detailQueries = [];
        $orderMetadataQueries = [];
        $detailCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings) use (&$detailQueries, &$orderMetadataQueries, &$detailCalls): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                $detailQueries[] = $sql;

                return $detailCalls++ === 0
                    ? [['id' => 5000, 'inner_order_id' => 'order-1', 'send_time' => null, 'logistic_id' => '']]
                    : [];
            }
            if (str_contains($sql, 'FROM rpt.jst_Orders om WHERE om.o_id IN')) {
                $orderMetadataQueries[] = [$sql, $bindings];

                return [['o_id' => 'order-1', 'l_id' => 'order-tracking', 'send_date' => '2026-09-02 12:00:00']];
            }
            if (str_contains($sql, 'FROM rpt.jst_Orders o WHERE o.status =')) {
                return [];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertNotEmpty($detailQueries);
        $this->assertStringNotContainsString('LEFT JOIN rpt.jst_Orders', $detailQueries[0]);
        $this->assertCount(1, $orderMetadataQueries);
        $this->assertSame(['order-1'], $orderMetadataQueries[0][1]);
        $this->assertSame(1, $result['processed']);
        $row = DB::connection('business_mirror')->table('jst_oms_shipping')->where('source_id', 5000)->sole();
        $this->assertSame('order-tracking', $row->express_no);
        $this->assertSame('2026-09-02 12:00:00', $row->ship_time);
    }

    public function test_windowed_outbound_lookup_preloads_date_range_and_reuses_cache_without_in_queries(): void
    {
        $from = '2026-09-22 00:00:00';
        $to = '2026-09-25 00:00:00';
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->expects($this->once())
            ->method('select')
            ->with(
                $this->callback(fn (string $sql): bool => str_contains($sql, 'WHERE io_date >= ? AND io_date < ?')
                    && ! str_contains($sql, 'l_id IN')
                    && ! str_contains($sql, 'o_id IN')),
                [$from, $to],
            )
            ->willReturn([
                ['l_id' => 'logistic-1', 'o_id' => 'order-1', 'io_id' => 'outbound-1'],
                ['l_id' => 'logistic-2', 'o_id' => 'order-2', 'io_id' => 'outbound-2'],
            ]);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $synchronizer = new ShippingReplicaSynchronizer($factory);
        $preload = new \ReflectionMethod($synchronizer, 'preloadOutboundMap');
        $preload->setAccessible(true);
        $preload->invoke($synchronizer, $from, $to);

        $lookup = new \ReflectionMethod($synchronizer, 'ioMapFor');
        $lookup->setAccessible(true);
        $map = $lookup->invoke($synchronizer, true, ['logistic-1'], ['order-2']);

        $this->assertSame('outbound-1', $map['byL']['logistic-1']);
        $this->assertSame('outbound-2', $map['byO']['order-2']);
    }

    public function test_windowed_item_lookup_skips_unindexed_raw_so_id_fallback(): void
    {
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->expects($this->once())
            ->method('select')
            ->with(
                $this->stringContains('WHERE o_id IN'),
                ['order-1'],
            )
            ->willReturn([
                [
                    'o_id' => 'order-1',
                    'raw_so_id' => 'trade-1',
                    'sku_id' => 'sku-1',
                    'i_id' => 'item-1',
                    'shop_i_id' => 'shop-item-1',
                    'name' => '商品',
                    'qty' => 1,
                    'amount' => 10,
                    'send_warehouse' => '仓库',
                    'Id' => 1,
                ],
            ]);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $synchronizer = new ShippingReplicaSynchronizer($factory);
        $method = new \ReflectionMethod($synchronizer, 'loadAllItems');
        $method->setAccessible(true);
        [$byOrder, $byTrade] = $method->invoke($synchronizer, ['order-1'], ['trade-1'], false);

        $this->assertArrayHasKey('order-1', $byOrder);
        $this->assertSame([], $byTrade);
    }

    public function test_windowed_scan_continues_past_an_old_page_to_find_later_shipments(): void
    {
        $pages = [
            array_map(fn (int $id): array => ['id' => $id, 'send_time' => $id === 4500 ? '2026-09-02 12:00:00' : null], range(4500, 3001)),
            array_map(fn (int $id): array => ['id' => $id, 'send_time' => '2026-08-31 12:00:00'], range(3000, 1501)),
            [['id' => 1500, 'send_time' => '2026-09-03 12:00:00']],
        ];
        $scanCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls, $pages): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return $pages[$scanCalls++] ?? [];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );

        $this->assertSame(3, $scanCalls);
        $this->assertSame(2, $result['processed']);
        $this->assertSame(2, $result['unique_packages']);
        $this->assertSame([1500, 4500], DB::connection('business_mirror')->table('jst_oms_shipping')
            ->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all());
    }

    public function test_express_page_reconnects_once_after_transient_dblib_cursor_failure(): void
    {
        $firstPage = array_map(fn (int $id): array => [
            'id' => $id,
            'send_time' => $id === 3000 ? '2026-09-02 12:00:00' : null,
        ], range(3000, 1501));
        $lastPage = [['id' => 1500, 'send_time' => '2026-09-03 12:00:00']];
        $scanCalls = 0;
        $bindings = [];
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bind) use (&$scanCalls, &$bindings, $firstPage, $lastPage): array {
            if (! str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [];
            }
            $scanCalls++;
            $bindings[] = $bind;
            if ($scanCalls === 2) {
                throw new PDOException('SQLSTATE[HY000]: General error: PDO_DBLIB: dbnextrow() returned FAIL');
            }

            return $scanCalls === 1 ? $firstPage : $lastPage;
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertSame(3, $scanCalls);
        $this->assertSame([[], [1501], [1501]], $bindings);
        $this->assertSame(2, $result['processed']);
        $this->assertSame(2, $result['unique_packages']);
        $this->assertSame([1500, 3000], DB::connection('business_mirror')->table('jst_oms_shipping')
            ->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all());
    }

    public function test_non_transient_upstream_failure_is_not_retried(): void
    {
        $calls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$calls): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                $calls++;
                throw new PDOException('SQLSTATE[HY000]: unrelated upstream error');
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        try {
            (new ShippingReplicaSynchronizer($factory))->sync(
                Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
            );
            $this->fail('Expected the upstream error to be propagated.');
        } catch (PDOException $e) {
            $this->assertStringContainsString('unrelated upstream error', $e->getMessage());
        }

        $this->assertSame(1, $calls);
        $this->assertSame('failed', DB::connection('business_mirror')->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade')->value('status'));
    }

    public function test_adaptive_server_timeout_reconnects_and_replays_the_same_page(): void
    {
        $page = array_map(fn (int $id): array => [
            'id' => $id, 'send_time' => $id === 1500 ? '2026-09-02 12:00:00' : null,
        ], range(1500, 1));
        $scanCalls = 0;
        $bindings = [];
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bind) use (&$scanCalls, &$bindings, $page): array {
            if (! str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [];
            }
            $scanCalls++;
            $bindings[] = $bind;
            if ($scanCalls <= 2) {
                throw new PDOException('SQLSTATE[HY000]: General error: 20003 Adaptive Server connection timed out [20003] (severity 6)');
            }
            return $scanCalls === 3 ? $page : [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertSame(4, $scanCalls);
        $this->assertSame([[], [], [], [1]], $bindings);
        $this->assertSame(1, $result['processed']);
        $this->assertSame([1500], DB::connection('business_mirror')->table('jst_oms_shipping')
            ->pluck('source_id')->map(fn ($id) => (int) $id)->all());
    }

    public function test_window_result_distinguishes_duplicate_package_rows_from_unique_source_ids(): void
    {
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                $row = ['id' => 5000, 'send_time' => '2026-09-02 12:00:00'];
                return [$row, $row];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertSame(2, $result['packages']);
        $this->assertSame(1, $result['unique_packages']);
        $this->assertSame(1, DB::connection('business_mirror')->table('jst_oms_shipping')
            ->distinct()->count('source_id'));
    }

    public function test_express_scan_fails_when_more_rows_remain_after_batch_limit(): void
    {
        DB::connection('business_mirror')->table('jst_oms_shipping')->insert([
            'source_id' => 3000, 'sku_source_id' => 3000, 'ship_time' => '2026-09-02 12:00:00',
            'express_no' => 'old-tracking', 'freight' => 18.75,
        ]);
        $pages = [
            array_map(fn (int $id): array => ['id' => $id, 'send_time' => $id === 3000 ? '2026-09-02 12:00:00' : null], range(3000, 1501)),
            [['id' => 1500]],
        ];
        $scanCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls, $pages): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return $pages[$scanCalls++] ?? [];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);
        $synchronizer = new ShippingReplicaSynchronizer($factory);
        $this->setScanBatchLimit($synchronizer, 'maxExpressBatches', 1);

        try {
            $synchronizer->sync(Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true);
            $this->fail('Expected an incomplete express scan to fail.');
        } catch (RuntimeException $e) {
            $this->assertStringContainsString('发货扫描达到批次上限', $e->getMessage());
        }

        $this->assertSame(2, $scanCalls);
        $this->assertSame('failed', DB::connection('business_mirror')->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade')->value('status'));
        $this->assertSame('pending', DB::connection('business_mirror')->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade.freight.20260901000000-20260905000000')->value('status'));
        $original = DB::connection('business_mirror')->table('jst_oms_shipping')->where('source_id', 3000)->first();
        $this->assertSame('old-tracking', $original->express_no);
        $this->assertEquals(18.75, $original->freight);
    }

    public function test_full_last_batch_succeeds_when_no_more_express_rows_exist(): void
    {
        $scanCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return $scanCalls++ === 0
                    ? array_map(fn (int $id): array => ['id' => $id], range(1500, 1))
                    : [];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);
        $synchronizer = new ShippingReplicaSynchronizer($factory);
        $this->setScanBatchLimit($synchronizer, 'maxExpressBatches', 1);

        $result = $synchronizer->sync(Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'));

        $this->assertSame(2, $scanCalls);
        $this->assertSame(0, $result['processed']);
        $this->assertSame('succeeded', DB::connection('business_mirror')->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade')->value('status'));
    }

    public function test_auto_ship_scan_fails_when_more_rows_remain_after_batch_limit(): void
    {
        $pages = [
            array_map(fn (int $id): array => ['rid' => $id, 'o_id' => ''], range(1000, 601)),
            [['rid' => 600, 'o_id' => '']],
        ];
        $scanCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls, $pages): array {
            if (str_contains($sql, 'FROM rpt.jst_Orders o WHERE')) {
                return $pages[$scanCalls++] ?? [];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);
        $synchronizer = new ShippingReplicaSynchronizer($factory);
        $this->setScanBatchLimit($synchronizer, 'maxAutoShipBatches', 1);

        try {
            $synchronizer->sync(Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'));
            $this->fail('Expected an incomplete auto-ship scan to fail.');
        } catch (RuntimeException $e) {
            $this->assertStringContainsString('代发仓扫描达到批次上限', $e->getMessage());
        }

        $this->assertSame(2, $scanCalls);
        $this->assertSame('failed', DB::connection('business_mirror')->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade')->value('status'));
    }

    public function test_auto_ship_excludes_every_order_with_express_detail_even_when_detail_is_outside_window_or_label_excluded(): void
    {
        $orderQueries = [];
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings) use (&$orderQueries): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [
                    ['id' => 1001, 'inner_order_id' => '123', 'jo_send' => '2026-09-02 13:00:00',
                        'send_time' => '2026-08-31 12:00:00'],
                    ['id' => 1002, 'inner_order_id' => '456', 'jo_send' => '2026-09-02 13:00:00',
                        'send_time' => '2026-09-02 12:00:00', 'order_labels' => '线上发货'],
                    ['id' => 1003, 'inner_order_id' => '789', 'jo_send' => '2026-09-02 13:00:00',
                        'send_time' => '2026-09-02 12:00:00'],
                ];
            }
            if (str_contains($sql, 'FROM rpt.jst_Orders o WHERE')) {
                $orderQueries[] = [$sql, $bindings];
                return array_map(fn (int $oid): array => [
                    'rid' => $oid, 'o_id' => (string) $oid, 'send_date' => '2026-09-02 13:00:00',
                ], [789, 456, 321, 123]);
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertCount(1, $orderQueries);
        $this->assertStringNotContainsString('NOT EXISTS', $orderQueries[0][0]);
        $this->assertSame(['2026-09-01 00:00:00', '2026-09-05 00:00:00'], $orderQueries[0][1]);
        $this->assertSame(1, $result['processed']);
        $this->assertSame(1, $result['autoship']);
        $this->assertSame([1003, 1000000321], DB::connection('business_mirror')->table('jst_oms_shipping')
            ->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all());
    }

    public function test_auto_ship_advances_raw_page_cursor_when_all_400_orders_have_express_details(): void
    {
        $orderBindings = [];
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings) use (&$orderBindings): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return array_map(fn (int $oid): array => [
                    'id' => $oid, 'inner_order_id' => (string) $oid,
                    'jo_send' => '2026-09-02 13:00:00', 'send_time' => '2026-08-31 12:00:00',
                ], range(1000, 601));
            }
            if (str_contains($sql, 'FROM rpt.jst_Orders o WHERE')) {
                $orderBindings[] = $bindings;
                if (count($orderBindings) === 1) {
                    return array_map(fn (int $oid): array => [
                        'rid' => $oid, 'o_id' => (string) $oid, 'send_date' => '2026-09-02 13:00:00',
                    ], range(1000, 601));
                }
                return [['rid' => 600, 'o_id' => '500', 'send_date' => '2026-09-02 13:00:00']];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertSame([
            ['2026-09-01 00:00:00', '2026-09-05 00:00:00'],
            ['2026-09-01 00:00:00', '2026-09-05 00:00:00', 601],
        ], $orderBindings);
        $this->assertSame(0, $result['processed']);
        $this->assertSame(1, $result['autoship']);
        $this->assertSame([1000000500], DB::connection('business_mirror')->table('jst_oms_shipping')
            ->pluck('source_id')->map(fn ($id) => (int) $id)->all());
    }

    public function test_skip_freight_preserves_existing_amounts_and_marks_new_amounts_incomplete(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            ['source_id' => 5000, 'sku_source_id' => 5000, 'ship_time' => '2026-09-02 12:00:00',
                'express_no' => 'existing-tracking', 'express_name2' => 'existing-express',
                'freight' => 12.34, 'settle_freight' => 13.45, 'weight_kg' => 1.2, 'settle_weight_kg' => 1.3],
            ['source_id' => 6000, 'sku_source_id' => 6000, 'ship_time' => '2026-09-03 12:00:00',
                'express_no' => 'weight-only-tracking', 'express_name2' => null,
                'freight' => null, 'settle_freight' => null, 'weight_kg' => null, 'settle_weight_kg' => 4.5],
        ]);

        $scanCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array {
            if (str_contains($sql, 'fys_jstFreightCostDetail')) {
                $this->fail('Skip-freight mode must not query the freight view.');
            }
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return $scanCalls++ === 0 ? [
                    ['id' => 6000, 'send_time' => '2026-09-03 12:00:00', 'weight' => 2, 'logistic_id' => 'new-weight-tracking'],
                    ['id' => 5000, 'send_time' => '2026-09-02 12:00:00', 'weight' => 2, 'logistic_id' => 'new-tracking'],
                    ['id' => 4000, 'send_time' => '2026-09-03 12:00:00', 'weight' => 2],
                ] : [];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertSame(3, $result['processed']);
        $this->assertSame(3, $result['packages']);
        $this->assertSame(3, $result['packages_without_items']);
        $this->assertTrue($result['freight_incomplete']);
        $this->assertSame('succeeded', $db->table('jst_sync_state')->where('dataset_key', 'jst.sctrade')->value('status'));
        $this->assertSame('pending', $db->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade.freight.20260901000000-20260905000000')->value('status'));
        $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']);
        $existing = $db->table('jst_oms_shipping')->where('source_id', 5000)->first();
        $this->assertSame('existing-tracking', $existing->express_no);
        $this->assertSame('existing-express', $existing->express_name2);
        $this->assertEquals(12.34, $existing->freight);
        $this->assertEquals(13.45, $existing->settle_freight);
        $this->assertEquals(1.2, $existing->weight_kg);
        $this->assertEquals(1.3, $existing->settle_weight_kg);
        $weightOnly = $db->table('jst_oms_shipping')->where('source_id', 6000)->first();
        $this->assertSame('weight-only-tracking', $weightOnly->express_no);
        $this->assertNull($weightOnly->freight);
        $this->assertEquals(4.5, $weightOnly->settle_weight_kg);
        $new = $db->table('jst_oms_shipping')->where('source_id', 4000)->first();
        $this->assertNull($new->freight);
        $this->assertNull($new->settle_freight);
        $this->assertNull($new->weight_kg);
        $this->assertNull($new->settle_weight_kg);
        $this->assertEquals(2, $new->actual_weight);

        (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-04'), Carbon::parse('2026-09-05'),
        );
        $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']);

        (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );
        $this->assertSame('pending', $db->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade.freight.20260901000000-20260905000000')->value('status'));
        $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']);
    }

    public function test_normal_window_preserves_old_charges_when_freight_is_missing_but_uses_new_cost(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            ['source_id' => 5000, 'sku_source_id' => 5000, 'ship_time' => '2026-09-02 12:00:00',
                'express_no' => 'old-5000', 'freight' => 12.34, 'settle_freight' => 13.45,
                'weight_kg' => 1.2, 'settle_weight_kg' => 1.3],
            ['source_id' => 6000, 'sku_source_id' => 6000, 'ship_time' => '2026-09-03 12:00:00',
                'express_no' => 'old-6000', 'freight' => 15.00, 'settle_freight' => 15.00,
                'weight_kg' => 1.5, 'settle_weight_kg' => 1.5],
        ]);
        $db->table('jst_sync_state')->insert([
            'dataset_key' => 'jst.sctrade.freight.20260901000000-20260905000000',
            'status' => 'pending',
        ]);

        $scanCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return $scanCalls++ === 0 ? [
                    ['id' => 6000, 'so_id' => 'so-6000', 'logistic_id' => 'new-6000',
                        'send_time' => '2026-09-03 12:00:00', 'weight' => 2],
                    ['id' => 5000, 'so_id' => 'so-5000', 'logistic_id' => 'new-5000',
                        'send_time' => '2026-09-02 12:00:00', 'weight' => 2],
                ] : [];
            }
            if (str_contains($sql, 'fys_jstFreightCostDetail')) {
                return [['so_id' => 'so-6000', 'expressNo' => 'new-6000', 'cost' => 7.50,
                    'w' => 2.5, 'name' => '快递公司']];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );

        $this->assertSame(0, $result['stale_mirror_packages']);
        $missing = $db->table('jst_oms_shipping')->where('source_id', 5000)->first();
        $this->assertSame('new-5000', $missing->express_no);
        $this->assertEquals(12.34, $missing->freight);
        $this->assertEquals(13.45, $missing->settle_freight);
        $this->assertEquals(1.2, $missing->weight_kg);
        $this->assertEquals(1.3, $missing->settle_weight_kg);
        $updated = $db->table('jst_oms_shipping')->where('source_id', 6000)->first();
        $this->assertEquals(7.50, $updated->freight);
        $this->assertEquals(7.50, $updated->settle_freight);
        $this->assertEquals(2.5, $updated->settle_weight_kg);
        $this->assertSame('succeeded', $db->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade.freight.20260901000000-20260905000000')->value('status'));
    }

    public function test_window_replacement_preserves_unreconstructed_package_and_matching_item_fields(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            'source_id' => 5000, 'sku_source_id' => 101, 'ship_time' => '2026-09-02 12:00:00',
            'package_no' => 'old-package', 'download_time' => '2026-09-02 13:00:00',
            'freight_rate' => 1.25, 'freight_formula' => 'legacy-formula',
            'settle_volume_cm3' => 100, 'recon_weight_kg' => 1.5, 'recon_volume_cm3' => 105,
            'finished_barcode' => 'finished-101', 'finished_name' => '成品甲',
            'finished_supplier' => '供应商甲',
        ]);
        $db->table('jst_oms_shipping')->insert([
            'source_id' => 5000, 'sku_source_id' => 102, 'ship_time' => '2026-09-02 12:00:00',
            'buyer_nick' => 'old-buyer', 'seller_memo' => 'old-seller-memo', 'front_no' => 'old-front-no',
            'receiver_name' => 'old-recipient', 'receiver_mobile' => 'old-mobile', 'receiver_phone' => 'old-phone',
            'finished_barcode' => 'finished-102', 'finished_name' => '成品乙',
            'finished_supplier' => '供应商乙',
        ]);
        $scanCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return $scanCalls++ === 0 ? [[
                    'id' => 5000, 'inner_order_id' => 'order-1', 'send_time' => '2026-09-02 12:00:00',
                ]] : [];
            }
            if (str_contains($sql, 'FROM rpt.jst_OrderItem WHERE o_id IN')) {
                return [
                    ['o_id' => 'order-1', 'Id' => 101, 'name' => '新商品甲', 'qty' => 1],
                    ['o_id' => 'order-1', 'Id' => 103, 'name' => '新商品丙', 'qty' => 1],
                ];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );

        $this->assertSame([101, 103], $db->table('jst_oms_shipping')->where('source_id', 5000)
            ->orderBy('sku_source_id')->pluck('sku_source_id')->map(fn ($id) => (int) $id)->all());
        $first = $db->table('jst_oms_shipping')->where('source_id', 5000)->where('sku_source_id', 101)->first();
        $this->assertSame('old-buyer', $first->buyer_nick);
        $this->assertSame('old-seller-memo', $first->seller_memo);
        $this->assertSame('old-recipient', $first->receiver_name);
        $this->assertSame('old-mobile', $first->receiver_mobile);
        $this->assertSame('old-phone', $first->receiver_phone);
        $this->assertSame('old-package', $first->package_no);
        $this->assertSame('2026-09-02 13:00:00', $first->download_time);
        $this->assertEquals(1.25, $first->freight_rate);
        $this->assertSame('legacy-formula', $first->freight_formula);
        $this->assertEquals(100, $first->settle_volume_cm3);
        $this->assertEquals(1.5, $first->recon_weight_kg);
        $this->assertEquals(105, $first->recon_volume_cm3);
        $this->assertSame('old-front-no', $first->front_no);
        $this->assertSame('finished-101', $first->finished_barcode);
        $this->assertSame('成品甲', $first->finished_name);
        $this->assertSame('供应商甲', $first->finished_supplier);
        $second = $db->table('jst_oms_shipping')->where('source_id', 5000)->where('sku_source_id', 103)->first();
        $this->assertNull($second->seller_memo);
        $this->assertNull($second->settle_volume_cm3);
        $this->assertNull($second->finished_barcode);
        $this->assertNull($second->finished_name);
        $this->assertNull($second->finished_supplier);
    }

    public function test_missing_new_items_preserves_old_product_rows_and_marks_items_pending(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            ['source_id' => 5000, 'sku_source_id' => 101, 'ship_time' => '2026-09-02 12:00:00',
                'barcode' => 'sku-101', 'product_name' => '旧商品甲', 'qty' => 1, 'amount' => 10.50,
                'freight' => 5.25],
            ['source_id' => 5000, 'sku_source_id' => 102, 'ship_time' => '2026-09-02 12:00:00',
                'barcode' => 'sku-102', 'product_name' => '旧商品乙', 'qty' => 2, 'amount' => 20.75,
                'freight' => null],
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings): array {
            if (str_contains($sql, 'rpt.dws_sn_sales_full_info_di') && $bindings === ['2026-09-02']) {
                throw new RuntimeException('DWS read timed out');
            }
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [['id' => 5000, 'inner_order_id' => 'order-1',
                    'send_time' => '2026-09-02 12:00:00']];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertSame(1, $result['packages_without_items']);
        $this->assertSame(1, $result['packages_retained_for_missing_items']);
        $this->assertContains('2026-09-02', $result['dws_failed_days']);
        $this->assertSame('pending', $db->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade.items.20260901000000-20260905000000')->value('status'));
        $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']);
        $rows = $db->table('jst_oms_shipping')->where('source_id', 5000)->orderBy('sku_source_id')->get();
        $this->assertCount(2, $rows);
        $this->assertSame(['sku-101', 'sku-102'], $rows->pluck('barcode')->all());
        $this->assertSame(['旧商品甲', '旧商品乙'], $rows->pluck('product_name')->all());
        $this->assertEquals(10.50, $rows[0]->amount);
        $this->assertEquals(20.75, $rows[1]->amount);
        $this->assertEquals(5.25, $rows[0]->freight);
    }

    public function test_view_item_with_null_amount_preserves_matching_synthetic_sku_amount(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            'source_id' => 5000, 'sku_source_id' => 500000, 'ship_time' => '2026-09-02 12:00:00',
            'barcode' => 'sku-101', 'product_name' => '旧商品', 'amount' => 10.50,
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [['id' => 5000, 'inner_order_id' => 'order-1',
                    'send_time' => '2026-09-02 12:00:00']];
            }
            if (str_contains($sql, 'FROM rpt.jst_salesOutboundInfo WHERE io_date')) {
                return [['o_id' => 'order-1', 'io_id' => 'outbound-1']];
            }
            if (str_contains($sql, 'FROM rpt.Jst_SendData_View_extension')) {
                return [['io_id' => 'outbound-1', 'sku_ids' => 'sku-101', 'qty' => 1,
                    'type' => 0, 'sku_name' => '实发商品', 'volume' => null]];
            }
            if (str_contains($sql, 'FROM rpt.jst_OrderItem WHERE o_id IN')) {
                return [['o_id' => 'order-1', 'Id' => 101, 'sku_id' => 'other-sku',
                    'name' => '订单商品', 'qty' => 1, 'amount' => 999]];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );

        $this->assertSame(0, $result['packages_retained_for_unmatched_amount']);
        $row = $db->table('jst_oms_shipping')->where('source_id', 5000)->sole();
        $this->assertSame(500000, (int) $row->sku_source_id);
        $this->assertSame('实发商品', $row->product_name);
        $this->assertEquals(10.50, $row->amount);
    }

    public function test_view_item_with_unmatched_synthetic_sku_retains_old_package_and_marks_items_pending(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            ['source_id' => 5000, 'sku_source_id' => 500000, 'ship_time' => '2026-09-02 12:00:00',
                'barcode' => null, 'product_name' => '旧商品甲', 'amount' => 10.50],
            ['source_id' => 6000, 'sku_source_id' => 600000, 'ship_time' => '2026-09-02 13:00:00',
                'barcode' => 'old-sku', 'product_name' => '旧商品乙', 'amount' => 20.75],
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [
                    ['id' => 5000, 'inner_order_id' => 'order-1', 'send_time' => '2026-09-02 12:00:00'],
                    ['id' => 6000, 'inner_order_id' => 'order-2', 'send_time' => '2026-09-02 13:00:00'],
                ];
            }
            if (str_contains($sql, 'FROM rpt.jst_salesOutboundInfo WHERE io_date')) {
                return [
                    ['o_id' => 'order-1', 'io_id' => 'outbound-1'],
                    ['o_id' => 'order-2', 'io_id' => 'outbound-2'],
                ];
            }
            if (str_contains($sql, 'FROM rpt.Jst_SendData_View_extension')) {
                return [
                    ['io_id' => 'outbound-1', 'sku_ids' => 'new-sku', 'qty' => 1, 'type' => 0,
                        'sku_name' => '新商品甲', 'volume' => null],
                    ['io_id' => 'outbound-2', 'sku_ids' => 'new-sku', 'qty' => 1, 'type' => 0,
                        'sku_name' => '新商品乙', 'volume' => null],
                ];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );

        // 旧条码为空的包裹无法判断是否错配：保留旧记录；旧条码 old-sku 与实发 new-sku 完全不同：视为商品已更正，覆盖旧记录
        $this->assertSame(1, $result['packages_retained_for_unmatched_amount']);
        $this->assertSame(1, $result['packages_items_corrected']);
        $this->assertSame('pending', $db->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade.items.20260901000000-20260905000000')->value('status'));
        $this->assertSame('partial', (new ShippingOrderRepository())->syncStatus()['code']);
        $rows = $db->table('jst_oms_shipping')->orderBy('source_id')->get();
        $this->assertSame(['旧商品甲', '新商品乙'], $rows->pluck('product_name')->all());
        $this->assertEquals(10.50, $rows[0]->amount);
    }

    public function test_explicit_zero_item_amount_replaces_old_amount(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            'source_id' => 5000, 'sku_source_id' => 101, 'ship_time' => '2026-09-02 12:00:00',
            'barcode' => 'sku-101', 'amount' => 10.50,
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [['id' => 5000, 'inner_order_id' => 'order-1',
                    'send_time' => '2026-09-02 12:00:00']];
            }
            if (str_contains($sql, 'FROM rpt.jst_OrderItem WHERE o_id IN')) {
                return [['o_id' => 'order-1', 'Id' => 101, 'sku_id' => 'sku-101',
                    'name' => '新商品', 'qty' => 1, 'amount' => 0]];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );

        $this->assertSame(0, $result['packages_retained_for_unmatched_amount']);
        $row = $db->table('jst_oms_shipping')->where('source_id', 5000)->sole();
        $this->assertSame('新商品', $row->product_name);
        $this->assertEquals(0, $row->amount);
    }

    public function test_shipping_command_reports_express_and_autoship_counts_separately(): void
    {
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [['id' => 5000, 'send_time' => '2026-09-02 12:00:00']];
            }
            if (str_contains($sql, 'FROM rpt.jst_Orders o WHERE')) {
                return [['rid' => 100, 'o_id' => '123', 'send_date' => '2026-09-02 13:00:00']];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);
        $this->app->instance(UpstreamReadOnlyQueryFactory::class, $factory);

        $this->artisan('jushuitan:sync-shipping', [
            '--from' => '2026-09-01', '--to' => '2026-09-05', '--skip-freight' => true,
        ])->expectsOutputToContain('唯一包裹源键 2 个，包裹记录处理 2 次；发货明细 1 条，代发仓 1 条，销售出库补漏 0 条，采购出库补漏 0 条，合计处理 2 条明细')
            ->assertExitCode(0);
    }

    public function test_normal_window_failure_preserves_unprocessed_old_packages(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            ['source_id' => 3000, 'sku_source_id' => 3000, 'ship_time' => '2026-09-02 12:00:00',
                'express_no' => 'processed-old', 'freight' => 11.11],
            ['source_id' => 1000, 'sku_source_id' => 1000, 'ship_time' => '2026-09-02 12:00:00',
                'express_no' => 'unprocessed-old', 'freight' => 22.22],
        ]);

        $scanCalls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql) use (&$scanCalls): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                if ($scanCalls++ > 0) {
                    throw new RuntimeException('upstream page failed');
                }
                return array_map(fn (int $id): array => [
                    'id' => $id,
                    'send_time' => $id >= 2701 ? '2026-09-02 12:00:00' : '2026-08-31 12:00:00',
                ], range(3000, 1501));
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        try {
            (new ShippingReplicaSynchronizer($factory))->sync(
                Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
            );
            $this->fail('Expected the second upstream page to fail.');
        } catch (RuntimeException $e) {
            $this->assertSame('upstream page failed', $e->getMessage());
        }

        $this->assertSame(2, $scanCalls);
        $this->assertSame('failed', $db->table('jst_sync_state')->where('dataset_key', 'jst.sctrade')->value('status'));
        $processed = $db->table('jst_oms_shipping')->where('source_id', 3000)->first();
        $this->assertEquals(11.11, $processed->freight);
        $unprocessed = $db->table('jst_oms_shipping')->where('source_id', 1000)->first();
        $this->assertSame('unprocessed-old', $unprocessed->express_no);
        $this->assertEquals(22.22, $unprocessed->freight);
        $this->assertSame(301, $db->table('jst_oms_shipping')->count());
    }

    public function test_at_prefix_dedup_requires_same_trade_and_same_window(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            ['source_id' => 1, 'sku_source_id' => 1, 'ship_time' => '2026-09-02 12:00:00',
                'trade_no' => 'same-trade', 'express_no' => 'clean-1'],
            ['source_id' => 2, 'sku_source_id' => 2, 'ship_time' => '2026-09-02 12:00:00',
                'trade_no' => 'same-trade', 'express_no' => '@clean-1'],
            ['source_id' => 3, 'sku_source_id' => 3, 'ship_time' => '2026-09-02 12:00:00',
                'trade_no' => 'other-trade', 'express_no' => '@clean-1'],
            ['source_id' => 4, 'sku_source_id' => 4, 'ship_time' => '2026-09-08 12:00:00',
                'trade_no' => 'outside-trade', 'express_no' => 'clean-2'],
            ['source_id' => 5, 'sku_source_id' => 5, 'ship_time' => '2026-09-02 12:00:00',
                'trade_no' => 'outside-trade', 'express_no' => '@clean-2'],
            ['source_id' => 6, 'sku_source_id' => 6, 'ship_time' => '2026-09-03 12:00:00',
                'trade_no' => 'same-trade', 'express_no' => '@clean-1'],
        ]);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($this->createMock(UpstreamReadOnlyQuery::class));

        $removed = (new ShippingReplicaSynchronizer($factory))->dedupAtPrefix(
            '2026-09-01 00:00:00', '2026-09-05 00:00:00',
        );

        $this->assertSame(1, $removed);
        $this->assertSame([1, 3, 4, 5, 6], $db->table('jst_oms_shipping')
            ->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all());
        $this->assertSame(0, (new ShippingReplicaSynchronizer($factory))->dedupAtPrefix());
    }

    public function test_non_window_fast_sync_still_ignores_existing_package(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            'source_id' => 5000, 'sku_source_id' => 5000,
            'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'existing-tracking', 'freight' => 12.34,
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [[
                    'id' => 5000,
                    'send_time' => '2026-09-02 12:00:00',
                    'logistic_id' => 'new-tracking',
                ]];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync();

        $this->assertSame(1, $result['processed']);
        $this->assertSame(1, $db->table('jst_oms_shipping')->count());
        $existing = $db->table('jst_oms_shipping')->where('source_id', 5000)->first();
        $this->assertSame('existing-tracking', $existing->express_no);
        $this->assertEquals(12.34, $existing->freight);
    }

    public function test_failed_dws_preload_is_reported_separately_from_package_count(): void
    {
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings): array {
            if (str_contains($sql, 'rpt.dws_sn_sales_full_info_di') && $bindings === ['2026-08-31']) {
                throw new RuntimeException('read timed out');
            }
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [['id' => 1234, 'send_time' => '2026-09-02 12:00:00']];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );

        $this->assertSame(1, $result['packages']);
        $this->assertSame(1, $result['packages_without_items']);
        $this->assertSame(['2026-08-31'], $result['dws_failed_days']);
        $this->assertSame('pending', DB::connection('business_mirror')->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade.items.20260901000000-20260905000000')->value('status'));
    }

    public function test_skip_freight_rolls_back_old_package_when_replacement_insert_conflicts(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            'source_id' => 5000, 'sku_source_id' => 5000,
            'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'old-tracking', 'freight' => 12.34,
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($upstream);
        $synchronizer = new ShippingReplicaSynchronizer($factory);

        $skipFreight = new \ReflectionProperty($synchronizer, 'skipFreight');
        $skipFreight->setAccessible(true);
        $skipFreight->setValue($synchronizer, true);
        $replacements = new \ReflectionProperty($synchronizer, 'replacementPackages');
        $replacements->setAccessible(true);
        $newRow = [
            'source_id' => 5000, 'sku_source_id' => 123,
            'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'new-tracking', 'express_name2' => null,
            'freight' => null, 'settle_freight' => null, 'weight_kg' => null, 'settle_weight_kg' => null,
            'amount' => null,
            'receiver_name' => null, 'receiver_mobile' => null, 'receiver_phone' => null,
        ];
        $replacements->setValue($synchronizer, [5000 => [$newRow, $newRow]]);
        $flush = new \ReflectionMethod($synchronizer, 'flushReplacements');
        $flush->setAccessible(true);

        try {
            $flush->invoke($synchronizer);
            $this->fail('Expected a duplicate replacement row to roll back the package transaction.');
        } catch (\Illuminate\Database\QueryException $e) {
            $this->assertStringContainsString('UNIQUE constraint failed', $e->getMessage());
        }

        $rows = $db->table('jst_oms_shipping')->where('source_id', 5000)->get();
        $this->assertCount(1, $rows);
        $this->assertSame(5000, (int) $rows[0]->sku_source_id);
        $this->assertSame('old-tracking', $rows[0]->express_no);
        $this->assertEquals(12.34, $rows[0]->freight);
    }

    public function test_skip_freight_keeps_unseen_mirror_package_and_marks_window_partial(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            'source_id' => 7000, 'sku_source_id' => 7000,
            'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'unseen-tracking', 'freight' => 22.50,
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturn([]);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );

        $this->assertSame(1, $result['stale_mirror_packages']);
        $this->assertSame('partial', $db->table('jst_sync_state')->where('dataset_key', 'jst.sctrade')->value('status'));
        $this->assertStringContainsString('源键核对', $db->table('jst_sync_state')
            ->where('dataset_key', 'jst.sctrade')->value('last_error'));
        $old = $db->table('jst_oms_shipping')->where('source_id', 7000)->first();
        $this->assertSame('unseen-tracking', $old->express_no);
        $this->assertEquals(22.50, $old->freight);
    }

    public function test_partial_sync_status_uses_business_warning_instead_of_raw_error(): void
    {
        DB::connection('business_mirror')->table('jst_sync_state')->insert([
            'dataset_key' => 'jst.sctrade',
            'status' => 'partial',
            'last_error' => 'SQLSTATE[42S02]: SECRET_MIRROR_TABLE does not exist',
        ]);

        $status = (new ShippingOrderRepository())->syncStatus();

        $this->assertSame('partial', $status['code']);
        $this->assertStringContainsString('源键核对', $status['warning']);
        $this->assertStringNotContainsString('SQLSTATE', $status['warning']);
        $this->assertStringNotContainsString('SECRET_MIRROR_TABLE', $status['warning']);
    }

    public function test_sync_removes_only_the_known_demo_source_ids(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            ['source_id' => 9000001, 'sku_source_id' => 9000001],
            ['source_id' => 9000006, 'sku_source_id' => 9000006],
            ['source_id' => 9000007, 'sku_source_id' => 9000007],
            ['source_id' => 999999999, 'sku_source_id' => 999999999],
        ]);

        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->expects($this->once())->method('select')->willReturn([]);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->expects($this->once())
            ->method('forSource')
            ->with('operations_replica')
            ->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync();

        $this->assertSame(['processed' => 0, 'autoship' => 0], $result);
        $this->assertSame(
            [9000007, 999999999],
            $db->table('jst_oms_shipping')->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all(),
        );
    }

    public function test_outbound_fallback_flushes_replacements_and_preserves_freight_when_skipped(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            'source_id' => 2000000420, 'sku_source_id' => 2000000420,
            'express_no' => 'outbound-420', 'trade_no' => 'trade-420',
            'ship_time' => '2026-09-02 12:00:00', 'freight' => 18.50, 'settle_weight_kg' => 7,
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'MIN(Id) AS lo')) {
                return [['lo' => 1, 'hi' => 1]];
            }
            if (str_contains($sql, 'WHERE Id >= ?')) {
                return [['io_id' => '420', 'so_id' => 'trade-420', 'l_id' => 'outbound-420', 'io_date' => '2026-09-02 12:00:00']];
            }
            if (str_contains($sql, 'SELECT io_id, sku_ids')) {
                return [['io_id' => '420', 'sku_ids' => 'sku-420', 'qty' => 2, 'type' => '0', 'volume' => 100, 'l' => 10, 'w' => 5, 'h' => 2]];
            }
            $this->assertStringNotContainsString('fys_jstFreightCost', $sql);
            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($upstream);
        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true,
        );
        $this->assertSame(1, $result['outbound']);
        $row = $db->table('jst_oms_shipping')->where('source_id', 2000000420)->sole();
        $this->assertSame('sku-420', $row->barcode);
        $this->assertEquals(2, $row->qty);
        $this->assertEquals(18.50, $row->freight);
        $this->assertEquals(7, $row->settle_weight_kg);
    }

    public function test_outbound_mapping_keeps_shared_tracking_orders_distinct_and_ignores_cancelled(): void
    {
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($upstream);
        $sync = new ShippingReplicaSynchronizer($factory);
        $map = (new \ReflectionMethod($sync, 'mapOutboundRows'))->invoke($sync, [
            ['io_id' => '10', 'o_id' => 'A', 'l_id' => '@shared', 'io_date' => '2026-09-02'],
            ['io_id' => '20', 'o_id' => 'B', 'l_id' => 'shared', 'io_date' => '2026-09-03'],
            ['io_id' => '30', 'o_id' => 'A', 'l_id' => 'shared', 'io_date' => '2026-09-04', 'status' => 'Cancelled'],
        ]);
        $method = new \ReflectionMethod($sync, 'ioFor');
        $this->assertSame('10', $method->invoke($sync, $map, 'shared', 'A'));
        $this->assertSame('20', $method->invoke($sync, $map, '@shared', 'B'));
        // 作废出库单保留在映射里但排在有效单之后，不会被选中
        $this->assertSame('10', $map['byLO']['shared|A']);
        $this->assertArrayNotHasKey('30', $map['byOAll']['A']);
    }

    public function test_tracking_replacement_pairs_one_candidate_per_old_tracking(): void
    {
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturn([]);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($upstream);
        $sync = new ShippingReplicaSynchronizer($factory);
        $method = new \ReflectionMethod($sync, 'pairChangedTracking');
        $rows = [['id' => 1, 'inner_order_id' => 'A', 'logistic_id' => 'old']];
        // 同订单多张候选出库单时按出库单号顺序一对一配对(对齐工作平台换单只保留新单号)
        $this->assertSame(['1' => ['io' => '10', 'l_id' => 'new-a', 'old' => 'old']],
            $method->invoke($sync, $rows, ['byL' => [], 'byOAll' => ['A' => ['10' => 'new-a', '20' => 'new-b']]]));
        $this->assertSame(['1' => ['io' => '10', 'l_id' => 'new-a', 'old' => 'old']],
            $method->invoke($sync, $rows, ['byL' => [], 'byOAll' => ['A' => ['10' => 'new-a']]]));
    }

    public function test_cloud_order_label_is_included_and_volume_rules_are_bounded(): void
    {
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($this->createMock(UpstreamReadOnlyQuery::class));
        $sync = new ShippingReplicaSynchronizer($factory);
        $label = new \ReflectionMethod($sync, 'excludedByLabel');
        $this->assertTrue($label->invoke($sync, '线上发货'));
        $this->assertFalse($label->invoke($sync, '线上发货,云仓订单'));
        $volume = new \ReflectionMethod($sync, 'detailVolume');
        $this->assertSame(120.0, $volume->invoke($sync, 10, 3, 2, 2.0, '普通仓', '中通速递'));
        $this->assertSame(60.0, $volume->invoke($sync, 10, 3, 2, 2.0, '常州简尚家居有限公司', '自送'));
        $this->assertNull($volume->invoke($sync, 0, 3, 2, 2.0, '普通仓', '中通速递'));
        $this->assertSame(0.0, $volume->invoke($sync, 10, 3, 2, 2.0, '普通仓', '中通礼品单'));
    }

    public function test_stale_express_detail_carrier_and_warehouse_follow_the_actual_outbound(): void
    {
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturn([
            // 先分给工厂代发仓后作废，再由总仓按外采出库(实际发货单)
            ['l_id' => 'JT1', 'o_id' => 'o1', 'io_id' => '11', 'io_date' => '2026-08-29 17:52:09', 'status' => 'Delete', 'logistics_company' => '浩浦极兔', 'wms_co_id' => 15170678],
            ['l_id' => 'JT1', 'o_id' => 'o1', 'io_id' => '12', 'io_date' => '2026-09-01 10:17:16', 'status' => 'Confirmed', 'logistics_company' => '外采快递', 'wms_co_id' => 14138566],
            ['l_id' => 'DP2', 'o_id' => 'o2', 'io_id' => '21', 'io_date' => '2026-09-02 09:00:00', 'status' => 'Confirmed', 'logistics_company' => '外采快递', 'wms_co_id' => 14138566],
            ['l_id' => 'YT3', 'o_id' => 'o3', 'io_id' => '31', 'io_date' => '2026-09-02 09:00:00', 'status' => 'Confirmed', 'logistics_company' => '线下快递', 'wms_co_id' => 15137729],
            ['l_id' => '', 'o_id' => 'o4', 'io_id' => '41', 'io_date' => '2026-09-02 09:00:00', 'status' => 'Confirmed', 'logistics_company' => '无需物流', 'wms_co_id' => 15170001],
        ]);
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($upstream);
        $sync = new ShippingReplicaSynchronizer($factory);
        $preload = new \ReflectionMethod($sync, 'preloadOutboundMap');
        $preload->setAccessible(true);
        $preload->invoke($sync, '2026-08-25 00:00:00', '2026-09-05 00:00:00');
        $map = new \ReflectionProperty($sync, 'outboundMap');
        $map->setAccessible(true);
        $ioFor = new \ReflectionMethod($sync, 'ioFor');
        $ioFor->setAccessible(true);
        $this->assertSame('12', $ioFor->invoke($sync, $map->getValue($sync), 'JT1', 'o1'));
        $carrier = new \ReflectionMethod($sync, 'withOutboundCarrier');
        $carrier->setAccessible(true);
        $wmsName = new \ReflectionMethod($sync, 'wmsName');
        $wmsName->setAccessible(true);
        $pending = new \ReflectionProperty($sync, 'pendingWarehouse');
        $pending->setAccessible(true);

        // 出库单快递公司不同：快递公司按出库单，仓库登记待回填
        $early = $carrier->invoke($sync, '31', ['logistic_company_name' => '霸州百世快运-雅柏', 'warehouse_name' => '工厂-河北雅柏代发仓'],
            ['express_name1' => '霸州百世快运-雅柏', 'warehouse_name' => '工厂-河北雅柏代发仓'], 3);
        $this->assertSame('线下快递', $early['express_name1']);
        $this->assertSame('15137729', $pending->getValue($sync)[3]);

        // 发货明细停在作废前的工厂代发分配：快递公司按出库单；仓库按出库单发货主体回填
        $stale = $carrier->invoke($sync, '12', ['logistic_company_name' => '浩浦极兔', 'warehouse_name' => '工厂-河北浩浦代发仓'],
            ['express_name1' => '浩浦极兔', 'warehouse_name' => '工厂-河北浩浦代发仓'], 1);
        $this->assertSame('外采快递', $stale['express_name1']);
        $this->assertSame('14138566', $pending->getValue($sync)[1]);
        $this->assertNull($wmsName->invoke($sync, '14138566')); // 样本不足不回填
        foreach ([20, 22, 23] as $id) {
            $carrier->invoke($sync, '21', ['logistic_company_name' => '外采快递', 'warehouse_name' => '常州简尚家居有限公司'],
                ['express_name1' => '外采快递', 'warehouse_name' => '常州简尚家居有限公司'], $id);
        }
        $this->assertSame('常州简尚家居有限公司', $wmsName->invoke($sync, '14138566'));

        // 无需物流(补差/信誉/运费)：仓库统一自动发货仓，不按发货主体回填
        $none = $carrier->invoke($sync, '41', ['logistic_company_name' => '', 'warehouse_name' => '工厂-山东祥和代发仓'],
            ['express_name1' => '', 'warehouse_name' => '工厂-山东祥和代发仓'], 4);
        $this->assertSame('无需物流', $none['express_name1']);
        $this->assertSame('简尚-自动发货仓', $none['warehouse_name']);
        $this->assertArrayNotHasKey(4, $pending->getValue($sync));

        // 同一发货主体对应多个仓库名(无明显多数)时不回填
        $names = new \ReflectionProperty($sync, 'wmsNames');
        $names->setAccessible(true);
        $names->setValue($sync, ['9' => ['仓A' => 5, '仓B' => 4]]);
        $this->assertNull($wmsName->invoke($sync, '9'));
    }

    public function test_tracking_pairing_treats_deleted_only_outbound_as_changed_and_survives_lookup_timeout(): void
    {
        $calls = 0;
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function () use (&$calls): array {
            $calls++;
            if ($calls === 1) {
                return [];
            }
            throw new PDOException('SQLSTATE[HY000]: General error: 20003 Adaptive Server connection timed out [20003] (severity 6)');
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($upstream);
        $sync = new ShippingReplicaSynchronizer($factory);
        $ok = new \ReflectionProperty($sync, 'ioOk');
        $ok->setAccessible(true);
        $ok->setValue($sync, ['11' => false, '12' => true]);
        $pair = new \ReflectionMethod($sync, 'pairChangedTracking');
        $pair->setAccessible(true);
        $ioMap = ['byL' => ['DPK1' => '11', 'SF2' => '12'], 'byO' => ['o1' => '12'], 'byLO' => [], 'byOAll' => ['o1' => ['12' => 'SF2']], 'ioL' => ['11' => 'DPK1', '12' => 'SF2']];
        $rows = [['id' => 7, 'logistic_id' => 'DPK1', 'inner_order_id' => 'o1']];

        $swap = $pair->invoke($sync, $rows, $ioMap);
        $this->assertSame(['io' => '12', 'l_id' => 'SF2', 'old' => 'DPK1'], $swap['7']);

        // 副本库超时：本页跳过配对，不抛错
        $this->assertSame([], $pair->invoke($sync, $rows, $ioMap));
        $skipped = new \ReflectionProperty($sync, 'pairingSkipped');
        $skipped->setAccessible(true);
        $this->assertSame(1, $skipped->getValue($sync));
    }

    public function test_view_express_and_warehouse_override_inferred_carrier_like_the_work_platform(): void
    {
        $chunks = [];
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings) use (&$chunks): array {
            $this->assertStringContainsString('expressName', $sql);
            $chunks[] = count($bindings);

            return [
                ['io_id' => '12', 'sku_ids' => 'a', 'qty' => 1, 'type' => 0, 'volume' => null, 'expressName' => '常州德邦', 'wareHoursName' => '工厂-河南千优代发仓'],
                ['io_id' => '12', 'sku_ids' => 'b', 'qty' => 1, 'type' => 0, 'volume' => null, 'expressName' => '', 'wareHoursName' => ''],
                ['io_id' => '41', 'sku_ids' => 'c', 'qty' => 1, 'type' => 0, 'volume' => null, 'expressName' => '无需物流', 'wareHoursName' => '简尚-自动发货仓'],
            ];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($upstream);
        $sync = new ShippingReplicaSynchronizer($factory);
        $load = new \ReflectionMethod($sync, 'loadViewItems');
        $load->setAccessible(true);
        $load->invoke($sync, array_merge(['12', '41', '99'], array_map('strval', range(1000, 1400))));
        $this->assertSame([300, 104], $chunks); // 视图按 300 个出库单一批

        $pending = new \ReflectionProperty($sync, 'pendingWarehouse');
        $pending->setAccessible(true);
        $pending->setValue($sync, [1 => '14138566', 2 => '15170678']);
        $carrier = new \ReflectionMethod($sync, 'withViewCarrier');
        $carrier->setAccessible(true);

        // 视图有值：快递名称、仓库按视图(工作平台口径)，不再按发货主体回填
        // 快递名称：出库单/发货明细已有当前值(换单后)时不被视图旧值覆盖；没有时用视图兜底
        $pkg = $carrier->invoke($sync, '12', ['express_name1' => '崔桥顺丰干配', 'warehouse_name' => '工厂-河北浩浦代发仓'], 1);
        $this->assertSame('崔桥顺丰干配', $pkg['express_name1']);
        $blank = $carrier->invoke($sync, '12', ['express_name1' => '', 'warehouse_name' => ''], 5);
        $this->assertSame('常州德邦', $blank['express_name1']);
        $this->assertSame('工厂-河南千优代发仓', $pkg['warehouse_name']);
        $this->assertArrayNotHasKey(1, $pending->getValue($sync));

        $none = $carrier->invoke($sync, '41', ['express_name1' => '无需物流', 'warehouse_name' => '工厂-山东恒思代发仓'], 4);
        $this->assertSame('简尚-自动发货仓', $none['warehouse_name']);

        // 视图没有该出库单：保留出库单/发货明细推断值和待回填仓库
        $keep = $carrier->invoke($sync, '99', ['express_name1' => '外采快递', 'warehouse_name' => '常州简尚家居有限公司'], 2);
        $this->assertSame('外采快递', $keep['express_name1']);
        $this->assertSame('常州简尚家居有限公司', $keep['warehouse_name']);
        $this->assertSame('15170678', $pending->getValue($sync)[2]);
    }

    public function test_retained_package_refreshes_address_and_superseded_fallback_package_is_removed(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            ['source_id' => 5000, 'sku_source_id' => 500000, 'ship_time' => '2026-09-02 12:00:00', 'express_no' => 'YT5',
                'trade_no' => null, 'barcode' => null, 'product_name' => '旧商品甲', 'amount' => 10.50, 'receiver_state' => null],
            // 凌晨先按自动发货仓兜底入库的旧包裹，之后发货明细到齐由 6000 吐出同一单号
            ['source_id' => 1000000777, 'sku_source_id' => 1000000777, 'ship_time' => '2026-09-02 13:00:00', 'express_no' => 'YT6',
                'trade_no' => null, 'barcode' => null, 'product_name' => null, 'amount' => null, 'receiver_state' => null],
            ['source_id' => 1000000888, 'sku_source_id' => 1000000888, 'ship_time' => '2026-09-02 14:00:00', 'express_no' => 'ZZ9',
                'trade_no' => null, 'barcode' => null, 'product_name' => null, 'amount' => null, 'receiver_state' => null],
            // 采购退货出库：上游已作废，本次未再出现
            ['source_id' => 3000000555, 'sku_source_id' => 3000000555, 'ship_time' => '2026-09-02 15:00:00', 'express_no' => 'SF555',
                'trade_no' => null, 'barcode' => null, 'product_name' => null, 'amount' => null, 'receiver_state' => null],
            // 凌晨入库时还没有快递单号的自动发货仓包裹，同一交易单号本次已由 6000 吐出
            ['source_id' => 1000000999, 'sku_source_id' => 1000000999, 'ship_time' => '2026-09-02 13:30:00', 'express_no' => null,
                'trade_no' => 'so-2', 'barcode' => null, 'product_name' => null, 'amount' => null, 'receiver_state' => null],
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'FROM rpt.jst_order_express_details e')) {
                return [
                    ['id' => 5000, 'inner_order_id' => 'order-1', 'logistic_id' => 'YT5', 'send_time' => '2026-09-02 12:00:00',
                        'receiver_address' => '河南省-郑州市-金水区'],
                    ['id' => 6000, 'inner_order_id' => 'order-2', 'so_id' => 'so-2', 'logistic_id' => 'YT6', 'send_time' => '2026-09-02 13:00:00',
                        'receiver_address' => '江苏省-常州市-武进区'],
                ];
            }
            if (str_contains($sql, 'FROM rpt.jst_salesOutboundInfo WHERE io_date')) {
                return [
                    ['o_id' => 'order-1', 'io_id' => 'outbound-1', 'l_id' => 'YT5'],
                    ['o_id' => 'order-2', 'io_id' => 'outbound-2', 'l_id' => 'YT6'],
                ];
            }
            if (str_contains($sql, 'FROM rpt.jst_purchaseout WHERE io_date')) {
                return [['io_id' => '555', 'io_date' => '2026-09-02 15:00:00', 'status' => 'Cancelled', 'l_id' => 'SF555']];
            }
            if (str_contains($sql, 'FROM rpt.Jst_SendData_View_extension')) {
                return [
                    ['io_id' => 'outbound-1', 'sku_ids' => 'new-sku', 'qty' => 1, 'type' => 0, 'sku_name' => '新商品甲', 'volume' => null],
                    ['io_id' => 'outbound-2', 'sku_ids' => 'new-sku', 'qty' => 1, 'type' => 0, 'sku_name' => '新商品乙', 'volume' => null],
                ];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $result = (new ShippingReplicaSynchronizer($factory))->sync(
            Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'),
        );

        $this->assertSame(1, $result['packages_retained_for_unmatched_amount']);
        $this->assertSame(3, $result['superseded_packages_removed']);
        $this->assertSame(0, $db->table('jst_oms_shipping')->where('source_id', 3000000555)->count());
        $this->assertSame(0, $db->table('jst_oms_shipping')->where('source_id', 1000000999)->count());
        $this->assertSame(1, $result['stale_mirror_packages']);
        // 保留的旧包裹：商品金额不动，省市区按上游刷新
        $kept = $db->table('jst_oms_shipping')->where('source_id', 5000)->sole();
        $this->assertSame('旧商品甲', $kept->product_name);
        $this->assertEquals(10.50, $kept->amount);
        $this->assertSame('河南省', $kept->receiver_state);
        $this->assertSame('郑州市', $kept->receiver_city);
        // 被发货明细取代的兜底包裹删除；无对应的仍保留待核对
        $this->assertSame(0, $db->table('jst_oms_shipping')->where('source_id', 1000000777)->count());
        $this->assertSame(1, $db->table('jst_oms_shipping')->where('source_id', 1000000888)->count());
        $this->assertSame('江苏省', $db->table('jst_oms_shipping')->where('source_id', 6000)->value('receiver_state'));
    }

    public function test_zero_quantity_order_item_placeholder_is_dropped_when_package_has_shipped_items(): void
    {
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($this->createMock(UpstreamReadOnlyQuery::class));
        $synchronizer = new ShippingReplicaSynchronizer($factory);
        $method = new \ReflectionMethod($synchronizer, 'dropZeroQtyPlaceholders');
        $method->setAccessible(true);

        $items = [
            ['id' => 4343977, 'sku' => '923141324442', 'shopid' => '3780268746917151177', 'qty' => 0, 'amount' => '0.00'],
            ['id' => 4344020, 'sku' => 'CBZ000216N1071V01', 'shopid' => '3780268746917151177', 'qty' => 1, 'amount' => '81.66'],
        ];
        $kept = $method->invoke($synchronizer, $items);
        $this->assertCount(1, $kept);
        $this->assertSame('CBZ000216N1071V01', $kept[0]['sku']);

        // 只有占位行时保持原样(不把包裹变成无商品)
        $onlyZero = [['id' => 1, 'sku' => 'X', 'qty' => 0, 'amount' => 0]];
        $this->assertSame($onlyZero, $method->invoke($synchronizer, $onlyZero));

        // 数量未知(null)的行不当作占位行
        $unknownQty = [['id' => 2, 'sku' => 'A', 'qty' => null, 'amount' => null], ['id' => 3, 'sku' => 'B', 'qty' => 1, 'amount' => 5]];
        $this->assertCount(2, $method->invoke($synchronizer, $unknownQty));

        // 数量 0 但有金额的行保留
        $paidZero = [['id' => 4, 'sku' => 'A', 'qty' => 0, 'amount' => 3.5], ['id' => 5, 'sku' => 'B', 'qty' => 1, 'amount' => 5]];
        $this->assertCount(2, $method->invoke($synchronizer, $paidZero));
    }

    public function test_overlong_text_values_are_truncated_to_column_limits(): void
    {
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($this->createMock(UpstreamReadOnlyQuery::class));
        $synchronizer = new ShippingReplicaSynchronizer($factory);
        $method = new \ReflectionMethod($synchronizer, 'cleanPayload');
        $method->setAccessible(true);

        $itemId = 'JST15438196_14138566_ZH-GBSFY-YMJ011901AN1403V01-1-GBSFY-YMJ011901BN1403V01-1';
        $this->assertGreaterThan(64, mb_strlen($itemId));
        $out = $method->invoke($synchronizer, ['item_id' => $itemId, 'barcode' => 'ABC', 'qty' => 2]);
        $this->assertSame(64, mb_strlen($out['item_id']));
        $this->assertSame(mb_substr($itemId, 0, 64), $out['item_id']);
        $this->assertSame('ABC', $out['barcode']);
        $this->assertSame(2, $out['qty']);
    }

    public function test_ship_time_is_moved_to_outbound_io_date_when_warehouse_ships_next_day(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            // 晚上打单(发货明细时间)、次日早上出库：应改成出库时间(一个包裹两行商品都改)
            ['source_id' => 7001, 'sku_source_id' => 1, 'ship_time' => '2026-10-20 20:40:51', 'express_no' => '79041235735054'],
            ['source_id' => 7001, 'sku_source_id' => 2, 'ship_time' => '2026-10-20 20:40:51', 'express_no' => '79041235735054'],
            // 已经是出库时间：不动
            ['source_id' => 7002, 'sku_source_id' => 1, 'ship_time' => '2026-10-21 09:00:00', 'express_no' => 'YT2'],
            // 单号带 @ 前缀、出库单有作废+有效两张：取有效那张
            ['source_id' => 7003, 'sku_source_id' => 1, 'ship_time' => '2026-10-22 10:00:00', 'express_no' => '@JT3'],
            // 出库时间比现有时间早：不往前改
            ['source_id' => 7004, 'sku_source_id' => 1, 'ship_time' => '2026-10-23 10:00:00', 'express_no' => 'SF4'],
            // 相差超过 7 天：视为单号复用，不改
            ['source_id' => 7005, 'sku_source_id' => 1, 'ship_time' => '2026-10-10 10:00:00', 'express_no' => 'DPK5'],
        ]);
        $queries = [];
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql, array $bindings) use (&$queries): array {
            if (str_contains($sql, 'FROM rpt.jst_salesOutboundInfo so WHERE so.io_date')) {
                $queries[] = $bindings;

                return [
                    ['l_id' => '79041235735054', 'io_date' => '2026-10-21 08:46:09', 'status' => 'Confirmed'],
                    ['l_id' => 'YT2', 'io_date' => '2026-10-21 09:00:00', 'status' => 'Confirmed'],
                    ['l_id' => 'JT3', 'io_date' => '2026-10-23 07:00:00', 'status' => 'Confirmed'],
                    ['l_id' => '@JT3', 'io_date' => '2026-10-24 07:00:00', 'status' => 'Cancelled'],
                    ['l_id' => 'SF4', 'io_date' => '2026-10-22 10:00:00', 'status' => 'Confirmed'],
                    ['l_id' => 'DPK5', 'io_date' => '2026-10-20 10:00:00', 'status' => 'Confirmed'],
                ];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $n = (new ShippingReplicaSynchronizer($factory))->fixShipTimes('2026-10-20 00:00:00', '2026-10-26 00:00:00');

        $this->assertSame(2, $n);
        $this->assertSame([['2026-10-20 00:00:00', '2026-10-29 00:00:00']], $queries);
        $this->assertSame(['2026-10-21 08:46:09'], $db->table('jst_oms_shipping')->where('source_id', 7001)->distinct()->pluck('ship_time')->all());
        $this->assertSame(2, $db->table('jst_oms_shipping')->where('source_id', 7001)->where('ship_time', '2026-10-21 08:46:09')->count());
        $this->assertSame('2026-10-21 09:00:00', $db->table('jst_oms_shipping')->where('source_id', 7002)->value('ship_time'));
        $this->assertSame('2026-10-23 07:00:00', $db->table('jst_oms_shipping')->where('source_id', 7003)->value('ship_time'));
        $this->assertSame('2026-10-23 10:00:00', $db->table('jst_oms_shipping')->where('source_id', 7004)->value('ship_time'));
        $this->assertSame('2026-10-10 10:00:00', $db->table('jst_oms_shipping')->where('source_id', 7005)->value('ship_time'));
    }

    public function test_online_and_actual_weight_follow_sales_outbound_weight_and_f_weight(): void
    {
        $db = DB::connection('business_mirror');
        $db->table('jst_oms_shipping')->insert([
            // 没称重：发货明细重量(=预估)被写进了实称、线上是商品单重×数量 -> 线上=出库单预估，实称置空
            ['source_id' => 8001, 'sku_source_id' => 1, 'ship_time' => '2026-10-10 10:00:00', 'express_no' => '302359667733',
                'online_weight' => 7.0, 'actual_weight' => 29.9],
            // 两行商品：第二行残留的单重清空，重量只记在第一行
            ['source_id' => 8002, 'sku_source_id' => 1, 'ship_time' => '2026-10-10 11:00:00', 'express_no' => 'YT7650735469496',
                'online_weight' => 47.6, 'actual_weight' => 7.08],
            ['source_id' => 8002, 'sku_source_id' => 2, 'ship_time' => '2026-10-10 11:00:00', 'express_no' => 'YT7650735469496',
                'online_weight' => 47.6, 'actual_weight' => null],
            // 称过重：线上=预估，实称=称重重量
            ['source_id' => 8003, 'sku_source_id' => 1, 'ship_time' => '2026-10-19 15:46:24', 'express_no' => '428196459565777',
                'online_weight' => 1.344, 'actual_weight' => 1.34],
            // 已经一致：不更新
            ['source_id' => 8004, 'sku_source_id' => 1, 'ship_time' => '2026-10-19 16:00:00', 'express_no' => 'OK4',
                'online_weight' => 2.195, 'actual_weight' => 2.04],
            // 出库单里没有的包裹：不动
            ['source_id' => 8005, 'sku_source_id' => 1, 'ship_time' => '2026-10-19 16:00:00', 'express_no' => 'NOIO5',
                'online_weight' => 3.0, 'actual_weight' => 3.0],
        ]);
        $upstream = $this->createMock(UpstreamReadOnlyQuery::class);
        $upstream->method('select')->willReturnCallback(function (string $sql): array {
            if (str_contains($sql, 'FROM rpt.jst_salesOutboundInfo so WHERE so.io_date')) {
                return [
                    ['l_id' => '302359667733', 'io_date' => '2026-10-10 10:00:00', 'status' => '已出库', 'weight' => 29.9, 'f_weight' => null],
                    ['l_id' => 'YT7650735469496', 'io_date' => '2026-10-10 11:00:00', 'status' => '已出库', 'weight' => 7.08, 'f_weight' => null],
                    ['l_id' => '428196459565777', 'io_date' => '2026-10-19 15:46:24', 'status' => '已出库', 'weight' => 1.3442, 'f_weight' => 2.31],
                    ['l_id' => 'OK4', 'io_date' => '2026-10-19 16:00:00', 'status' => '已出库', 'weight' => 2.195, 'f_weight' => 2.04],
                ];
            }

            return [];
        });
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->with('operations_replica')->willReturn($upstream);

        $res = (new ShippingReplicaSynchronizer($factory))->alignWithOutbound('2026-10-01 00:00:00', '2026-10-26 00:00:00');

        $this->assertSame(['ship_time' => 0, 'weight' => 3], $res);
        $w = fn (int $sid, int $sku) => $db->table('jst_oms_shipping')->where('source_id', $sid)->where('sku_source_id', $sku)->first(['online_weight', 'actual_weight']);
        $this->assertEquals(29.9, $w(8001, 1)->online_weight);
        $this->assertNull($w(8001, 1)->actual_weight);
        $this->assertEquals(7.08, $w(8002, 1)->online_weight);
        $this->assertNull($w(8002, 1)->actual_weight);
        $this->assertNull($w(8002, 2)->online_weight);
        $this->assertNull($w(8002, 2)->actual_weight);
        $this->assertEquals(1.3442, $w(8003, 1)->online_weight);
        $this->assertEquals(2.31, $w(8003, 1)->actual_weight);
        $this->assertEquals(2.195, $w(8004, 1)->online_weight);
        $this->assertEquals(2.04, $w(8004, 1)->actual_weight);
        $this->assertEquals(3.0, $w(8005, 1)->online_weight);
        $this->assertEquals(3.0, $w(8005, 1)->actual_weight);
    }

    public function test_stale_zero_quantity_order_rows_are_purged_and_package_fields_move_to_first_kept_row(): void
    {
        $db = DB::connection('business_mirror');
        foreach ([
            // 组合套装原行(数量0、金额0)排在首行，且带着包裹级运费/重量；实发只有挂钩一行
            ['source_id' => 9001, 'sku_source_id' => 1, 'ship_time' => '2026-10-07 09:54:45', 'express_no' => 'JT2213401161843', 'barcode' => '运营组合套装13462',
                'qty' => 0, 'amount' => 0, 'pkg_count' => '1/3', 'freight' => 2.5, 'weight_kg' => 0.01, 'online_weight' => 0.01],
            ['source_id' => 9001, 'sku_source_id' => 2, 'ship_time' => '2026-10-07 09:54:45', 'express_no' => 'JT2213401161843', 'barcode' => 'SJ038059N1410V01',
                'qty' => 0, 'amount' => 0, 'pkg_count' => '1/3'],
            ['source_id' => 9001, 'sku_source_id' => 3, 'ship_time' => '2026-10-07 09:54:45', 'express_no' => 'JT2213401161843', 'barcode' => 'GG003-SZ',
                'qty' => 1, 'amount' => 568, 'pkg_count' => '1/3'],
            // 被替换的原商品(数量0、金额空)在后面
            ['source_id' => 9002, 'sku_source_id' => 1, 'ship_time' => '2026-10-22 11:02:57', 'express_no' => 'JT3181752322961', 'barcode' => 'HCZDYMJ-50-KFSBD-292',
                'qty' => 1, 'amount' => 73, 'pkg_count' => '1/2', 'freight' => 4.1],
            ['source_id' => 9002, 'sku_source_id' => 2, 'ship_time' => '2026-10-22 11:02:57', 'express_no' => 'JT3181752322961', 'barcode' => 'HCZDYMJ-50-KFSBD-TJ',
                'qty' => 0, 'amount' => null, 'pkg_count' => '1/2'],
            // 只有数量0的行：不动(不把包裹删成无商品)
            ['source_id' => 9003, 'sku_source_id' => 1, 'ship_time' => '2026-10-10 10:00:00', 'express_no' => 'ZERO3', 'barcode' => 'X',
                'qty' => 0, 'amount' => 0, 'pkg_count' => '1/1'],
            // 窗口外：不动
            ['source_id' => 9004, 'sku_source_id' => 1, 'ship_time' => '2026-09-10 10:00:00', 'express_no' => 'OLD4', 'barcode' => 'A',
                'qty' => 0, 'amount' => 0, 'pkg_count' => '1/2'],
            ['source_id' => 9004, 'sku_source_id' => 2, 'ship_time' => '2026-09-10 10:00:00', 'express_no' => 'OLD4', 'barcode' => 'B',
                'qty' => 1, 'amount' => 5, 'pkg_count' => '1/2'],
        ] as $row) {
            $db->table('jst_oms_shipping')->insert($row);
        }
        $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
        $factory->method('forSource')->willReturn($this->createMock(UpstreamReadOnlyQuery::class));

        $n = (new ShippingReplicaSynchronizer($factory))->purgeZeroQtyPlaceholders('2026-10-01 00:00:00', '2026-10-26 00:00:00');

        $this->assertSame(2, $n);
        $kept = $db->table('jst_oms_shipping')->where('source_id', 9001)->get();
        $this->assertCount(1, $kept);
        $this->assertSame('GG003-SZ', $kept[0]->barcode);
        $this->assertSame('1/1', $kept[0]->pkg_count);
        $this->assertEquals(2.5, $kept[0]->freight);
        $this->assertEquals(0.01, $kept[0]->weight_kg);
        $this->assertEquals(0.01, $kept[0]->online_weight);
        $this->assertSame(['HCZDYMJ-50-KFSBD-292'], $db->table('jst_oms_shipping')->where('source_id', 9002)->pluck('barcode')->all());
        $this->assertSame('1/1', $db->table('jst_oms_shipping')->where('source_id', 9002)->value('pkg_count'));
        $this->assertEquals(4.1, $db->table('jst_oms_shipping')->where('source_id', 9002)->value('freight'));
        $this->assertSame(1, $db->table('jst_oms_shipping')->where('source_id', 9003)->count());
        $this->assertSame(2, $db->table('jst_oms_shipping')->where('source_id', 9004)->count());
    }

    private function setScanBatchLimit(ShippingReplicaSynchronizer $synchronizer, string $property, int $limit): void
    {
        $reflection = new \ReflectionProperty($synchronizer, $property);
        $reflection->setAccessible(true);
        $reflection->setValue($synchronizer, $limit);
    }
}