feat(kline): 首次加载时让最新K线居中显示

- 新增 _scrollToCenter() 方法，计算让最新K线居中的滚动位置
- 初始化时调用 _scrollToCenter() 替代 _scrollToEnd()
- 如果K线总宽度小于屏幕宽度，不滚动，从左开始显示
- 保留 _scrollToEnd() 方法供刷新按钮等场景使用

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

backend/services/planting-service/prisma/migrations/20260120000000_add_debezium_heartbeat/migration.sql

@@ -0,0 +1,14 @@
+-- CreateTable: Debezium Heartbeat
+-- 用于 CDC replication slot 的 WAL 位置推进
+-- 由 Debezium heartbeat.action.query 自动更新
+-- 防止因业务表长期无写入导致 WAL 堆积
+
+CREATE TABLE "debezium_heartbeat" (
+    "id" INTEGER NOT NULL DEFAULT 1,
+    "ts" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "debezium_heartbeat_pkey" PRIMARY KEY ("id")
+);
+
+-- 插入初始记录
+INSERT INTO "debezium_heartbeat" ("id", "ts") VALUES (1, NOW());

backend/services/planting-service/prisma/schema.prisma

@@ -377,3 +377,16 @@ model ContractSigningTask {
   @@index([status, expiresAt])
   @@map("contract_signing_tasks")
 }
+
+// ============================================
+// Debezium 心跳表
+// 用于 CDC replication slot 的 WAL 位置推进
+// 由 Debezium heartbeat.action.query 自动更新
+// 防止因业务表长期无写入导致 WAL 堆积
+// ============================================
+model DebeziumHeartbeat {
+  id Int      @id @default(1)
+  ts DateTime @default(now()) @updatedAt @map("ts")
+
+  @@map("debezium_heartbeat")
+}

backend/services/referral-service/prisma/migrations/20260120000000_add_debezium_heartbeat/migration.sql

@@ -0,0 +1,14 @@
+-- CreateTable: Debezium Heartbeat
+-- 用于 CDC replication slot 的 WAL 位置推进
+-- 由 Debezium heartbeat.action.query 自动更新
+-- 防止因业务表长期无写入导致 WAL 堆积
+
+CREATE TABLE "debezium_heartbeat" (
+    "id" INTEGER NOT NULL DEFAULT 1,
+    "ts" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "debezium_heartbeat_pkey" PRIMARY KEY ("id")
+);
+
+-- 插入初始记录
+INSERT INTO "debezium_heartbeat" ("id", "ts") VALUES (1, NOW());

backend/services/referral-service/prisma/schema.prisma

@@ -199,3 +199,16 @@ model ReferralEvent {
   @@index([userId], name: "idx_event_user")
   @@index([occurredAt], name: "idx_event_occurred")
 }
+
+// ============================================
+// Debezium 心跳表
+// 用于 CDC replication slot 的 WAL 位置推进
+// 由 Debezium heartbeat.action.query 自动更新
+// 防止因业务表长期无写入导致 WAL 堆积
+// ============================================
+model DebeziumHeartbeat {
+  id Int      @id @default(1)
+  ts DateTime @default(now()) @updatedAt @map("ts")
+
+  @@map("debezium_heartbeat")
+}

backend/services/scripts/debezium/planting-connector.json

