CaiWuXiTong/tests/Feature/Console/StoreReportMirrorSyncTest.php

<?php

namespace Tests\Feature\Console;

use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Repositories\StoreReportRepository;
use Modules\EcommerceFinance\Repositories\StoreReportUpstreamRepository;
use Modules\EcommerceFinance\Services\StoreReportMirrorSynchronizer;
use Tests\TestCase;

class StoreReportMirrorSyncTest extends TestCase
{
    protected function setUp(): void
    {
        parent::setUp();
        config([
            'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
            'upstream_mirrors.connection' => 'business_mirror',
            'upstream_mirrors.store_report.batch_size' => 100,
            'upstream_mirrors.store_report.max_batches_per_run' => 2,
            'upstream_mirrors.store_report.lifecycle_batch_size' => 100,
            'upstream_mirrors.store_report.lifecycle_max_batches_per_run' => 2,
            'upstream_mirrors.store_report.daily_batch_size' => 100,
            'upstream_mirrors.store_report.daily_max_batches_per_run' => 2,
            'upstream_mirrors.store_report.overlap_rows' => 50,
        ]);
        DB::purge('business_mirror');
        $this->createTables();
    }

    public function test_it_automatically_completes_first_sync_and_maps_income_direction(): void
    {
        $upstream = $this->createMock(StoreReportUpstreamRepository::class);
        $upstream->expects($this->once())->method('lifecycleBatch')->with(0, 100)->willReturn([
            ['source_id' => 1, 'erp_store_name' => '店铺A', 'unified_social_credit_code' => 'CREDIT-A', 'effective_at' => '2026-01-01', 'company_name' => '主体甲'],
        ]);
        $upstream->expects($this->once())->method('billBatch')->with(0, 100)->willReturn([
            $this->bill(10, '1', 120.5),
            $this->bill(11, '-1', 8.2),
        ]);

        $result = (new StoreReportMirrorSynchronizer($upstream, app('db')))->sync(100, 2);

        $this->assertSame('succeeded', $result['lifecycles']['status']);
        $this->assertSame('succeeded', $result['bills']['status']);
        $this->assertSame('succeeded', $result['daily']['status']);
        $this->assertDatabaseHas('store_report_bill_rows', [
            'source_id' => 10, 'store_name' => '店铺A', 'income' => 120.5, 'expense' => 0,
        ], 'business_mirror');
        $this->assertDatabaseHas('store_report_bill_rows', [
            'source_id' => 11, 'income' => 0, 'expense' => 8.2,
        ], 'business_mirror');
        $this->assertDatabaseHas('store_report_import_states', [
            'dataset_key' => StoreReportRepository::BILLS_DATASET,
            'status' => 'succeeded',
            'cursor_id' => 11,
            'row_count' => 2,
        ], 'business_mirror');
        $this->assertNotNull(DB::connection('business_mirror')->table('store_report_import_states')
            ->where('dataset_key', StoreReportRepository::BILLS_DATASET)->value('active_generation'));
        $this->assertDatabaseHas('store_report_daily_rows', [
            'store_name' => '店铺A', 'occurred_on' => '2026-08-01',
            'income' => 120.5, 'expense' => 8.2,
        ], 'business_mirror');
    }

    public function test_incremental_sync_reuses_active_generation_and_overlaps_recent_ids(): void
    {
        $this->activeState(StoreReportRepository::LIFECYCLES_DATASET, 'lifecycle-live', 20, 1);
        $this->activeState(StoreReportRepository::BILLS_DATASET, 'bill-live', 100, 10);
        $this->activeState(StoreReportRepository::DAILY_DATASET, 'bill-live', 100, 1);
        DB::connection('business_mirror')->table('store_report_bill_rows')->insert([
            'generation' => 'bill-live', 'source_id' => 101, 'shop_id' => 88,
            'platform' => 'TouTiaoFXGZJ', 'platform_label' => '抖音', 'store_name' => '店铺A',
            'occurred_at' => '2026-08-01 10:00:00', 'income' => 10, 'expense' => 0,
            'bill_item' => '订单收入', 'record_code' => '101', 'source_file' => 'source',
            'row_hash' => 'old', 'source_updated_at' => null, 'imported_at' => now(),
        ]);
        DB::connection('business_mirror')->table('store_report_daily_rows')->insert([
            'generation' => 'bill-live', 'store_name' => '店铺A', 'occurred_on' => '2026-08-01',
            'income' => 10, 'expense' => 0, 'updated_at' => now(),
        ]);
        $upstream = $this->createMock(StoreReportUpstreamRepository::class);
        $upstream->expects($this->once())->method('lifecycleBatch')->with(0, 100)->willReturn([]);
        $upstream->expects($this->once())->method('billBatch')->with(50, 100)->willReturn([
            $this->bill(101, '1', 33.0),
        ]);

        (new StoreReportMirrorSynchronizer($upstream, app('db')))->sync(100, 2);

        $this->assertDatabaseHas('store_report_bill_rows', [
            'generation' => 'bill-live', 'source_id' => 101, 'income' => 33,
        ], 'business_mirror');
        $this->assertDatabaseHas('store_report_import_states', [
            'dataset_key' => StoreReportRepository::BILLS_DATASET,
            'active_generation' => 'bill-live',
            'cursor_id' => 101,
            'row_count' => 1,
        ], 'business_mirror');
        $this->assertDatabaseHas('store_report_daily_rows', [
            'generation' => 'bill-live', 'occurred_on' => '2026-08-01', 'income' => 33,
        ], 'business_mirror');
    }

