package processor import ( "context" "os" "testing" "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "gorm.io/driver/postgres" "gorm.io/gorm" "gorm.io/gorm/logger" "apskel-pos-be/internal/constants" "apskel-pos-be/internal/repository" ) // Needs TEST_DATABASE_URL pointing at a migrated database; see // internal/repository/wallet_repository_test.go. Other packages' tests may use the // same database at the same time, so everything here is scoped to its own customers. func TestWalletMigrationProcessor_AgainstPostgres(t *testing.T) { dsn := os.Getenv("TEST_DATABASE_URL") if dsn == "" { t.Skip("TEST_DATABASE_URL not set") } db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)}) require.NoError(t, err) ctx := context.Background() org := uuid.New() full, tokensOnly, pointsOnly, none := uuid.New(), uuid.New(), uuid.New(), uuid.New() customers := []uuid.UUID{full, tokensOnly, pointsOnly, none} exec := func(q string, args ...any) { t.Helper() require.NoError(t, db.Exec(q, args...).Error) } exec(`INSERT INTO organizations (id, name, plan_type) VALUES (?, 'migration test', 'basic')`, org) for _, c := range customers { exec(`INSERT INTO customers (id, organization_id, name) VALUES (?, ?, 'migration test')`, c, org) } t.Cleanup(func() { db.Exec(`DELETE FROM wallet_lot_allocations WHERE lot_id IN (SELECT id FROM wallet_lots WHERE customer_id IN ?)`, customers) db.Exec(`DELETE FROM wallet_lots WHERE customer_id IN ?`, customers) db.Exec(`DELETE FROM wallet_transactions WHERE customer_id IN ?`, customers) db.Exec(`DELETE FROM customer_wallets WHERE customer_id IN ?`, customers) db.Exec(`DELETE FROM customers WHERE id IN ?`, customers) db.Exec(`DELETE FROM organizations WHERE id = ?`, org) }) // The example from §10: SPIN 5 + RAFFLE 2 + MINIGAME 1 = 8 EnakCoin. exec(`INSERT INTO customer_points (customer_id, balance) VALUES (?, 100), (?, 0), (?, 40)`, full, tokensOnly, pointsOnly) exec(`INSERT INTO customer_tokens (customer_id, token_type, balance) VALUES (?, 'SPIN', 5), (?, 'RAFFLE', 2), (?, 'MINIGAME', 1), (?, 'SPIN', 3)`, full, full, full, tokensOnly) migrator := NewWalletMigrationProcessor( repository.NewWalletMigrationRepository(db), NewWalletProcessor(repository.NewWalletRepository(db)), repository.NewTxManager(db), ) type balance struct{ Point, Coin int64 } balances := func() map[uuid.UUID]balance { t.Helper() var rows []struct { CustomerID uuid.UUID PointBalance, CoinBalance int64 } require.NoError(t, db.Raw(`SELECT customer_id, point_balance, coin_balance FROM customer_wallets WHERE customer_id IN ?`, customers).Scan(&rows).Error) out := map[uuid.UUID]balance{} for _, r := range rows { out[r.CustomerID] = balance{r.PointBalance, r.CoinBalance} } return out } countRows := func() int64 { t.Helper() var n int64 require.NoError(t, db.Raw(`SELECT COUNT(*) FROM wallet_transactions WHERE customer_id IN ?`, customers).Scan(&n).Error) return n } // A dry run reports and writes nothing, not even the wallets. report, err := migrator.Run(ctx, true, 2) require.NoError(t, err) assert.GreaterOrEqual(t, report.PointsCredited, int64(140)) assert.GreaterOrEqual(t, report.CoinsCredited, int64(11)) assert.Empty(t, balances()) assert.Zero(t, countRows()) // The real run. A batch of 2 makes it page through the customers. _, err = migrator.Run(ctx, false, 2) require.NoError(t, err) assert.Equal(t, map[uuid.UUID]balance{ full: {Point: 100, Coin: 8}, tokensOnly: {Point: 0, Coin: 3}, pointsOnly: {Point: 40, Coin: 0}, }, balances(), "a customer without legacy rows gets no wallet") assert.Equal(t, int64(4), countRows(), "one row per customer per currency with a balance") var coinRow struct { ReferenceType string ReferenceID uuid.UUID Metadata string } require.NoError(t, db.Raw(`SELECT reference_type, reference_id, metadata::text AS metadata FROM wallet_transactions WHERE customer_id = ? AND currency = 'COIN'`, full).Scan(&coinRow).Error) assert.Equal(t, constants.WalletRefTypeLegacyTokens, coinRow.ReferenceType) assert.Equal(t, full, coinRow.ReferenceID) for _, part := range []string{`"token_type": "SPIN"`, `"token_type": "RAFFLE"`, `"token_type": "MINIGAME"`, `"legacy_balance": 8`} { assert.Contains(t, coinRow.Metadata, part) } var pointRef, pointsRowID string require.NoError(t, db.Raw(`SELECT reference_id::text FROM wallet_transactions WHERE customer_id = ? AND currency = 'POINT'`, full).Scan(&pointRef).Error) require.NoError(t, db.Raw(`SELECT id::text FROM customer_points WHERE customer_id = ?`, full).Scan(&pointsRowID).Error) assert.NotEmpty(t, pointRef) assert.Equal(t, pointsRowID, pointRef, "points row points at the customer_points row") var expiring int64 require.NoError(t, db.Raw(`SELECT COUNT(*) FROM wallet_lots WHERE customer_id IN ? AND expires_at IS NOT NULL`, customers).Scan(&expiring).Error) assert.Zero(t, expiring, "migrated lots never expire") // Running again changes nothing. report, err = migrator.Run(ctx, false, 2) require.NoError(t, err) assert.Equal(t, int64(4), countRows()) assert.Empty(t, discrepanciesFor(report, customers)) // The old code kept writing: one balance grew, one shrank. Only the growth is // migrated; the shrink is reported and left alone. exec(`UPDATE customer_tokens SET balance = 9 WHERE customer_id = ? AND token_type = 'SPIN'`, full) exec(`UPDATE customer_points SET balance = 30 WHERE customer_id = ?`, pointsOnly) report, err = migrator.Run(ctx, false, 2) require.NoError(t, err) assert.Equal(t, balance{Point: 100, Coin: 12}, balances()[full]) assert.Equal(t, balance{Point: 40, Coin: 0}, balances()[pointsOnly]) assert.Equal(t, []WalletMigrationDiscrepancy{{CustomerID: pointsOnly, Currency: constants.WalletCurrencyPoint, Legacy: 30, Migrated: 40}}, discrepanciesFor(report, customers)) assert.Equal(t, int64(5), countRows()) // §7.5 for these customers. var broken int64 require.NoError(t, db.Raw(` SELECT COUNT(*) FROM customer_wallets w WHERE w.customer_id IN ? AND ( w.point_balance <> (SELECT COALESCE(SUM(amount), 0) FROM wallet_transactions t WHERE t.customer_id = w.customer_id AND t.currency = 'POINT') OR w.coin_balance <> (SELECT COALESCE(SUM(amount), 0) FROM wallet_transactions t WHERE t.customer_id = w.customer_id AND t.currency = 'COIN') OR w.point_balance <> (SELECT COALESCE(SUM(remaining_amount), 0) FROM wallet_lots l WHERE l.customer_id = w.customer_id AND l.currency = 'POINT') OR w.coin_balance <> (SELECT COALESCE(SUM(remaining_amount), 0) FROM wallet_lots l WHERE l.customer_id = w.customer_id AND l.currency = 'COIN'))`, customers).Scan(&broken).Error) assert.Zero(t, broken) } func discrepanciesFor(report *WalletMigrationReport, customers []uuid.UUID) []WalletMigrationDiscrepancy { mine := map[uuid.UUID]bool{} for _, c := range customers { mine[c] = true } var out []WalletMigrationDiscrepancy for _, d := range report.Discrepancies { if mine[d.CustomerID] { out = append(out, d) } } return out }