@@ -2,26 +2,40 @@
   "name": "planting-postgres-connector",
   "config": {
     "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
+    "tasks.max": "1",
+
     "database.hostname": "postgres",
     "database.port": "5432",
-    "database.user": "debezium",
+    "database.user": "${POSTGRES_USER:-rwa_user}",
-    "database.password": "debezium_password",
+    "database.password": "${POSTGRES_PASSWORD:-rwa_secure_password}",
     "database.dbname": "rwa_planting",
-    "database.server.name": "planting",
+
     "topic.prefix": "cdc.planting",
+
+    "table.include.list": "public.planting_orders,public.planting_positions,public.contract_signing_tasks,public.fund_allocations,public.debezium_heartbeat",
+
     "plugin.name": "pgoutput",
-    "publication.name": "planting_cdc_publication",
+    "publication.name": "debezium_planting_publication",
-    "slot.name": "planting_cdc_slot",
+    "publication.autocreate.mode": "filtered",
-    "table.include.list": "public.planting_orders,public.planting_positions,public.contract_signing_tasks,public.fund_allocations",
+
-    "transforms": "unwrap",
+    "slot.name": "debezium_planting_slot",
-    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
+
-    "transforms.unwrap.add.fields": "op,table,source.ts_ms",
-    "transforms.unwrap.delete.handling.mode": "rewrite",
     "key.converter": "org.apache.kafka.connect.json.JsonConverter",
     "key.converter.schemas.enable": "false",
     "value.converter": "org.apache.kafka.connect.json.JsonConverter",
     "value.converter.schemas.enable": "false",
+
+    "transforms": "unwrap",
+    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
+    "transforms.unwrap.drop.tombstones": "true",
+    "transforms.unwrap.delete.handling.mode": "rewrite",
+    "transforms.unwrap.add.fields": "op,table,source.ts_ms",
+
+    "heartbeat.interval.ms": "60000",
+    "heartbeat.action.query": "UPDATE debezium_heartbeat SET ts = NOW() WHERE id = 1",
+
     "snapshot.mode": "initial",
+
     "decimal.handling.mode": "string",
     "time.precision.mode": "connect"
   }

backend/services/scripts/debezium/referral-connector.json

@@ -12,7 +12,7 @@
 
     "topic.prefix": "cdc.referral",
 
-    "table.include.list": "public.referral_relationships",
+    "table.include.list": "public.referral_relationships,public.debezium_heartbeat",
 
     "plugin.name": "pgoutput",
     "publication.name": "debezium_referral_publication",
@@ -31,7 +31,8 @@
     "transforms.unwrap.delete.handling.mode": "rewrite",
     "transforms.unwrap.add.fields": "op,table,source.ts_ms",
 
-    "heartbeat.interval.ms": "10000",
+    "heartbeat.interval.ms": "60000",
+    "heartbeat.action.query": "UPDATE debezium_heartbeat SET ts = NOW() WHERE id = 1",
 
     "snapshot.mode": "initial",
 

backend/services/scripts/debezium/wallet-connector.json

@@ -2,26 +2,40 @@
   "name": "wallet-postgres-connector",
   "config": {
     "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
+    "tasks.max": "1",
+
     "database.hostname": "postgres",
     "database.port": "5432",
-    "database.user": "debezium",
+    "database.user": "${POSTGRES_USER:-rwa_user}",
-    "database.password": "debezium_password",
+    "database.password": "${POSTGRES_PASSWORD:-rwa_secure_password}",
     "database.dbname": "rwa_wallet",
-    "database.server.name": "wallet",
+
     "topic.prefix": "cdc.wallet",
+
+    "table.include.list": "public.wallet_accounts,public.withdrawal_orders,public.fiat_withdrawal_orders,public.wallet_ledger_entries,public.debezium_heartbeat",
+
     "plugin.name": "pgoutput",
-    "publication.name": "wallet_cdc_publication",
+    "publication.name": "debezium_wallet_publication",
-    "slot.name": "wallet_cdc_slot",
+    "publication.autocreate.mode": "filtered",
-    "table.include.list": "public.wallet_accounts,public.withdrawal_orders,public.fiat_withdrawal_orders,public.wallet_ledger_entries",
+
-    "transforms": "unwrap",
+    "slot.name": "debezium_wallet_slot",
-    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
+
-    "transforms.unwrap.add.fields": "op,table,source.ts_ms",
-    "transforms.unwrap.delete.handling.mode": "rewrite",
     "key.converter": "org.apache.kafka.connect.json.JsonConverter",
     "key.converter.schemas.enable": "false",
     "value.converter": "org.apache.kafka.connect.json.JsonConverter",
     "value.converter.schemas.enable": "false",
+
+    "transforms": "unwrap",
+    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
+    "transforms.unwrap.drop.tombstones": "true",
+    "transforms.unwrap.delete.handling.mode": "rewrite",
+    "transforms.unwrap.add.fields": "op,table,source.ts_ms",
+
+    "heartbeat.interval.ms": "60000",
+    "heartbeat.action.query": "UPDATE debezium_heartbeat SET ts = NOW() WHERE id = 1",
+
     "snapshot.mode": "initial",
+
     "decimal.handling.mode": "string",
     "time.precision.mode": "connect"
   }

