modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php

@@ -23,7 +23,7 @@ class BusinessMirrorSchemaManager
                 $table->char('_mirror_key', 64)->primary();
 
                 foreach ($columns as $column) {
-                    $this->addColumn($table, $column);
+                    $this->addColumn($table, $column, $dataset);
                 }
 
                 $table->uuid('_mirror_generation')->nullable()->index();
@@ -49,9 +49,9 @@ class BusinessMirrorSchemaManager
         ));
 
         if ($missing !== []) {
-            $schema->table($dataset->targetTable(), function (Blueprint $table) use ($missing): void {
+            $schema->table($dataset->targetTable(), function (Blueprint $table) use ($missing, $dataset): void {
                 foreach ($missing as $column) {
-                    $this->addColumn($table, $column);
+                    $this->addColumn($table, $column, $dataset);
                 }
             });
         }
@@ -120,7 +120,7 @@ class BusinessMirrorSchemaManager
         return $rows;
     }
 
-    private function addColumn(Blueprint $table, array $column): void
+    private function addColumn(Blueprint $table, array $column, BusinessMirrorDataset $dataset): void
     {
         $name = (string) $column['column_name'];
         $type = strtolower((string) $column['data_type']);
@@ -144,9 +144,11 @@ class BusinessMirrorSchemaManager
             // MariaDB counts every inline varchar column towards the 65,535-byte
             // row limit. Wide legacy SQL Server tables can contain dozens of
             // nvarchar columns even though each individual column looks small.
-            // Keep only genuinely short values inline; TEXT is stored off-page
-            // and lets the mirror preserve the complete source schema.
-            'char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying' => $length > 0 && $length <= 64
+            // Keep only genuinely short values inline. A declared lookup index
+            // may opt a bounded legacy identifier into VARCHAR; every other
+            // wide string remains TEXT and is stored off-page.
+            'char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying' => $length > 0
+                && ($length <= 64 || ($length <= 191 && $this->isConfiguredIndexColumn($dataset, $name)))
                 ? $table->string($name, $length)
                 : $table->longText($name),
             default => $table->longText($name),
@@ -155,7 +157,7 @@ class BusinessMirrorSchemaManager
         $definition->nullable();
     }
 
-    private function isIndexable(array $columns, string $name): bool
+    private function isIndexable(array $columns, string $name, ?BusinessMirrorDataset $dataset = null): bool
     {
         foreach ($columns as $column) {
             if (strcasecmp((string) $column['column_name'], $name) !== 0) {
@@ -166,7 +168,8 @@ class BusinessMirrorSchemaManager
             $length = (int) ($column['max_length'] ?? 0);
 
             return ! in_array($type, ['text', 'ntext', 'xml', 'image', 'binary', 'varbinary', 'bytea', 'json', 'jsonb', 'array', 'user-defined'], true)
-                && (! in_array($type, ['char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying'], true) || ($length > 0 && $length <= 64));
+                && (! in_array($type, ['char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying'], true)
+                    || ($length > 0 && ($length <= 64 || ($length <= 191 && $dataset !== null && $this->isConfiguredIndexColumn($dataset, $name)))));
         }
 
         return false;
@@ -188,7 +191,7 @@ class BusinessMirrorSchemaManager
         foreach ($dataset->indexDefinitions() as $index) {
             foreach ($index as $column) {
                 if (! in_array(strtolower($column), $columnNames, true)
-                    || ! $this->isIndexable($columns, $column)) {
+                    || ! $this->isIndexable($columns, $column, $dataset)) {
                     throw new RuntimeException('business_mirror_index_column_unsupported:' . $dataset->key());
                 }
             }
@@ -197,6 +200,19 @@ class BusinessMirrorSchemaManager
         return $dataset->indexDefinitions();
     }
 
+    private function isConfiguredIndexColumn(BusinessMirrorDataset $dataset, string $column): bool
+    {
+        foreach ($dataset->indexDefinitions() as $index) {
+            foreach ($index as $indexedColumn) {
+                if (strcasecmp((string) $indexedColumn, $column) === 0) {
+                    return true;
+                }
+            }
+        }
+
+        return false;
+    }
+
     private function configuredIndexName(string $table, array $columns, int $position): string
     {
         $signature = implode('_', $columns) . '_' . $position;

tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php

@@ -246,6 +246,77 @@ class BusinessMirrorSynchronizerTest extends TestCase
         $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_empty_full_read_does_not_delete_previous_mirror_rows(): void
     {
         $this->query->batches = [[