    private function bill(int $id, string $direction, float $amount): array
    {
        return [
            'source_id' => $id, 'shop_id' => 88, 'channel' => 'TouTiaoFXGZJ',
            'fee_time' => '2026-08-01 10:00:00', 'del_flag' => 0,
            'primary_key' => 'PK-' . $id, 'source_updated_at' => '2026-08-01 11:00:00',
            'fee_name' => '订单收入', 'inout_type' => $direction, 'amount' => $amount,
            'shop_name' => '店铺A',
        ];
    }

    private function activeState(string $dataset, string $generation, int $cursor, int $rows): void
    {
        DB::connection('business_mirror')->table('store_report_import_states')->insert([
            'dataset_key' => $dataset, 'status' => 'succeeded', 'active_generation' => $generation,
            'cursor_id' => $cursor, 'row_count' => $rows, 'initial_completed_at' => now(),
        ]);
    }

    private function createTables(): void
    {
        $schema = Schema::connection('business_mirror');
        $schema->create('store_report_import_states', function (Blueprint $table): void {
            $table->string('dataset_key')->primary();
            $table->string('status')->default('pending');
            $table->string('run_mode')->nullable();
            $table->string('active_generation')->nullable();
            $table->string('pending_generation')->nullable();
            $table->unsignedBigInteger('cursor_id')->nullable();
            $table->unsignedBigInteger('row_count')->default(0);
            $table->text('source_files_json')->nullable();
            $table->dateTime('last_attempted_at')->nullable();
            $table->dateTime('last_succeeded_at')->nullable();
            $table->dateTime('initial_completed_at')->nullable();
            $table->string('last_error_code')->nullable();
            $table->timestamps();
        });
        $schema->create('store_report_bill_rows', function (Blueprint $table): void {
            $table->increments('id');
            $table->string('generation');
            $table->unsignedBigInteger('source_id');
            $table->unsignedBigInteger('shop_id')->nullable();
            $table->string('platform');
            $table->string('platform_label');
            $table->string('store_name');
            $table->dateTime('occurred_at')->nullable();
            $table->decimal('income', 20, 4);
            $table->decimal('expense', 20, 4);
            $table->string('bill_item')->nullable();
            $table->string('record_code');
            $table->string('source_file');
            $table->string('row_hash');
            $table->dateTime('source_updated_at')->nullable();
            $table->dateTime('imported_at');
            $table->unique(['generation', 'source_id']);
        });
        $schema->create('store_report_lifecycle_rows', function (Blueprint $table): void {
            $table->increments('id');
            $table->string('generation');
            $table->unsignedBigInteger('source_id');
            $table->string('business_key');
            $table->string('erp_store_name');
            $table->string('company_name')->nullable();
            $table->string('unified_social_credit_code')->nullable();
            $table->dateTime('effective_at')->nullable();
            $table->dateTime('expires_at')->nullable();
            $table->string('source_file');
            $table->string('row_hash');
            $table->dateTime('imported_at');
            $table->unique(['generation', 'source_id']);
        });
        $schema->create('store_report_daily_rows', function (Blueprint $table): void {
            $table->increments('id');
            $table->string('generation');
            $table->string('store_name');
            $table->date('occurred_on');
            $table->decimal('income', 20, 4)->default(0);
            $table->decimal('expense', 20, 4)->default(0);
            $table->dateTime('updated_at')->nullable();
            $table->unique(['generation', 'store_name', 'occurred_on']);
        });
    }
}