billing: add db migrations (#437)

Signed-off-by: Sander Pick <sanderpick@gmail.com>

Author: sanderpick

api/billingd/migrations/migrations.go

@@ -0,0 +1,72 @@
+package migrations
+
+import (
+	"context"
+	"time"
+
+	logging "github.com/ipfs/go-log"
+	migrate "github.com/xakep666/mongo-migrate"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+var (
+	log            = logging.Logger("migrations")
+	migrateTimeout = time.Hour
+)
+
+var m001 = migrate.Migration{
+	Version:     1,
+	Description: "make customer_id index sparse",
+	Up: func(db *mongo.Database) error {
+		log.Info("migrating 001 up")
+		ctx, cancel := context.WithTimeout(context.Background(), migrateTimeout)
+		defer cancel()
+		count, err := db.Collection("customers").CountDocuments(ctx, bson.M{})
+		if err != nil {
+			return err
+		}
+		if count == 0 {
+			return nil // namespace doesn't exist
+		}
+		_, err = db.Collection("customers").Indexes().DropOne(ctx, "customer_id_1")
+		if err != nil {
+			return err
+		}
+		_, err = db.Collection("customers").Indexes().CreateOne(ctx, mongo.IndexModel{
+			Keys:    bson.D{{"customer_id", 1}},
+			Options: options.Index().SetUnique(true).SetSparse(true),
+		})
+		return err
+	},
+	Down: func(db *mongo.Database) error {
+		log.Info("migrating 001 down")
+		ctx, cancel := context.WithTimeout(context.Background(), migrateTimeout)
+		defer cancel()
+		count, err := db.Collection("customers").CountDocuments(ctx, bson.M{})
+		if err != nil {
+			return err
+		}
+		if count == 0 {
+			return nil // namespace doesn't exist
+		}
+		_, err = db.Collection("customers").Indexes().DropOne(ctx, "customer_id_1")
+		if err != nil {
+			return err
+		}
+		_, err = db.Collection("customers").Indexes().CreateOne(ctx, mongo.IndexModel{
+			Keys:    bson.D{{"customer_id", 1}},
+			Options: options.Index().SetUnique(true),
+		})
+		return err
+	},
+}
+
+func Migrate(db *mongo.Database) error {
+	m := migrate.NewMigrate(
+		db,
+		m001,
+	)
+	return m.Up(migrate.AllAvailable)
+}

api/billingd/migrations/migrations_test.go

@@ -0,0 +1,67 @@
+package migrations
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	migrate "github.com/xakep666/mongo-migrate"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+// Test make customer_id index sparse
+func TestMigrations_m001(t *testing.T) {
+	ctx := context.Background()
+	db := setup(t, ctx)
+
+	// Preload collection
+	_, err := db.Collection("customers").Indexes().CreateOne(ctx, mongo.IndexModel{
+		Keys:    bson.D{{"customer_id", 1}},
+		Options: options.Index().SetUnique(true),
+	})
+	require.NoError(t, err)
+	_, err = db.Collection("customers").InsertMany(ctx, []interface{}{
+		bson.M{"customer_id": "one"},
+		bson.M{"customer_id": "two"},
+		bson.M{"customer_id": "three"},
+	})
+	require.NoError(t, err)
+
+	// Test that nil customer_id causes duplicate key error
+	docs := []interface{}{
+		bson.M{"foo": 1}, // nil customer_id
+		bson.M{"bar": 1}, // nil customer_id
+	}
+	_, err = db.Collection("customers").InsertMany(ctx, docs)
+	require.Error(t, err) // Duplicate key error
+
+	// Run up
+	err = migrate.NewMigrate(db, m001).Up(migrate.AllAvailable)
+	require.NoError(t, err)
+
+	// No duplicate key error this time
+	_, err = db.Collection("customers").InsertMany(ctx, docs)
+	require.NoError(t, err)
+
+	// Clean up
+	_, err = db.Collection("customers").DeleteMany(ctx, bson.M{})
+	require.NoError(t, err)
+
+	// Run down
+	err = migrate.NewMigrate(db, m001).Down(migrate.AllAvailable)
+	require.NoError(t, err)
+}
+
+func setup(t *testing.T, ctx context.Context) *mongo.Database {
+	uri := "mongodb://127.0.0.1:27017"
+	client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
+	require.NoError(t, err)
+	db := client.Database("test_billing_migrations")
+	t.Cleanup(func() {
+		err := db.Drop(ctx)
+		require.NoError(t, err)
+	})
+	return db
+}

api/billingd/service/service.go

@@ -20,6 +20,7 @@ import (
 	"github.com/textileio/go-threads/util"
 	"github.com/textileio/textile/v2/api/billingd/common"
 	"github.com/textileio/textile/v2/api/billingd/gateway"
+	"github.com/textileio/textile/v2/api/billingd/migrations"
 	pb "github.com/textileio/textile/v2/api/billingd/pb"
 	mdb "github.com/textileio/textile/v2/mongodb"
 	"go.mongodb.org/mongo-driver/bson"
@@ -235,13 +236,16 @@ func NewService(ctx context.Context, config Config) (*Service, error) {
 		return nil, err
 	}
 	db := client.Database(config.DBName)
+	if err = migrations.Migrate(db); err != nil {
+		return nil, err
+	}
 
 	pdb := db.Collection("products")
 	cdb := db.Collection("customers")
 	indexes, err := cdb.Indexes().CreateMany(ctx, []mongo.IndexModel{
 		{
 			Keys:    bson.D{{"customer_id", 1}},
-			Options: options.Index().SetUnique(true),
+			Options: options.Index().SetUnique(true).SetSparse(true),
 		},
 		{
 			Keys: bson.D{{"parent_key", 1}, {"created_at", 1}},

mongodb/migrations/migrations_test.go

@@ -141,7 +141,7 @@ func setup(t *testing.T, ctx context.Context) *mongo.Database {
 	uri := "mongodb://127.0.0.1:27017"
 	client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
 	require.NoError(t, err)
-	db := client.Database("test_migrations")
+	db := client.Database("test_textile_migrations")
 	t.Cleanup(func() {
 		err := db.Drop(ctx)
 		require.NoError(t, err)