backend/services/wallet-service/prisma/migrations/20260120000000_add_debezium_heartbeat/migration.sql

@@ -0,0 +1,14 @@
+-- CreateTable: Debezium Heartbeat
+-- 用于 CDC replication slot 的 WAL 位置推进
+-- 由 Debezium heartbeat.action.query 自动更新
+-- 防止因业务表长期无写入导致 WAL 堆积
+
+CREATE TABLE "debezium_heartbeat" (
+    "id" INTEGER NOT NULL DEFAULT 1,
+    "ts" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "debezium_heartbeat_pkey" PRIMARY KEY ("id")
+);
+
+-- 插入初始记录
+INSERT INTO "debezium_heartbeat" ("id", "ts") VALUES (1, NOW());

backend/services/wallet-service/prisma/schema.prisma

@@ -498,3 +498,16 @@ model OutboxEvent {
   @@index([topic])
   @@map("outbox_events")
 }
+
+// ============================================
+// Debezium 心跳表
+// 用于 CDC replication slot 的 WAL 位置推进
+// 由 Debezium heartbeat.action.query 自动更新
+// 防止因业务表长期无写入导致 WAL 堆积
+// ============================================
+model DebeziumHeartbeat {
+  id Int      @id @default(1)
+  ts DateTime @default(now()) @updatedAt @map("ts")
+
+  @@map("debezium_heartbeat")
+}

frontend/mining-app/lib/presentation/widgets/kline_chart/kline_chart_widget.dart

@@ -83,13 +83,38 @@ class _KlineChartWidgetState extends State<KlineChartWidget> {
     _candleWidth = idealWidth.clamp(_minCandleWidth, _maxCandleWidth);
     _prevCandleWidth = _candleWidth;
 
-    // 滚动到最右边（显示最新数据）
+    // 首次加载时让最新 K 线居中显示
-    _scrollToEnd();
+    _scrollToCenter();
   }
 
+  /// 滚动使最新 K 线居中显示
+  ///
+  /// 计算逻辑：
+  /// - 如果 K 线总宽度小于屏幕宽度，不滚动，从左开始显示
+  /// - 否则计算让最新 K 线中心位于屏幕中央的滚动位置
+  void _scrollToCenter() {
+    if (widget.klines.isEmpty || _chartWidth == 0) return;
+
+    final totalWidth = widget.klines.length * _candleWidth;
+
+    if (totalWidth <= _chartWidth) {
+      // K 线总宽度小于等于屏幕宽度，不需要滚动，从左开始显示
+      _scrollX = 0;
+    } else {
+      // 计算让最新 K 线居中的滚动位置
+      // 最新 K 线的中心位置 = (K线数量 - 0.5) * 单根K线宽度
+      final lastKlineCenter = (widget.klines.length - 0.5) * _candleWidth;
+      // 目标滚动位置 = 最新K线中心 - 屏幕宽度的一半
+      final targetScroll = lastKlineCenter - _chartWidth / 2;
+      // 限制在有效滚动范围内
+      final maxScroll = totalWidth - _chartWidth;
+      _scrollX = targetScroll.clamp(0.0, maxScroll);
+    }
+  }
+
+  /// 滚动到最右边（显示最新数据在右侧）
   void _scrollToEnd() {
     if (widget.klines.isEmpty || _chartWidth == 0) return;
-    // 滚动到最右边（显示最新数据）
     final totalWidth = widget.klines.length * _candleWidth;
     final maxScroll = math.max(0.0, totalWidth - _chartWidth);
     _scrollX = maxScroll;