## 任务占用与边界 - 任务名称:全项目上游业务镜像目录与通用同步框架 - 实际开发者:WuYongJie - 目标结果:登记当前代码实际使用的上游原表,提供低压、可断点的首次全量、增量重叠和全量校准框架。 - 实际修改:13 个文件;新增通用数据集目录、动态镜像建表、同步状态、同步命令、生产低压轮询、测试和团队清单。 - 将写入的公共契约:`business_mirror_table_states`、`business-mirror:sync-tables`、`BusinessMirrorDatasetRegistry`、`BusinessMirrorSchemaManager`。 - 仅复用、明确不修改:`UpstreamReadOnlyQueryFactory`、`upstream_schema_columns`、既有 `product_price_mirror_rows`。 - 明确不包含:本 PR 不切换其余业务页面 Repository,不执行生产迁移或首次全量,不提交真实业务数据或凭据。 ## 合并前自审 - [x] 已重新拉取最新 `origin/main`,分支仅领先 1 个提交 - [x] `git merge-tree --write-tree origin/main HEAD` 无冲突 - [x] 完整 diff 共 13 个文件,未发现越界文件或敏感内容 - [x] 生产共享数据字典核对 38 个数据集,缺失对象/字段为 0 - [x] 临时 MariaDB 空库迁移实测成功,验证库已删除 - [x] 上游异常空结果不会清空已有镜像 - [x] 定时任务只轮转已完成首次全量的数据集,不会自动扫描未初始化大表 ## 测试 - `BusinessMirrorSynchronizerTest` 等相关测试:14 tests / 65 assertions 全通过 - 新同步器专项:5 tests / 25 assertions 全通过 - 新增 PHP 文件语法检查全部通过 - 两份生产激活脚本 `bash -n` 通过 - `git diff --check` 通过 ## 风险与回滚 - 包含新增迁移,但发布脚本遇到迁移仍会要求独立高风险授权;合并本身不会执行生产迁移。 - 首次全量必须逐数据集显式执行并做数量/翻页/筛选验收,特别是约 280 万行的大表不得随发布自动全扫。 - 回滚代码可回退本合并提交;生产迁移执行前必须备份 `caiwu_system_config` 与 `caiwu_business_mirror`。 Reviewed-on: http://git.1080p.hk/WuYongJie/CaiWuXiTong/pulls/53
146 lines
5.7 KiB
PHP
146 lines
5.7 KiB
PHP
<?php
|
|
|
|
namespace Modules\EcommerceFinance\Services;
|
|
|
|
use Illuminate\Database\Schema\Blueprint;
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Facades\Schema;
|
|
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
|
|
use RuntimeException;
|
|
|
|
class BusinessMirrorSchemaManager
|
|
{
|
|
/** @return list<array<string,mixed>> */
|
|
public function ensure(BusinessMirrorDataset $dataset): array
|
|
{
|
|
$columns = $this->columns($dataset);
|
|
$connection = (string) config('upstream_mirrors.connection', 'business_mirror');
|
|
$schema = Schema::connection($connection);
|
|
|
|
if (! $schema->hasTable($dataset->targetTable())) {
|
|
$schema->create($dataset->targetTable(), function (Blueprint $table) use ($columns, $dataset): void {
|
|
$table->char('_mirror_key', 64)->primary();
|
|
|
|
foreach ($columns as $column) {
|
|
$this->addColumn($table, $column);
|
|
}
|
|
|
|
$table->uuid('_mirror_generation')->nullable()->index();
|
|
$table->dateTime('_mirror_synced_at', 6)->index();
|
|
|
|
$watermark = $dataset->watermarkColumn();
|
|
if ($watermark !== null && $this->isIndexable($columns, $watermark)) {
|
|
$table->index($watermark, $this->indexName($dataset->targetTable(), 'watermark'));
|
|
}
|
|
});
|
|
|
|
return $columns;
|
|
}
|
|
|
|
$existing = array_map('strtolower', $schema->getColumnListing($dataset->targetTable()));
|
|
$missing = array_values(array_filter(
|
|
$columns,
|
|
fn (array $column): bool => ! in_array(strtolower((string) $column['column_name']), $existing, true),
|
|
));
|
|
|
|
if ($missing !== []) {
|
|
$schema->table($dataset->targetTable(), function (Blueprint $table) use ($missing): void {
|
|
foreach ($missing as $column) {
|
|
$this->addColumn($table, $column);
|
|
}
|
|
});
|
|
}
|
|
|
|
return $columns;
|
|
}
|
|
|
|
/** @return list<array<string,mixed>> */
|
|
public function columns(BusinessMirrorDataset $dataset): array
|
|
{
|
|
$rows = DB::table('upstream_schema_columns')
|
|
->where('source_key', $dataset->sourceKey())
|
|
->whereRaw('LOWER(schema_name) = ?', [strtolower($dataset->sourceSchema())])
|
|
->whereRaw('LOWER(object_name) = ?', [strtolower($dataset->sourceObject())])
|
|
->orderBy('ordinal_position')
|
|
->get()
|
|
->map(fn (object $row): array => (array) $row)
|
|
->all();
|
|
|
|
if ($rows === []) {
|
|
throw new RuntimeException('business_mirror_schema_catalog_missing:' . $dataset->key());
|
|
}
|
|
|
|
$names = array_map(fn (array $row): string => strtolower((string) $row['column_name']), $rows);
|
|
if (count($names) !== count(array_unique($names))) {
|
|
throw new RuntimeException('business_mirror_duplicate_column:' . $dataset->key());
|
|
}
|
|
|
|
foreach ($rows as $row) {
|
|
$name = (string) $row['column_name'];
|
|
if (strlen($name) > 64 || str_starts_with(strtolower($name), '_mirror_')) {
|
|
throw new RuntimeException('business_mirror_unsupported_column:' . $dataset->key());
|
|
}
|
|
}
|
|
|
|
foreach ([...$dataset->keyColumns(), $dataset->watermarkColumn()] as $required) {
|
|
if ($required !== null && ! in_array(strtolower($required), $names, true)) {
|
|
throw new RuntimeException('business_mirror_required_column_missing:' . $dataset->key());
|
|
}
|
|
}
|
|
|
|
return $rows;
|
|
}
|
|
|
|
private function addColumn(Blueprint $table, array $column): void
|
|
{
|
|
$name = (string) $column['column_name'];
|
|
$type = strtolower((string) $column['data_type']);
|
|
$length = (int) ($column['max_length'] ?? 0);
|
|
$precision = min(max((int) ($column['numeric_precision'] ?? 20), 1), 65);
|
|
$scale = min(max((int) ($column['numeric_scale'] ?? 0), 0), min($precision, 30));
|
|
|
|
$definition = match ($type) {
|
|
'bit' => $table->boolean($name),
|
|
'tinyint' => $table->unsignedTinyInteger($name),
|
|
'smallint' => $table->smallInteger($name),
|
|
'int' => $table->integer($name),
|
|
'bigint' => $table->bigInteger($name),
|
|
'decimal', 'numeric', 'money', 'smallmoney' => $table->decimal($name, $precision, $scale),
|
|
'float', 'real' => $table->double($name),
|
|
'date' => $table->date($name),
|
|
'datetime', 'datetime2', 'smalldatetime' => $table->dateTime($name, 6),
|
|
'time' => $table->time($name, 6),
|
|
'uniqueidentifier' => $table->char($name, 36),
|
|
'binary', 'varbinary', 'image', 'timestamp', 'rowversion' => $table->binary($name),
|
|
'char', 'nchar', 'varchar', 'nvarchar' => $length > 0 && $length <= 1024
|
|
? $table->string($name, $length)
|
|
: $table->longText($name),
|
|
default => $table->longText($name),
|
|
};
|
|
|
|
$definition->nullable();
|
|
}
|
|
|
|
private function isIndexable(array $columns, string $name): bool
|
|
{
|
|
foreach ($columns as $column) {
|
|
if (strcasecmp((string) $column['column_name'], $name) !== 0) {
|
|
continue;
|
|
}
|
|
|
|
$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 <= 191));
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
private function indexName(string $table, string $suffix): string
|
|
{
|
|
return substr($table . '_' . $suffix, 0, 64);
|
|
}
|
|
}
|