payments/internal/storage/migrations/4-migrate-balances-events-from-v2.go

package migrations

import (
	"context"
	"math/big"
	"time"

	"github.com/hanzo-fi/go-libs/v5/pkg/observe/log"
	"github.com/hanzo-fi/go-libs/v5/pkg/storage/bun/paginate"
	"github.com/hanzo-fi/payments/pkg/domain/models"
	"github.com/uptrace/bun"
)

type v2Balance struct {
	bun.BaseModel `bun:"accounts.balances"`

	AccountID     models.AccountID `bun:"account_id"`
	Asset         string           `bun:"currency"`
	Balance       *big.Int         `bun:"balance"`
	CreatedAt     time.Time        `bun:"created_at"`
	LastUpdatedAt time.Time        `bun:"last_updated_at"`
}

func MigrateBalancesFromV2(ctx context.Context, logger logging.Logger, db bun.IDB) error {
	exist, err := isTableExisting(ctx, db, "accounts", "balances")
	if err != nil {
		return err
	}

	if !exist {
		// Nothing to migrate
		return nil
	}

	_, err = db.ExecContext(ctx, `
		ALTER TABLE accounts.balances ADD COLUMN IF NOT EXISTS sort_id bigserial;
	`)
	if err != nil {
		return err
	}

	q := paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[any]]{
		Order:    paginate.OrderAsc,
		PageSize: 1000,
		Options: paginate.PaginatedQueryOptions[any]{
			PageSize: 1000,
		},
	}
	for {
		cursor, err := paginateWithOffset[paginate.PaginatedQueryOptions[any], v2Balance](
			ctx,
			db,
			(*paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[any]])(&q),
			func(query *bun.SelectQuery) *bun.SelectQuery {
				return query.Order("created_at ASC", "sort_id ASC")
			},
		)
		if err != nil {
			return err
		}

		logger.WithField("balances", len(cursor.Data)).Info("migrating balances batch...")

		events := make([]v3eventSent, 0, len(cursor.Data))
		for _, balance := range cursor.Data {
			b := models.Balance{
				AccountID:     balance.AccountID,
				CreatedAt:     balance.CreatedAt.UTC(),
				LastUpdatedAt: balance.LastUpdatedAt.UTC(),
				Asset:         balance.Asset,
				Balance:       balance.Balance,
			}

			events = append(events, v3eventSent{
				ID: models.EventID{
					EventIdempotencyKey: b.IdempotencyKey(),
					ConnectorID:         &balance.AccountID.ConnectorID,
				},
				ConnectorID: &balance.AccountID.ConnectorID,
				SentAt:      balance.LastUpdatedAt.UTC(),
			})
		}

		if len(events) > 0 {
			_, err = db.NewInsert().
				Model(&events).
				On("conflict (id) do nothing").
				Exec(ctx)
			if err != nil {
				return err
			}
		}

		logger.WithField("balances", len(cursor.Data)).Info("finished migrating balances batch")

		if !cursor.HasMore {
			break
		}

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		if err != nil {
			return err
		}
	}

	return nil
}