<?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'),
        );

        $this->assertSame(2, $result['packages_retained_for_unmatched_amount']);
        $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);
        $this->assertEquals(20.75, $rows[1]->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));
    }

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