diff --git a/app/Console/Commands/BusinessMirrorStatus.php b/app/Console/Commands/BusinessMirrorStatus.php new file mode 100644 index 00000000..298e02d0 --- /dev/null +++ b/app/Console/Commands/BusinessMirrorStatus.php @@ -0,0 +1,40 @@ +argument('dataset'); + $selected = $requested ? [$requested => $datasets->get($requested)] : $datasets->all(); + $rows = []; + foreach ($selected as $key => $dataset) { + $state = $mirror->state($dataset); + $rows[] = [ + 'dataset' => $key, 'enabled' => $datasets->scheduled($key), + 'initial_completed' => ($state['initial_completed_at'] ?? null) !== null, + 'status' => $state['status'] ?? ($datasets->automaticInitialSyncAllowed($key) ? 'queued' : 'not_initialized'), + 'last_attempted_at' => $state['last_attempted_at'] ?? null, + 'last_succeeded_at' => $state['last_succeeded_at'] ?? null, + 'error_code' => $state['last_error_code'] ?? null, + ]; + } + if ($this->option('json')) $this->line(json_encode($rows, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR)); + else $this->table(['数据集', '启用', '首次完成', '状态', '最近尝试', '最近完成', '错误类型'], array_map(fn ($r) => array_values($r), $rows)); + return self::SUCCESS; + } catch (Throwable) { + $this->error('无法读取镜像状态，请确认共享系统数据库和业务镜像库已接入。'); + return self::FAILURE; + } + } +} diff --git a/app/Console/Commands/InspectUpstream.php b/app/Console/Commands/InspectUpstream.php new file mode 100644 index 00000000..6613f338 --- /dev/null +++ b/app/Console/Commands/InspectUpstream.php @@ -0,0 +1,66 @@ +argument('source'); + $schema = (string) $this->argument('schema'); + $object = (string) $this->argument('object'); + try { + if ($source === '') { + $rows = []; + foreach ($sources->all() as $key => $definition) $rows[] = [$key, $definition['label'], $sources->driver($key), $sources->enabled($key) ? '启用' : '停用']; + $this->table(['内部标识', '数据库名称', '类型', '状态'], $rows); + return self::SUCCESS; + } + $sources->definition($source); + if ($schema === '' || $object === '') { + $objects = DB::table('upstream_schema_columns')->where('source_key', $source) + ->when($schema !== '', fn ($q) => $q->where('schema_name', $schema)) + ->select('schema_name', 'object_name')->distinct()->orderBy('schema_name')->orderBy('object_name')->get(); + $this->table(['Schema', '表 / 视图'], $objects->map(fn ($row) => [$row->schema_name, $row->object_name])->all()); + return self::SUCCESS; + } + $catalog = DB::table('upstream_schema_columns')->where('source_key', $source) + ->where('schema_name', $schema)->where('object_name', $object)->orderBy('ordinal_position') + ->get(['column_name', 'data_type', 'is_nullable']); + if ($catalog->isEmpty()) { + $this->error('尚无该对象的字段目录，请先核实来源并同步数据字典。'); + return self::FAILURE; + } + if (! $this->option('columns')) { + $this->table(['字段', '类型', '允许空值'], $catalog->map(fn ($c) => [$c->column_name, $c->data_type, $c->is_nullable ? '是' : '否'])->all()); + return self::SUCCESS; + } + $columns = array_values(array_unique(array_filter(array_map('trim', explode(',', (string) $this->option('columns')))))); + if ($columns === [] || array_diff($columns, $catalog->pluck('column_name')->all())) { + $this->error('样本字段必须来自该对象的数据字典。'); + return self::FAILURE; + } + $dialect = new UpstreamSqlDialect($sources->driver($source)); + $sql = $dialect->select(implode(', ', array_map($dialect->quote(...), $columns)), + 'FROM ' . $dialect->quote($schema) . '.' . $dialect->quote($object), + min(max((int) $this->option('limit'), 1), 20)); + $rows = $queries->forSource($source)->select($sql); + $this->table($columns, array_map(fn ($row) => array_values($row), $rows)); + $this->info('样本仅用于字段核实，不能作为完整业务数据。'); + return self::SUCCESS; + } catch (Throwable) { + $this->error('只读检查未完成，请检查共享连接、来源启用状态和访问权限。'); + return self::FAILURE; + } + } +} diff --git a/app/Console/Commands/RegisterBusinessMirror.php b/app/Console/Commands/RegisterBusinessMirror.php new file mode 100644 index 00000000..a6c96d17 --- /dev/null +++ b/app/Console/Commands/RegisterBusinessMirror.php @@ -0,0 +1,94 @@ +error('尚未执行镜像数据集登记表迁移。'); + return self::FAILURE; + } + $source = (string) $this->argument('source'); + $schema = (string) $this->argument('schema'); + $object = (string) $this->argument('object'); + $key = $source . '.' . $schema . '.' . $object; + try { + $sources->definition($source); + if (strlen($key) > 191 || isset(config('upstream_mirrors.datasets', [])[$key])) { + throw new \RuntimeException('数据集键过长或由现有代码维护，不能覆盖。'); + } + if ($this->option('disable-sync')) { + if ($this->option('enable-sync')) throw new \RuntimeException('不能同时启用和停用。'); + if (DB::table('business_mirror_datasets')->where('dataset_key', $key)->update(['sync_enabled' => false, 'updated_at' => now()]) !== 1) { + throw new \RuntimeException('未找到该登记数据集。'); + } + $this->info('已停止自动同步，现有镜像保留。'); + return self::SUCCESS; + } + $keys = $this->csv((string) $this->option('keys')); + $columns = $this->csv((string) $this->option('columns')); + $watermark = trim((string) $this->option('watermark')); + if ($keys === [] || $columns === [] || array_diff([...$keys, ...($watermark === '' ? [] : [$watermark])], $columns)) { + throw new \RuntimeException('请明确字段和稳定键；同步字段必须包含稳定键及增量字段。'); + } + $catalog = DB::table('upstream_schema_columns')->where('source_key', $source) + ->where('schema_name', $schema)->where('object_name', $object)->pluck('column_name')->all(); + if (array_diff($columns, $catalog)) { + throw new \RuntimeException('字段不在数据字典中，请先同步该来源字典并核实字段。'); + } + $definition = ['source_key' => $source, 'schema' => $schema, 'object' => $object, + 'target' => 'mirror_' . substr(hash('sha256', $key), 0, 24), + 'keys' => $keys, 'columns' => $columns, 'watermark' => $watermark]; + $datasets->validate($key, $definition); + $enable = (bool) $this->option('enable-sync'); + DB::transaction(function () use ($key, $source, $definition, $enable): void { + $existing = DB::table('business_mirror_datasets')->where('dataset_key', $key)->lockForUpdate()->first(); + if ($existing !== null && json_decode($existing->definition, true) != $definition) { + throw new \RuntimeException('已有数据集定义不同，不能覆盖稳定键、字段或同步游标。请先完成变更评估。'); + } + if ($existing !== null) { + if ($enable) DB::table('business_mirror_datasets')->where('dataset_key', $key)->update([ + 'sync_enabled' => true, 'initial_sync_approved' => true, 'updated_at' => now(), + ]); + return; + } + DB::table('business_mirror_datasets')->insert([ + 'dataset_key' => $key, 'source_key' => $source, 'definition' => json_encode($definition, JSON_THROW_ON_ERROR), + 'sync_enabled' => $enable, 'initial_sync_approved' => $enable, 'created_at' => now(), 'updated_at' => now(), + ]); + }); + $this->info($key . ($enable ? '：已启用，生产调度将自动执行首次同步并持续增量。' : '：已登记，自动同步未启用。')); + return self::SUCCESS; + } catch (\Illuminate\Database\QueryException) { + $this->error('登记失败，请检查系统数据库状态；原数据集保留。'); + } catch (\RuntimeException|\InvalidArgumentException $e) { + $this->error($e->getMessage()); + } catch (Throwable) { + $this->error('登记失败，未启用新同步。请检查系统数据库状态。'); + } + return self::FAILURE; + } + + private function csv(string $value): array + { + return array_values(array_unique(array_filter(array_map('trim', explode(',', $value))))); + } +} diff --git a/app/Console/Commands/SyncBusinessMirrors.php b/app/Console/Commands/SyncBusinessMirrors.php index 46772a7c..7c3bb48c 100644 --- a/app/Console/Commands/SyncBusinessMirrors.php +++ b/app/Console/Commands/SyncBusinessMirrors.php @@ -35,23 +35,27 @@ class SyncBusinessMirrors extends Command $requested = $this->argument('dataset'); $keys = $requested === null ? $this->scheduledKeys($datasets, $mirror) - : [(string) $requested]; + : [['key' => (string) $requested, 'full' => (bool) $this->option('full'), 'catchup' => false]]; $maxDatasets = max((int) ($this->option('max-datasets') - ?? config('upstream_mirrors.table_sync.max_datasets_per_run', 1)), 1); + ?? config('upstream_pipeline.sync.max_datasets', 8)), 1); $processed = 0; $failed = false; + $deadline = microtime(true) + max((int) config('upstream_pipeline.sync.cycle_seconds', 30), 1); - foreach ($keys as $key) { - if ($processed >= $maxDatasets) { + foreach ($keys as $task) { + if ($processed >= $maxDatasets || ($processed > 0 && microtime(true) >= $deadline)) { break; } + $key = $task['key']; try { $result = $synchronizer->sync( $key, - (bool) $this->option('full'), + (bool) $task['full'], $this->option('batch') === null ? null : (int) $this->option('batch'), - $this->option('max-batches') === null ? null : (int) $this->option('max-batches'), + $this->option('max-batches') === null + ? ($task['catchup'] ? max((int) config('upstream_pipeline.sync.catchup_batches', 6), 1) : null) + : (int) $this->option('max-batches'), ); } catch (Throwable $exception) { $failed = true; @@ -105,7 +109,7 @@ class SyncBusinessMirrors extends Command return $failed ? self::FAILURE : self::SUCCESS; } - /** @return list */ + /** @return list */ private function scheduledKeys( BusinessMirrorDatasetRegistry $datasets, GenericBusinessMirrorRepository $mirror, @@ -117,17 +121,20 @@ class SyncBusinessMirrors extends Command ), true); foreach ($datasets->all() as $key => $dataset) { - if (isset($dedicated[$key])) { + if (isset($dedicated[$key]) || ! $datasets->scheduled($key)) { continue; } $state = $mirror->state($dataset); - if (! is_array($state) || ($state['initial_completed_at'] ?? null) === null) { - continue; - } + $initial = ! is_array($state) || ($state['initial_completed_at'] ?? null) === null; + if ($initial && ! $datasets->automaticInitialSyncAllowed($key)) continue; + if (($state['status'] ?? '') === 'failed' && ! empty($state['last_attempted_at']) + && now()->diffInSeconds(\Illuminate\Support\Carbon::parse($state['last_attempted_at'])) < 300) continue; $scheduled[] = [ 'key' => $key, + 'full' => $initial || (bool) $this->option('full'), + 'catchup' => $initial || ($state['status'] ?? '') === 'running', 'last_attempted_at' => (string) ($state['last_attempted_at'] ?? ''), ]; } @@ -136,6 +143,15 @@ class SyncBusinessMirrors extends Command [$left['last_attempted_at'], $left['key']] <=> [$right['last_attempted_at'], $right['key']] ); - return array_column($scheduled, 'key'); + // Alternate unfinished windows and routine refreshes: catch-up progresses + // every cycle without starving small tables or monopolizing a source. + $catchup = array_values(array_filter($scheduled, fn ($task) => $task['catchup'])); + $routine = array_values(array_filter($scheduled, fn ($task) => ! $task['catchup'])); + $ordered = []; + while ($catchup !== [] || $routine !== []) { + if ($routine !== []) $ordered[] = array_shift($routine); + if ($catchup !== []) $ordered[] = array_shift($catchup); + } + return $ordered; } } diff --git a/app/Console/Commands/UpstreamSyncSchemaCatalog.php b/app/Console/Commands/UpstreamSyncSchemaCatalog.php index 25318160..cc78fbc1 100644 --- a/app/Console/Commands/UpstreamSyncSchemaCatalog.php +++ b/app/Console/Commands/UpstreamSyncSchemaCatalog.php @@ -32,7 +32,7 @@ class UpstreamSyncSchemaCatalog extends Command $requested = $this->argument('source'); $sourceKeys = $requested ? [(string) $requested] - : array_keys((array) config('upstreams.sources', [])); + : array_keys($sources->all()); $cache = $cacheManager->store((string) config('upstreams.snapshots.cache_store', 'upstream_shared')); $failed = false; diff --git a/app/Http/Controllers/SystemSettings/Databases.php b/app/Http/Controllers/SystemSettings/Databases.php index 581ba4b8..abaa3df5 100644 --- a/app/Http/Controllers/SystemSettings/Databases.php +++ b/app/Http/Controllers/SystemSettings/Databases.php @@ -211,7 +211,7 @@ class Databases extends Controller $data = $this->validateConnection($request); $connection = DatabaseConnection::create($this->attributes($data, null, $request)); - $this->resetSource($connection->source_key, $storedConfiguration); + $this->resetSource($connection->sourceIdentifier(), $storedConfiguration); flash('数据库配置已新增。')->success(); @@ -234,12 +234,12 @@ class Databases extends Controller { abort_unless($storedConfiguration->editable() && $storedConfiguration->enabled(), 403); $connection = $this->connection($connection); - $oldSource = $connection->source_key; + $oldSource = $connection->sourceIdentifier(); $data = $this->validateConnection($request, $connection); $connection->update($this->attributes($data, $connection, $request)); $this->resetSource($oldSource, $storedConfiguration); - $this->resetSource($connection->source_key, $storedConfiguration); + $this->resetSource($connection->sourceIdentifier(), $storedConfiguration); flash('数据库配置已修改。')->success(); @@ -250,7 +250,7 @@ class Databases extends Controller { abort_unless($storedConfiguration->editable() && $storedConfiguration->enabled(), 403); $connection = $this->connection($connection); - $source = $connection->source_key; + $source = $connection->sourceIdentifier(); $connection->delete(); $this->resetSource($source, $storedConfiguration); @@ -272,13 +272,14 @@ class Databases extends Controller ]); } catch (ServiceUnavailableHttpException $exception) { throw ValidationException::withMessages([ - 'connection' => $connection->driver === 'pgsql' + 'connection' => $connection->driver !== 'sqlsrv' ? $exception->getMessage() : 'SQL Server 连接失败。请检查地址、端口、数据库、账号、网络白名单和 ODBC 驱动。', ]); } - flash($connection->name . ($connection->driver === 'pgsql' ? ' PostgreSQL SSL 连接成功，仅执行了 SELECT 1 只读测试。' : '连接成功，仅执行了 SELECT 1 只读测试。'))->success(); + $protocol = match ($connection->driver) { 'pgsql' => ' PostgreSQL SSL ', 'mysql' => ' MySQL/MariaDB TLS ', default => '' }; + flash($connection->name . $protocol . '连接成功，仅执行了 SELECT 1 只读测试。')->success(); return redirect()->route('system-settings.databases.index'); } @@ -287,16 +288,16 @@ class Databases extends Controller { $sourceKeys = array_keys($this->sourceLabels()); $request->merge(['driver' => $request->input('driver', $connection?->driver ?? 'sqlsrv')]); - $postgres = $request->input('driver') === 'pgsql'; + $postgres = in_array($request->input('driver'), ['pgsql', 'mysql'], true); if ($postgres && $request->input('source_key', $connection?->source_key)) { throw ValidationException::withMessages([ - 'source_key' => '现有业务来源使用 SQL Server，不能改绑到 PostgreSQL。请为 PostgreSQL 单独保存连接。', + 'source_key' => '现有业务来源使用 SQL Server，不能更换其数据库类型。请单独添加新数据库。', ]); } return $request->validate([ - 'driver' => ['required', Rule::in(['sqlsrv', 'pgsql'])], + 'driver' => ['required', Rule::in(['sqlsrv', 'pgsql', 'mysql'])], 'name' => [ 'required', 'string', @@ -342,11 +343,12 @@ class Databases extends Controller 'port' => (int) $data['port'], 'database_name' => trim($data['database_name']), 'username' => trim($data['username']), - 'odbc_driver' => $data['driver'] === 'pgsql' ? null : trim($data['odbc_driver']), - 'tds_version' => $data['driver'] === 'pgsql' ? null : trim($data['tds_version']), + 'odbc_driver' => $data['driver'] !== 'sqlsrv' ? null : trim($data['odbc_driver']), + 'tds_version' => $data['driver'] !== 'sqlsrv' ? null : trim($data['tds_version']), 'connect_timeout' => (int) $data['connect_timeout'], 'query_timeout' => (int) $data['query_timeout'], - 'allowed_scopes' => $this->parseScopes((string) ($data['allowed_scopes'] ?? '')), + 'allowed_scopes' => array_key_exists('allowed_scopes', $data) + ? $this->parseScopes((string) $data['allowed_scopes']) : ($connection?->allowed_scopes ?? []), 'charset' => 'UTF-8', 'enabled' => $request->boolean('enabled'), ]; @@ -382,7 +384,7 @@ class Databases extends Controller private function sourceLabels(): array { - return collect((array) config('upstreams.sources', [])) + return collect(app(UpstreamSourceRegistry::class)->all()) ->mapWithKeys(fn (array $source, string $key) => [$key => $source['label'] ?? $key]) ->all(); } @@ -393,7 +395,7 @@ class Databases extends Controller 'name' => '', 'driver' => 'sqlsrv', 'source_key' => '', - 'enabled' => false, + 'enabled' => true, 'host' => '', 'port' => 1433, 'database_name' => '', @@ -460,6 +462,6 @@ class Databases extends Controller } $storedConfiguration->forget($source); - UpstreamOdbcConnection::reset($source); + \Modules\EcommerceFinance\Services\UpstreamConnection::reset($source); } } diff --git a/app/Models/Settings/DatabaseConnection.php b/app/Models/Settings/DatabaseConnection.php index de430301..62df065b 100644 --- a/app/Models/Settings/DatabaseConnection.php +++ b/app/Models/Settings/DatabaseConnection.php @@ -65,4 +65,10 @@ class DatabaseConnection extends Model { return $this->password; } + + public function sourceIdentifier(): string + { + // Legacy names stay stable; new connections need no manually assigned binding. + return $this->source_key ?: 'database_' . $this->id; + } } diff --git a/app/Utilities/SystemDatabaseConfiguration.php b/app/Utilities/SystemDatabaseConfiguration.php index e95eb153..f48b2e60 100644 --- a/app/Utilities/SystemDatabaseConfiguration.php +++ b/app/Utilities/SystemDatabaseConfiguration.php @@ -178,6 +178,12 @@ class SystemDatabaseConfiguration $updated = $this->replaceEnvironmentValues($current, [ 'DB_CONNECTION' => 'mysql', 'DB_HOST' => (string) $settings['host'], + // Laravel prioritizes split read/write hosts, URL and socket over + // DB_HOST. They must point to the same newly tested connection. + 'DB_HOST_READ' => (string) $settings['host'], + 'DB_HOST_WRITE' => (string) $settings['host'], + 'DATABASE_URL' => '', + 'DB_SOCKET' => '', 'DB_PORT' => (string) $settings['port'], 'DB_DATABASE' => (string) $settings['database'], 'DB_USERNAME' => (string) $settings['username'], @@ -555,10 +561,9 @@ class SystemDatabaseConfiguration $found = false; foreach ($lines as $index => $line) { - if (preg_match('/^' . preg_quote($key, '/') . '=/', $line) === 1) { + if (preg_match('/^\s*(?:export\s+)?' . preg_quote($key, '/') . '\s*=/', $line) === 1) { $lines[$index] = $replacement; $found = true; - break; } } diff --git a/config/upstream_pipeline.php b/config/upstream_pipeline.php new file mode 100644 index 00000000..b1c515bd --- /dev/null +++ b/config/upstream_pipeline.php @@ -0,0 +1,12 @@ + ['enabled' => false, 'connection' => ['connect_timeout' => 5, 'query_timeout' => 15]], + 'sync' => [ + 'max_datasets' => (int) env('BUSINESS_MIRROR_MAX_DATASETS', 8), + 'cycle_seconds' => (int) env('BUSINESS_MIRROR_CYCLE_SECONDS', 30), + 'dataset_seconds' => (int) env('BUSINESS_MIRROR_DATASET_SECONDS', 8), + 'catchup_batches' => (int) env('BUSINESS_MIRROR_CATCHUP_BATCHES', 6), + ], +]; diff --git a/database/migrations/2026_08_31_180000_create_business_mirror_datasets_table.php b/database/migrations/2026_08_31_180000_create_business_mirror_datasets_table.php new file mode 100644 index 00000000..7bef0672 --- /dev/null +++ b/database/migrations/2026_08_31_180000_create_business_mirror_datasets_table.php @@ -0,0 +1,24 @@ +string('dataset_key', 191)->primary(); + $table->string('source_key', 64)->index(); + $table->json('definition'); + $table->boolean('sync_enabled')->default(false); + $table->boolean('initial_sync_approved')->default(false); + $table->timestamps(); + }); + } + + public function down(): void + { + Schema::dropIfExists('business_mirror_datasets'); + } +}; diff --git a/docs/开发与部署SOP.md b/docs/开发与部署SOP.md index 146c0a6d..37511d8a 100644 --- a/docs/开发与部署SOP.md +++ b/docs/开发与部署SOP.md @@ -163,6 +163,8 @@ git pull --ff-only ## 8. 多上游财务数据库同步 +新数据库和新数据集优先使用 [通用上游数据接入](通用上游数据接入.md)：来源由连接记录自动登记，已批准的声明式数据集由调度自动首次同步和增量同步。下面的固定来源、逐项 `--full` 和旧快照说明适用于既有兼容路径，不应当作为新增普通开发任务的重复审批要求。 + ### 8.1 统一接入边界 本系统按 3 个以上上游数据库设计。生产系统数据库的启动连接只存在于服务器私有 `runtime/.env`；本地新人只填写系统数据库密码，即可自动解锁两套上游的固定主机、端口、库名、账号和密码，不需要复制生产 `APP_KEY`、填写上游环境变量或等待负责人开放范围。 diff --git a/docs/通用上游数据接入.md b/docs/通用上游数据接入.md new file mode 100644 index 00000000..94c33592 --- /dev/null +++ b/docs/通用上游数据接入.md @@ -0,0 +1,49 @@ +# 通用上游数据接入 + +## 使用边界 + +财务主库仍为团队 MariaDB/MySQL。SQL Server、PostgreSQL、MySQL/MariaDB 上游共用来源发现、字典、有限样本查询和表级镜像入口。添加同类型的第二个数据库不需要修改来源枚举、工厂或刷新器。 + +数据库连接记录未设置历史 `source_key` 时，系统使用稳定标识 `database_<记录ID>`。已有 `jiansun`、`operations_replica`、`js_data_middle_platform` 等映射保留。普通连接表单不再要求手工绑定来源或填写旧快照 scope；编辑旧连接仍保留这些历史字段。 + +原生 PostgreSQL/MySQL 连接统一要求 TLS、只读会话、非持久连接及最长 15 秒查询。SQL Server 继续复用已验证的 ODBC/DBLIB 入口。应用限制不替代上游只读账号；不提供接受用户原始 SQL 的网页接口。 + +## 开发流程（由 Codex 执行） + +1. 新人通过标准初始化启动本地环境，首次输入一次团队系统数据库密码。Docker healthy 只证明当前程序可用，`system-database:verify` 成功才表示共享接入通过。普通页面、计算逻辑和自动测试可先使用明确的隔离 fixture；不得把模拟数据当作真实财务数据。 +2. 在数据库管理页面添加并启用连接；“测试”与实际读取共用同一驱动。生产及现有字典刷新任务会发现新来源，不再要求逐库添加专用刷新器。 +3. `php artisan upstream:inspect` 列出来源标识，不输出凭据。`upstream:inspect ` 列出已有字典对象；`upstream:inspect ` 查看字段。 +4. 缺少目录时运行 `upstream:schema:sync `。此命令只读上游元数据、写入系统库目录，失败保留原目录。 +5. 需要样本时明确指定 `--columns=id,updated_at,label --limit=5`，最多返回20行。只选本任务有权读取的非敏感字段；真实样本不得进入 Git、PR 或公开日志。样本不能证明全量完整性。 +6. 确认业务键、需要的字段、增量依据和业务含义后，登记数据集。以下仅为示例，表名必须来自实际目录： + +```sh +php artisan business-mirror:register database_17 public example_orders \ + --keys=id --columns=id,updated_at,label --watermark=updated_at +``` + +登记默认不启动同步。生产首次同步涉及系统库登记和业务镜像表创建/写入，应先说明精确字段、批次、影响及备份，再取得负责人授权。授权后对同一声明追加 `--enable-sync`，生产调度即可自动完成或续传首次全量，完成后转为增量；不再为每次部署、分页、筛选重新审批或生成快照。`--disable-sync` 停止后续自动执行，保留已有镜像。 + +声明式登记不能覆盖代码维护的数据集，也不能悄悄修改已有声明的稳定键、字段和增量口径；此类变化需评估历史数据/游标兼容性。来源表没有可靠更新时间时只按主键追加，不能承诺自动发现历史修改或删除，需另行声明校准策略。 + +## 页面读取与同步状态 + +新页面在原有公司、角色和数据权限校验之后调用 `BusinessMirrorDataReader::paginate($datasetKey, $filters, $page, $perPage)`。它只读业务镜像，返回完整总数、稳定分页、实际同步状态和最近完成时间。来源暂时关闭或不可达时，已完成镜像仍可读取；无首次镜像时明确报“尚未完成首次同步”，不登记逐页快照、不从页面直接访问上游。 + +通用筛选只接受已声明字段的等值条件；复杂业务联表、金额口径、公司行级过滤继续由固定 Repository 实现。不能把此内部读取器当作绕过业务权限的公开接口。 + +`php artisan business-mirror:status [dataset] --json` 只读输出启用、首次完成、最近尝试、最近完成及错误类型。`running` 表示一个同步窗口尚未完成，不保证此刻正有进程运行。最后尝试时间不是最后数据完成时间。 + +通用调度默认每轮最多8个数据集、30秒周期预算；未完成窗口与常规数据集交替处理，单数据集默认8秒软预算，积压任务默认最多6批。预算在批次之间检查，单条查询仍受驱动超时限制；不提高上游并发。失败任务至少冷却5分钟。专用商品/价格/钉钉调度保持既有路径。 + +只对明确批准的登记数据集自动首次同步；既有代码数据集仍保留原首次授权流程。未完成的旧页面切换不等于已经上线，必须逐页验收并与正在开发的 PR 协调。 + +## 类型及验收 + +- PostgreSQL 带时区时间戳在镜像中保存为 UTC；业务展示按自己的明确时区换算。无时区日期时间保留原值。`bytea` 使用十六进制文本表示；不自动解释 JSON、数组、枚举中的业务语义。 +- 验收包含：没有修改来源枚举也能发现新库；TLS/只读保护；字段目录；只复制明确字段；首次同步、第二页、末页、总数和筛选；增量插入与更新；源不可达时读旧镜像；失败和未授权状态。 +- SQL Server 既有 Repository 与专用同步器需回归。不得把“新连接测试成功”当作全项目页面已迁移或所有业务表已同步。 + +## 系统连接保存后回滚的排查 + +截图中的“新配置未通过应用级验证”表示已尝试保存并复核，不等于未输入密码。保存系统连接必须同时统一 `DB_HOST`、读写地址，清除旧 `DATABASE_URL` 和 socket 覆盖，并处理重复赋值；验证失败时恢复完整旧配置。不要只改 `SYSTEM_DATABASE_MODE` 标志，也不要关闭失败回滚来制造“连接成功”。 diff --git a/modules/EcommerceFinance/Services/BusinessMirrorDataReader.php b/modules/EcommerceFinance/Services/BusinessMirrorDataReader.php new file mode 100644 index 00000000..a4c4f5bc --- /dev/null +++ b/modules/EcommerceFinance/Services/BusinessMirrorDataReader.php @@ -0,0 +1,46 @@ +datasets->get($key); + if (! $this->mirror->ready($dataset)) { + throw new ServiceUnavailableHttpException(30, '该数据集尚未完成首次同步，请查看同步状态；页面不会临时访问上游。'); + } + $columns = array_column($this->schemas->columns($dataset), 'column_name'); + $query = $this->mirror->connection()->table($dataset->targetTable()); + foreach ($filters as $column => $value) { + if (! in_array($column, $columns, true) || (! is_scalar($value) && $value !== null)) { + throw new InvalidArgumentException('镜像筛选字段或值无效。'); + } + $query->where($column, $value); + } + $total = (clone $query)->count(); + foreach ($dataset->keyColumns() as $column) $query->orderBy($column); + $page = max($page, 1); + $perPage = min(max($perPage, 1), 100); + $state = $this->mirror->state($dataset); + return [ + 'rows' => $query->select($columns)->offset(($page - 1) * $perPage)->limit($perPage)->get() + ->map(fn ($row) => (array) $row)->all(), + 'total' => $total, 'page' => $page, 'per_page' => $perPage, + 'last_page' => max((int) ceil($total / $perPage), 1), + 'sync_status' => $state['status'] ?? 'unknown', + 'last_succeeded_at' => $state['last_succeeded_at'] ?? null, + ]; + } +} diff --git a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php index 6464d7a8..52029d6d 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php @@ -3,6 +3,8 @@ namespace Modules\EcommerceFinance\Services; use InvalidArgumentException; +use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Schema; use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset; class BusinessMirrorDatasetRegistry @@ -12,7 +14,7 @@ class BusinessMirrorDatasetRegistry { $datasets = []; - foreach ((array) config('upstream_mirrors.datasets', []) as $key => $definition) { + foreach ($this->definitions() as $key => $definition) { $datasets[(string) $key] = $this->make((string) $key, (array) $definition); } @@ -21,7 +23,7 @@ class BusinessMirrorDatasetRegistry public function get(string $key): BusinessMirrorDataset { - $definitions = (array) config('upstream_mirrors.datasets', []); + $definitions = $this->definitions(); $definition = (array) ($definitions[$key] ?? []); if ($definition === []) { @@ -31,6 +33,43 @@ class BusinessMirrorDatasetRegistry return $this->make($key, $definition); } + private function definitions(): array + { + $definitions = (array) config('upstream_mirrors.datasets', []); + if (Schema::hasTable('business_mirror_datasets')) { + foreach (DB::table('business_mirror_datasets')->get() as $row) { + // Existing code-owned datasets cannot be redirected by a new registration. + $definitions[$row->dataset_key] ??= json_decode($row->definition, true, 32, JSON_THROW_ON_ERROR); + } + } + return $definitions; + } + + public function automaticInitialSyncAllowed(string $key): bool + { + return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets') + ->where('dataset_key', $key)->where('sync_enabled', true)->where('initial_sync_approved', true)->exists(); + } + + public function scheduled(string $key): bool + { + $definition = $this->definitions()[$key] ?? null; + if (! $definition || ! app(UpstreamSourceRegistry::class)->enabled($definition['source_key'])) return false; + if (array_key_exists($key, (array) config('upstream_mirrors.datasets', []))) return true; + return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets') + ->where('dataset_key', $key)->where('sync_enabled', true)->exists(); + } + + public function validate(string $key, array $definition): BusinessMirrorDataset + { + return $this->make($key, $definition); + } + + public function columnSelection(string $key): ?array + { + return $this->definitions()[$key]['columns'] ?? null; + } + private function make(string $key, array $definition): BusinessMirrorDataset { $source = trim((string) ($definition['source_key'] ?? '')); @@ -45,17 +84,23 @@ class BusinessMirrorDatasetRegistry )); $indexColumns = array_merge(...($indexes ?: [[]])); - foreach ([$source, $schema, $object, $target, ...$keys, ...$indexColumns] as $identifier) { + foreach ([$source, $target] as $identifier) { if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) { throw new InvalidArgumentException('业务镜像数据集包含不安全的标识符：' . $key); } } + foreach ([$schema, $object, ...$keys, ...$indexColumns] as $identifier) { + if ($identifier === '' || mb_strlen($identifier) > 128 || str_contains($identifier, "\0")) { + throw new InvalidArgumentException('业务镜像数据集包含无效的标识符：' . $key); + } + } + if ($keys === [] || in_array([], $indexes, true) || strlen($target) > 64) { throw new InvalidArgumentException('业务镜像数据集缺少稳定键或目标表名过长：' . $key); } - if ($watermark !== '' && preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $watermark) !== 1) { + if ($watermark !== '' && (mb_strlen($watermark) > 128 || str_contains($watermark, "\0"))) { throw new InvalidArgumentException('业务镜像数据集的增量字段不安全：' . $key); } diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php index 547b5401..7f011d71 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php @@ -6,6 +6,7 @@ use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Schema; use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset; +use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry; use RuntimeException; class BusinessMirrorSchemaManager @@ -87,6 +88,13 @@ class BusinessMirrorSchemaManager ->map(fn (object $row): array => (array) $row) ->all(); + $selected = app(BusinessMirrorDatasetRegistry::class)->columnSelection($dataset->key()); + if ($selected !== null) { + $selected = array_map('strtolower', $selected); + $rows = array_values(array_filter($rows, fn (array $row): bool => in_array(strtolower($row['column_name']), $selected, true))); + if (count($rows) !== count($selected)) throw new RuntimeException('business_mirror_selected_column_missing:' . $dataset->key()); + } + if ($rows === []) { throw new RuntimeException('business_mirror_schema_catalog_missing:' . $dataset->key()); } @@ -121,24 +129,24 @@ class BusinessMirrorSchemaManager $scale = min(max((int) ($column['numeric_scale'] ?? 0), 0), min($precision, 30)); $definition = match ($type) { - 'bit' => $table->boolean($name), + 'bit', 'boolean' => $table->boolean($name), 'tinyint' => $table->unsignedTinyInteger($name), 'smallint' => $table->smallInteger($name), - 'int' => $table->integer($name), + 'int', 'integer' => $table->integer($name), 'bigint' => $table->bigInteger($name), 'decimal', 'numeric', 'money', 'smallmoney' => $table->decimal($name, $precision, $scale), - 'float', 'real' => $table->double($name), + 'float', 'real', 'double', 'double precision' => $table->double($name), 'date' => $table->date($name), - 'datetime', 'datetime2', 'smalldatetime' => $table->dateTime($name, 6), + 'datetime', 'datetime2', 'smalldatetime', 'timestamp without time zone', 'timestamp with time zone' => $table->dateTime($name, 6), 'time' => $table->time($name, 6), - 'uniqueidentifier' => $table->char($name, 36), + 'uniqueidentifier', 'uuid' => $table->char($name, 36), 'binary', 'varbinary', 'image', 'timestamp', 'rowversion' => $table->binary($name), // 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' => $length > 0 && $length <= 64 + 'char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying' => $length > 0 && $length <= 64 ? $table->string($name, $length) : $table->longText($name), default => $table->longText($name), @@ -157,8 +165,8 @@ class BusinessMirrorSchemaManager $type = strtolower((string) $column['data_type']); $length = (int) ($column['max_length'] ?? 0); - return ! in_array($type, ['text', 'ntext', 'xml', 'image', 'binary', 'varbinary'], true) - && (! in_array($type, ['char', 'nchar', 'varchar', 'nvarchar'], true) || ($length > 0 && $length <= 64)); + 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)); } return false; diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php index 7577037c..673340dd 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php @@ -11,6 +11,7 @@ use Throwable; class BusinessMirrorSynchronizer { + private string $driver = 'sqlsrv'; public function __construct( private readonly BusinessMirrorDatasetRegistry $datasets, private readonly BusinessMirrorSchemaManager $schemas, @@ -23,10 +24,11 @@ class BusinessMirrorSynchronizer public function sync(string $datasetKey, bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array { $dataset = $this->datasets->get($datasetKey); + $this->driver = app(UpstreamSourceRegistry::class)->driver($dataset->sourceKey()); // A long-running scheduler may have retained a timed-out PDO instance // from the previous dataset. Each dataset starts with a fresh upstream // connection so one source failure cannot poison subsequent mirrors. - UpstreamOdbcConnection::reset($dataset->sourceKey()); + UpstreamConnection::reset($dataset->sourceKey()); $state = $this->mirror->state($dataset); if (! $full && ($state['initial_completed_at'] ?? null) === null) { @@ -60,6 +62,7 @@ class BusinessMirrorSynchronizer $batchSize = min(max($batchSize ?? (int) config('upstream_mirrors.table_sync.batch_size', 500), 1), 5000); $maxBatches = max($maxBatches ?? (int) config('upstream_mirrors.table_sync.max_batches_per_run', 2), 0); $batches = 0; + $deadline = microtime(true) + max((int) config('upstream_pipeline.sync.dataset_seconds', 8), 1); $rows = (int) ($state['last_upserted_rows'] ?? 0); try { @@ -92,7 +95,7 @@ class BusinessMirrorSynchronizer return $this->result($dataset, 'succeeded', $mode, $batches, $rows); } - if ($maxBatches > 0 && $batches >= $maxBatches) { + if ($maxBatches > 0 && ($batches >= $maxBatches || microtime(true) >= $deadline)) { return $this->result($dataset, 'running', $mode, $batches, $rows); } } @@ -120,8 +123,11 @@ class BusinessMirrorSynchronizer if ($mode === 'incremental' && $dataset->watermarkColumn() !== null) { $watermark = $this->quote($dataset->watermarkColumn()); $where[] = $watermark . ' >= ? AND ' . $watermark . ' < ?'; - $bindings[] = $this->sqlServerDateTime($windowStart); - $bindings[] = $this->sqlServerDateTime($windowEnd); + $timezoneAware = $this->driver === 'pgsql' && collect($columns)->contains(fn ($c) => + $c['column_name'] === $dataset->watermarkColumn() && $c['data_type'] === 'timestamp with time zone' + ); + $bindings[] = $timezoneAware ? $windowStart?->format('Y-m-d\TH:i:s.uP') : $this->sqlServerDateTime($windowStart); + $bindings[] = $timezoneAware ? $windowEnd?->format('Y-m-d\TH:i:s.uP') : $this->sqlServerDateTime($windowEnd); } elseif ($mode === 'incremental' && $cursor === null && $highWatermark !== []) { [$predicate, $values] = $this->keyset($dataset->keyColumns(), $highWatermark); $where[] = $predicate; @@ -134,11 +140,12 @@ class BusinessMirrorSynchronizer array_push($bindings, ...$values); } - $sql = 'SELECT TOP (' . $limit . ') ' - . implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns)) - . ' FROM ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject()) + $tail = 'FROM ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject()) . ($where === [] ? '' : ' WHERE (' . implode(') AND (', $where) . ')') . ' ORDER BY ' . implode(', ', array_map(fn (string $name): string => $this->quote($name), $order)); + $sql = (new UpstreamSqlDialect($this->driver))->select( + implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns)), $tail, $limit, + ); return $this->queries->forSource($dataset->sourceKey())->select($sql, $bindings); } @@ -149,6 +156,17 @@ class BusinessMirrorSynchronizer $name = (string) $column['column_name']; $quoted = $this->quote($name); + if ($this->driver !== 'sqlsrv') { + if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'timestamp with time zone') { + // Store timezone-aware PostgreSQL instants as UTC in the shared mirror. + return "to_char({$quoted} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}"; + } + if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'bytea') { + return "encode({$quoted}, 'hex') AS {$quoted}"; + } + return $quoted; + } + // FreeTDS may return SQL Server date values using the connection locale // (for example "Jan 1 2026"). Convert temporal columns at the source so // every mirror receives deterministic ISO values that MariaDB accepts. @@ -244,7 +262,7 @@ class BusinessMirrorSynchronizer $type = $types[strtolower((string) $name)] ?? ''; if ($type === 'date') { $row[$name] = substr($value, 0, 10); - } elseif (in_array($type, ['datetime', 'datetime2', 'smalldatetime'], true)) { + } elseif (in_array($type, ['datetime', 'datetime2', 'smalldatetime', 'timestamp without time zone', 'timestamp with time zone'], true)) { $row[$name] = preg_replace('/(\.\d{6})\d+/', '$1', str_replace('T', ' ', $value)); } elseif (in_array($type, ['char', 'nchar', 'varchar', 'nvarchar', 'text', 'ntext', 'xml'], true)) { $row[$name] = $this->normalizeText($value); @@ -271,11 +289,7 @@ class BusinessMirrorSynchronizer private function quote(string $identifier): string { - if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) { - throw new RuntimeException('business_mirror_invalid_identifier'); - } - - return '[' . $identifier . ']'; + return (new UpstreamSqlDialect($this->driver))->quote($identifier); } private function cursorValue(array $cursor, string $column): mixed diff --git a/modules/EcommerceFinance/Services/UpstreamConnection.php b/modules/EcommerceFinance/Services/UpstreamConnection.php new file mode 100644 index 00000000..8ebc7f08 --- /dev/null +++ b/modules/EcommerceFinance/Services/UpstreamConnection.php @@ -0,0 +1,73 @@ + */ + private static array $native = []; + + public static function get(string $source): PDO + { + $registry = app(UpstreamSourceRegistry::class); + if ($registry->driver($source) === 'sqlsrv') { + return UpstreamOdbcConnection::get($source); + } + if (! $registry->enabled($source)) { + throw new ServiceUnavailableHttpException(30, '该数据源尚未启用。'); + } + return self::$native[$source] ??= self::openNative($registry->driver($source), $registry->value($source, 'connection', [])); + } + + public static function reset(?string $source = null): void + { + if ($source === null) self::$native = []; + else unset(self::$native[$source]); + UpstreamOdbcConnection::reset($source); + } + + public static function openNative(string $driver, #[SensitiveParameter] array $config): PDO + { + if (! in_array($driver, ['pgsql', 'mysql'], true) || ! in_array($driver, PDO::getAvailableDrivers(), true)) { + throw new ServiceUnavailableHttpException(30, '应用缺少该数据库类型的驱动，请更新应用镜像。'); + } + $timeout = min(max((int) ($config['connect_timeout'] ?? 5), 1), 30); + $seconds = min(max((int) ($config['query_timeout'] ?? 15), 1), 15); + $options = [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_PERSISTENT => false, PDO::ATTR_TIMEOUT => $timeout]; + $host = (string) $config['host']; + $database = (string) $config['database']; + if (preg_match('/[;\x00-\x1F]/', $host . $database)) { + throw new ServiceUnavailableHttpException(30, '主机或数据库名称包含不支持的字符。'); + } + $port = min(max((int) $config['port'], 1), 65535); + if ($driver === 'pgsql') { + $quote = static fn (string $s): string => "'" . str_replace(['\\', "'"], ['\\\\', "\\'"], $s) . "'"; + $dsn = 'pgsql:host=' . $quote($host) . ';port=' . $port . ';dbname=' . $quote($database) + . ';sslmode=require;connect_timeout=' . $timeout + . ";options='-c default_transaction_read_only=on -c statement_timeout=" . ($seconds * 1000) . "'"; + } else { + $dsn = "mysql:host={$host};port={$port};dbname={$database};charset=utf8mb4"; + $options[PDO::MYSQL_ATTR_SSL_CIPHER] = 'DEFAULT'; + $options[PDO::MYSQL_ATTR_SSL_VERIFY_SERVER_CERT] = false; + } + try { + $pdo = new PDO($dsn, (string) $config['username'], (string) $config['password'], $options); + if ($driver === 'mysql') { + $tls = $pdo->query("SHOW SESSION STATUS LIKE 'Ssl_cipher'")->fetch(PDO::FETCH_NUM); + if (empty($tls[1])) throw new ServiceUnavailableHttpException(30, '上游数据库未建立 TLS 加密连接。'); + $pdo->exec('SET SESSION TRANSACTION READ ONLY'); + $pdo->exec(str_contains($pdo->getAttribute(PDO::ATTR_SERVER_VERSION), 'MariaDB') + ? 'SET SESSION max_statement_time=' . $seconds + : 'SET SESSION MAX_EXECUTION_TIME=' . ($seconds * 1000)); + } + return $pdo; + } catch (PDOException) { + throw new ServiceUnavailableHttpException(30, '上游只读连接失败，请检查账号、权限、网络和 TLS 配置。'); + } + } +} diff --git a/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php b/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php index a41b9ff1..7cd0adbe 100644 --- a/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php +++ b/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php @@ -7,62 +7,32 @@ use PDOException; use SensitiveParameter; use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException; -/** Connection tests only: never used to route business queries or synchronize tables. */ +/** The test button and business readers share one driver implementation. */ class UpstreamDatabaseProbe { public function probe(string $driver, #[SensitiveParameter] array $connection, string $label): void { if ($driver === 'sqlsrv') { UpstreamOdbcConnection::probeConfiguration($connection, $label); - return; } - - if ($driver !== 'pgsql') { - throw new ServiceUnavailableHttpException(30, '不支持此数据库类型，请选择 SQL Server 或 PostgreSQL。'); + if (! in_array($driver, ['pgsql', 'mysql'], true)) { + throw new ServiceUnavailableHttpException(30, '不支持此数据库类型。'); } - - $timeout = min(max((int) ($connection['connect_timeout'] ?? 5), 1), 30); - $queryTimeout = min(max((int) ($connection['query_timeout'] ?? 15), 1), 15) * 1000; - $dsn = 'pgsql:host=' . $this->quote((string) $connection['host']) - . ';port=' . min(max((int) $connection['port'], 1), 65535) - . ';dbname=' . $this->quote((string) $connection['database']) - . ';sslmode=require;connect_timeout=' . $timeout - . ";options='-c default_transaction_read_only=on -c statement_timeout=" . $queryTimeout . "'"; - try { - // One short-lived connection, TLS and read-only enforced at session startup. - $pdo = $this->connect($dsn, (string) $connection['username'], (string) $connection['password']); + $pdo = $this->connect($driver, $connection); $statement = $pdo->query('SELECT 1'); if ($statement === false || (int) $statement->fetchColumn() !== 1) { - throw new ServiceUnavailableHttpException(30, 'PostgreSQL 只读测试未返回预期结果。'); + throw new ServiceUnavailableHttpException(30, '只读测试未返回预期结果。'); } $statement->closeCursor(); } catch (PDOException) { - // Do not expose raw driver errors, DSNs or credentials in logs/session flashes. - throw new ServiceUnavailableHttpException(30, 'PostgreSQL 连接失败。请检查账号、密码、数据库访问权限，以及服务器网络和 SSL 配置。'); + throw new ServiceUnavailableHttpException(30, '上游只读连接失败，请检查账号、权限、网络和 TLS 配置。'); } } - protected function connect(string $dsn, string $username, #[SensitiveParameter] string $password): PDO + protected function connect(string $driver, #[SensitiveParameter] array $connection): PDO { - if (! in_array('pgsql', PDO::getAvailableDrivers(), true)) { - throw new ServiceUnavailableHttpException(30, '应用缺少 PostgreSQL 驱动，请部署包含 pdo_pgsql 的新版应用镜像。'); - } - - return new PDO($dsn, $username, $password, [ - PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, - PDO::ATTR_PERSISTENT => false, - ]); - } - - private function quote(string $value): string - { - // PDO replaces all DSN semicolons with spaces; reject rather than alter database names. - if (preg_match('/[;\x00-\x1F]/', $value)) { - throw new ServiceUnavailableHttpException(30, '主机或数据库名称包含不支持的字符。'); - } - - return "'" . str_replace(['\\', "'"], ['\\\\', "\\'"], $value) . "'"; + return UpstreamConnection::openNative($driver, $connection); } } diff --git a/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php b/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php index 30825d79..8d2f6ea1 100644 --- a/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php +++ b/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php @@ -27,7 +27,7 @@ class UpstreamReadOnlyQuery { self::assertReadOnlySql($sql); - $statement = UpstreamOdbcConnection::get($this->source)->prepare($sql); + $statement = UpstreamConnection::get($this->source)->prepare($sql); $statement->execute($bindings); return $statement->fetchAll(PDO::FETCH_ASSOC); @@ -45,7 +45,7 @@ class UpstreamReadOnlyQuery { self::assertReadOnlySql($sql); - $statement = UpstreamOdbcConnection::get($this->source)->prepare($sql); + $statement = UpstreamConnection::get($this->source)->prepare($sql); $statement->execute($bindings); return $statement->fetchColumn(); @@ -68,6 +68,7 @@ class UpstreamReadOnlyQuery $inspect = preg_replace("/'(?:''|[^'])*'/s", "''", $sql) ?? $sql; $inspect = preg_replace('/\[(?:\]\]|[^\]])*\]/s', '[]', $inspect) ?? $inspect; $inspect = preg_replace('/"(?:""|[^"])*"/s', '""', $inspect) ?? $inspect; + $inspect = preg_replace('/`(?:``|[^`])*`/s', '``', $inspect) ?? $inspect; $inspect = strtoupper(ltrim($inspect)); if (preg_match('/^(SELECT|WITH)\b/', $inspect) !== 1) { @@ -76,7 +77,7 @@ class UpstreamReadOnlyQuery $forbidden = '/\b(?:INSERT|UPDATE|DELETE|MERGE|DROP|ALTER|CREATE|TRUNCATE|EXEC|EXECUTE|' . 'GRANT|REVOKE|DENY|BACKUP|RESTORE|DBCC|BULK|OPENROWSET|OPENDATASOURCE|INTO|' - . 'WAITFOR|KILL|SHUTDOWN)\b|\bNEXT\s+VALUE\s+FOR\b/i'; + . 'WAITFOR|KILL|SHUTDOWN|SET_CONFIG|NEXTVAL|SETVAL|SLEEP|BENCHMARK|GET_LOCK|RELEASE_LOCK|LOAD_FILE|PG_SLEEP)\b|\bNEXT\s+VALUE\s+FOR\b/i'; if (preg_match($forbidden, $inspect) === 1) { throw new InvalidArgumentException('上游查询包含禁止的写入或高风险语法。'); diff --git a/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php b/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php index a70f0a05..62093dc6 100644 --- a/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php +++ b/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php @@ -7,7 +7,7 @@ use Illuminate\Support\Facades\Schema; use RuntimeException; /** - * 将上游 SQL Server 的表、视图和字段结构同步到共享系统数据库。 + * 将已支持类型的上游表、视图和字段结构同步到共享系统数据库。 * * 这里只读取 INFORMATION_SCHEMA 元数据，不读取业务行，也不保存连接资料。 * 获取完整结果成功后才替换单个来源的旧目录，避免网络故障清空现有字典。 @@ -28,25 +28,15 @@ class UpstreamSchemaCatalogSynchronizer throw new RuntimeException('该上游数据库尚未启用。'); } - $rows = $this->queries->forSource($source)->select( - "SELECT - c.TABLE_SCHEMA AS schema_name, - c.TABLE_NAME AS object_name, - CASE WHEN t.TABLE_TYPE = 'VIEW' THEN 'view' ELSE 'table' END AS object_type, - c.COLUMN_NAME AS column_name, - c.ORDINAL_POSITION AS ordinal_position, - c.DATA_TYPE AS data_type, - c.CHARACTER_MAXIMUM_LENGTH AS max_length, - c.NUMERIC_PRECISION AS numeric_precision, - c.NUMERIC_SCALE AS numeric_scale, - c.IS_NULLABLE AS is_nullable - FROM INFORMATION_SCHEMA.COLUMNS c - INNER JOIN INFORMATION_SCHEMA.TABLES t - ON t.TABLE_SCHEMA = c.TABLE_SCHEMA - AND t.TABLE_NAME = c.TABLE_NAME - WHERE c.TABLE_SCHEMA NOT IN ('sys', 'INFORMATION_SCHEMA') - ORDER BY c.TABLE_SCHEMA, c.TABLE_NAME, c.ORDINAL_POSITION" - ); + $driver = $this->sources->driver($source); + $rows = $this->queries->forSource($source)->select((new UpstreamSqlDialect($driver))->catalogSql()); + if ($driver === 'mysql') { + foreach ($rows as &$row) { + // SQL Server timestamp is binary; MySQL timestamp is temporal. + if (strtolower($row['data_type'] ?? '') === 'timestamp') $row['data_type'] = 'datetime'; + } + unset($row); + } if ($rows === []) { throw new RuntimeException('上游数据库没有返回可用的表字段目录。'); diff --git a/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php b/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php index d5287524..3380924f 100644 --- a/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php +++ b/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php @@ -28,7 +28,11 @@ class UpstreamSourceRegistry $definition = config('upstreams.sources.' . $source); if (! is_array($definition)) { - throw new InvalidArgumentException('未登记的上游数据源：' . $source); + $connection = $this->storedConfiguration->find($source); + if ($connection === null) { + throw new InvalidArgumentException('未登记的上游数据源：' . $source); + } + $definition = ['label' => $connection->name, 'config' => 'upstream_pipeline.defaults']; } $configKey = trim((string) ($definition['config'] ?? '')); @@ -77,6 +81,25 @@ class UpstreamSourceRegistry return (bool) $this->value($source, 'enabled', false); } + public function driver(string $source): string + { + $definition = $this->definition($source); + $default = config($definition['config'] . '.connection.driver', 'sqlsrv'); + return (string) (($this->usesLocalConfiguration() + ? $default : $this->storedConfiguration->value($source, 'connection.driver', $default)) ?: 'sqlsrv'); + } + + /** @return array */ + public function all(): array + { + $keys = array_unique(array_merge(array_keys((array) config('upstreams.sources', [])), array_keys($this->storedConfiguration->all()))); + $definitions = []; + foreach ($keys as $key) { + $definitions[$key] = $this->definition($key); + } + return $definitions; + } + private function usesLocalConfiguration(): bool { return (bool) config('upstreams.local_configuration.enabled', false); @@ -85,6 +108,7 @@ class UpstreamSourceRegistry private function connection(string $source, array $configured): array { $fields = array_values(array_unique(array_merge(array_keys($configured), [ + 'driver', 'host', 'port', 'database', diff --git a/modules/EcommerceFinance/Services/UpstreamSqlDialect.php b/modules/EcommerceFinance/Services/UpstreamSqlDialect.php new file mode 100644 index 00000000..1f813875 --- /dev/null +++ b/modules/EcommerceFinance/Services/UpstreamSqlDialect.php @@ -0,0 +1,55 @@ +driver) { + 'pgsql' => '"' . str_replace('"', '""', $identifier) . '"', + 'mysql' => '`' . str_replace('`', '``', $identifier) . '`', + default => '[' . str_replace(']', ']]', $identifier) . ']', + }; + } + + public function select(string $columns, string $tail, int $limit): string + { + $limit = min(max($limit, 1), 5000); + return $this->driver === 'sqlsrv' + ? 'SELECT TOP (' . $limit . ') ' . $columns . ' ' . $tail + : 'SELECT ' . $columns . ' ' . $tail . ' LIMIT ' . $limit; + } + + public function catalogSql(): string + { + // Same normalized catalog shape for all engines; MySQL never enumerates other databases. + $schema = $this->driver === 'sqlsrv' ? 'INFORMATION_SCHEMA' : 'information_schema'; + $sql = "SELECT c.table_schema AS schema_name, c.table_name AS object_name, + CASE WHEN t.table_type = 'VIEW' THEN 'view' ELSE 'table' END AS object_type, + c.column_name AS column_name, c.ordinal_position AS ordinal_position, + c.data_type AS data_type, c.character_maximum_length AS max_length, + c.numeric_precision AS numeric_precision, c.numeric_scale AS numeric_scale, + c.is_nullable AS is_nullable + FROM {$schema}." . ($this->driver === 'sqlsrv' ? 'COLUMNS' : 'columns') . " c + INNER JOIN {$schema}." . ($this->driver === 'sqlsrv' ? 'TABLES' : 'tables') . " t + ON t.table_schema = c.table_schema AND t.table_name = c.table_name + WHERE c.table_schema NOT IN ('sys', 'INFORMATION_SCHEMA', 'information_schema', 'pg_catalog', 'mysql', 'performance_schema')"; + if ($this->driver === 'mysql') $sql .= ' AND c.table_schema = DATABASE()'; + $sql .= ' ORDER BY c.table_schema, c.table_name, c.ordinal_position'; + if ($this->driver === 'sqlsrv') { + $sql = preg_replace_callback('/\b([ct])\.([a-z_]+)\b/', fn ($m) => $m[1] . '.' . strtoupper($m[2]), $sql); + } + return $sql; + } +} diff --git a/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php b/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php index 82e5c4cd..cd9feee3 100644 --- a/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php +++ b/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php @@ -42,10 +42,13 @@ class UpstreamStoredConfiguration return $this->connections[$source] = null; } - $this->connections[$source] = DatabaseConnection::query() - ->where('company_id', self::SYSTEM_COMPANY_ID) - ->where('source_key', $source) - ->first(); + $query = DatabaseConnection::query()->where('company_id', self::SYSTEM_COMPANY_ID); + if (preg_match('/^database_([1-9][0-9]*)$/', $source, $matches)) { + $query->where('id', $matches[1])->whereNull('source_key'); + } else { + $query->where('source_key', $source); + } + $this->connections[$source] = $query->first(); return $this->connections[$source]; } @@ -55,6 +58,16 @@ class UpstreamStoredConfiguration unset($this->connections[$source]); } + /** @return array */ + public function all(): array + { + if (! $this->enabled() || ! Schema::hasTable('system_database_connections')) { + return []; + } + return DatabaseConnection::query()->where('company_id', self::SYSTEM_COMPANY_ID) + ->get()->keyBy(fn (DatabaseConnection $row) => $row->sourceIdentifier())->all(); + } + public function value(string $source, string $path, mixed $default = null): mixed { $connection = $this->find($source); @@ -68,6 +81,7 @@ class UpstreamStoredConfiguration 'allowed_scopes' => $connection->allowed_scopes ?? [], 'connection' => $this->connection($connection, $default), 'connection.host' => $connection->host, + 'connection.driver' => $connection->driver, 'connection.port' => $connection->port, 'connection.database' => $connection->database_name, 'connection.username' => $connection->username, @@ -85,6 +99,7 @@ class UpstreamStoredConfiguration $configured = is_array($default) ? $default : []; return array_replace($configured, [ + 'driver' => $connection->driver, 'host' => $connection->host, 'port' => $connection->port, 'database' => $connection->database_name, diff --git a/resources/views/system_settings/databases/form.blade.php b/resources/views/system_settings/databases/form.blade.php index 7e3e5cdf..e16fff02 100644 --- a/resources/views/system_settings/databases/form.blade.php +++ b/resources/views/system_settings/databases/form.blade.php @@ -5,7 +5,8 @@
@include('system_settings.partials.navigation') - @php($postgres = old('driver', $values['driver']) === 'pgsql') + @php($selectedDriver = old('driver', $values['driver'])) + @php($postgres = $selectedDriver !== 'sqlsrv')
@csrf @if ($connection) @method('PATCH') @endif @@ -20,11 +21,12 @@ -

通过 SSL 加密连接（sslmode=require）；仅执行只读测试，查询最长 15 秒。保存连接不会自动同步业务表。

+

通过 TLS 加密连接（PostgreSQL 使用 sslmode=require）；只读查询最长 15 秒。启用后自动发现字段目录，业务表按已批准的数据集同步。

-
@@ -69,7 +61,6 @@ -
@@ -90,11 +81,12 @@ (() => { const form = document.getElementById('cloud-database-form'); const driver = form.elements.driver; + const ports = {sqlsrv: '1433', pgsql: '5432', mysql: '3306'}; let previous = driver.value; const update = (changePort) => { - const postgres = driver.value === 'pgsql'; - if (changePort && form.elements.port.value === (previous === 'pgsql' ? '5432' : '1433')) { - form.elements.port.value = postgres ? '5432' : '1433'; + const postgres = driver.value !== 'sqlsrv'; + if (changePort && form.elements.port.value === ports[previous]) { + form.elements.port.value = ports[driver.value]; } form.querySelectorAll('[data-sqlserver-only]').forEach((group) => { group.style.display = postgres ? 'none' : ''; diff --git a/resources/views/system_settings/databases/index.blade.php b/resources/views/system_settings/databases/index.blade.php index d38520ea..be88516b 100644 --- a/resources/views/system_settings/databases/index.blade.php +++ b/resources/views/system_settings/databases/index.blade.php @@ -88,7 +88,6 @@ 名称 地址 数据库 - 业务绑定 状态 操作 @@ -96,10 +95,9 @@ @forelse ($connections as $connection) - {{ $connection->name }}{{ $connection->driver === 'pgsql' ? 'PostgreSQL · SSL' : 'SQL Server' }} + {{ $connection->name }}{{ match ($connection->driver) { 'pgsql' => 'PostgreSQL · SSL', 'mysql' => 'MySQL / MariaDB · TLS', default => 'SQL Server' } }} {{ $connection->host }}:{{ $connection->port }} {{ $connection->database_name }} - {{ $sourceLabels[$connection->source_key] ?? '未绑定' }} @empty - + {{ $systemDatabase['mode'] === 'bootstrap' ? '连接团队共享系统数据库后，两套云数据库配置会自动显示在这里。' : '共享系统数据库中还没有云数据库配置。' }} diff --git a/tests/Feature/Console/BusinessMirrorScheduleTest.php b/tests/Feature/Console/BusinessMirrorScheduleTest.php index bccd7d41..1b525ae5 100644 --- a/tests/Feature/Console/BusinessMirrorScheduleTest.php +++ b/tests/Feature/Console/BusinessMirrorScheduleTest.php @@ -12,6 +12,49 @@ use Tests\TestCase; class BusinessMirrorScheduleTest extends TestCase { + public function test_unapproved_first_sync_is_skipped(): void + { + config(['upstream_mirrors.refresh_enabled' => true]); + $dataset = Mockery::mock(BusinessMirrorDataset::class); + $registry = Mockery::mock(BusinessMirrorDatasetRegistry::class); + $registry->shouldReceive('all')->once()->andReturn(['pending' => $dataset]); + $registry->shouldReceive('scheduled')->with('pending')->andReturn(true); + $registry->shouldReceive('automaticInitialSyncAllowed')->with('pending')->andReturn(false); + $mirror = Mockery::mock(GenericBusinessMirrorRepository::class); + $mirror->shouldReceive('state')->with($dataset)->andReturn(null); + $sync = Mockery::mock(BusinessMirrorSynchronizer::class); + $sync->shouldNotReceive('sync'); + $this->app->instance(BusinessMirrorDatasetRegistry::class, $registry); + $this->app->instance(GenericBusinessMirrorRepository::class, $mirror); + $this->app->instance(BusinessMirrorSynchronizer::class, $sync); + $this->artisan('business-mirror:sync-tables')->assertSuccessful(); + } + + public function test_catchup_and_routine_datasets_both_receive_a_turn(): void + { + config(['upstream_mirrors.refresh_enabled' => true, 'upstream_pipeline.sync.max_datasets' => 2]); + $registry = Mockery::mock(BusinessMirrorDatasetRegistry::class); + $mirror = Mockery::mock(GenericBusinessMirrorRepository::class); + $objects = []; + foreach (['routine-a', 'routine-b', 'catchup-z'] as $key) { + $objects[$key] = Mockery::mock(BusinessMirrorDataset::class); + $registry->shouldReceive('scheduled')->with($key)->andReturn(true); + $mirror->shouldReceive('state')->with($objects[$key])->andReturn([ + 'initial_completed_at' => '2026-08-01 00:00:00', 'last_attempted_at' => '2026-08-01 00:00:00', + 'status' => $key === 'catchup-z' ? 'running' : 'succeeded', + ]); + } + $registry->shouldReceive('all')->andReturn($objects); + $sync = Mockery::mock(BusinessMirrorSynchronizer::class); + $result = ['status' => 'succeeded', 'batches' => 1, 'rows' => 1]; + $sync->shouldReceive('sync')->once()->with('routine-a', false, null, null)->ordered()->andReturn($result); + $sync->shouldReceive('sync')->once()->with('catchup-z', false, null, 6)->ordered()->andReturn($result); + $this->app->instance(BusinessMirrorDatasetRegistry::class, $registry); + $this->app->instance(GenericBusinessMirrorRepository::class, $mirror); + $this->app->instance(BusinessMirrorSynchronizer::class, $sync); + $this->artisan('business-mirror:sync-tables')->assertSuccessful(); + } + public function test_generic_rotation_excludes_datasets_with_a_dedicated_scheduler(): void { config([ @@ -28,6 +71,7 @@ class BusinessMirrorScheduleTest extends TestCase ProductCatalogMirrorRepository::DATASET_KEY => $catalog, 'test.dbo.other' => $other, ]); + $datasets->shouldReceive('scheduled')->once()->with('test.dbo.other')->andReturn(true); $mirror = Mockery::mock(GenericBusinessMirrorRepository::class); $mirror->shouldReceive('state')->once()->with($other)->andReturn([ 'initial_completed_at' => '2026-08-01 02:00:00', diff --git a/tests/Feature/SystemSettings/UpstreamDatabasesTest.php b/tests/Feature/SystemSettings/UpstreamDatabasesTest.php index 1812ff0f..254257fc 100644 --- a/tests/Feature/SystemSettings/UpstreamDatabasesTest.php +++ b/tests/Feature/SystemSettings/UpstreamDatabasesTest.php @@ -578,6 +578,30 @@ class UpstreamDatabasesTest extends FeatureTestCase ], $overrides); } + public function test_mysql_connection_appears_in_dictionary_sources_without_a_binding(): void + { + $payload = $this->payload(['driver' => 'mysql', 'port' => 3306]); + unset($payload['source_key'], $payload['allowed_scopes'], $payload['odbc_driver'], $payload['tds_version']); + $this->loginAs()->post(route('system-settings.databases.store'), $payload)->assertSessionHasNoErrors(); + $row = DatabaseConnection::where('driver', 'mysql')->firstOrFail(); + $this->assertArrayHasKey($row->sourceIdentifier(), app(UpstreamSourceRegistry::class)->all()); + $this->get(route('system-settings.databases.create'))->assertOk() + ->assertSee('MySQL / MariaDB')->assertDontSee('name="source_key"', false)->assertDontSee('name="allowed_scopes"', false); + } + + public function test_editing_simplified_form_preserves_legacy_binding_and_scopes(): void + { + $row = $this->connection('keep-secret'); + $row->update(['source_key' => 'jiansun']); + $payload = $this->payload(['name' => $row->name, 'password' => '']); + unset($payload['source_key'], $payload['allowed_scopes']); + $this->loginAs()->patch(route('system-settings.databases.update', $row->id), $payload)->assertSessionHasNoErrors(); + $row->refresh(); + $this->assertSame('jiansun', $row->source_key); + $this->assertSame(['supplier.options'], $row->allowed_scopes); + $this->assertSame('keep-secret', $row->password); + } + private function connection(string $password): DatabaseConnection { return DatabaseConnection::create([ diff --git a/tests/Unit/EcommerceFinance/GenericUpstreamPipelineTest.php b/tests/Unit/EcommerceFinance/GenericUpstreamPipelineTest.php new file mode 100644 index 00000000..01cbfde2 --- /dev/null +++ b/tests/Unit/EcommerceFinance/GenericUpstreamPipelineTest.php @@ -0,0 +1,184 @@ + true, + 'upstreams.local_configuration.enabled' => false, + 'database.connections.sqlite.password' => 'fixture-cipher-key', + 'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''], + 'upstream_mirrors.connection' => 'business_mirror', + 'upstream_mirrors.datasets' => [], + 'upstream_mirrors.refresh_enabled' => true, + ]); + DB::purge('business_mirror'); + Schema::connection('business_mirror')->create('business_mirror_table_states', function (Blueprint $t): void { + $t->string('dataset_key')->primary(); + foreach (['source_key', 'source_object', 'target_table', 'status', 'run_mode', 'generation', 'last_error_code'] as $c) $t->string($c)->nullable(); + foreach (['cursor', 'high_watermark'] as $c) $t->json($c)->nullable(); + foreach (['window_started_at', 'window_ends_at', 'initial_completed_at', 'last_full_succeeded_at', 'last_attempted_at', 'last_succeeded_at'] as $c) $t->dateTime($c)->nullable(); + $t->unsignedBigInteger('last_upserted_rows')->default(0); + $t->unsignedBigInteger('mirror_row_count')->default(0); + $t->timestamps(); + }); + } + + protected function tearDown(): void + { + UpstreamConnection::reset(); + parent::tearDown(); + } + + private function connection(string $driver = 'pgsql'): DatabaseConnection + { + return DatabaseConnection::create([ + 'company_id' => 0, 'name' => 'Isolated fixture', 'driver' => $driver, 'host' => 'fixture.invalid', + 'port' => $driver === 'pgsql' ? 5432 : 3306, 'database_name' => 'fixture', + 'username' => 'fixture', 'password' => 'fixture-password', 'enabled' => true, + 'connect_timeout' => 5, 'query_timeout' => 15, + ]); + } + + public function test_new_connection_is_immediately_a_source_without_changing_config(): void + { + $row = $this->connection(); + $key = $row->sourceIdentifier(); + $sources = app(UpstreamSourceRegistry::class); + $this->assertNull($row->source_key); + $this->assertNull(config('upstreams.sources.' . $key)); + $this->assertSame('pgsql', $sources->driver($key)); + $this->assertTrue($sources->enabled($key)); + $this->assertSame('fixture-password', $sources->value($key, 'connection.password')); + $this->assertArrayHasKey($key, $sources->all()); + $this->assertSame('jiansun', $sources->definition('jiansun')['key']); + } + + public function test_dialects_use_engine_specific_limits_and_quotes(): void + { + $this->assertSame('SELECT TOP (5) [id] FROM [dbo].[records]', (new UpstreamSqlDialect('sqlsrv'))->select('[id]', 'FROM [dbo].[records]', 5)); + $this->assertSame('SELECT "id" FROM "public"."records" LIMIT 5', (new UpstreamSqlDialect('pgsql'))->select('"id"', 'FROM "public"."records"', 5)); + $this->assertSame('SELECT `id` FROM `records` LIMIT 5', (new UpstreamSqlDialect('mysql'))->select('`id`', 'FROM `records`', 5)); + $this->assertStringContainsString('c.table_schema = DATABASE()', (new UpstreamSqlDialect('mysql'))->catalogSql()); + } + + public function test_source_discovery_does_not_decrypt_credentials(): void + { + $row = $this->connection(); + DB::table('system_database_connections')->where('id', $row->id)->update(['shared_password' => 'invalid-fixture-envelope']); + $sources = app(UpstreamSourceRegistry::class); + $this->assertArrayHasKey($row->sourceIdentifier(), $sources->all()); + $this->assertSame('pgsql', $sources->driver($row->sourceIdentifier())); + } + + public function test_declarations_require_fields_and_explicit_initial_sync_approval(): void + { + $row = $this->connection(); + $source = $row->sourceIdentifier(); + DB::table('upstream_schema_columns')->insert([ + 'source_key' => $source, 'schema_name' => 'public', 'object_name' => 'records', + 'object_type' => 'table', 'column_name' => 'id', 'ordinal_position' => 1, 'data_type' => 'bigint', + 'is_nullable' => false, 'synchronized_at' => now(), 'created_at' => now(), 'updated_at' => now(), + ]); + $args = ['source' => $source, 'schema' => 'public', 'object' => 'records', '--keys' => 'id', '--columns' => 'id']; + $this->artisan('business-mirror:register', $args)->assertSuccessful(); + $key = $source . '.public.records'; + $registry = app(BusinessMirrorDatasetRegistry::class); + $this->assertFalse($registry->automaticInitialSyncAllowed($key)); + $this->artisan('business-mirror:sync-tables')->assertSuccessful(); + $this->assertSame(0, DB::connection('business_mirror')->table('business_mirror_table_states')->count()); + $this->artisan('business-mirror:register', $args + ['--enable-sync' => true])->assertSuccessful(); + $this->assertTrue($registry->automaticInitialSyncAllowed($key)); + $this->assertSame(['id'], $registry->columnSelection($key)); + $this->artisan('business-mirror:register', ['source' => $source, 'schema' => 'public', 'object' => 'records', '--disable-sync' => true])->assertSuccessful(); + $this->assertFalse($registry->scheduled($key)); + } + + /** @dataProvider nativeDrivers */ + public function test_native_tls_catalog_first_sync_pagination_and_incremental(string $driver, bool $timezoneAware = false, ?string $fixtureHost = null): void + { + if (getenv('CAIWU_PIPELINE_INTEGRATION') !== '1') $this->markTestSkipped('Requires isolated pipeline fixture containers.'); + $host = $fixtureHost ?? ($driver === 'pgsql' ? 'caiwu-pipeline-pg' : 'caiwu-pipeline-mysql'); + $database = $driver === 'pgsql' ? 'postgres' : 'pipeline_fixture'; + $user = $driver === 'pgsql' ? 'postgres' : 'root'; + $schema = $driver === 'pgsql' ? 'public' : $database; + $dsn = $driver === 'pgsql' ? "pgsql:host={$host};dbname={$database};sslmode=require" : "mysql:host={$host};dbname={$database}"; + $options = [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]; + if ($driver === 'mysql') { + $options[PDO::MYSQL_ATTR_SSL_CIPHER] = 'DEFAULT'; + $options[PDO::MYSQL_ATTR_SSL_VERIFY_SERVER_CERT] = false; + } + $admin = new PDO($dsn, $user, 'fixture-password', $options); + $admin->exec('DROP TABLE IF EXISTS pipeline_records'); + $type = $driver === 'pgsql' ? ($timezoneAware ? 'timestamp with time zone' : 'timestamp without time zone') : 'datetime(6)'; + $admin->exec('CREATE TABLE pipeline_records (id bigint primary key, modified_at ' . $type . ', amount decimal(18,2), label varchar(50), private_note text)'); + $date = now()->subMinute()->format($timezoneAware ? 'Y-m-d H:i:sP' : 'Y-m-d H:i:s'); + $insert = $admin->prepare('INSERT INTO pipeline_records (id, modified_at, amount, label) VALUES (?, ?, ?, ?)'); + $insert->execute([1, $date, '12.30', 'first']); + $insert->execute([2, $date, '45.60', 'second']); + $row = $this->connection($driver); + $row->update(['host' => $host, 'database_name' => $database, 'username' => $user]); + $source = $row->sourceIdentifier(); + $result = app(UpstreamSchemaCatalogSynchronizer::class)->sync($source); + $this->assertGreaterThanOrEqual(5, $result['columns']); + $this->artisan('business-mirror:register', [ + 'source' => $source, 'schema' => $schema, 'object' => 'pipeline_records', '--keys' => 'id', + '--columns' => 'id,modified_at,amount,label', '--watermark' => 'modified_at', '--enable-sync' => true, + ])->assertSuccessful(); + $this->artisan('business-mirror:sync-tables')->assertSuccessful(); + $dataset = app(BusinessMirrorDatasetRegistry::class)->get($source . '.' . $schema . '.pipeline_records'); + $mirror = DB::connection('business_mirror')->table($dataset->targetTable()); + $this->assertSame(2, $mirror->count()); + $this->assertSame(2, (int) $mirror->orderBy('id')->offset(1)->value('id')); + $this->assertFalse(Schema::connection('business_mirror')->hasColumn($dataset->targetTable(), 'private_note')); + $admin->exec("UPDATE pipeline_records SET amount=99.50 WHERE id=1"); + $insert->execute([3, $date, '6.70', 'third']); + $insert->execute([4, now()->addHour()->format($timezoneAware ? 'Y-m-d H:i:sP' : 'Y-m-d H:i:s'), '1.00', 'future']); + $this->artisan('business-mirror:sync-tables')->assertSuccessful(); + $this->assertSame(3, DB::connection('business_mirror')->table($dataset->targetTable())->count()); + $this->assertEquals(99.5, DB::connection('business_mirror')->table($dataset->targetTable())->where('id', 1)->value('amount')); + $pdo = UpstreamConnection::get($source); + $row->update(['enabled' => false]); + $page = app(BusinessMirrorDataReader::class)->paginate($dataset->key(), [], 2, 2); + $this->assertSame(3, $page['total']); + $this->assertSame(3, (int) $page['rows'][0]['id']); + $this->assertArrayNotHasKey('_mirror_key', $page['rows'][0]); + $filtered = app(BusinessMirrorDataReader::class)->paginate($dataset->key(), ['label' => 'first']); + $this->assertSame(1, $filtered['total']); + if ($driver === 'pgsql') { + $this->assertSame('on', $pdo->query('SHOW default_transaction_read_only')->fetchColumn()); + $this->assertTrue((bool) $pdo->query('SELECT ssl FROM pg_stat_ssl WHERE pid=pg_backend_pid()')->fetchColumn()); + } else { + $this->assertNotEmpty($pdo->query("SHOW SESSION STATUS LIKE 'Ssl_cipher'")->fetch(PDO::FETCH_NUM)[1]); + } + try { + $pdo->exec("INSERT INTO pipeline_records (id) VALUES (99)"); + $this->fail('The native upstream connection must reject writes.'); + } catch (PDOException $e) { + $this->assertStringContainsString('read', strtolower($e->getMessage())); + } + } + + public static function nativeDrivers(): array + { + return [['pgsql'], ['mysql'], ['pgsql', true], ['mysql', false, 'caiwu-pipeline-mariadb']]; + } +} diff --git a/tests/Unit/EcommerceFinance/UpstreamDatabaseProbeTest.php b/tests/Unit/EcommerceFinance/UpstreamDatabaseProbeTest.php index 0f66222e..3ff1304b 100644 --- a/tests/Unit/EcommerceFinance/UpstreamDatabaseProbeTest.php +++ b/tests/Unit/EcommerceFinance/UpstreamDatabaseProbeTest.php @@ -21,7 +21,7 @@ class UpstreamDatabaseProbeTest extends TestCase 'username' => 'reader', 'password' => 'fixture-secret', 'connect_timeout' => 5, 'query_timeout' => 120]; } - public function test_postgres_uses_tls_read_only_timeout_and_one_select(): void + public function test_postgres_probe_uses_the_shared_connector_and_one_select(): void { $statement = Mockery::mock(PDOStatement::class); $statement->shouldReceive('fetchColumn')->once()->andReturn(1); @@ -29,16 +29,7 @@ class UpstreamDatabaseProbeTest extends TestCase $pdo = Mockery::mock(PDO::class); $pdo->shouldReceive('query')->once()->with('SELECT 1')->andReturn($statement); $probe = Mockery::mock(UpstreamDatabaseProbe::class)->makePartial()->shouldAllowMockingProtectedMethods(); - $probe->shouldReceive('connect')->once()->withArgs(function ($dsn, $user, $password) { - $this->assertStringContainsString('sslmode=require', $dsn); - $this->assertStringContainsString('default_transaction_read_only=on', $dsn); - $this->assertStringContainsString('statement_timeout=15000', $dsn); - $this->assertStringContainsString("dbname='test\\' database'", $dsn); - $this->assertStringContainsString('connect_timeout=5', $dsn); - $this->assertStringNotContainsString('fixture-secret', $dsn); - - return $user === 'reader' && $password === 'fixture-secret'; - })->andReturn($pdo); + $probe->shouldReceive('connect')->once()->with('pgsql', $this->connection())->andReturn($pdo); $probe->probe('pgsql', $this->connection(), '测试库'); } @@ -50,7 +41,7 @@ class UpstreamDatabaseProbeTest extends TestCase $probe->probe('pgsql', $this->connection(), '测试库'); $this->fail('Expected sanitized connection error'); } catch (ServiceUnavailableHttpException $exception) { - $this->assertStringContainsString('PostgreSQL 连接失败', $exception->getMessage()); + $this->assertStringContainsString('上游只读连接失败', $exception->getMessage()); $this->assertStringNotContainsString('fixture-secret', $exception->getMessage()); $this->assertNull($exception->getPrevious()); } diff --git a/tests/Unit/Utilities/SystemDatabasePersistenceTest.php b/tests/Unit/Utilities/SystemDatabasePersistenceTest.php index f68754c5..7d90e8a6 100644 --- a/tests/Unit/Utilities/SystemDatabasePersistenceTest.php +++ b/tests/Unit/Utilities/SystemDatabasePersistenceTest.php @@ -9,6 +9,42 @@ use Tests\TestCase; class SystemDatabasePersistenceTest extends TestCase { + public function test_switch_replaces_all_legacy_connection_routes_and_duplicate_keys(): void + { + $environmentPath = storage_path('framework/testing-legacy-routes.env'); + $original = "DB_HOST=mysql\nDB_HOST_READ=old-read\nDB_HOST_WRITE=old-write\n" + . "DATABASE_URL=mysql://fixture:fixture@old-url/old_database\nDB_SOCKET=/tmp/old.sock\n" + . "DB_HOST=old-duplicate\nSYSTEM_DATABASE_MODE=bootstrap\nUNRELATED_OPTION=keep\n"; + file_put_contents($environmentPath, $original); + $configuration = new class($environmentPath) extends SystemDatabaseConfiguration { + public array $verified = []; + protected function runFreshVerification(string $command): void + { + $this->verified = $this->freshVerificationEnvironment(); + } + protected function clearConfigurationCache(): void {} + }; + try { + $backup = $configuration->persist([ + 'host' => 'shared.example.test', 'port' => 8443, 'database' => 'shared_system', + 'username' => 'fixture_user', 'password' => 'fixture_password', 'prefix' => '', + 'tls_required' => true, 'tls_verify_server_cert' => false, + ]); + $saved = \Dotenv\Dotenv::parse(file_get_contents($environmentPath)); + foreach (['DB_HOST', 'DB_HOST_READ', 'DB_HOST_WRITE'] as $key) { + $this->assertSame('shared.example.test', $saved[$key]); + $this->assertSame('shared.example.test', $configuration->verified[$key]); + } + $this->assertSame('', $saved['DATABASE_URL']); + $this->assertSame('', $saved['DB_SOCKET']); + $this->assertSame('keep', $saved['UNRELATED_OPTION']); + $this->assertSame($original, file_get_contents($backup)); + } finally { + @unlink($environmentPath); + if (isset($backup)) @unlink($backup); + } + } + public function test_failed_fresh_process_verification_restores_original_environment(): void { $environmentPath = storage_path('framework/testing-system-database.env');