CaiWuXiTong/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php

<?php

namespace Tests\Unit\EcommerceFinance;

use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
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 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,
        ]);
        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->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->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' => '一号'],
            ['Id' => 2, 'updateTime' => '2026-08-14 11:00:00.0000000', 'displayName' => '二号'],
        ]];

        $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('ORDER BY [Id]', $this->query->calls[0]['sql']);
    }

    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 [updateTime], [Id]', $this->query->calls[0]['sql']);
        $this->assertCount(2, $this->query->calls[0]['bindings']);
    }

    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_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_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->assertCount(38, $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));
    }

    /** @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>>> */
    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');

        return array_shift($this->batches) ?? [];
    }
}