<?php

namespace Tests\Unit\EcommerceFinance;

use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
use Modules\EcommerceFinance\Services\BusinessMirrorSchemaManager;
use Modules\EcommerceFinance\Services\BusinessMirrorSynchronizer;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
use PHPUnit\Framework\Attributes\DataProvider;
use Tests\TestCase;

class BusinessMirrorSynchronizerTest extends TestCase
{
    private FakeBusinessMirrorQuery $query;
    private BusinessMirrorSynchronizer $synchronizer;

    protected function setUp(): void
    {
        parent::setUp();

        config([
            'database.connections.business_mirror' => [
                'driver' => 'sqlite',
                'database' => ':memory:',
                'prefix' => '',
            ],
            'upstream_mirrors.connection' => 'business_mirror',
            'upstream_mirrors.datasets' => [
                'test.dbo.records' => [
                    'source_key' => 'test',
                    'schema' => 'dbo',
                    'object' => 'records',
                    'target' => 'test_records',
                    'keys' => ['Id'],
                    'watermark' => 'updateTime',
                ],
            ],
            'upstream_mirrors.table_sync.overlap_hours' => 24,
            'upstreams.sources.test' => [
                'label' => 'Test source',
                'config' => 'test',
            ],
        ]);
        DB::purge('business_mirror');

        Schema::connection('business_mirror')->create('business_mirror_table_states', function (Blueprint $table): void {
            $table->string('dataset_key')->primary();
            $table->string('source_key');
            $table->string('source_object');
            $table->string('target_table');
            $table->string('status')->default('pending');
            $table->string('run_mode')->nullable();
            $table->string('generation')->nullable();
            $table->json('cursor')->nullable();
            $table->json('high_watermark')->nullable();
            $table->dateTime('window_started_at')->nullable();
            $table->dateTime('window_ends_at')->nullable();
            $table->dateTime('initial_completed_at')->nullable();
            $table->dateTime('last_full_succeeded_at')->nullable();
            $table->dateTime('last_attempted_at')->nullable();
            $table->dateTime('last_succeeded_at')->nullable();
            $table->unsignedBigInteger('last_upserted_rows')->default(0);
            $table->unsignedBigInteger('mirror_row_count')->default(0);
            $table->boolean('mirror_row_count_is_exact')->default(false);
            $table->string('last_error_code')->nullable();
            $table->timestamps();
        });

        DB::table('upstream_schema_columns')->insert([
            $this->column('Id', 1, 'int', null),
            $this->column('updateTime', 2, 'datetime2', null),
            $this->column('displayName', 3, 'nvarchar', 120),
            $this->column('businessDate', 4, 'date', null),
        ]);

        $this->query = new FakeBusinessMirrorQuery();
        $factory = new class($this->query) extends UpstreamReadOnlyQueryFactory {
            public function __construct(private readonly UpstreamReadOnlyQuery $query) {}
            public function forSource(string $source): UpstreamReadOnlyQuery { return $this->query; }
        };
        $registry = new BusinessMirrorDatasetRegistry();
        $this->synchronizer = new BusinessMirrorSynchronizer(
            $registry,
            new BusinessMirrorSchemaManager(),
            new GenericBusinessMirrorRepository(app('db')),
            $factory,
        );
    }

