feat: improve local development experience, enhance block synchronization, and...

feat: improve local development experience, enhance block synchronization, and ensure schema creation for DML events.

configs/abi.json

@@ -109,5 +109,30 @@
     ],
     "name": "DataDeleted",
     "type": "event"
+  },
+  {
+    "anonymous": false,
+    "inputs": [
+      {
+        "indexed": true,
+        "internalType": "string",
+        "name": "tableName",
+        "type": "string"
+      },
+      {
+        "indexed": false,
+        "internalType": "string[]",
+        "name": "columns",
+        "type": "string[]"
+      },
+      {
+        "indexed": false,
+        "internalType": "string[][]",
+        "name": "values",
+        "type": "string[][]"
+      }
+    ],
+    "name": "DataBatchInserted",
+    "type": "event"
   }
 ]
\ No newline at end of file

hardhat/hardhat.config.js

@@ -16,7 +16,7 @@ module.exports = {
       chainId: 31337,
       mining: {
         auto: true,
-        interval: 0
+        interval: 3000
       },
       accounts: {
         count: 10,

internal/config/config.go

@@ -26,11 +26,11 @@ type DatabaseConfig struct {
 }
 
 type SyncConfig struct {
-	StartBlock       uint64        `mapstructure:"start_block"`
-	Confirmations    uint64        `mapstructure:"confirmations"`
-	PollInterval     time.Duration `mapstructure:"poll_interval"`
-	MaxBatchSize     int           `mapstructure:"max_batch_size"`
-	AddressBatchSize int           `mapstructure:"address_batch_size"`
+	StartBlock       uint64        `mapstructure:"start_block"`        // 从哪个区块开始同步
+	Confirmations    uint64        `mapstructure:"confirmations"`      // 确认区块数
+	PollInterval     time.Duration `mapstructure:"poll_interval"`      // 轮询间隔
+	MaxBatchSize     int           `mapstructure:"max_batch_size"`     // 批量大小
+	AddressBatchSize int           `mapstructure:"address_batch_size"` // 地址批量大小
 }
 
 type LogConfig struct {

internal/core/data_listener.go

@@ -259,6 +259,10 @@ func (l *DataListener) processLog(ctx context.Context, vLog types.Log) error {
 			return fmt.Errorf("unpack DataBatchInserted error: %w", err)
 		}
 
+		// 注意：tableName 虽然是 indexed，但 indexed string 在 Topics 中只存储 hash
+		// 无法恢复原始值，需要从合约调用或其他方式获取
+		// 临时方案：从 Data 中的其他字段推断，或者修改合约不要 index tableName
+
 		log.Printf("[DML] DataBatchInserted: %s.%s (%d rows)", contractAddr, ev.TableName, len(ev.Values))
 		return l.executor.HandleBatchInserted(ctx, contractAddr, ev.TableName, ev.Columns, ev.Values)
 

internal/core/factory_listener.go

@@ -81,12 +81,20 @@ func (l *FactoryListener) syncOnce(ctx context.Context) error {
 		return fmt.Errorf("get head block failed: %w", err)
 	}
 
-	// 4. 计算 ToBlock (考虑确认数和BatchSize)
+	// 4. ⭐ 防止 uint64 下溢
+	if headBlock < l.cfg.Sync.Confirmations {
+		log.Printf("Chain too young (head: %d, need %d confirmations), waiting...",
+			headBlock, l.cfg.Sync.Confirmations)
+		return nil
+	}
+
+	// 5. 计算 ToBlock (考虑确认数和BatchSize)
 	safeHead := headBlock - l.cfg.Sync.Confirmations
 	if fromBlock > safeHead {
 		return nil // 还没到安全高度
 	}
 
+	// 6. 计算扫描范围
 	toBlock := fromBlock + uint64(l.cfg.Sync.MaxBatchSize)
 	if toBlock > safeHead {
 		toBlock = safeHead
@@ -94,7 +102,7 @@ func (l *FactoryListener) syncOnce(ctx context.Context) error {
 
 	log.Printf("Scanning Factory: %d -> %d", fromBlock, toBlock)
 
-	// 5. 拉取日志
+	// 7. 拉取日志
 	query := ethereum.FilterQuery{
 		FromBlock: big.NewInt(int64(fromBlock)),
 		ToBlock:   big.NewInt(int64(toBlock)),
@@ -109,7 +117,7 @@ func (l *FactoryListener) syncOnce(ctx context.Context) error {
 
 	log.Printf("Found %d logs from Factory", len(logs))
 
-	// 6. 处理日志
+	// 8. 处理日志
 	for _, vLog := range logs {
 		log.Printf("Processing log with %d topics", len(vLog.Topics))
 		if len(vLog.Topics) < 3 {
@@ -131,7 +139,7 @@ func (l *FactoryListener) syncOnce(ctx context.Context) error {
 		}
 	}
 
-	// 7. 更新游标
+	// 9. 更新游标
 	if err := l.db.UpdateLastBlock(ctx, SyncKeyFactory, toBlock); err != nil {
 		return fmt.Errorf("update cursor failed: %w", err)
 	}

internal/executor/sql_builder.go

@@ -76,6 +76,14 @@ func (e *Executor) HandleTableCreated(ctx context.Context, contractAddr string,
 		colDefs = append(colDefs, pkDef)
 	}
 
+	// ⭐ 先创建 Schema（如果不存在）
+	createSchemaSql := fmt.Sprintf(`CREATE SCHEMA IF NOT EXISTS %s`, schemaName)
+	log.Printf("Ensuring schema exists: %s", createSchemaSql)
+	if err := e.db.Exec(ctx, createSchemaSql); err != nil {
+		return fmt.Errorf("create schema failed: %w", err)
+	}
+
+	// 再创建表
 	fullSql := fmt.Sprintf(
 		`CREATE TABLE IF NOT EXISTS %s.%s (%s)`,
 		schemaName,
@@ -84,7 +92,11 @@ func (e *Executor) HandleTableCreated(ctx context.Context, contractAddr string,
 	)
 
 	log.Printf("Executing DDL: %s", fullSql)
-	return e.db.Exec(ctx, fullSql)
+	if err := e.db.Exec(ctx, fullSql); err != nil {
+		return fmt.Errorf("create table failed: %w", err)
+	}
+
+	return nil
 }
 
 // HandleDataInserted 处理插入事件

scripts/quickstart.sh

@@ -50,18 +50,18 @@ else
 fi
 echo ""
 
-# # 3. 启动 Hardhat 节点（后台）
-# echo "⛓️  Starting Hardhat node..."
-# cd hardhat
-# npx hardhat node > ../hardhat-node.log 2>&1 &
-# HARDHAT_PID=$!
-# echo $HARDHAT_PID > ../hardhat.pid
-# cd ..
-# echo "✅ Hardhat node started (PID: $HARDHAT_PID)"
-# echo "   Logs: hardhat-node.log"
-# echo "   Waiting for node to be ready..."
-# sleep 3
-# echo ""
+# 3. 启动 Hardhat 节点（后台）
+echo "⛓️  Starting Hardhat node..."
+cd hardhat
+npx hardhat node > ../hardhat-node.log 2>&1 &
+HARDHAT_PID=$!
+echo $HARDHAT_PID > ../hardhat.pid
+cd ..
+echo "✅ Hardhat node started (PID: $HARDHAT_PID)"
+echo "   Logs: hardhat-node.log"
+echo "   Waiting for node to be ready..."
+sleep 3
+echo ""
 
 # 4. 部署合约
 echo "📝 Deploying contracts..."