Do not migrate expired welcome messages (#1620)

fbac · web-flow · commit 676047cccfb9 · 2026-02-11T19:37:47.000+01:00
diff --git a/pkg/config/migrator.go b/pkg/config/migrator.go
@@ -14,7 +14,6 @@ type MigrationServerOptions struct {
 	BatchSize              int32         `long:"batch-size"               env:"XMTPD_MIGRATION_DB_BATCH_SIZE"               description:"Batch size for migration"                       default:"1000"`
 	PollInterval           time.Duration `long:"process-interval"         env:"XMTPD_MIGRATION_DB_PROCESS_INTERVAL"         description:"Interval for processing migration"              default:"10s"`
 	Namespace              string        `long:"namespace"                env:"XMTPD_MIGRATION_DB_NAMESPACE"                description:"Namespace for migration"                        default:""`
-	StartDate              time.Time     `long:"start-date"               env:"XMTPD_MIGRATION_START_DATE"                  description:"Start date for migration"                       default:"2025-10-01T00:00:00Z"`
 }
 
 type MigrationClientOptions struct {
diff --git a/pkg/migrator/migrator.go b/pkg/migrator/migrator.go
@@ -185,14 +185,11 @@ func NewMigrationService(opts ...DBMigratorOption) (*Migrator, error) {
 	readDB := db.NewDBHandler(reader, db.WithReadReplica(reader))
 
 	readers := map[string]ISourceReader{
-		groupMessagesTableName: NewGroupMessageReader(readDB.DB(), cfg.options.StartDate.Unix()),
-		inboxLogTableName:      NewInboxLogReader(readDB.DB(), cfg.options.StartDate.UnixNano()),
-		keyPackagesTableName:   NewKeyPackageReader(readDB.DB()),
-		welcomeMessagesTableName: NewWelcomeMessageReader(
-			readDB.DB(),
-			cfg.options.StartDate.Unix(),
-		),
-		commitMessagesTableName: NewCommitMessageReader(readDB.DB(), cfg.options.StartDate.Unix()),
+		groupMessagesTableName:   NewGroupMessageReader(readDB.DB()),
+		inboxLogTableName:        NewInboxLogReader(readDB.DB()),
+		keyPackagesTableName:     NewKeyPackageReader(readDB.DB()),
+		welcomeMessagesTableName: NewWelcomeMessageReader(readDB.DB()),
+		commitMessagesTableName:  NewCommitMessageReader(readDB.DB()),
 	}
 
 	transformer := NewTransformer(cfg.feeCalculator, payerPrivateKey, nodeSigningKey)
diff --git a/pkg/migrator/migrator_test.go b/pkg/migrator/migrator_test.go
@@ -35,8 +35,8 @@ const (
 	// Identity updates go to blockchain, not database.
 	inboxLogAmount       int64 = 0
 	inboxLogLastID       int64 = 19
-	welcomeMessageAmount int64 = 19
-	welcomeMessageLastID int64 = 19
+	welcomeMessageAmount int64 = 18
+	welcomeMessageLastID int64 = 150000017
 	keyPackageAmount     int64 = 19
 	keyPackageLastID     int64 = 19
 )
@@ -75,7 +75,6 @@ func newMigratorTest(t *testing.T) *migratorTest {
 			WaitForDB:              5 * time.Second,
 			BatchSize:              1000,
 			PollInterval:           500 * time.Millisecond,
-			StartDate:              startDate,
 		}),
 		migrator.WithContractsOptions(chainConfig),
 		migrator.WithFeeCalculator(feeCalculator),
@@ -404,7 +403,8 @@ func verifyGroupMessagesIntegrity(t *testing.T, ctx context.Context, sourceDB, d
 func verifyWelcomeMessagesIntegrity(t *testing.T, ctx context.Context, sourceDB, destDB *sql.DB) {
 	sourceRows, err := sourceDB.QueryContext(ctx, `
 		SELECT id, installation_key, data 
-		FROM welcome_messages 
+		FROM welcome_messages
+		WHERE id > 150000000
 		ORDER BY id
 	`)
 	require.NoError(t, err)
diff --git a/pkg/migrator/reader.go b/pkg/migrator/reader.go
@@ -29,14 +29,12 @@ func NewDBReader[T ISourceRecord](
 	query string,
 	queryHeight string,
 	factory func() T,
-	startDate int64,
 ) *DBReader[T] {
 	return &DBReader[T]{
 		db:          db,
 		query:       query,
 		queryHeight: queryHeight,
 		factory:     factory,
-		startDate:   startDate,
 		heightEvery: 10 * time.Minute,
 	}
 }
@@ -127,11 +125,11 @@ type GroupMessageReader struct {
 	*DBReader[*GroupMessage]
 }
 
-func NewGroupMessageReader(db *sql.DB, startDate int64) *GroupMessageReader {
+func NewGroupMessageReader(db *sql.DB) *GroupMessageReader {
 	query := `
 		SELECT id, created_at, group_id, data, group_id_data_hash, is_commit, sender_hmac, should_push
 		FROM group_messages
-		WHERE id > $1 AND is_commit = false AND created_at > to_timestamp($3)
+		WHERE id > $1 AND is_commit = false
 		ORDER BY id ASC
 		LIMIT $2
 	`
@@ -149,7 +147,6 @@ func NewGroupMessageReader(db *sql.DB, startDate int64) *GroupMessageReader {
 			query,
 			queryHeight,
 			func() *GroupMessage { return &GroupMessage{} },
-			startDate,
 		),
 	}
 }
@@ -158,11 +155,11 @@ type CommitMessageReader struct {
 	*DBReader[*CommitMessage]
 }
 
-func NewCommitMessageReader(db *sql.DB, startDate int64) *CommitMessageReader {
+func NewCommitMessageReader(db *sql.DB) *CommitMessageReader {
 	query := `
 		SELECT id, created_at, group_id, data, group_id_data_hash, is_commit, sender_hmac, should_push
 		FROM group_messages
-		WHERE id > $1 AND is_commit = true AND created_at > to_timestamp($3)
+		WHERE id > $1 AND is_commit = true
 		ORDER BY id ASC
 		LIMIT $2
 	`
@@ -181,7 +178,6 @@ func NewCommitMessageReader(db *sql.DB, startDate int64) *CommitMessageReader {
 			query,
 			queryHeight,
 			func() *CommitMessage { return &CommitMessage{} },
-			startDate,
 		),
 	}
 }
@@ -190,11 +186,11 @@ type InboxLogReader struct {
 	*DBReader[*InboxLog]
 }
 
-func NewInboxLogReader(db *sql.DB, startDate int64) *InboxLogReader {
+func NewInboxLogReader(db *sql.DB) *InboxLogReader {
 	query := `
 		SELECT sequence_id, inbox_id, server_timestamp_ns, identity_update_proto
 		FROM inbox_log
-		WHERE sequence_id > $1 AND server_timestamp_ns > $3
+		WHERE sequence_id > $1
 		ORDER BY sequence_id ASC
 		LIMIT $2
 	`
@@ -212,7 +208,6 @@ func NewInboxLogReader(db *sql.DB, startDate int64) *InboxLogReader {
 			query,
 			queryHeight,
 			func() *InboxLog { return &InboxLog{} },
-			startDate,
 		),
 	}
 }
@@ -243,7 +238,6 @@ func NewKeyPackageReader(db *sql.DB) *KeyPackageReader {
 			query,
 			queryHeight,
 			func() *KeyPackage { return &KeyPackage{} },
-			0,
 		),
 	}
 }
@@ -252,11 +246,11 @@ type WelcomeMessageReader struct {
 	*DBReader[*WelcomeMessage]
 }
 
-func NewWelcomeMessageReader(db *sql.DB, startDate int64) *WelcomeMessageReader {
+func NewWelcomeMessageReader(db *sql.DB) *WelcomeMessageReader {
 	query := `
 		SELECT id, created_at, installation_key, data, hpke_public_key, installation_key_data_hash, wrapper_algorithm, welcome_metadata
 		FROM welcome_messages
-		WHERE id > $1 AND created_at > to_timestamp($3)
+		WHERE id > 150000000 AND id > $1
 		ORDER BY id ASC
 		LIMIT $2
 	`
@@ -274,7 +268,6 @@ func NewWelcomeMessageReader(db *sql.DB, startDate int64) *WelcomeMessageReader
 			query,
 			queryHeight,
 			func() *WelcomeMessage { return &WelcomeMessage{} },
-			startDate,
 		),
 	}
 }
diff --git a/pkg/migrator/reader_test.go b/pkg/migrator/reader_test.go
@@ -2,22 +2,19 @@ package migrator_test
 
 import (
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
 	"github.com/xmtp/xmtpd/pkg/migrator"
 	"github.com/xmtp/xmtpd/pkg/migrator/testdata"
 )
 
-var startDate time.Time
-
 func TestGroupMessageReader(t *testing.T) {
 	ctx := t.Context()
 
 	db, _, cleanup := testdata.NewMigratorTestDB(t, ctx)
 	defer cleanup()
 
-	reader := migrator.NewGroupMessageReader(db, startDate.Unix())
+	reader := migrator.NewGroupMessageReader(db)
 
 	cases := []struct {
 		name      string
@@ -69,7 +66,7 @@ func TestInboxLogReader(t *testing.T) {
 	db, _, cleanup := testdata.NewMigratorTestDB(t, ctx)
 	defer cleanup()
 
-	reader := migrator.NewInboxLogReader(db, startDate.UnixNano())
+	reader := migrator.NewInboxLogReader(db)
 
 	cases := []struct {
 		name      string
@@ -173,7 +170,7 @@ func TestCommitMessageReader(t *testing.T) {
 	db, _, cleanup := testdata.NewMigratorTestDB(t, ctx)
 	defer cleanup()
 
-	reader := migrator.NewCommitMessageReader(db, startDate.Unix())
+	reader := migrator.NewCommitMessageReader(db)
 
 	cases := []struct {
 		name   string
@@ -220,7 +217,7 @@ func TestWelcomeMessageReader(t *testing.T) {
 	db, _, cleanup := testdata.NewMigratorTestDB(t, ctx)
 	defer cleanup()
 
-	reader := migrator.NewWelcomeMessageReader(db, startDate.Unix())
+	reader := migrator.NewWelcomeMessageReader(db)
 
 	cases := []struct {
 		name      string
diff --git a/pkg/migrator/testdata/welcome_messages.csv b/pkg/migrator/testdata/welcome_messages.csv
diff --git a/pkg/migrator/transformer_test.go b/pkg/migrator/transformer_test.go

Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,6 @@ type MigrationServerOptions struct {`
`14`	`14`	BatchSize int32 `long:"batch-size" env:"XMTPD_MIGRATION_DB_BATCH_SIZE" description:"Batch size for migration" default:"1000"`
`15`	`15`	PollInterval time.Duration `long:"process-interval" env:"XMTPD_MIGRATION_DB_PROCESS_INTERVAL" description:"Interval for processing migration" default:"10s"`
`16`	`16`	Namespace string `long:"namespace" env:"XMTPD_MIGRATION_DB_NAMESPACE" description:"Namespace for migration" default:""`
`17`		- StartDate time.Time `long:"start-date" env:"XMTPD_MIGRATION_START_DATE" description:"Start date for migration" default:"2025-10-01T00:00:00Z"`
`18`	`17`	`}`
`19`	`18`
`20`	`19`	`type MigrationClientOptions struct {`
Original file line number	Diff line number	Diff line change
`@@ -29,14 +29,12 @@ func NewDBReader[T ISourceRecord](`
`29`	`29`	`query string,`
`30`	`30`	`queryHeight string,`
`31`	`31`	`factory func() T,`
`32`		`- startDate int64,`
`33`	`32`	`) *DBReader[T] {`
`34`	`33`	`return &DBReader[T]{`
`35`	`34`	`db: db,`
`36`	`35`	`query: query,`
`37`	`36`	`queryHeight: queryHeight,`
`38`	`37`	`factory: factory,`
`39`		`- startDate: startDate,`
`40`	`38`	`heightEvery: 10 * time.Minute,`
`41`	`39`	`}`
`42`	`40`	`}`
`@@ -127,11 +125,11 @@ type GroupMessageReader struct {`
`127`	`125`	`DBReader[GroupMessage]`
`128`	`126`	`}`
`129`	`127`
`130`		`-func NewGroupMessageReader(db sql.DB, startDate int64) GroupMessageReader {`
	`128`	`+func NewGroupMessageReader(db sql.DB) GroupMessageReader {`
`131`	`129`	query := `
`132`	`130`	`SELECT id, created_at, group_id, data, group_id_data_hash, is_commit, sender_hmac, should_push`
`133`	`131`	`FROM group_messages`
`134`		`- WHERE id > $1 AND is_commit = false AND created_at > to_timestamp($3)`
	`132`	`+ WHERE id > $1 AND is_commit = false`
`135`	`133`	`ORDER BY id ASC`
`136`	`134`	`LIMIT $2`
`137`	`135`	`
`@@ -149,7 +147,6 @@ func NewGroupMessageReader(db sql.DB, startDate int64) GroupMessageReader {`
`149`	`147`	`query,`
`150`	`148`	`queryHeight,`
`151`	`149`	`func() *GroupMessage { return &GroupMessage{} },`
`152`		`- startDate,`
`153`	`150`	`),`
`154`	`151`	`}`
`155`	`152`	`}`
`@@ -158,11 +155,11 @@ type CommitMessageReader struct {`
`158`	`155`	`DBReader[CommitMessage]`
`159`	`156`	`}`
`160`	`157`
`161`		`-func NewCommitMessageReader(db sql.DB, startDate int64) CommitMessageReader {`
	`158`	`+func NewCommitMessageReader(db sql.DB) CommitMessageReader {`
`162`	`159`	query := `
`163`	`160`	`SELECT id, created_at, group_id, data, group_id_data_hash, is_commit, sender_hmac, should_push`
`164`	`161`	`FROM group_messages`
`165`		`- WHERE id > $1 AND is_commit = true AND created_at > to_timestamp($3)`
	`162`	`+ WHERE id > $1 AND is_commit = true`
`166`	`163`	`ORDER BY id ASC`
`167`	`164`	`LIMIT $2`
`168`	`165`	`
`@@ -181,7 +178,6 @@ func NewCommitMessageReader(db sql.DB, startDate int64) CommitMessageReader {`
`181`	`178`	`query,`
`182`	`179`	`queryHeight,`
`183`	`180`	`func() *CommitMessage { return &CommitMessage{} },`
`184`		`- startDate,`
`185`	`181`	`),`
`186`	`182`	`}`
`187`	`183`	`}`
`@@ -190,11 +186,11 @@ type InboxLogReader struct {`
`190`	`186`	`DBReader[InboxLog]`
`191`	`187`	`}`
`192`	`188`
`193`		`-func NewInboxLogReader(db sql.DB, startDate int64) InboxLogReader {`
	`189`	`+func NewInboxLogReader(db sql.DB) InboxLogReader {`
`194`	`190`	query := `
`195`	`191`	`SELECT sequence_id, inbox_id, server_timestamp_ns, identity_update_proto`
`196`	`192`	`FROM inbox_log`
`197`		`- WHERE sequence_id > $1 AND server_timestamp_ns > $3`
	`193`	`+ WHERE sequence_id > $1`
`198`	`194`	`ORDER BY sequence_id ASC`
`199`	`195`	`LIMIT $2`
`200`	`196`	`
`@@ -212,7 +208,6 @@ func NewInboxLogReader(db sql.DB, startDate int64) InboxLogReader {`
`212`	`208`	`query,`
`213`	`209`	`queryHeight,`
`214`	`210`	`func() *InboxLog { return &InboxLog{} },`
`215`		`- startDate,`
`216`	`211`	`),`
`217`	`212`	`}`
`218`	`213`	`}`
`@@ -243,7 +238,6 @@ func NewKeyPackageReader(db sql.DB) KeyPackageReader {`
`243`	`238`	`query,`
`244`	`239`	`queryHeight,`
`245`	`240`	`func() *KeyPackage { return &KeyPackage{} },`
`246`		`- 0,`
`247`	`241`	`),`
`248`	`242`	`}`
`249`	`243`	`}`
`@@ -252,11 +246,11 @@ type WelcomeMessageReader struct {`
`252`	`246`	`DBReader[WelcomeMessage]`
`253`	`247`	`}`
`254`	`248`
`255`		`-func NewWelcomeMessageReader(db sql.DB, startDate int64) WelcomeMessageReader {`
	`249`	`+func NewWelcomeMessageReader(db sql.DB) WelcomeMessageReader {`
`256`	`250`	query := `
`257`	`251`	`SELECT id, created_at, installation_key, data, hpke_public_key, installation_key_data_hash, wrapper_algorithm, welcome_metadata`
`258`	`252`	`FROM welcome_messages`
`259`		`- WHERE id > $1 AND created_at > to_timestamp($3)`
	`253`	`+ WHERE id > 150000000 AND id > $1`
`260`	`254`	`ORDER BY id ASC`
`261`	`255`	`LIMIT $2`
`262`	`256`	`
`@@ -274,7 +268,6 @@ func NewWelcomeMessageReader(db sql.DB, startDate int64) WelcomeMessageReader`
`274`	`268`	`query,`
`275`	`269`	`queryHeight,`
`276`	`270`	`func() *WelcomeMessage { return &WelcomeMessage{} },`
`277`		`- startDate,`
`278`	`271`	`),`
`279`	`272`	`}`
`280`	`273`	`}`