diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php index a3fa313d..136f56b8 100644 --- a/config/upstream_mirrors.php +++ b/config/upstream_mirrors.php @@ -246,6 +246,7 @@ return [ ['Id'], ['shop_id', 'fee_time', 'Id'], ['fee_time', 'Id'], + ['channel', 'fee_time', 'Id', 'inout_type', 'amount'], ], ], diff --git a/database/migrations/2026_09_09_170000_index_original_bill_channel_queries.php b/database/migrations/2026_09_09_170000_index_original_bill_channel_queries.php new file mode 100644 index 00000000..5fd2f935 --- /dev/null +++ b/database/migrations/2026_09_09_170000_index_original_bill_channel_queries.php @@ -0,0 +1,32 @@ +hasTable(self::TABLE) || $schema->hasIndex(self::TABLE, self::INDEX)) return; + $schema->table(self::TABLE, static function (Blueprint $table): void { + $table->index(['channel', 'fee_time', 'Id', 'inout_type', 'amount'], self::INDEX); + }); + } + + public function down(): void + { + if (config('system_database.mode', 'bootstrap') === 'bootstrap') return; + $schema = Schema::connection((string) config('upstream_mirrors.connection', 'business_mirror')); + if (! $schema->hasTable(self::TABLE) || ! $schema->hasIndex(self::TABLE, self::INDEX)) return; + $schema->table(self::TABLE, static function (Blueprint $table): void { + $table->dropIndex(self::INDEX); + }); + } +}; diff --git a/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php b/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php index 291c982a..201f83f6 100644 --- a/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/StoreReceivingAlipayBillMirrorRepository.php @@ -139,9 +139,48 @@ class StoreReceivingAlipayBillMirrorRepository 'balance' => 'b.amount', ][$sort] ?? 'b.fee_time'; - return $query->select('b.*') - ->orderBy($sortColumn, $direction)->orderBy('b.Id', $direction)->cursor() - ->map(fn (object $row): array => $this->mapRowFor($row, $shopNames[(string) $row->shop_id] ?? null, $mapping)); + return $this->exportBatches($query, $sortColumn, $direction, $shopNames, $mapping); + } + + /** Read bounded pages so a full export does not sort/buffer millions of rows at once. */ + private function exportBatches(Builder $query, string $sortColumn, string $direction, array $shopNames, string $mapping): iterable + { + $operator = $direction === 'asc' ? '>' : '<'; + $sortField = substr($sortColumn, 2); + $last = null; + $maximumId = $this->billQuery()->max('b.Id'); + if ($maximumId === null) return; + + do { + $batch = (clone $query)->where('b.Id', '<=', $maximumId); + if ($last !== null) { + if ($sortColumn === 'b.Id') { + $batch->where('b.Id', $operator, $last->Id); + } else { + $value = $last->{$sortField}; + $batch->where(function (Builder $after) use ($sortColumn, $direction, $operator, $last, $value): void { + if ($value === null) { + $after->where(function (Builder $ties) use ($sortColumn, $operator, $last): void { + $ties->whereNull($sortColumn)->where('b.Id', $operator, $last->Id); + }); + if ($direction === 'asc') $after->orWhereNotNull($sortColumn); + } else { + $after->where($sortColumn, $operator, $value) + ->orWhere(function (Builder $ties) use ($sortColumn, $operator, $last, $value): void { + $ties->where($sortColumn, $value)->where('b.Id', $operator, $last->Id); + }); + if ($direction === 'desc') $after->orWhereNull($sortColumn); + } + }); + } + } + $rows = $batch->select('b.*')->orderBy($sortColumn, $direction) + ->orderBy('b.Id', $direction)->limit(1000)->get(); + foreach ($rows as $row) { + yield $this->mapRowFor($row, $shopNames[(string) $row->shop_id] ?? null, $mapping); + } + $last = $rows->last(); + } while ($rows->count() === 1000); } private function billQuery(): Builder diff --git a/tests/Unit/EcommerceFinance/StoreReceivingAlipayBillMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/StoreReceivingAlipayBillMirrorRepositoryTest.php index 576937b1..19368199 100644 --- a/tests/Unit/EcommerceFinance/StoreReceivingAlipayBillMirrorRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/StoreReceivingAlipayBillMirrorRepositoryTest.php @@ -126,6 +126,68 @@ class StoreReceivingAlipayBillMirrorRepositoryTest extends TestCase $this->assertSame(20.10, $rows[0]['expense']); } + public function test_export_batches_preserve_sort_ties_nulls_and_channel_filter(): void + { + $this->markReady(); + $db = DB::connection('business_mirror'); + $table = $db->table('ops_jst_original_bill_info'); + $fixtures = []; + for ($id = 1; $id <= 2105; $id++) { + $row = $this->row($id, 7, '2026-08-01 09:00:00', '1', 10); + $row['fee_time'] = $id <= 1005 ? null : '2026-08-01 09:00:00'; + $row['amount'] = $id <= 1005 ? null : 10; + $fixtures[] = $row; + } + foreach (array_chunk($fixtures, 100) as $chunk) $table->insert($chunk); + $table->insert($this->row(9999, 7, '2026-08-01 09:00:00', '1', 10, 'OtherChannel')); + foreach (['occurred_at' => 'fee_time', 'income' => 'amount', 'record_code' => 'Id'] as $sort => $column) { + foreach (['asc', 'desc'] as $direction) { + $expected = (clone $table)->where('channel', 'CommonBillTGCPay') + ->orderBy($column, $direction)->orderBy('Id', $direction)->pluck('Id')->map('strval')->all(); + $db->flushQueryLog(); + $db->enableQueryLog(); + $rows = iterator_to_array($this->repository->exportChannelRows([], $sort, $direction, 'CommonBillTGCPay')); + $this->assertSame($expected, array_column($rows, 'record_code')); + $reads = array_filter($db->getQueryLog(), fn ($query) => str_contains($query['query'], 'select "b".*')); + $this->assertCount(3, $reads); + foreach ($reads as $read) $this->assertStringContainsString('limit 1000', $read['query']); + $db->disableQueryLog(); + } + } + } + + public function test_export_does_not_chase_new_rows_inserted_after_it_starts(): void + { + $this->markReady(); + $table = DB::connection('business_mirror')->table('ops_jst_original_bill_info'); + for ($id = 1; $id <= 1001; $id++) { + $table->insert($this->row($id, 7, '2026-08-01 09:00:00', '1', 10)); + } + $ids = []; + foreach ($this->repository->exportRows([], 'record_code', 'asc') as $row) { + $ids[] = $row['record_code']; + if (count($ids) === 1) $table->insert($this->row(9999, 7, '2026-08-01 09:00:00', '1', 10)); + } + $this->assertCount(1001, $ids); + $this->assertNotContains('9999', $ids); + } + + public function test_channel_index_migration_and_rollback_preserve_rows(): void + { + config(['system_database.mode' => 'connected']); + $table = DB::connection('business_mirror')->table('ops_jst_original_bill_info'); + $table->insert($this->row(1, 7, '2026-08-01 09:00:00', '1', 10, str_repeat('A', 101))); + $migration = require base_path('database/migrations/2026_09_09_170000_index_original_bill_channel_queries.php'); + $before = (array) $table->first(); + $migration->up(); + $migration->up(); + $this->assertTrue(Schema::connection('business_mirror')->hasIndex('ops_jst_original_bill_info', 'idx_original_bill_channel_time')); + $this->assertEquals($before, (array) $table->first()); + $migration->down(); + $this->assertFalse(Schema::connection('business_mirror')->hasIndex('ops_jst_original_bill_info', 'idx_original_bill_channel_time')); + $this->assertEquals($before, (array) $table->first()); + } + public function test_common_filters_use_the_existing_daily_summary_acceleration(): void { $this->markReady();