feat: migrate bridgesync claim data into claimsync DB on startup (#1554)

## 🔄 Changes Summary

- Add `ImportDataFromBridgesyncer` which copies `block`, `claim`,
`set_claim` and `unset_claim` rows from a bridgesync SQLite DB into the
claimsync DB. Handles old schema versions (missing `block.hash`,
`claim.tx_hash`, `claim.block_timestamp`, `claim.type`) gracefully via
column detection.
- Add `ImportKeyValueFromBridgesyncer` which copies the single
`key_value` row, replacing the `owner` field with the claimsync syncer
ID.
- Both functions are idempotent (`INSERT OR IGNORE`) and no-ops when the
source DB has no relevant data (claimDB is not created unless data
exists).
- A migration guard validates that `bridgesync0012` (the last migration
touching the imported tables) has been applied before proceeding.
- In `cmd/run.go`, wire the migration via
`runImportFromBridgeSyncerIfNeeded` at the call site, before both the
bridge and claim syncers start:
  - L1: triggered when `BRIDGE` or `L1BRIDGESYNC` are active
- L2: triggered when any of `BRIDGE`, `L2BRIDGESYNC`, `L2CLAIMSYNC`,
`AGGSENDER`, `AGGSENDERVALIDATOR` or `AGGCHAINPROOFGEN` are active
- It removes the old tables from bridgesyncer

Cases:
| # | Condition | Shuld Migrate | Outcome 
| -- | ---------------|----------------------|----------------
| 0 | Componentsdoesn't imply run Bridge syncer neither Claim sycner | ❌
| Skip
| 1 | BridgeDB doesn't exist | ❌ | Skip 
| 2 | ClaimDB exists | ❌| Skip. ClaimDB has data so no migration
required
| 3 | BridgeDB doesn't have claims tables or they are empty |❌ | Skip.
No data to migrate
| 4 | BridgeDB migrations below `bridgesync0012` | ⚠️| Abort. It's not
possible to migrate it, update to an intermediate version
| 5 | All conditions met | ✅| Migration runs 

## ⚠️ Breaking Changes
None

## 📋 Config Updates
None

## ✅ Testing
- Unit tests
- Manually upgrade v0.8.2 to this version

## 🐞 Issues
- Closes #1546

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: joanestebanr

bridgesync/backfill_tx_sender_test.go

@@ -79,23 +79,6 @@ func TestBackfillTxnSender(t *testing.T) {
 	require.NoError(t, meddler.Insert(tx, bridgeTableName, newTestBridge(1, 0,
 		"0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890")))
 
-	// Insert test claim record
-	_, err = tx.Exec(`
-		INSERT INTO claim (
-			block_num, block_pos, global_index, origin_network, origin_address,
-			destination_address, amount, proof_local_exit_root, proof_rollup_exit_root,
-			mainnet_exit_root, rollup_exit_root, global_exit_root, destination_network,
-			metadata, is_message, block_timestamp, tx_hash
-		) VALUES (
-			1, 1, '1', 1, '0x1234567890123456789012345678901234567890',
-			'0x0987654321098765432109876543210987654321', '1000000000000000000',
-			'', '', '', '', '', 2,
-			'', false, 1234567890,
-			'0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890'
-		)
-	`)
-	require.NoError(t, err)
-
 	err = tx.Commit()
 	require.NoError(t, err)
 

bridgesync/migrations/bridgesync0013.sql

@@ -7,6 +7,7 @@ DROP TABLE _bridge_to_address_backup;
 
 -- +migrate Up
 CREATE TABLE _bridge_to_address_backup AS SELECT block_num, block_pos, to_address FROM bridge;
+CREATE INDEX _bridge_to_address_backup_idx ON _bridge_to_address_backup(block_num, block_pos);
 ALTER TABLE bridge DROP COLUMN to_address;
 ALTER TABLE bridge ADD COLUMN to_address VARCHAR DEFAULT '';
 -- Only set to_address for rows != null

bridgesync/migrations/bridgesync0015.sql

@@ -0,0 +1,50 @@
+-- +migrate Down
+CREATE TABLE IF NOT EXISTS claim (
+	block_num               INTEGER NOT NULL REFERENCES block(num) ON DELETE CASCADE,
+	block_pos               INTEGER NOT NULL,
+	global_index            TEXT NOT NULL,
+	origin_network          INTEGER NOT NULL,
+	origin_address          VARCHAR NOT NULL,
+	destination_address     VARCHAR NOT NULL,
+	amount                  TEXT NOT NULL,
+	proof_local_exit_root   VARCHAR,
+	proof_rollup_exit_root  VARCHAR,
+	mainnet_exit_root       VARCHAR,
+	rollup_exit_root        VARCHAR,
+	global_exit_root        VARCHAR,
+	destination_network     INTEGER NOT NULL,
+	metadata                BLOB,
+	is_message              BOOLEAN,
+	tx_hash                 VARCHAR,
+	block_timestamp         INTEGER,
+	type                    TEXT NOT NULL DEFAULT '',
+	PRIMARY KEY (block_num, block_pos)
+);
+
+CREATE INDEX IF NOT EXISTS idx_claim_block_num_block_pos_desc ON claim (block_num DESC, block_pos DESC);
+CREATE INDEX IF NOT EXISTS idx_claim_block_num_block_pos_asc ON claim (block_num ASC, block_pos ASC);
+CREATE INDEX IF NOT EXISTS idx_claim_type_block ON claim (type, block_num);
+
+CREATE TABLE IF NOT EXISTS unset_claim (
+	block_num                     INTEGER NOT NULL REFERENCES block(num) ON DELETE CASCADE,
+	block_pos                     INTEGER NOT NULL,
+	tx_hash                       VARCHAR NOT NULL,
+	global_index                  TEXT NOT NULL,
+	unset_global_index_hash_chain VARCHAR NOT NULL,
+	created_at                    INTEGER NOT NULL DEFAULT (strftime('%s', 'now')),
+	PRIMARY KEY (block_num, block_pos)
+);
+
+CREATE TABLE IF NOT EXISTS set_claim (
+	block_num    INTEGER NOT NULL REFERENCES block(num) ON DELETE CASCADE,
+	block_pos    INTEGER NOT NULL,
+	tx_hash      VARCHAR NOT NULL,
+	global_index TEXT NOT NULL,
+	created_at   INTEGER NOT NULL DEFAULT (strftime('%s', 'now')),
+	PRIMARY KEY (block_num, block_pos)
+);
+
+-- +migrate Up
+DROP TABLE IF EXISTS set_claim;
+DROP TABLE IF EXISTS unset_claim;
+DROP TABLE IF EXISTS claim;

bridgesync/migrations/migrations.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"embed"
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 
@@ -54,7 +55,7 @@ func addSourceField(database *sql.DB) error {
 		return fmt.Errorf("error getting applied migrations: %w", err)
 	}
 	// This code is to undo the change where bridgesync0014 drops the field
-	if !contains(migrations, "bridgesync0014") {
+	if !slices.Contains(migrations, "bridgesync0014") {
 		log.Warn("migration 'bridgesync0014' not applied, skipping addSourceField." +
 			" This means that the 'source' column on 'bridge' table will not be added.")
 		return nil
@@ -74,15 +75,6 @@ func addSourceField(database *sql.DB) error {
 	return nil
 }
 
-func contains(slice []string, item string) bool {
-	for _, s := range slice {
-		if s == item {
-			return true
-		}
-	}
-	return false
-}
-
 func GetFullMigrations() []types.Migration {
 	baseMigrations := dbmigrations.GetBaseMigrations()
 	total := len(baseMigrations) + len(migrations) + len(treemigrations.Migrations)

bridgesync/migrations/migrations_test.go

@@ -21,6 +21,13 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestRunMigrationsExploratory(t *testing.T) {
+	t.Skip("This test is for exploratory testing of migrations during development. It is not meant to be run as part of automated tests.")
+	dbPath := "/tmp/bridgel1sync.sqlite"
+	err := RunMigrations(dbPath)
+	require.NoError(t, err)
+}
+
 func TestMigration0001(t *testing.T) {
 	dbPath := path.Join(t.TempDir(), "bridgesyncTest001.sqlite")
 
@@ -49,24 +56,6 @@ func TestMigration0001(t *testing.T) {
 			metadata,
 			deposit_count
 		) VALUES (1, 0, 0, 0, '0x0000', 0, '0x0000', 0, NULL, 0);
-
-		INSERT INTO claim (
-			block_num,
-			block_pos,
-    		global_index,
-			origin_network,
-			origin_address,
-			destination_address,
-			amount,
-			proof_local_exit_root,
-			proof_rollup_exit_root,
-			mainnet_exit_root,
-			rollup_exit_root,
-			global_exit_root,
-			destination_network,
-			metadata,
-			is_message
-		) VALUES (1, 0, 0, 0, '0x0000', '0x0000', 0, '0x000,0x000', '0x000,0x000', '0x000', '0x000', '0x0', 0, NULL, FALSE);
 	`)
 	require.NoError(t, err)
 	err = tx.Commit()
@@ -118,20 +107,6 @@ func TestMigration0002(t *testing.T) {
 			from_address
 		) VALUES (1, 0, 0, 0, '0x3', 0, '0x0000', 0, NULL, 0, 1739270804, '0xabcd', '0x123');
 
-		INSERT INTO claim (
-			block_num,
-			block_pos,
-    		global_index,
-			origin_network,
-			origin_address,
-			destination_address,
-			amount,
-			destination_network,
-			metadata,
-			is_message,
-			block_timestamp,
-			tx_hash
-		) VALUES (1, 0, 0, 0, '0x3', '0x0000', 0, 0, NULL, FALSE, 1739270804, '0xabcd');
 	`)
 	require.NoError(t, err)
 	err = tx.Commit()
@@ -185,27 +160,6 @@ func TestMigration0002(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, bridge)
 	require.Equal(t, uint64(1739270804), bridge.BlockTimestamp)
-
-	var claim struct {
-		BlockNum           uint64   `meddler:"block_num"`
-		BlockPos           uint64   `meddler:"block_pos"`
-		GlobalIndex        *big.Int `meddler:"global_index,bigint"`
-		OriginNetwork      uint32   `meddler:"origin_network"`
-		OriginAddress      string   `meddler:"origin_address"`
-		DestinationAddress string   `meddler:"destination_address"`
-		Amount             *big.Int `meddler:"amount,bigint"`
-		DestinationNetwork uint32   `meddler:"destination_network"`
-		Metadata           []byte   `meddler:"metadata"`
-		IsMessage          bool     `meddler:"is_message"`
-		BlockTimestamp     uint64   `meddler:"block_timestamp"`
-		TxHash             string   `meddler:"tx_hash"`
-	}
-
-	err = meddler.QueryRow(db, &claim,
-		`SELECT * FROM claim`)
-	require.NoError(t, err)
-	require.NotNil(t, claim)
-	require.Equal(t, uint64(1739270804), claim.BlockTimestamp)
 }
 
 func TestMigrations0003(t *testing.T) {
@@ -763,6 +717,40 @@ func TestMigration0013(t *testing.T) {
 	err = tx.Commit()
 	require.NoError(t, err)
 }
+func TestMigration0015(t *testing.T) {
+	dbPath := path.Join(t.TempDir(), "bridgesyncTest0015.sqlite")
+
+	database, err := db.NewSQLiteDB(dbPath)
+	require.NoError(t, err)
+	defer database.Close()
+
+	// Run migrations up to 0014 — claim, set_claim and unset_claim still exist.
+	err = db.RunMigrationsDBExtended(log.GetDefaultLogger(),
+		database, GetUpTo("bridgesync0014"), nil, migrate.Up, db.NoLimitMigrations)
+	require.NoError(t, err)
+
+	tableExists := func(name string) bool {
+		var count int
+		err := database.QueryRow(
+			`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?`, name).Scan(&count)
+		require.NoError(t, err)
+		return count > 0
+	}
+
+	require.True(t, tableExists("claim"), "claim table should exist before migration 0015")
+	require.True(t, tableExists("set_claim"), "set_claim table should exist before migration 0015")
+	require.True(t, tableExists("unset_claim"), "unset_claim table should exist before migration 0015")
+
+	// Apply migration 0015.
+	err = db.RunMigrationsDBExtended(log.GetDefaultLogger(),
+		database, GetUpTo("bridgesync0015"), nil, migrate.Up, db.NoLimitMigrations)
+	require.NoError(t, err)
+
+	require.False(t, tableExists("claim"), "claim table should be dropped by migration 0015")
+	require.False(t, tableExists("set_claim"), "set_claim table should be dropped by migration 0015")
+	require.False(t, tableExists("unset_claim"), "unset_claim table should be dropped by migration 0015")
+}
+
 func TestMigrationsDown(t *testing.T) {
 	dbPath := path.Join(t.TempDir(), "bridgesyncTestDown.sqlite")
 	err := RunMigrations(dbPath)
@@ -788,8 +776,23 @@ func TestMigrationFromPreviousVersion(t *testing.T) {
 
 	referenceHash := schemaHash(t, freshDB)
 
+	// Build the expected set of migration IDs from the full migration list.
+	expectedIDs := make(map[string]struct{})
+	for _, m := range GetFullMigrations() {
+		expectedIDs[m.ID] = struct{}{}
+	}
+
+	// Verify the fresh DB itself has all expected migrations applied.
+	appliedIDs, err := db.GetMigrationsIDsApplied(freshDB)
+	require.NoError(t, err)
+	for _, id := range appliedIDs {
+		delete(expectedIDs, id)
+	}
+	require.Empty(t, expectedIDs, "fresh DB is missing migrations: %v", expectedIDs)
+
 	// For each testdata/*.sqlite, copy it, apply all remaining migrations, and
-	// verify that the resulting schema hash matches the reference.
+	// verify that the resulting schema hash matches the reference and that all
+	// expected migrations are recorded in gorp_migrations.
 	testdataEntries, err := os.ReadDir("testdata")
 	require.NoError(t, err)
 
@@ -810,6 +813,18 @@ func TestMigrationFromPreviousVersion(t *testing.T) {
 
 			require.Equal(t, referenceHash, schemaHash(t, migratedDB),
 				"schema mismatch for %s after applying all migrations", entry.Name())
+
+			// Verify all expected migrations are recorded in gorp_migrations.
+			applied, err := db.GetMigrationsIDsApplied(migratedDB)
+			require.NoError(t, err)
+			appliedSet := make(map[string]struct{}, len(applied))
+			for _, id := range applied {
+				appliedSet[id] = struct{}{}
+			}
+			for _, m := range GetFullMigrations() {
+				require.Contains(t, appliedSet, m.ID,
+					"migration %q missing from gorp_migrations after migrating %s", m.ID, entry.Name())
+			}
 		})
 	}
 }

bridgesync/migrations/testdata/0.7.3.sqlite

@@ -45,62 +45,6 @@ func newTestProcessor(dbPath string, syncerID string, logger *log.Logger, dbQuer
 	return newProcessor(database, syncerID, logger, dbQueryTimeout)
 }
 
-func TestBigIntString(t *testing.T) {
-	globalIndex := GenerateGlobalIndex(true, 0, 1093)
-	fmt.Println(globalIndex.String())
-
-	_, ok := new(big.Int).SetString(globalIndex.String(), 10)
-	require.True(t, ok)
-
-	dbPath := filepath.Join(t.TempDir(), "bridgesyncTestBigIntString.sqlite")
-
-	err := migrations.RunMigrations(dbPath)
-	require.NoError(t, err)
-	db, err := db.NewSQLiteDB(dbPath)
-	require.NoError(t, err)
-
-	ctx := context.Background()
-	tx, err := db.BeginTx(ctx, nil)
-	require.NoError(t, err)
-
-	claim := &claimsynctypes.Claim{
-		BlockNum:            1,
-		BlockPos:            0,
-		GlobalIndex:         GenerateGlobalIndex(true, 0, 1093),
-		OriginNetwork:       11,
-		Amount:              big.NewInt(11),
-		OriginAddress:       common.HexToAddress("0x11"),
-		DestinationAddress:  common.HexToAddress("0x11"),
-		ProofLocalExitRoot:  types.Proof{},
-		ProofRollupExitRoot: types.Proof{},
-		MainnetExitRoot:     common.Hash{},
-		RollupExitRoot:      common.Hash{},
-		GlobalExitRoot:      common.Hash{},
-		DestinationNetwork:  12,
-		Type:                claimsynctypes.ClaimEvent,
-	}
-
-	_, err = tx.Exec(`INSERT INTO block (num) VALUES ($1)`, claim.BlockNum)
-	require.NoError(t, err)
-	require.NoError(t, meddler.Insert(tx, "claim", claim))
-
-	require.NoError(t, tx.Commit())
-
-	tx, err = db.BeginTx(ctx, nil)
-	require.NoError(t, err)
-
-	rows, err := tx.Query(`
-		SELECT * FROM claim
-		WHERE block_num >= $1 AND block_num <= $2;
-	`, claim.BlockNum, claim.BlockNum)
-	require.NoError(t, err)
-
-	claimsFromDB := []*claimsynctypes.Claim{}
-	require.NoError(t, meddler.ScanAll(rows, &claimsFromDB))
-	require.Len(t, claimsFromDB, 1)
-	require.Equal(t, claim, claimsFromDB[0])
-}
-
 func TestProcessor(t *testing.T) {
 	path := path.Join(t.TempDir(), "bridgeSyncerProcessor.db")
 	logger := log.WithFields("module", "bridge-syncer")