Db migration

cmd/main.go

@@ -7,6 +7,7 @@ import (
 	"airdao-mobile-api/pkg/logger"
 	"airdao-mobile-api/pkg/mongodb"
 	"airdao-mobile-api/services/health"
+	"airdao-mobile-api/services/migration"
 	"airdao-mobile-api/services/watcher"
 	"context"
 	"errors"
@@ -57,6 +58,11 @@ func main() {
 	}
 	zapLogger.Info("DB connected successfully")
 
+	err = migration.RunMigrations(db, cfg.MongoDb.MongoDbName, zapLogger)
+	if err != nil {
+		zapLogger.Fatalf("failed to run migration: %s", err)
+	}
+
 	// Firebase
 	firebaseClient, err := firebase.NewClient(cfg.Firebase.CredPath)
 	if err != nil {
@@ -103,20 +109,25 @@ func main() {
 		ServerHeader: "AIRDAO-Mobile-Api", // add custom server header
 	}
 
+	zapLogger.Info("Deleting watchers with stale data...")
+
 	// Run DeleteWatchersWithStaleData on start for check and delete stale data
 	if err := watcherService.DeleteWatchersWithStaleData(context.Background()); err != nil {
 		zapLogger.Errorf("failed to delete watchers with stale data - %v", err)
 	}
 
+	zapLogger.Info("Deleted watchers with stale data successfully")
+
 	// Run DeleteWatchersWithStaleData every 24 hours for check and delete stale data
 	go func() {
 		for {
+			time.Sleep(24 * time.Hour)
+
 			err := watcherService.DeleteWatchersWithStaleData(context.Background())
 			if err != nil {
 				zapLogger.Errorf("failed to delete watchers with stale data - %v", err)
 			}
-
-			time.Sleep(24 * time.Hour)
+			zapLogger.Info("Deleted watchers with stale data successfully")
 		}
 	}()
 
@@ -149,7 +160,7 @@ func main() {
 		}
 	}()
 
-	zapLogger.Infoln("Server started on port %v", cfg.Port)
+	zapLogger.Infoln("Server started on port", cfg.Port)
 
 	// Create a context that will be used to gracefully shut down the server
 	ctx, cancel = signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)

services/migration/migration.go

@@ -0,0 +1,124 @@
+package migration
+
+import (
+	"airdao-mobile-api/services/watcher"
+	"errors"
+
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.uber.org/zap"
+)
+
+import (
+	"context"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+type HistoryNotificationDocument struct {
+	WatcherID    primitive.ObjectID           `json:"watcher_id" bson:"watcher_id"`
+	Notification *watcher.HistoryNotification `json:"notification" bson:"notification"`
+}
+
+type Migration struct {
+	ID        primitive.ObjectID `json:"id" bson:"_id"`
+	Name      string             `json:"name" bson:"name"`
+	IsApplied bool               `json:"is_applied" bson:"is_applied"`
+}
+
+func RunMigrations(db *mongo.Client, dbName string, logger *zap.SugaredLogger) error {
+	err := historicalNotificationMigration(db, dbName, logger)
+	// Add more migrations here
+	return err
+}
+
+func historicalNotificationMigration(db *mongo.Client, dbName string, logger *zap.SugaredLogger) error {
+	if db == nil {
+		return errors.New("[watcher_repository] invalid user database")
+	}
+	name := "historical_notification_migration"
+
+	migrationsCollection := db.Database(dbName).Collection("migrations")
+	watcherCollection := db.Database(dbName).Collection(watcher.CollectionName)
+	historyCollection := db.Database(dbName).Collection(watcher.HistoricalNotificationsCollectionName)
+
+	// Check if migration has already been run
+	var migration Migration
+	if err := migrationsCollection.FindOne(context.Background(), bson.M{"name": name}).Decode(&migration); err == nil {
+		logger.Info("Migration already applied.")
+		return nil
+	}
+
+	// Create index { watcher_id: 1, notification.timestamp: -1 }
+
+	indexModel := mongo.IndexModel{
+		Keys: bson.D{
+			{Key: "watcher_id", Value: 1},
+			{Key: "notification.timestamp", Value: -1},
+		},
+	}
+	_, err := historyCollection.Indexes().CreateOne(context.Background(), indexModel)
+	if err != nil {
+		return err
+	}
+
+	cursor, err := watcherCollection.Find(context.Background(), bson.M{})
+	if err != nil {
+		return err
+	}
+	logger.Info("Starting migration...", name)
+	defer cursor.Close(context.Background())
+
+	for cursor.Next(context.Background()) {
+		var watcher watcher.Watcher
+		if err := cursor.Decode(&watcher); err != nil {
+			return err
+		}
+
+		// Check if historical_notifications field is nil
+		if watcher.HistoricalNotifications != nil {
+			logger.Info("Migrating historical notifications...", watcher.ID)
+
+			// Iterate over historical notifications only if it's not nil
+			for _, notification := range *watcher.HistoricalNotifications {
+				_, err := historyCollection.InsertOne(context.Background(), &HistoryNotificationDocument{
+					WatcherID:    watcher.ID,
+					Notification: notification,
+				})
+				if err != nil {
+					return err
+				}
+			}
+
+			// Update watcher to set historical_notifications to nil
+			_, err := watcherCollection.UpdateOne(
+				context.Background(),
+				bson.M{"_id": watcher.ID},
+				bson.M{"$set": bson.M{"historical_notifications": nil}},
+			)
+			if err != nil {
+				return err
+			}
+		} else {
+			// Log a message or handle the case when historical_notifications is nil
+			logger.Info("No historical notifications found for watcher:", watcher.ID)
+		}
+
+	}
+
+	if err := cursor.Err(); err != nil {
+		return err
+	}
+
+	logger.Info("Migration completed successfully.")
+
+	// Mark migration as applied
+	_, err = migrationsCollection.InsertOne(context.Background(), &Migration{
+		Name:      name,
+		IsApplied: true,
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}