CaiWuXiTong/tests/Unit/EcommerceFinance/DingPayableWorkflowMirrorSynchronizerTest.php

<?php

namespace Tests\Unit\EcommerceFinance;

use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Services\DingPayableWorkflowMirrorSynchronizer;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
use Tests\TestCase;

class DingPayableWorkflowMirrorSynchronizerTest extends TestCase
{
    private FakeDingPayableWorkflowQuery $query;

    private DingPayableWorkflowMirrorSynchronizer $synchronizer;

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

        config([
            'database.connections.business_mirror' => [
                'driver' => 'sqlite',
                'database' => ':memory:',
                'prefix' => '',
            ],
            'upstream_mirrors.connection' => 'business_mirror',
            'upstream_mirrors.table_sync.overlap_hours' => 48,
            'upstream_mirrors.ding_payable_workflows' => [
                'dataset_key' => 'operations_replica.rpt.ding_payable_workflows',
                'source_key' => 'operations_replica',
                'process_code' => 'PROC-TEST',
                'target_table' => 'ding_payable_workflow_mirror',
                'batch_size' => 100,
                'max_batches_per_run' => 2,
            ],
        ]);
        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();
        });

        $this->query = new FakeDingPayableWorkflowQuery();
        $factory = new class($this->query) extends UpstreamReadOnlyQueryFactory {
            public function __construct(private readonly UpstreamReadOnlyQuery $query) {}

            public function forSource(string $source): UpstreamReadOnlyQuery
            {
                return $this->query;
            }
        };
        $this->synchronizer = new DingPayableWorkflowMirrorSynchronizer($factory);
    }

    public function test_full_sync_resumes_with_update_time_and_process_id_tie_breaker(): void
    {
        $sameUpdateAt = '2026-08-29 10:00:00.1234567';
        $this->query->workflowBatches = [[
            $this->workflow('A-process', $sameUpdateAt),
            $this->workflow('B-process', $sameUpdateAt),
        ]];

        $result = $this->synchronizer->sync(true, 2, 1);

        $this->assertSame('running', $result['status']);
        $this->assertStringContainsString('ORDER BY COALESCE([updateAt], [createAt]), [processInstanceId]', $this->query->calls[0]['sql']);
        $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
        $this->assertSame([
            'updateAt' => $sameUpdateAt,
            'processInstanceId' => 'B-process',
        ], json_decode((string) $state->cursor, true));

        $this->query->calls = [];
        $this->query->workflowBatches = [[
            $this->workflow('C-process', $sameUpdateAt),
        ]];

        $result = $this->synchronizer->sync(true, 2, 0);

        $this->assertSame('succeeded', $result['status']);
        $this->assertStringContainsString(
            'COALESCE([updateAt], [createAt]) > ? OR (COALESCE([updateAt], [createAt]) = ? AND [processInstanceId] > ?',
            $this->query->calls[0]['sql'],
        );
        $this->assertSame([$sameUpdateAt, $sameUpdateAt, 'B-process'], array_slice($this->query->calls[0]['bindings'], -3));
        $this->assertSame(3, DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->count());
    }

    public function test_incremental_sync_uses_overlap_window_and_keeps_fixed_process_code(): void
    {
        $this->query->workflowBatches = [[
            $this->workflow('existing-process', now()->subHour()->format('Y-m-d H:i:s.u')),
        ]];
        $this->synchronizer->sync(true, 10, 0);

        $this->query->calls = [];
        $this->query->workflowBatches = [[
            $this->workflow('new-process', now()->format('Y-m-d H:i:s.u'), 'RUNNING'),
        ]];

        $result = $this->synchronizer->sync(false, 10, 0);

        $this->assertSame('succeeded', $result['status']);
        $sql = $this->query->calls[0]['sql'];
        $this->assertStringContainsString('[processCode] = ?', $sql);
        $this->assertStringContainsString('COALESCE([updateAt], [createAt]) >= ? AND COALESCE([updateAt], [createAt]) < ?', $sql);
        $this->assertStringContainsString('ORDER BY COALESCE([updateAt], [createAt]), [processInstanceId]', $sql);
        $this->assertCount(3, $this->query->calls[0]['bindings']);
        $this->assertSame('PROC-TEST', $this->query->calls[0]['bindings'][0]);
        $this->assertSame('RUNNING', json_decode(
            (string) DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->where('process_instance_id', 'new-process')->value('workflow_payload'),
            true,
        )['status']);

        $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
        $this->assertArrayHasKey('updateAt', json_decode((string) $state->high_watermark, true));
    }

    public function test_legacy_process_id_only_watermark_does_not_skip_new_processes(): void
    {
        $this->query->workflowBatches = [[
            $this->workflow('existing-process', now()->subHour()->format('Y-m-d H:i:s.u')),
        ]];
        $this->synchronizer->sync(true, 10, 0);

        DB::connection('business_mirror')->table('business_mirror_table_states')->update([
            'high_watermark' => json_encode(['processInstanceId' => 'z-last-process']),
        ]);
        $this->query->calls = [];
        $this->query->workflowBatches = [[
            $this->workflow('a-new-process', now()->format('Y-m-d H:i:s.u')),
        ]];

        $this->synchronizer->sync(false, 10, 0);

        $this->assertStringNotContainsString('AND [processInstanceId] > ?', $this->query->calls[0]['sql']);
        $this->assertSame(2, DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->count());
    }

    /** @return array<string,string> */
    private function workflow(string $id, string $updateAt, string $status = 'COMPLETED'): array
    {
        return [
            'processInstanceId' => $id,
            'processCode' => 'PROC-TEST',
            'updateAt' => $updateAt,
            'createAt' => $updateAt,
            'status' => $status,
            'originatorUserId' => '',
        ];
    }
}

class FakeDingPayableWorkflowQuery extends UpstreamReadOnlyQuery
{
    /** @var list<list<array<string,mixed>>> */
    public array $workflowBatches = [];

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

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

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

        if (str_contains($sql, 'FROM [rpt].[ding_Workflow]')) {
            return array_shift($this->workflowBatches) ?? [];
        }

        return [];
    }
}