diff --git a/Makefile b/Makefile index 6b79f01..48f2b97 100644 --- a/Makefile +++ b/Makefile @@ -42,6 +42,7 @@ help: @echo " migration-up Up migrations" @echo " migration-up ENV=production Up migrations (production DB)" @echo " migration-down Down last migration" + @echo " wallet-migrate args=-dry-run Move legacy points/tokens into the wallet" @echo @echo " docker-up Up docker services" @echo " docker-down Down docker services" @@ -105,6 +106,13 @@ migration-down: migration-force: @migrate -database $(DB_URL) -path ./migrations force $(version) +# Move customer_points / customer_tokens into the wallet. Safe to run again. +# Pass args=-dry-run to only report. + +.SILENT: wallet-migrate +wallet-migrate: + ENV_MODE=$(ENV) go run ./cmd/wallet-migrate $(args) + .SILENT: seeder-create seeder-create: @migrate create -ext sql -dir ./seeders -seq $(name) diff --git a/cmd/wallet-migrate/main.go b/cmd/wallet-migrate/main.go new file mode 100644 index 0000000..7cd2652 --- /dev/null +++ b/cmd/wallet-migrate/main.go @@ -0,0 +1,73 @@ +// Command wallet-migrate moves customer_points and customer_tokens balances into the +// EnakPoint & EnakCoin wallet (docs/prd-point-coin.md §10). Safe to run more than once: +// each run credits only what earlier runs have not. +// +// ENV_MODE=staging go run ./cmd/wallet-migrate -dry-run +// ENV_MODE=staging go run ./cmd/wallet-migrate +// +// It exits non-zero when the legacy totals and the migrated totals disagree. +package main + +import ( + "context" + "flag" + "fmt" + "log" + "os" + + "apskel-pos-be/config" + "apskel-pos-be/internal/db" + "apskel-pos-be/internal/processor" + "apskel-pos-be/internal/repository" +) + +func main() { + dryRun := flag.Bool("dry-run", false, "report what would be credited without writing anything") + batchSize := flag.Int("batch", 500, "customers read per batch") + flag.Parse() + + cfg := config.LoadConfig() + pg, err := db.NewPostgres(cfg.Database) + if err != nil { + log.Fatal(err) + } + + migrator := processor.NewWalletMigrationProcessor( + repository.NewWalletMigrationRepository(pg), + processor.NewWalletProcessor(repository.NewWalletRepository(pg)), + repository.NewTxManager(pg), + ) + report, err := migrator.Run(context.Background(), *dryRun, *batchSize) + if err != nil { + log.Fatalf("wallet migration failed: %v", err) + } + + verb := "credited" + if report.DryRun { + verb = "would credit" + } + t := report.Totals + fmt.Printf("customers scanned: %d\n", report.CustomersScanned) + fmt.Printf("EnakPoint: %s %d in %d rows\n", verb, report.PointsCredited, report.PointCredits) + fmt.Printf("EnakCoin: %s %d in %d rows\n", verb, report.CoinsCredited, report.CoinCredits) + fmt.Printf("\n%-10s %15s %15s %15s\n", "", "legacy", "migrated", "wallet balance") + fmt.Printf("%-10s %15d %15d %15d\n", "EnakPoint", t.LegacyPoints, t.MigratedPoints, t.WalletPoints) + fmt.Printf("%-10s %15d %15d %15d\n", "EnakCoin", t.LegacyCoins, t.MigratedCoins, t.WalletCoins) + + if len(report.Discrepancies) > 0 { + fmt.Printf("\n%d balances shrank in the legacy tables after they were migrated.\n", len(report.Discrepancies)) + fmt.Println("They were left as they are in the wallet; correct them with an adjustment if needed:") + for _, d := range report.Discrepancies { + fmt.Printf(" customer %s %s: legacy %d, migrated %d\n", d.CustomerID, d.Currency, d.Legacy, d.Migrated) + } + } + + if report.DryRun { + return + } + if !report.Balanced() { + fmt.Println("\nNOT BALANCED: legacy and migrated totals differ") + os.Exit(1) + } + fmt.Println("\nbalanced: legacy totals = migrated totals") +} diff --git a/internal/contract/campaign_contract.go b/internal/contract/campaign_contract.go index bb5e479..f19d4a2 100644 --- a/internal/contract/campaign_contract.go +++ b/internal/contract/campaign_contract.go @@ -12,7 +12,7 @@ import ( type CreateCampaignRequest struct { Name string `json:"name" binding:"required,min=1,max=150"` Description *string `json:"description,omitempty"` - Type string `json:"type" binding:"required,oneof=REWARD POINTS TOKENS MIXED"` + Type string `json:"type" binding:"required,oneof=REWARD POINTS COINS TOKENS MIXED"` StartDate time.Time `json:"start_date" binding:"required"` EndDate time.Time `json:"end_date" binding:"required"` IsActive bool `json:"is_active"` @@ -25,7 +25,7 @@ type UpdateCampaignRequest struct { ID uuid.UUID `json:"id" binding:"required"` Name string `json:"name" binding:"required,min=1,max=150"` Description *string `json:"description,omitempty"` - Type string `json:"type" binding:"required,oneof=REWARD POINTS TOKENS MIXED"` + Type string `json:"type" binding:"required,oneof=REWARD POINTS COINS TOKENS MIXED"` StartDate time.Time `json:"start_date" binding:"required"` EndDate time.Time `json:"end_date" binding:"required"` IsActive bool `json:"is_active"` @@ -58,7 +58,7 @@ type CreateCampaignRuleRequest struct { CampaignID uuid.UUID `json:"campaign_id" binding:"required"` RuleType string `json:"rule_type" binding:"required,oneof=TIER SPEND PRODUCT CATEGORY DAY LOCATION"` ConditionValue *string `json:"condition_value,omitempty"` - RewardType string `json:"reward_type" binding:"required,oneof=POINTS TOKENS REWARD"` + RewardType string `json:"reward_type" binding:"required,oneof=POINTS COINS TOKENS REWARD"` RewardValue *int64 `json:"reward_value,omitempty"` RewardSubtype *string `json:"reward_subtype,omitempty"` RewardRefID *uuid.UUID `json:"reward_ref_id,omitempty"` @@ -70,7 +70,7 @@ type UpdateCampaignRuleRequest struct { CampaignID uuid.UUID `json:"campaign_id" binding:"required"` RuleType string `json:"rule_type" binding:"required,oneof=TIER SPEND PRODUCT CATEGORY DAY LOCATION"` ConditionValue *string `json:"condition_value,omitempty"` - RewardType string `json:"reward_type" binding:"required,oneof=POINTS TOKENS REWARD"` + RewardType string `json:"reward_type" binding:"required,oneof=POINTS COINS TOKENS REWARD"` RewardValue *int64 `json:"reward_value,omitempty"` RewardSubtype *string `json:"reward_subtype,omitempty"` RewardRefID *uuid.UUID `json:"reward_ref_id,omitempty"` @@ -142,7 +142,7 @@ type ListCampaignRulesResponse struct { type CampaignRuleStruct struct { RuleType string `json:"rule_type" binding:"required,oneof=TIER SPEND PRODUCT CATEGORY DAY LOCATION"` ConditionValue *string `json:"condition_value,omitempty"` - RewardType string `json:"reward_type" binding:"required,oneof=POINTS TOKENS REWARD"` + RewardType string `json:"reward_type" binding:"required,oneof=POINTS COINS TOKENS REWARD"` RewardValue *int64 `json:"reward_value,omitempty"` RewardSubtype *string `json:"reward_subtype,omitempty"` RewardRefID *uuid.UUID `json:"reward_ref_id,omitempty"` diff --git a/internal/entities/campaign.go b/internal/entities/campaign.go index bf76ac5..5eabd76 100644 --- a/internal/entities/campaign.go +++ b/internal/entities/campaign.go @@ -12,10 +12,31 @@ type CampaignType string const ( CampaignTypeReward CampaignType = "REWARD" CampaignTypePoints CampaignType = "POINTS" - CampaignTypeTokens CampaignType = "TOKENS" + CampaignTypeCoins CampaignType = "COINS" CampaignTypeMixed CampaignType = "MIXED" ) +// legacyCampaignTokens is what COINS was called before tokens became EnakCoin +// (docs/prd-point-coin.md §10). Requests may still send it; it is stored as COINS. +const legacyCampaignTokens = "TOKENS" + +// NormalizeCampaignType maps the legacy TOKENS to COINS and leaves anything else as is. +func NormalizeCampaignType(t string) CampaignType { + if t == legacyCampaignTokens { + return CampaignTypeCoins + } + return CampaignType(t) +} + +// NormalizeCampaignRewardType maps the legacy TOKENS to COINS and leaves anything +// else as is. +func NormalizeCampaignRewardType(t string) CampaignRewardType { + if t == legacyCampaignTokens { + return CampaignRewardTypeCoins + } + return CampaignRewardType(t) +} + type RuleType string const ( @@ -31,7 +52,7 @@ type CampaignRewardType string const ( CampaignRewardTypePoints CampaignRewardType = "POINTS" - CampaignRewardTypeTokens CampaignRewardType = "TOKENS" + CampaignRewardTypeCoins CampaignRewardType = "COINS" CampaignRewardTypeReward CampaignRewardType = "REWARD" ) diff --git a/internal/entities/campaign_test.go b/internal/entities/campaign_test.go new file mode 100644 index 0000000..5dd1cd0 --- /dev/null +++ b/internal/entities/campaign_test.go @@ -0,0 +1,18 @@ +package entities + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNormalizeCampaignTypesMapsLegacyTokensToCoins(t *testing.T) { + assert.Equal(t, CampaignTypeCoins, NormalizeCampaignType("TOKENS")) + assert.Equal(t, CampaignTypeCoins, NormalizeCampaignType("COINS")) + assert.Equal(t, CampaignTypePoints, NormalizeCampaignType("POINTS")) + assert.Equal(t, CampaignType(""), NormalizeCampaignType(""), "an empty filter stays empty") + + assert.Equal(t, CampaignRewardTypeCoins, NormalizeCampaignRewardType("TOKENS")) + assert.Equal(t, CampaignRewardTypeReward, NormalizeCampaignRewardType("REWARD")) + assert.Equal(t, CampaignRewardType(""), NormalizeCampaignRewardType("")) +} diff --git a/internal/mappers/campaign_mapper.go b/internal/mappers/campaign_mapper.go index f662108..ffd2940 100644 --- a/internal/mappers/campaign_mapper.go +++ b/internal/mappers/campaign_mapper.go @@ -71,7 +71,7 @@ func ToCampaignEntity(request *contract.CreateCampaignRequest) *entities.Campaig return &entities.Campaign{ Name: request.Name, Description: request.Description, - Type: entities.CampaignType(request.Type), + Type: entities.NormalizeCampaignType(request.Type), StartDate: request.StartDate, EndDate: request.EndDate, IsActive: request.IsActive, @@ -91,7 +91,7 @@ func ToCampaignEntityFromUpdate(request *contract.UpdateCampaignRequest) *entiti ID: request.ID, Name: request.Name, Description: request.Description, - Type: entities.CampaignType(request.Type), + Type: entities.NormalizeCampaignType(request.Type), StartDate: request.StartDate, EndDate: request.EndDate, IsActive: request.IsActive, @@ -118,7 +118,7 @@ func ToCampaignRuleEntity(request *contract.CampaignRuleStruct, campaignID uuid. CampaignID: campaignID, RuleType: entities.RuleType(request.RuleType), ConditionValue: request.ConditionValue, - RewardType: entities.CampaignRewardType(request.RewardType), + RewardType: entities.NormalizeCampaignRewardType(request.RewardType), RewardValue: request.RewardValue, RewardSubtype: rewardSubtype, RewardRefID: request.RewardRefID, @@ -142,7 +142,7 @@ func ToCampaignRuleEntityFromUpdate(request *contract.CampaignRuleStruct, campai CampaignID: campaignID, RuleType: entities.RuleType(request.RuleType), ConditionValue: request.ConditionValue, - RewardType: entities.CampaignRewardType(request.RewardType), + RewardType: entities.NormalizeCampaignRewardType(request.RewardType), RewardValue: request.RewardValue, RewardSubtype: rewardSubtype, RewardRefID: request.RewardRefID, diff --git a/internal/models/campaign.go b/internal/models/campaign.go index 9c65dc6..9720694 100644 --- a/internal/models/campaign.go +++ b/internal/models/campaign.go @@ -11,7 +11,7 @@ import ( type CreateCampaignRequest struct { Name string `json:"name" binding:"required,min=1,max=150"` Description *string `json:"description,omitempty"` - Type string `json:"type" binding:"required,oneof=REWARD POINTS TOKENS MIXED"` + Type string `json:"type" binding:"required,oneof=REWARD POINTS COINS TOKENS MIXED"` StartDate time.Time `json:"start_date" binding:"required"` EndDate time.Time `json:"end_date" binding:"required"` IsActive bool `json:"is_active"` @@ -25,7 +25,7 @@ type UpdateCampaignRequest struct { ID uuid.UUID `json:"id" binding:"required"` Name string `json:"name" binding:"required,min=1,max=150"` Description *string `json:"description,omitempty"` - Type string `json:"type" binding:"required,oneof=REWARD POINTS TOKENS MIXED"` + Type string `json:"type" binding:"required,oneof=REWARD POINTS COINS TOKENS MIXED"` StartDate time.Time `json:"start_date" binding:"required"` EndDate time.Time `json:"end_date" binding:"required"` IsActive bool `json:"is_active"` @@ -88,7 +88,7 @@ type CreateCampaignRuleRequest struct { CampaignID uuid.UUID `json:"campaign_id" binding:"required"` RuleType string `json:"rule_type" binding:"required,oneof=TIER SPEND PRODUCT CATEGORY DAY LOCATION"` ConditionValue *string `json:"condition_value,omitempty"` - RewardType string `json:"reward_type" binding:"required,oneof=POINTS TOKENS REWARD"` + RewardType string `json:"reward_type" binding:"required,oneof=POINTS COINS TOKENS REWARD"` RewardValue *int64 `json:"reward_value,omitempty"` RewardSubtype *string `json:"reward_subtype,omitempty"` RewardRefID *uuid.UUID `json:"reward_ref_id,omitempty"` @@ -100,7 +100,7 @@ type UpdateCampaignRuleRequest struct { CampaignID uuid.UUID `json:"campaign_id" binding:"required"` RuleType string `json:"rule_type" binding:"required,oneof=TIER SPEND PRODUCT CATEGORY DAY LOCATION"` ConditionValue *string `json:"condition_value,omitempty"` - RewardType string `json:"reward_type" binding:"required,oneof=POINTS TOKENS REWARD"` + RewardType string `json:"reward_type" binding:"required,oneof=POINTS COINS TOKENS REWARD"` RewardValue *int64 `json:"reward_value,omitempty"` RewardSubtype *string `json:"reward_subtype,omitempty"` RewardRefID *uuid.UUID `json:"reward_ref_id,omitempty"` @@ -126,7 +126,7 @@ type ListCampaignRulesResponse struct { type CampaignRuleStruct struct { RuleType string `json:"rule_type" binding:"required,oneof=TIER SPEND PRODUCT CATEGORY DAY LOCATION"` ConditionValue *string `json:"condition_value,omitempty"` - RewardType string `json:"reward_type" binding:"required,oneof=POINTS TOKENS REWARD"` + RewardType string `json:"reward_type" binding:"required,oneof=POINTS COINS TOKENS REWARD"` RewardValue *int64 `json:"reward_value,omitempty"` RewardSubtype *string `json:"reward_subtype,omitempty"` RewardRefID *uuid.UUID `json:"reward_ref_id,omitempty"` diff --git a/internal/processor/campaign_processor.go b/internal/processor/campaign_processor.go index 2823a26..57f2c8f 100644 --- a/internal/processor/campaign_processor.go +++ b/internal/processor/campaign_processor.go @@ -70,7 +70,7 @@ func (p *campaignProcessor) ListCampaigns(ctx context.Context, req *contract.Lis Page: req.Page, Limit: req.Limit, Search: req.Search, - Type: req.Type, + Type: string(entities.NormalizeCampaignType(req.Type)), IsActive: req.IsActive, ShowOnApp: req.ShowOnApp, StartDate: req.StartDate, @@ -178,7 +178,7 @@ func (p *campaignRuleProcessor) CreateCampaignRule(ctx context.Context, req *con CampaignID: req.CampaignID, RuleType: entities.RuleType(req.RuleType), ConditionValue: req.ConditionValue, - RewardType: entities.CampaignRewardType(req.RewardType), + RewardType: entities.NormalizeCampaignRewardType(req.RewardType), RewardValue: req.RewardValue, RewardSubtype: (*entities.RewardSubtype)(req.RewardSubtype), RewardRefID: req.RewardRefID, @@ -218,7 +218,7 @@ func (p *campaignRuleProcessor) ListCampaignRules(ctx context.Context, req *cont Limit: req.Limit, CampaignID: req.CampaignID, RuleType: req.RuleType, - RewardType: req.RewardType, + RewardType: string(entities.NormalizeCampaignRewardType(req.RewardType)), } // Get from repository @@ -247,7 +247,7 @@ func (p *campaignRuleProcessor) UpdateCampaignRule(ctx context.Context, req *con CampaignID: req.CampaignID, RuleType: entities.RuleType(req.RuleType), ConditionValue: req.ConditionValue, - RewardType: entities.CampaignRewardType(req.RewardType), + RewardType: entities.NormalizeCampaignRewardType(req.RewardType), RewardValue: req.RewardValue, RewardSubtype: (*entities.RewardSubtype)(req.RewardSubtype), RewardRefID: req.RewardRefID, diff --git a/internal/processor/wallet_migration_processor.go b/internal/processor/wallet_migration_processor.go new file mode 100644 index 0000000..16aaef6 --- /dev/null +++ b/internal/processor/wallet_migration_processor.go @@ -0,0 +1,184 @@ +package processor + +import ( + "context" + "fmt" + + "github.com/google/uuid" + + "apskel-pos-be/internal/constants" + "apskel-pos-be/internal/entities" + "apskel-pos-be/internal/repository" +) + +// TxRunner runs fn inside a database transaction. repository.TxManager is one. +type TxRunner interface { + WithTransaction(ctx context.Context, fn func(ctx context.Context) error) error +} + +// WalletMigrationDiscrepancy is a customer whose legacy balance is now lower than +// what was already migrated: the old code spent from it after the migration ran. +// The wallet is left alone, because only an admin adjustment can take balance away. +type WalletMigrationDiscrepancy struct { + CustomerID uuid.UUID + Currency string + Legacy int64 + Migrated int64 +} + +type WalletMigrationReport struct { + DryRun bool + CustomersScanned int + // Ledger rows written (or, on a dry run, that would be written) and their sum. + PointCredits int + PointsCredited int64 + CoinCredits int + CoinsCredited int64 + Discrepancies []WalletMigrationDiscrepancy + // Taken after the run. On a dry run they show the state before it. + Totals *repository.WalletMigrationTotals +} + +// Balanced reports whether everything in the legacy tables is now in the wallet. +func (r *WalletMigrationReport) Balanced() bool { + return len(r.Discrepancies) == 0 && r.Totals != nil && + r.Totals.LegacyPoints == r.Totals.MigratedPoints && + r.Totals.LegacyCoins == r.Totals.MigratedCoins +} + +// WalletMigrationProcessor moves the balances in customer_points and customer_tokens +// into the wallet (docs/prd-point-coin.md §10, PC-105). Each customer gets a MIGRATION +// ledger row and a non-expiring lot per currency, through WalletProcessor like any +// other credit, so the wallet reconciles from the first row. +// +// It credits the difference between the legacy balance and what earlier runs already +// migrated, so running it again never doubles a balance, and a run after the old code +// kept writing to the legacy tables picks up only what was added since. +type WalletMigrationProcessor struct { + repo repository.WalletMigrationRepository + wallet *WalletProcessor + tx TxRunner +} + +func NewWalletMigrationProcessor(repo repository.WalletMigrationRepository, wallet *WalletProcessor, tx TxRunner) *WalletMigrationProcessor { + return &WalletMigrationProcessor{repo: repo, wallet: wallet, tx: tx} +} + +// Run migrates every customer with a legacy balance, one transaction per customer. +// With dryRun it only reports what it would credit. +func (p *WalletMigrationProcessor) Run(ctx context.Context, dryRun bool, batchSize int) (*WalletMigrationReport, error) { + if batchSize <= 0 { + batchSize = 500 + } + report := &WalletMigrationReport{DryRun: dryRun} + + after := uuid.Nil + for { + ids, err := p.repo.ListLegacyCustomers(ctx, after, batchSize) + if err != nil { + return nil, err + } + if len(ids) == 0 { + break + } + for _, id := range ids { + if dryRun { + err = p.migrateCustomer(ctx, id, true, report) + } else { + err = p.tx.WithTransaction(ctx, func(ctx context.Context) error { + return p.migrateCustomer(ctx, id, false, report) + }) + } + if err != nil { + return nil, fmt.Errorf("customer %s: %w", id, err) + } + report.CustomersScanned++ + } + after = ids[len(ids)-1] + } + + totals, err := p.repo.Totals(ctx) + if err != nil { + return nil, err + } + report.Totals = totals + return report, nil +} + +func (p *WalletMigrationProcessor) migrateCustomer(ctx context.Context, customerID uuid.UUID, dryRun bool, report *WalletMigrationReport) error { + // Lock before reading what was migrated, so two runs at once cannot both see the + // same gap and fill it twice. + if !dryRun { + if err := p.wallet.LockWallet(ctx, customerID); err != nil { + return err + } + } + legacy, err := p.repo.GetLegacyBalance(ctx, customerID) + if err != nil { + return err + } + + // Points come from the single customer_points row. Tokens come from several rows, + // one per type, so the ledger row points at the customer and lists the rows. + pointsRef := customerID + if legacy.PointsRowID != nil { + pointsRef = *legacy.PointsRowID + } + tokens := make([]map[string]any, 0, len(legacy.Tokens)) + for _, t := range legacy.Tokens { + tokens = append(tokens, map[string]any{"id": t.ID, "token_type": string(t.TokenType), "balance": t.Balance}) + } + + for _, c := range []struct { + currency, refType string + refID uuid.UUID + legacy int64 + metadata entities.Metadata + credits *int + credited *int64 + }{ + {constants.WalletCurrencyPoint, constants.WalletRefTypeLegacyPoints, pointsRef, legacy.Points, + entities.Metadata{}, &report.PointCredits, &report.PointsCredited}, + {constants.WalletCurrencyCoin, constants.WalletRefTypeLegacyTokens, customerID, legacy.Coins(), + entities.Metadata{"legacy_tokens": tokens}, &report.CoinCredits, &report.CoinsCredited}, + } { + migrated, err := p.repo.SumMigrated(ctx, customerID, c.currency) + if err != nil { + return err + } + delta := c.legacy - migrated + if delta < 0 { + report.Discrepancies = append(report.Discrepancies, WalletMigrationDiscrepancy{ + CustomerID: customerID, Currency: c.currency, Legacy: c.legacy, Migrated: migrated, + }) + continue + } + if delta == 0 { + continue + } + + if !dryRun { + c.metadata["legacy_balance"] = c.legacy + c.metadata["previously_migrated"] = migrated + _, err = p.wallet.Credit(ctx, WalletCreditInput{WalletEntry: WalletEntry{ + CustomerID: customerID, + Currency: c.currency, + Type: constants.WalletTxTypeMigration, + Amount: delta, + ReferenceType: c.refType, + ReferenceID: c.refID, + Description: "Saldo awal dari sistem lama", + Metadata: c.metadata, + // The legacy total in the key lets a later run top up a balance that + // grew, while a retry of the same run is still recognised. + IdempotencyKey: fmt.Sprintf("migration:%s:%s:%d", c.currency, customerID, c.legacy), + }}) + if err != nil { + return err + } + } + *c.credits++ + *c.credited += delta + } + return nil +} diff --git a/internal/processor/wallet_migration_processor_db_test.go b/internal/processor/wallet_migration_processor_db_test.go new file mode 100644 index 0000000..c0c3f3d --- /dev/null +++ b/internal/processor/wallet_migration_processor_db_test.go @@ -0,0 +1,167 @@ +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 +} diff --git a/internal/repository/wallet_migration_repository.go b/internal/repository/wallet_migration_repository.go new file mode 100644 index 0000000..cfef523 --- /dev/null +++ b/internal/repository/wallet_migration_repository.go @@ -0,0 +1,134 @@ +package repository + +import ( + "context" + "fmt" + + "github.com/google/uuid" + "gorm.io/gorm" + + "apskel-pos-be/internal/constants" + "apskel-pos-be/internal/entities" +) + +// LegacyBalance is what one customer holds in customer_points and customer_tokens, +// the tables the wallet replaces (docs/prd-point-coin.md §10). +type LegacyBalance struct { + CustomerID uuid.UUID + // Nil when the customer has no customer_points row. + PointsRowID *uuid.UUID + Points int64 + Tokens []entities.CustomerTokens +} + +// Coins is the sum of every token type: all of them become EnakCoin (Q6). +func (b LegacyBalance) Coins() int64 { + var total int64 + for _, t := range b.Tokens { + total += t.Balance + } + return total +} + +// WalletMigrationTotals compares the legacy tables with what has been migrated. +type WalletMigrationTotals struct { + LegacyPoints int64 + LegacyCoins int64 + MigratedPoints int64 + MigratedCoins int64 + WalletPoints int64 + WalletCoins int64 +} + +// WalletMigrationRepository reads the legacy balances for the one-time move into the +// wallet. The writes go through the wallet processor like any other credit. +type WalletMigrationRepository interface { + // ListLegacyCustomers returns, in id order, up to limit customers after the given + // id that have a row in customer_points or customer_tokens. + ListLegacyCustomers(ctx context.Context, after uuid.UUID, limit int) ([]uuid.UUID, error) + GetLegacyBalance(ctx context.Context, customerID uuid.UUID) (*LegacyBalance, error) + // SumMigrated returns how much has already been credited to the customer by + // MIGRATION ledger rows in the currency. + SumMigrated(ctx context.Context, customerID uuid.UUID, currency string) (int64, error) + Totals(ctx context.Context) (*WalletMigrationTotals, error) +} + +type walletMigrationRepository struct { + db *gorm.DB +} + +func NewWalletMigrationRepository(db *gorm.DB) WalletMigrationRepository { + return &walletMigrationRepository{db: db} +} + +func (r *walletMigrationRepository) ListLegacyCustomers(ctx context.Context, after uuid.UUID, limit int) ([]uuid.UUID, error) { + var ids []uuid.UUID + err := DBFromContext(ctx, r.db).WithContext(ctx).Raw(` + SELECT customer_id FROM ( + SELECT customer_id FROM customer_points + UNION + SELECT customer_id FROM customer_tokens + ) legacy + WHERE customer_id > ? + ORDER BY customer_id + LIMIT ?`, after, limit). + Scan(&ids).Error + if err != nil { + return nil, fmt.Errorf("failed to list legacy customers: %w", err) + } + return ids, nil +} + +func (r *walletMigrationRepository) GetLegacyBalance(ctx context.Context, customerID uuid.UUID) (*LegacyBalance, error) { + db := DBFromContext(ctx, r.db).WithContext(ctx) + balance := &LegacyBalance{CustomerID: customerID} + + // Find rather than First: many customers have tokens but no points row, and First + // would log each of them as a "record not found" error. + var points []entities.CustomerPoints + if err := db.Where("customer_id = ?", customerID).Limit(1).Find(&points).Error; err != nil { + return nil, fmt.Errorf("failed to get legacy points: %w", err) + } + if len(points) > 0 { + balance.PointsRowID = &points[0].ID + balance.Points = points[0].Balance + } + + err := db.Where("customer_id = ?", customerID).Order("token_type").Find(&balance.Tokens).Error + if err != nil { + return nil, fmt.Errorf("failed to get legacy tokens: %w", err) + } + return balance, nil +} + +func (r *walletMigrationRepository) SumMigrated(ctx context.Context, customerID uuid.UUID, currency string) (int64, error) { + var total int64 + err := DBFromContext(ctx, r.db).WithContext(ctx). + Model(&entities.WalletTransaction{}). + Where("customer_id = ? AND currency = ? AND type = ?", customerID, currency, constants.WalletTxTypeMigration). + Select("COALESCE(SUM(amount), 0)"). + Scan(&total).Error + if err != nil { + return 0, fmt.Errorf("failed to sum migrated balance: %w", err) + } + return total, nil +} + +func (r *walletMigrationRepository) Totals(ctx context.Context) (*WalletMigrationTotals, error) { + var totals WalletMigrationTotals + err := DBFromContext(ctx, r.db).WithContext(ctx).Raw(` + SELECT + (SELECT COALESCE(SUM(balance), 0) FROM customer_points) AS legacy_points, + (SELECT COALESCE(SUM(balance), 0) FROM customer_tokens) AS legacy_coins, + (SELECT COALESCE(SUM(amount), 0) FROM wallet_transactions WHERE type = ? AND currency = ?) AS migrated_points, + (SELECT COALESCE(SUM(amount), 0) FROM wallet_transactions WHERE type = ? AND currency = ?) AS migrated_coins, + (SELECT COALESCE(SUM(point_balance), 0) FROM customer_wallets) AS wallet_points, + (SELECT COALESCE(SUM(coin_balance), 0) FROM customer_wallets) AS wallet_coins`, + constants.WalletTxTypeMigration, constants.WalletCurrencyPoint, + constants.WalletTxTypeMigration, constants.WalletCurrencyCoin). + Scan(&totals).Error + if err != nil { + return nil, fmt.Errorf("failed to compute wallet migration totals: %w", err) + } + return &totals, nil +} diff --git a/internal/service/campaign_service.go b/internal/service/campaign_service.go index 95a671a..bc42ba2 100644 --- a/internal/service/campaign_service.go +++ b/internal/service/campaign_service.go @@ -177,7 +177,7 @@ func (s *campaignService) CreateCampaignRule(ctx context.Context, req *contract. } // Validate reward value based on reward type - if req.RewardType == "POINTS" || req.RewardType == "TOKENS" { + if req.RewardType == "POINTS" || req.RewardType == "COINS" || req.RewardType == "TOKENS" { if req.RewardValue == nil || *req.RewardValue <= 0 { return nil, fmt.Errorf("reward value must be positive for %s type", req.RewardType) } @@ -247,7 +247,7 @@ func (s *campaignService) UpdateCampaignRule(ctx context.Context, req *contract. } // Validate reward value based on reward type - if req.RewardType == "POINTS" || req.RewardType == "TOKENS" { + if req.RewardType == "POINTS" || req.RewardType == "COINS" || req.RewardType == "TOKENS" { if req.RewardValue == nil || *req.RewardValue <= 0 { return nil, fmt.Errorf("reward value must be positive for %s type", req.RewardType) } @@ -295,13 +295,13 @@ func (s *campaignService) GetCampaignRulesByCampaignID(ctx context.Context, camp } func (s *campaignService) validateCampaignType(campaignType string) error { - validTypes := []string{"REWARD", "POINTS", "TOKENS", "MIXED"} + validTypes := []string{"REWARD", "POINTS", "COINS", "TOKENS", "MIXED"} for _, validType := range validTypes { if campaignType == validType { return nil } } - return fmt.Errorf("invalid campaign type: %s. Valid types are: REWARD, POINTS, TOKENS, MIXED", campaignType) + return fmt.Errorf("invalid campaign type: %s. Valid types are: REWARD, POINTS, COINS, MIXED", campaignType) } func (s *campaignService) validateDateRange(startDate, endDate time.Time) error { @@ -335,7 +335,7 @@ func (s *campaignService) validateCampaignRules(rules []contract.CampaignRuleStr } // Validate reward value based on reward type - if rule.RewardType == "POINTS" || rule.RewardType == "TOKENS" { + if rule.RewardType == "POINTS" || rule.RewardType == "COINS" || rule.RewardType == "TOKENS" { if rule.RewardValue == nil || *rule.RewardValue <= 0 { return fmt.Errorf("reward value must be positive for %s type in rule %d", rule.RewardType, i+1) } @@ -365,11 +365,11 @@ func (s *campaignService) validateRuleType(ruleType string) error { } func (s *campaignService) validateRewardType(rewardType string) error { - validTypes := []string{"POINTS", "TOKENS", "REWARD"} + validTypes := []string{"POINTS", "COINS", "TOKENS", "REWARD"} for _, validType := range validTypes { if rewardType == validType { return nil } } - return fmt.Errorf("invalid reward type: %s. Valid types are: POINTS, TOKENS, REWARD", rewardType) + return fmt.Errorf("invalid reward type: %s. Valid types are: POINTS, COINS, REWARD", rewardType) } diff --git a/internal/validator/campaign_validator.go b/internal/validator/campaign_validator.go index d46bae7..8618fc8 100644 --- a/internal/validator/campaign_validator.go +++ b/internal/validator/campaign_validator.go @@ -43,7 +43,7 @@ func (v *CampaignValidatorImpl) ValidateCreateCampaignRequest(req *contract.Crea // Validate campaign type if !v.isValidCampaignType(req.Type) { - return errors.New("invalid campaign type. Valid types are: REWARD, POINTS, TOKENS, MIXED"), constants.ValidationErrorCode + return errors.New("invalid campaign type. Valid types are: REWARD, POINTS, COINS, MIXED"), constants.ValidationErrorCode } // Validate date range @@ -79,7 +79,7 @@ func (v *CampaignValidatorImpl) ValidateUpdateCampaignRequest(req *contract.Upda // Validate campaign type if !v.isValidCampaignType(req.Type) { - return errors.New("invalid campaign type. Valid types are: REWARD, POINTS, TOKENS, MIXED"), constants.ValidationErrorCode + return errors.New("invalid campaign type. Valid types are: REWARD, POINTS, COINS, MIXED"), constants.ValidationErrorCode } // Validate date range @@ -113,7 +113,7 @@ func (v *CampaignValidatorImpl) ValidateListCampaignsRequest(req *contract.ListC // Validate campaign type filter if provided if req.Type != "" && !v.isValidCampaignType(req.Type) { - return errors.New("invalid campaign type filter. Valid types are: REWARD, POINTS, TOKENS, MIXED"), constants.ValidationErrorCode + return errors.New("invalid campaign type filter. Valid types are: REWARD, POINTS, COINS, MIXED"), constants.ValidationErrorCode } return nil, "" @@ -223,7 +223,7 @@ func (v *CampaignValidatorImpl) ValidateListCampaignRulesRequest(req *contract.L // Validate reward type filter if provided if req.RewardType != "" && !v.isValidRewardType(req.RewardType) { - return errors.New("invalid reward type filter. Valid types are: POINTS, TOKENS, REWARD"), constants.ValidationErrorCode + return errors.New("invalid reward type filter. Valid types are: POINTS, COINS, REWARD"), constants.ValidationErrorCode } return nil, "" @@ -254,7 +254,7 @@ func (v *CampaignValidatorImpl) ValidateDeleteCampaignRuleRequest(req *contract. } func (v *CampaignValidatorImpl) isValidCampaignType(campaignType string) bool { - validTypes := []string{"REWARD", "POINTS", "TOKENS", "MIXED"} + validTypes := []string{"REWARD", "POINTS", "COINS", "TOKENS", "MIXED"} return contains(validTypes, campaignType) } @@ -264,7 +264,7 @@ func (v *CampaignValidatorImpl) isValidRuleType(ruleType string) bool { } func (v *CampaignValidatorImpl) isValidRewardType(rewardType string) bool { - validTypes := []string{"POINTS", "TOKENS", "REWARD"} + validTypes := []string{"POINTS", "COINS", "TOKENS", "REWARD"} return contains(validTypes, rewardType) } @@ -293,11 +293,11 @@ func (v *CampaignValidatorImpl) validateCampaignRule(rule *contract.CampaignRule // Validate reward type if !v.isValidRewardType(rule.RewardType) { - return errors.New("invalid reward type in rule " + string(rune(ruleNumber)) + ". Valid types are: POINTS, TOKENS, REWARD") + return errors.New("invalid reward type in rule " + string(rune(ruleNumber)) + ". Valid types are: POINTS, COINS, REWARD") } // Validate reward value based on reward type - if rule.RewardType == "POINTS" || rule.RewardType == "TOKENS" { + if rule.RewardType == "POINTS" || rule.RewardType == "COINS" || rule.RewardType == "TOKENS" { if rule.RewardValue == nil || *rule.RewardValue <= 0 { return errors.New("reward value must be positive for " + rule.RewardType + " type in rule " + string(rune(ruleNumber))) } diff --git a/migrations/000092_rename_campaign_tokens_to_coins.down.sql b/migrations/000092_rename_campaign_tokens_to_coins.down.sql new file mode 100644 index 0000000..06a8ab1 --- /dev/null +++ b/migrations/000092_rename_campaign_tokens_to_coins.down.sql @@ -0,0 +1,5 @@ +UPDATE campaigns SET type = 'TOKENS' WHERE type = 'COINS'; +UPDATE campaign_rules SET reward_type = 'TOKENS' WHERE reward_type = 'COINS'; + +COMMENT ON COLUMN campaigns.type IS 'Type of campaign: REWARD, POINTS, TOKENS, MIXED'; +COMMENT ON COLUMN campaign_rules.reward_type IS 'Type of reward: POINTS, TOKENS, REWARD'; diff --git a/migrations/000092_rename_campaign_tokens_to_coins.up.sql b/migrations/000092_rename_campaign_tokens_to_coins.up.sql new file mode 100644 index 0000000..8ccf3fc --- /dev/null +++ b/migrations/000092_rename_campaign_tokens_to_coins.up.sql @@ -0,0 +1,7 @@ +-- Tokens become EnakCoin (docs/prd-point-coin.md §10). Campaigns that handed out +-- tokens now hand out coins; the API still accepts TOKENS and stores it as COINS. +UPDATE campaigns SET type = 'COINS' WHERE type = 'TOKENS'; +UPDATE campaign_rules SET reward_type = 'COINS' WHERE reward_type = 'TOKENS'; + +COMMENT ON COLUMN campaigns.type IS 'Type of campaign: REWARD, POINTS, COINS, MIXED'; +COMMENT ON COLUMN campaign_rules.reward_type IS 'Type of reward: POINTS, COINS, REWARD';