Track Migrations Per Database (vapor#271)

Package.swift

@@ -16,11 +16,13 @@ let package = Package(
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.0.0"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.0.0"),
         .package(url: "https://github.com/vapor/sql-kit.git", from: "3.0.0-rc.1"),
+        .package(url: "https://github.com/vapor/async-kit.git", from: "1.0.0-rc.1"),
     ],
     targets: [
         .target(name: "FluentKit", dependencies: [
             .product(name: "NIO", package: "swift-nio"),
             .product(name: "Logging", package: "swift-log"),
+            .product(name: "AsyncKit", package: "async-kit"),
         ]),
         .target(name: "FluentBenchmark", dependencies: [
             .target(name: "FluentKit"),

Sources/FluentBenchmark/FluentBenchmarker.swift

@@ -14,6 +14,7 @@ public final class FluentBenchmarker {
     }
 
     public init(databases: Databases) {
+        precondition(databases.ids().count >= 2, "FluentBenchmarker Databases instance must have 2 or more registered databases")
         self.databases = databases
     }
 

Sources/FluentBenchmark/SolarSystem/SolarSystem.swift

@@ -1,3 +1,5 @@
+import AsyncKit
+
 private let migrations: [Migration] = [
     GalaxyMigration(),
     StarMigration(),
@@ -29,9 +31,8 @@ public struct SolarSystem: Migration {
         } else {
             all = migrations
         }
-        return .andAllSync(all.map { migration in
-            { migration.prepare(on: database) }
-        }, on: database.eventLoop)
+
+        return EventLoopFutureQueue(eventLoop: database.eventLoop).append(each: all) { $0.prepare(on: database) }
     }
 
     public func revert(on database: Database) -> EventLoopFuture<Void> {
@@ -41,8 +42,7 @@ public struct SolarSystem: Migration {
         } else {
             all = migrations
         }
-        return .andAllSync(all.reversed().map { migration in
-            { migration.revert(on: database) }
-        }, on: database.eventLoop)
+
+        return EventLoopFutureQueue(eventLoop: database.eventLoop).append(each: all.reversed()) { $0.revert(on: database) }
     }
 }

Sources/FluentBenchmark/Tests/MigratorTests.swift

@@ -2,6 +2,7 @@ extension FluentBenchmarker {
     public func testMigrator() throws {
         try self.testMigrator_success()
         try self.testMigrator_error()
+        try self.testMigrator_sequence()
     }
 
     private func testMigrator_success() throws {
@@ -53,6 +54,83 @@ extension FluentBenchmarker {
             try migrator.revertAllBatches().wait()
         }
     }
+
+    private func testMigrator_sequence() throws {
+        try self.runTest(#function, []) {
+
+            // Setup
+            let ids = Array(self.databases.ids())
+            let databaseID = (ids[0], ids[1])
+
+            let database1 = try XCTUnwrap(
+                self.databases.database(
+                    databaseID.0,
+                    logger: Logger(label: "codes.vapor.tests"),
+                    on: self.databases.eventLoopGroup.next()
+                )
+            )
+            let database2 = try XCTUnwrap(
+                self.databases.database(
+                    databaseID.1,
+                    logger: Logger(label: "codes.vapor.tests"),
+                    on: self.databases.eventLoopGroup.next()
+                )
+            )
+
+            let migrations = Migrations()
+
+
+            // Migration #1
+            migrations.add(GalaxyMigration(), to: databaseID.0)
+
+            let migrator = Migrator(
+                databases: self.databases,
+                migrations: migrations,
+                logger: Logger(label: "codes.vapor.tests"),
+                on: self.databases.eventLoopGroup.next()
+            )
+
+            try migrator.setupIfNeeded().wait()
+            try migrator.prepareBatch().wait()
+
+            let logs1 = try MigrationLog.query(on: database1).all().wait()
+            XCTAssertEqual(logs1.count, 1)
+            let log1 = try XCTUnwrap(logs1.first)
+            XCTAssertEqual(log1.batch, 1)
+            XCTAssertEqual(log1.name, "\(GalaxyMigration.self)")
+
+            do {
+                let count = try MigrationLog.query(on: database2).count().wait()
+
+                // This is a valid state to enter. Unlike databases in the SQL family,
+                // some databases such as MongoDB won't throw an error if the table doesn't exist.
+                XCTAssertEqual(count, 0)
+            } catch {
+                // This is a valid state to enter. A SQL database will throw an error
+                // because the `_fluent_migrations` table on the `database2` database
+                // will have not been created yet.
+            }
+
+
+            // Migration #2
+            migrations.add(GalaxyMigration(), to: databaseID.1)
+
+            try migrator.setupIfNeeded().wait()
+            try migrator.prepareBatch().wait()
+
+            let logs2 = try MigrationLog.query(on: database2).all().wait()
+            XCTAssertEqual(logs2.count, 1)
+            let log2 = try XCTUnwrap(logs2.first)
+            XCTAssertEqual(log2.batch, 1)
+            XCTAssertEqual(log2.name, "\(GalaxyMigration.self)")
+
+            try XCTAssertEqual(MigrationLog.query(on: database1).count().wait(), 1)
+
+
+            // Teardown
+            try migrator.revertAllBatches().wait()
+        }
+    }
 }
 
 private struct ErrorMigration: Migration {

Sources/FluentKit/Database/Databases.swift

@@ -122,6 +122,10 @@ public final class Databases {
         }
     }
 
+    public func ids() -> Set<DatabaseID> {
+        return self.lock.withLock { Set(self.configurations.keys) }
+    }
+
     public func shutdown() {
         self.lock.lock()
         defer { self.lock.unlock() }

Sources/FluentKit/Migration/Migrations.swift

@@ -5,6 +5,7 @@ public final class Migrations {
     }
 
     var storage: [Item]
+    var databases: Set<DatabaseID?> { Set(self.storage.map(\.id)) }
 
     public init() {
         self.storage = []