    public function test_full_sync_materializes_complete_table_and_marks_state_ready(): void
    {
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00.1234567', 'displayName' => '一号', 'businessDate' => '2026-08-14'],
            ['Id' => 2, 'updateTime' => '2026-08-14 11:00:00.0000000', 'displayName' => '二号', 'businessDate' => '2026-08-15'],
        ]];

        $result = $this->synchronizer->sync('test.dbo.records', true, 10, 1);

        $this->assertSame('succeeded', $result['status']);
        $this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count());
        $this->assertSame(
            '2026-08-14 10:00:00.123456',
            DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('updateTime'),
        );
        $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
        $this->assertSame('succeeded', $state->status);
        $this->assertNotNull($state->initial_completed_at);
        $this->assertSame(2, (int) $state->mirror_row_count);
        $this->assertStringContainsString('SELECT TOP (10)', $this->query->calls[0]['sql']);
        $this->assertStringContainsString('CONVERT(varchar(27), [updateTime], 126) AS [updateTime]', $this->query->calls[0]['sql']);
        $this->assertStringContainsString('CONVERT(varchar(10), [businessDate], 23) AS [businessDate]', $this->query->calls[0]['sql']);
        $this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[0]['sql']);
    }

    public function test_excluded_year_keeps_january_2026_and_unknown_dates(): void
    {
        $definitions = config('upstream_mirrors.datasets');
        $definitions['test.dbo.records']['excluded_date_range'] = [
            'column' => 'businessDate', 'from' => '2025-01-01', 'to' => '2026-01-01',
        ];
        config(['upstream_mirrors.datasets' => $definitions]);
        $source = new \PDO('sqlite::memory:');
        $source->exec("CREATE TABLE records (Id INTEGER, businessDate TEXT);
            INSERT INTO records VALUES (1, '2024-12-31'), (2, '2025-01-01'),
                (3, '2025-12-31 23:59:59'), (4, '2026-01-01'), (5, NULL)");
        [$sql, $bindings] = (new \ReflectionMethod($this->synchronizer, 'excludedDateRange'))->invoke(
            $this->synchronizer, (new BusinessMirrorDatasetRegistry())->get('test.dbo.records'),
        );
        $statement = $source->prepare('SELECT Id FROM records WHERE ' . $sql . ' ORDER BY Id');
        $statement->execute($bindings);
        $this->assertSame([1, 4, 5], $statement->fetchAll(\PDO::FETCH_COLUMN));
    }

    public function test_excluded_year_details_keep_retained_shared_and_orphan_parents(): void
    {
        $definitions = config('upstream_mirrors.datasets');
        $definitions['test.dbo.parents'] = array_replace($definitions['test.dbo.records'], [
            'object' => 'parents', 'target' => 'test_parents',
        ]);
        $definitions['test.dbo.records']['excluded_date_range'] = [
            'column' => 'businessDate', 'from' => '2025-01-01', 'to' => '2026-01-01',
            'parent_dataset' => 'test.dbo.parents', 'key' => 'parent_id',
        ];
        config(['upstream_mirrors.datasets' => $definitions]);
        $source = new \PDO('sqlite::memory:');
        $source->exec("ATTACH DATABASE ':memory:' AS dbo;
            CREATE TABLE dbo.parents (parent_id INTEGER, businessDate TEXT);
            CREATE TABLE records (Id INTEGER, parent_id INTEGER);
            INSERT INTO dbo.parents VALUES (1, '2025-01-01'), (2, '2026-01-01'),
                (3, '2025-12-31'), (3, '2026-01-01'), (4, NULL), (5, '2024-12-31');
            INSERT INTO records VALUES (1, 1), (2, 2), (3, 3), (4, 4), (5, 5), (6, 99)");
        [$sql, $bindings] = (new \ReflectionMethod($this->synchronizer, 'excludedDateRange'))->invoke(
            $this->synchronizer, (new BusinessMirrorDatasetRegistry())->get('test.dbo.records'),
        );
        $statement = $source->prepare('SELECT Id FROM records WHERE ' . $sql . ' ORDER BY Id');
        $statement->execute($bindings);
        $this->assertSame([2, 3, 4, 5, 6], $statement->fetchAll(\PDO::FETCH_COLUMN));
    }

    public function test_minimum_datetime_filter_excludes_old_and_null_source_rows(): void
    {
        $definitions = config('upstream_mirrors.datasets');
        $definitions['test.dbo.records']['minimum_datetime'] = [
            'column' => 'businessDate',
            'value' => '2026-01-01 00:00:00',
        ];
        config(['upstream_mirrors.datasets' => $definitions]);
        $source = new \PDO('sqlite::memory:');
        $source->exec("CREATE TABLE records (Id INTEGER, businessDate TEXT);
            INSERT INTO records VALUES (1, '2025-12-31T23:59:59.000'),
                (2, '2026-01-01T00:00:00.000'), (3, '2026-06-01T12:00:00.000'), (4, NULL)");
        [$sql, $bindings] = (new \ReflectionMethod($this->synchronizer, 'minimumDatetimeFilter'))->invoke(
            $this->synchronizer,
            (new BusinessMirrorDatasetRegistry())->get('test.dbo.records'),
            [
                ['column_name' => 'Id'],
                ['column_name' => 'businessDate'],
            ],
        );
        $statement = $source->prepare('SELECT Id FROM records WHERE ' . $sql . ' ORDER BY Id');
        $statement->execute($bindings);
        $this->assertSame([2, 3], $statement->fetchAll(\PDO::FETCH_COLUMN));
        $this->assertSame(['2026-01-01T00:00:00.000'], $bindings);
    }

    public function test_parent_minimum_datetime_filter_excludes_old_null_and_orphan_details(): void
    {
        $definitions = config('upstream_mirrors.datasets');
        $definitions['test.dbo.parents'] = array_replace($definitions['test.dbo.records'], [
            'object' => 'parents',
            'target' => 'test_parents',
        ]);
        $definitions['test.dbo.children'] = array_replace($definitions['test.dbo.records'], [
            'object' => 'children',
            'target' => 'test_children',
            'parent_minimum_datetime' => [
                'column' => 'ShipTIme',
                'value' => '2026-01-01 00:00:00',
                'parent_dataset' => 'test.dbo.parents',
                'child_key' => 'guid',
                'parent_key' => 'SourceGuid',
            ],
        ]);
        config(['upstream_mirrors.datasets' => $definitions]);
        $source = new \PDO('sqlite::memory:');
        $source->exec("ATTACH DATABASE ':memory:' AS dbo;
            CREATE TABLE dbo.parents (SourceGuid TEXT, ShipTIme TEXT);
            CREATE TABLE children (Id INTEGER, guid TEXT);
            INSERT INTO dbo.parents VALUES ('A', '2025-12-31T23:59:59.000'),
                ('B', '2026-01-01T00:00:00.000'), ('C', NULL);
            INSERT INTO children VALUES (1, 'A'), (2, 'B'), (3, 'C'), (4, 'ORPHAN')");
        [$sql, $bindings] = (new \ReflectionMethod($this->synchronizer, 'parentMinimumDatetimeFilter'))->invoke(
            $this->synchronizer,
            (new BusinessMirrorDatasetRegistry())->get('test.dbo.children'),
        );
        $statement = $source->prepare('SELECT Id FROM children WHERE ' . $sql . ' ORDER BY Id');
        $statement->execute($bindings);
        $this->assertSame([2], $statement->fetchAll(\PDO::FETCH_COLUMN));
    }

    public function test_incremental_sync_uses_overlap_window_and_updates_existing_key(): void
    {
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '旧值'],
        ]];
        $this->synchronizer->sync('test.dbo.records', true, 10, 1);

        $this->query->calls = [];
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-14 12:00:00', 'displayName' => '新值'],
            ['Id' => 2, 'updateTime' => '2026-08-14 12:30:00', 'displayName' => '新增'],
        ]];

        $result = $this->synchronizer->sync('test.dbo.records', false, 10, 1);

        $this->assertSame('succeeded', $result['status']);
        $this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count());
        $this->assertSame('新值', DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('displayName'));
        $this->assertStringContainsString('[updateTime] >= ? AND [updateTime] < ?', $this->query->calls[0]['sql']);
        $this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[0]['sql']);
        $this->assertStringNotContainsString('ORDER BY [updateTime]', $this->query->calls[0]['sql']);
        $this->assertCount(2, $this->query->calls[0]['bindings']);
        foreach ($this->query->calls[0]['bindings'] as $binding) {
            $this->assertMatchesRegularExpression('/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}$/', $binding);
            $this->assertStringNotContainsString(' ', $binding);
        }
    }

    public function test_key_first_incremental_read_preserves_filters_projection_and_resume_keys(): void
    {
        $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '旧值']]];
        $this->synchronizer->sync('test.dbo.records', true, 10, 1);
        config(['upstream_mirrors.table_sync.key_first_datasets' => ['test.dbo.records']]);
        $this->query->calls = [];
        $this->query->batches = [
            [['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '更新']],
            [['Id' => 2, 'updateTime' => '2026-08-15 10:01:00', 'displayName' => '新增']],
            [],
        ];

        $this->assertSame('running', $this->synchronizer->sync('test.dbo.records', false, 1, 1)['status']);
        $this->assertSame('succeeded', $this->synchronizer->sync('test.dbo.records', false, 1, 3)['status']);
        $sql = $this->query->calls[0]['sql'];
        $this->assertStringContainsString('FROM (SELECT TOP (1) [Id] FROM [dbo].[records] WHERE ([updateTime] >= ? AND [updateTime] < ?) ORDER BY [Id]) AS [mirror_keys]', $sql);
        $this->assertStringContainsString('INNER LOOP JOIN [dbo].[records] AS [mirror_rows] ON [mirror_rows].[Id] = [mirror_keys].[Id]', $sql);
        $this->assertStringContainsString('CONVERT(varchar(27), [mirror_rows].[updateTime], 126) AS [updateTime]', $sql);
        $this->assertStringContainsString('ORDER BY [mirror_keys].[Id]', $sql);
        $this->assertCount(2, $this->query->calls[0]['bindings']);
        $this->assertSame(1, $this->query->calls[1]['bindings'][2]);
        UpstreamReadOnlyQuery::assertReadOnlySql($sql);
        $this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count());
        $this->assertSame('更新', DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('displayName'));

        $this->query->calls = [];
        $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '更新']]];
        $this->synchronizer->sync('test.dbo.records', true, 10, 1);
        $this->assertStringNotContainsString('LOOP JOIN', $this->query->calls[0]['sql']);
    }

    public function test_fixed_filters_are_validated_and_bound_to_every_source_batch(): void
    {
        DB::table('upstream_schema_columns')->insert($this->column('channel', 5, 'nvarchar', 50));
        $datasets = (array) config('upstream_mirrors.datasets', []);
        $datasets['test.dbo.records.pinduoduo'] = [
            'source_key' => 'test',
            'schema' => 'dbo',
            'object' => 'records',
            'target' => 'test_pinduoduo_records',
            'keys' => ['Id'],
            'watermark' => 'updateTime',
            'fixed_filters' => ['channel' => 'PinduoduoImport'],
        ];
        config(['upstream_mirrors.datasets' => $datasets]);
        $this->query->batches = [[[
            'Id' => 1,
            'updateTime' => '2026-08-14 10:00:00',
            'displayName' => '拼多多记录',
            'businessDate' => '2026-08-14',
            'channel' => 'PinduoduoImport',
        ]]];

        $this->synchronizer->sync('test.dbo.records.pinduoduo', true, 10, 1);

        $this->assertStringContainsString('WHERE ([channel] = ?)', $this->query->calls[0]['sql']);
        $this->assertSame(['PinduoduoImport'], $this->query->calls[0]['bindings']);
        $this->assertSame(
            ['channel' => 'PinduoduoImport'],
            (new BusinessMirrorDatasetRegistry())->fixedFilters('test.dbo.records.pinduoduo'),
        );
    }

    public function test_fixed_filter_lists_are_validated_and_bound_to_every_source_batch(): void
    {
        DB::table('upstream_schema_columns')->insert($this->column('channel', 5, 'nvarchar', 50));
        $datasets = (array) config('upstream_mirrors.datasets', []);
        $datasets['test.dbo.records.dewu'] = [
            'source_key' => 'test',
            'schema' => 'dbo',
            'object' => 'records',
            'target' => 'test_dewu_records',
            'keys' => ['Id'],
            'watermark' => 'updateTime',
            'fixed_filters' => ['channel' => ['Dewu', 'DewuTK', 'DewuKJ']],
        ];
        config(['upstream_mirrors.datasets' => $datasets]);
        $this->query->batches = [[[
            'Id' => 1,
            'updateTime' => '2026-08-14 10:00:00',
            'displayName' => '得物退款记录',
            'businessDate' => '2026-08-14',
            'channel' => 'DewuTK',
        ]]];

        $this->synchronizer->sync('test.dbo.records.dewu', true, 10, 1);

        $this->assertStringContainsString('WHERE ([channel] IN (?, ?, ?))', $this->query->calls[0]['sql']);
        $this->assertSame(['Dewu', 'DewuTK', 'DewuKJ'], $this->query->calls[0]['bindings']);
        $this->assertSame(
            ['channel' => ['Dewu', 'DewuTK', 'DewuKJ']],
            (new BusinessMirrorDatasetRegistry())->fixedFilters('test.dbo.records.dewu'),
        );
    }

    #[DataProvider('invalidFixedFilterLists')]
    public function test_registry_rejects_invalid_fixed_filter_lists(mixed $value): void
    {
        $datasets = (array) config('upstream_mirrors.datasets', []);
        $datasets['test.dbo.invalid_filter'] = [
            'source_key' => 'test',
            'schema' => 'dbo',
            'object' => 'records',
            'target' => 'test_invalid_filter',
            'keys' => ['Id'],
            'fixed_filters' => ['channel' => $value],
        ];
        config(['upstream_mirrors.datasets' => $datasets]);

        $this->expectException(\InvalidArgumentException::class);
        (new BusinessMirrorDatasetRegistry())->get('test.dbo.invalid_filter');
    }

    public static function invalidFixedFilterLists(): array
    {
        return [
            'empty list' => [[]],
            'associative array' => [['main' => 'Dewu']],
            'nested list' => [[['Dewu']]],
            'non-scalar item' => [[null]],
        ];
    }

    public function test_incremental_keyset_keeps_watermark_out_of_pagination_cursor(): void
    {
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '初始'],
        ]];
        $this->synchronizer->sync('test.dbo.records', true, 10, 1);

        $this->query->calls = [];
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-14 12:00:00', 'displayName' => '更新一'],
            ['Id' => 2, 'updateTime' => '2026-08-14 12:30:00', 'displayName' => '更新二'],
        ], []];

        $result = $this->synchronizer->sync('test.dbo.records', false, 2, 0);

        $this->assertSame('succeeded', $result['status']);
        $this->assertCount(2, $this->query->calls);
        $this->assertStringContainsString('[updateTime] >= ? AND [updateTime] < ?', $this->query->calls[1]['sql']);
        $this->assertStringContainsString('[Id] > ?', $this->query->calls[1]['sql']);
        $this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[1]['sql']);
        $this->assertStringNotContainsString('[updateTime] > ?', $this->query->calls[1]['sql']);
        $this->assertCount(3, $this->query->calls[1]['bindings']);
        $this->assertSame(2, $this->query->calls[1]['bindings'][2]);
    }

    public function test_incremental_sync_refuses_to_guess_before_first_full_sync(): void
    {
        $result = $this->synchronizer->sync('test.dbo.records', false, 10, 1);

        $this->assertSame('requires_full_sync', $result['status']);
        $this->assertSame([], $this->query->calls);
    }

    public function test_supervised_full_sync_caps_explicit_batch_size_at_five_thousand(): void
    {
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '一号', 'businessDate' => '2026-08-14'],
        ]];

        $this->synchronizer->sync('test.dbo.records', true, 50000, 1);

        $this->assertStringContainsString('SELECT TOP (5000)', $this->query->calls[0]['sql']);
    }

    public function test_full_sync_sanitizes_invalid_utf8_text_without_changing_valid_text(): void
    {
        $this->query->batches = [[
            [
                'Id' => 1,
                'updateTime' => '2026-08-14 10:00:00',
                'displayName' => hex2bin('e80044495900'),
            ],
            [
                'Id' => 2,
                'updateTime' => '2026-08-14 11:00:00',
                'displayName' => '合法名称',
            ],
        ]];

        $this->synchronizer->sync('test.dbo.records', true, 10, 1);

        $rows = DB::connection('business_mirror')->table('test_records')->orderBy('Id')->pluck('displayName')->all();
        $this->assertStringEndsWith('DIY', $rows[0]);
        $this->assertNotSame(hex2bin('e80044495900'), $rows[0]);
        $this->assertSame('合法名称', $rows[1]);
        $this->assertTrue(mb_check_encoding($rows[0], 'UTF-8'));
        $this->assertStringNotContainsString("\0", $rows[0]);
    }

    public function test_wide_source_schema_stores_only_short_strings_inline(): void
    {
        $datasets = (array) config('upstream_mirrors.datasets', []);
        $datasets['test.dbo.wide_records'] = [
            'source_key' => 'test',
            'schema' => 'dbo',
            'object' => 'wide_records',
            'target' => 'test_wide_records',
            'keys' => ['Id'],
            'watermark' => 'updateTime',
        ];
        config(['upstream_mirrors.datasets' => $datasets]);

        $columns = [
            $this->column('Id', 1, 'int', null),
            $this->column('updateTime', 2, 'datetime2', null),
            $this->column('shortCode', 3, 'nvarchar', 32),
            $this->column('longLabel', 4, 'nvarchar', 120),
        ];
        DB::table('upstream_schema_columns')->insert(array_map(
            fn (array $column): array => [...$column, 'object_name' => 'wide_records'],
            $columns,
        ));

        $dataset = (new BusinessMirrorDatasetRegistry())->get('test.dbo.wide_records');
        (new BusinessMirrorSchemaManager())->ensure($dataset);

        $types = DB::connection('business_mirror')->select('PRAGMA table_info(test_wide_records)');
        $types = collect($types)->pluck('type', 'name')->map(fn (string $type): string => strtolower($type));

        $this->assertSame('varchar', $types['shortCode']);
        $this->assertSame('text', $types['longLabel']);
    }

    public function test_small_payment_work_order_dataset_accepts_only_supported_indexes(): void
    {
        $datasets = (array) config('upstream_mirrors.datasets', []);
        $datasets['jiansun.dbo.jst_workOrderInfo'] = [
            'source_key' => 'jiansun',
            'schema' => 'dbo',
            'object' => 'jst_workOrderInfo',
            'target' => 'test_small_payment_work_orders',
            'keys' => ['Id'],
            'watermark' => 'updateTime',
            'columns' => [
                'Id', 'woNo', 'woTypeId', 'title', 'content', 'status', 'grade',
                'urging', 'createby', 'executors', 'beginTime', 'endTime',
                'completer', 'completeTime', 'updated', 'created',
                'businessFieldList', 'createTime', 'updateTime',
            ],
            'indexes' => [
                ['woTypeId', 'woNo', 'Id'],
                ['created', 'Id'],
            ],
        ];
        config(['upstream_mirrors.datasets' => $datasets]);

        $columns = [
            ['name' => 'Id', 'type' => 'int', 'length' => null],
            ['name' => 'woNo', 'type' => 'nvarchar', 'length' => 100],
            ['name' => 'woTypeId', 'type' => 'nvarchar', 'length' => 100],
            ['name' => 'urging', 'type' => 'int', 'length' => null],
            ['name' => 'created', 'type' => 'datetime2', 'length' => null],
            ['name' => 'updateTime', 'type' => 'datetime2', 'length' => null],
        ];
        $selected = $datasets['jiansun.dbo.jst_workOrderInfo']['columns'];
        foreach ($selected as $position => $name) {
            $definition = collect($columns)->firstWhere('name', $name) ?? [
                'name' => $name,
                'type' => 'nvarchar',
                'length' => 120,
            ];
            DB::table('upstream_schema_columns')->insert(array_replace($this->column(
                $definition['name'],
                $position + 1,
                $definition['type'],
                $definition['length'],
            ), [
                'source_key' => 'jiansun',
                'schema_name' => 'dbo',
                'object_name' => 'jst_workOrderInfo',
            ]));
        }

        $dataset = (new BusinessMirrorDatasetRegistry())->get('jiansun.dbo.jst_workOrderInfo');
        $this->assertSame([
            ['woTypeId', 'woNo', 'Id'],
            ['created', 'Id'],
        ], $dataset->indexDefinitions());

        (new BusinessMirrorSchemaManager())->ensure($dataset);

        $indexes = collect(Schema::connection('business_mirror')->getIndexes($dataset->targetTable()))
            ->map(static fn (array $index): array => array_map('strtolower', (array) ($index['columns'] ?? [])))
            ->all();
        $types = collect(DB::connection('business_mirror')->select('PRAGMA table_info(test_small_payment_work_orders)'))
            ->pluck('type', 'name')
            ->map(static fn (string $type): string => strtolower($type));
        $this->assertSame('varchar', $types['woNo']);
        $this->assertSame('varchar', $types['woTypeId']);
        $this->assertSame('text', $types['businessFieldList']);
        $this->assertContains(['wotypeid', 'wono', 'id'], $indexes);
        $this->assertContains(['created', 'id'], $indexes);
    }

    public function test_declared_255_character_lookup_preserves_the_value_and_builds_an_index(): void
    {
        DB::table('upstream_schema_columns')->where('source_key', 'test')->where('column_name', 'displayName')
            ->update(['max_length' => 255]);
        $definitions = config('upstream_mirrors.datasets');
        $definitions['test.dbo.records']['indexes'] = [['displayName', 'Id']];
        config(['upstream_mirrors.datasets' => $definitions]);
        $dataset = (new BusinessMirrorDatasetRegistry())->get('test.dbo.records');
        (new BusinessMirrorSchemaManager())->ensure($dataset);
        $indexes = collect(Schema::connection('business_mirror')->getIndexes($dataset->targetTable()))
            ->pluck('columns')->all();
        $this->assertContains(['displayName', 'Id'], $indexes);
        $value = str_repeat('x', 255);
        $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => $value]]];
        $this->synchronizer->sync('test.dbo.records', true);
        $this->assertSame($value, DB::connection('business_mirror')->table($dataset->targetTable())->value('displayName'));
    }

    public function test_incremental_batches_keep_exact_counts_without_counting_updates_as_inserts(): void
    {
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => 'one'],
            ['Id' => 2, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => 'two'],
        ]];
        $this->synchronizer->sync('test.dbo.records', true);
        $db = DB::connection('business_mirror');
        $state = fn () => $db->table('business_mirror_table_states')->where('dataset_key', 'test.dbo.records')->first();
        $this->assertTrue((bool) $state()->mirror_row_count_is_exact);
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => 'updated'],
            ['Id' => 3, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => 'new'],
        ]];
        $this->synchronizer->sync('test.dbo.records', false, 2, 1);
        $this->assertSame('running', $state()->status);
        $this->assertSame(3, (int) $state()->mirror_row_count);
        $this->assertTrue((bool) $state()->mirror_row_count_is_exact);
        $this->assertSame(3, $db->table('test_records')->count());
        $this->query->batches = [[
            ['Id' => 4, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => 'rollback-a'],
            ['Id' => 5, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => 'rollback-b'],
        ]];
        $db->beginTransaction();
        $this->synchronizer->sync('test.dbo.records', false, 2, 1);
        $this->assertSame(5, (int) $state()->mirror_row_count);
        $db->rollBack();
        $this->assertSame(3, (int) $state()->mirror_row_count);
        $this->assertSame(3, $db->table('test_records')->count());
        $db->enableQueryLog();
        $db->flushQueryLog();
        $this->query->batches = [[]];
        $this->synchronizer->sync('test.dbo.records');
        $this->assertSame(3, (int) $state()->mirror_row_count);
        $this->assertCount(0, array_filter($db->getQueryLog(), static fn ($q) => str_contains(strtolower($q['query']), 'count(*)') && str_contains($q['query'], 'test_records')));
        $db->disableQueryLog();
    }

    public function test_legacy_running_count_is_not_trusted_until_the_run_completes(): void
    {
        $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => 'one']]];
        $this->synchronizer->sync('test.dbo.records', true);
        $db = DB::connection('business_mirror');
        $query = $db->table('business_mirror_table_states')->where('dataset_key', 'test.dbo.records');
        $query->update(['status' => 'running', 'run_mode' => 'incremental', 'cursor' => '{"Id":1}', 'mirror_row_count' => 999, 'mirror_row_count_is_exact' => false]);
        $this->query->batches = [[['Id' => 2, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => 'two']]];
        $this->synchronizer->sync('test.dbo.records', false, 1, 1);
        $this->assertFalse((bool) $query->first()->mirror_row_count_is_exact);
        $this->query->batches = [[]];
        $this->synchronizer->sync('test.dbo.records');
        $this->assertSame(2, (int) $query->first()->mirror_row_count);
        $this->assertTrue((bool) $query->first()->mirror_row_count_is_exact);
    }

    public function test_empty_full_read_does_not_delete_previous_mirror_rows(): void
    {
        $this->query->batches = [[
            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留'],
        ]];
        $this->synchronizer->sync('test.dbo.records', true, 10, 1);

        $this->query->batches = [[]];

        try {
            $this->synchronizer->sync('test.dbo.records', true, 10, 1);
            $this->fail('空的上游响应不应覆盖已有镜像。');
        } catch (\RuntimeException $exception) {
            $this->assertSame('business_mirror_unexpected_empty_source', $exception->getMessage());
        }

        $this->assertSame(1, DB::connection('business_mirror')->table('test_records')->count());
        $this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName'));
        $this->assertSame(
            'failed',
            DB::connection('business_mirror')->table('business_mirror_table_states')->value('status'),
        );
    }

    public function test_upstream_timeout_keeps_previous_data_and_logs_only_safe_diagnostics(): void
    {
        $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留']]];
        $this->synchronizer->sync('test.dbo.records', true, 10, 1);
        $previousSuccess = DB::connection('business_mirror')->table('business_mirror_table_states')->value('last_succeeded_at');
        $exception = new \PDOException('secret DSN and financial SQL must never be logged');
        $exception->errorInfo = ['HY000', 20003, 'secret native message'];
        $this->query->batches = [$exception];
        Log::spy();

        try {
            $this->synchronizer->sync('test.dbo.records', false, 10, 1);
            $this->fail('Expected the upstream failure to propagate.');
        } catch (\PDOException $caught) {
            $this->assertSame($exception, $caught);
        }

        Log::shouldHaveReceived('warning')->once()->withArgs(function (string $message, array $context): bool {
            $this->assertSame('read_upstream', $context['stage']);
            $this->assertSame('HY000', $context['sql_state']);
            $this->assertSame(20003, $context['driver_code']);
            $this->assertStringNotContainsString('secret', json_encode($context));
            $this->assertArrayNotHasKey('message', $context);
            $this->assertArrayNotHasKey('sql', $context);
            return true;
        });
        $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
        $this->assertSame('failed', $state->status);
        $this->assertSame($previousSuccess, $state->last_succeeded_at);
        $this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName'));
    }

    public function test_local_pdo_failure_is_identified_as_a_mirror_write(): void
    {
        $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留']]];
        $this->synchronizer->sync('test.dbo.records', true, 10, 1);
        $exception = new \PDOException('private local database error');
        $exception->errorInfo = ['40001', 1213, 'private message'];
        $mirror = \Mockery::mock(GenericBusinessMirrorRepository::class, [app('db')])->makePartial();
        $mirror->shouldReceive('storeBatch')->once()->andThrow($exception);
        $factory = new class($this->query) extends UpstreamReadOnlyQueryFactory {
            public function __construct(private readonly UpstreamReadOnlyQuery $query) {}
            public function forSource(string $source): UpstreamReadOnlyQuery { return $this->query; }
        };
        $synchronizer = new BusinessMirrorSynchronizer(new BusinessMirrorDatasetRegistry(), new BusinessMirrorSchemaManager(), $mirror, $factory);
        $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '未写入']]];
        Log::spy();

        try {
            $synchronizer->sync('test.dbo.records', false, 10, 1);
            $this->fail('Expected the local failure to propagate.');
        } catch (\PDOException $caught) {
            $this->assertSame($exception, $caught);
        }

        Log::shouldHaveReceived('warning')->once()->withArgs(function (string $message, array $context): bool {
            $this->assertSame('write_mirror', $context['stage']);
            $this->assertSame('40001', $context['sql_state']);
            $this->assertSame(1213, $context['driver_code']);
            $this->assertStringNotContainsString('private', json_encode($context));
            return true;
        });
        $this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName'));
    }

    public function test_project_dataset_registry_has_unique_safe_targets(): void
    {
        $configured = require base_path('config/upstream_mirrors.php');
        config(['upstream_mirrors.datasets' => $configured['datasets']]);
        $datasets = (new BusinessMirrorDatasetRegistry())->all();
        $targets = array_map(fn ($dataset): string => $dataset->targetTable(), $datasets);

        $this->assertGreaterThanOrEqual(38, count($datasets));
        $this->assertCount(count($targets), array_unique($targets));
        $this->assertContains('operations_replica', array_map(fn ($dataset): string => $dataset->sourceKey(), $datasets));
        $this->assertContains('jiansun', array_map(fn ($dataset): string => $dataset->sourceKey(), $datasets));
        $this->assertNotContains('Bulk_Price_Limit', array_map(fn ($dataset): string => $dataset->sourceObject(), $datasets));
        $this->assertContains('fys_Ticketing_install_view', array_map(fn ($dataset): string => $dataset->sourceObject(), $datasets));
        $this->assertContains('fys_Ticketing_install_item', array_map(fn ($dataset): string => $dataset->sourceObject(), $datasets));

        $pinduoduo = $datasets['jiansun.dbo.jst_originalBillInfo.pinduoduo_receiving'];
        $this->assertSame('jiansun', $pinduoduo->sourceKey());
        $this->assertSame('jst_originalBillInfo', $pinduoduo->sourceObject());
        $this->assertSame('js_jst_pdd_receiving_bill_rows', $pinduoduo->targetTable());
        $this->assertSame(['channel' => 'PinduoduoImport'], $pinduoduo->fixedFilters());

        $dewu = $datasets['jiansun.dbo.jst_originalBillInfo.dewu_receiving'];
        $this->assertSame('jiansun', $dewu->sourceKey());
        $this->assertSame('jst_originalBillInfo', $dewu->sourceObject());
        $this->assertSame('js_jst_dewu_receiving_bill_rows', $dewu->targetTable());
        $this->assertSame(['channel' => ['Dewu', 'DewuTK', 'DewuKJ']], $dewu->fixedFilters());
    }

    /** @return array<string,mixed> */
    private function column(string $name, int $position, string $type, ?int $length): array
    {
        $now = now();

        return [
            'source_key' => 'test',
            'schema_name' => 'dbo',
            'object_name' => 'records',
            'object_type' => 'table',
            'column_name' => $name,
            'ordinal_position' => $position,
            'data_type' => $type,
            'max_length' => $length,
            'numeric_precision' => null,
            'numeric_scale' => null,
            'is_nullable' => true,
            'synchronized_at' => $now,
            'created_at' => $now,
            'updated_at' => $now,
        ];
    }
}

class FakeBusinessMirrorQuery extends UpstreamReadOnlyQuery
{
    /** @var list<list<array<string,mixed>>|\Throwable> */
    public array $batches = [];

    /** @var list<array{sql:string,bindings:array}> */
    public array $calls = [];

    public function __construct()
    {
        parent::__construct('test');
    }

    public function select(string $sql, array $bindings = []): array
    {
        $this->calls[] = compact('sql', 'bindings');

        $batch = array_shift($this->batches) ?? [];
        if ($batch instanceof \Throwable) {
            throw $batch;
        }

        return $batch;
    }
}