Fully overhaul package for modern Swift/Vapor/Fluent

.github/workflows/test.yml

@@ -1,26 +1,11 @@
 name: test
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
 on:
-  pull_request:
-  push:
-    branches:
-    - master
+  pull_request: { types: [opened, reopened, synchronize, ready_for_review] }
+  push: { branches: [ main ] }
+
 jobs:
-  linux:
-    runs-on: ubuntu-latest
-    container: swift:5.2-bionic
-    steps:
-      - name: Check out code
-        uses: actions/checkout@v2
-      - name: Run tests with Thread Sanitizer
-        run: swift test --enable-test-discovery --sanitize=thread
-  macOS:
-    runs-on: macos-latest
-    steps:
-      - name: Select latest available Xcode
-        uses: maxim-lobanov/[email protected]
-        with:
-          xcode-version: latest
-      - name: Check out code
-        uses: actions/checkout@v2
-      - name: Run tests with Thread Sanitizer
-        run: swift test --enable-test-discovery --sanitize=thread
+  unit-tests:
+     uses: vapor/ci/.github/workflows/run-unit-tests.yml@main

Package.swift

@@ -1,6 +1,4 @@
-// swift-tools-version:5.4
-// The swift-tools-version declares the minimum version of Swift required to build this package.
-
+// swift-tools-version:5.7
 import PackageDescription
 
 let package = Package(
@@ -9,30 +7,32 @@ let package = Package(
         .macOS(.v10_15)
     ],
     products: [
-        .library(
-            name: "QueuesFluentDriver",
-            targets: ["QueuesFluentDriver"]),
+        .library(name: "QueuesFluentDriver", targets: ["QueuesFluentDriver"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/vapor/vapor.git", from: "4.0.0"),
-        .package(url: "https://github.com/vapor/fluent.git", from: "4.0.0"),
-        .package(url: "https://github.com/vapor/sql-kit.git", from: "3.6.0"),
-        .package(url: "https://github.com/vapor/queues.git", from: "1.11.1"),
+        .package(url: "https://github.com/vapor/vapor.git", from: "4.92.1"),
+        .package(url: "https://github.com/vapor/fluent.git", from: "4.9.0"),
+        .package(url: "https://github.com/vapor/fluent-kit.git", from: "1.45.1"),
+        .package(url: "https://github.com/vapor/sql-kit.git", from: "3.28.0"),
+        .package(url: "https://github.com/vapor/queues.git", from: "1.13.0"),
     ],
     targets: [
         .target(
             name: "QueuesFluentDriver",
             dependencies: [
                 .product(name: "Vapor", package: "vapor"),
                 .product(name: "Fluent", package: "fluent"),
+                .product(name: "FluentKit", package: "fluent-kit"),
+                .product(name: "FluentSQL", package: "fluent-kit"),
                 .product(name: "SQLKit", package: "sql-kit"),
                 .product(name: "Queues", package: "queues")
-            ],
-            path: "Sources"
+            ]
         ),
         .testTarget(
             name: "QueuesFluentDriverTests",
-            dependencies: ["QueuesFluentDriver"]
+            dependencies: [
+                .target(name: "QueuesFluentDriver"),
+            ]
         ),
     ]
 )

Package@swift-5.9.swift

@@ -0,0 +1,51 @@
+// swift-tools-version:5.9
+import PackageDescription
+
+let swiftSettings: [SwiftSetting] = [
+    .enableUpcomingFeature("ForwardTrailingClosures"),
+    .enableUpcomingFeature("ExistentialAny"),
+    .enableUpcomingFeature("ConciseMagicFile"),
+    .enableUpcomingFeature("DisableOutwardActorInference"),
+    .enableExperimentalFeature("StrictConcurrency=complete"),
+]
+
+let package = Package(
+    name: "QueuesFluentDriver",
+    platforms: [
+        .macOS(.v10_15),
+        .iOS(.v13),
+        .watchOS(.v6),
+        .tvOS(.v13),
+    ],
+    products: [
+        .library(name: "QueuesFluentDriver", targets: ["QueuesFluentDriver"]),
+    ],
+    dependencies: [
+        .package(url: "https://github.com/vapor/vapor.git", from: "4.92.1"),
+        .package(url: "https://github.com/vapor/fluent.git", from: "4.9.0"),
+        .package(url: "https://github.com/vapor/fluent-kit.git", from: "1.45.1"),
+        .package(url: "https://github.com/vapor/sql-kit.git", from: "3.28.0"),
+        .package(url: "https://github.com/vapor/queues.git", from: "1.13.0"),
+    ],
+    targets: [
+        .target(
+            name: "QueuesFluentDriver",
+            dependencies: [
+                .product(name: "Vapor", package: "vapor"),
+                .product(name: "Fluent", package: "fluent"),
+                .product(name: "FluentKit", package: "fluent-kit"),
+                .product(name: "FluentSQL", package: "fluent-kit"),
+                .product(name: "SQLKit", package: "sql-kit"),
+                .product(name: "Queues", package: "queues")
+            ],
+            swiftSettings: swiftSettings
+        ),
+        .testTarget(
+            name: "QueuesFluentDriverTests",
+            dependencies: [
+                .target(name: "QueuesFluentDriver"),
+            ],
+            swiftSettings: swiftSettings
+        ),
+    ]
+)

README.md

@@ -25,7 +25,7 @@ Add it to the `Package.swift` of your Vapor4 project:
 
 ```swift
 
-// swift-tools-version:5.4
+// swift-tools-version:5.7
 import PackageDescription
 
 let package = Package(
@@ -36,7 +36,7 @@ let package = Package(
     ...
     dependencies: [
         ...
-        .package(url: "https://github.com/m-barthelemy/vapor-queues-fluent-driver.git", from: "3.0.0-beta1"),
+        .package(url: "https://github.com/vapor-community/vapor-queues-fluent-driver.git", from: "3.0.0-beta.2"),
         ...
     ],
     targets: [

Sources/QueuesFluentDriver/FluentQueue.swift

@@ -1,23 +1,22 @@
 import Foundation
-import Queues
-import Fluent
-import SQLKit
+@preconcurrency import Queues
+@preconcurrency import FluentKit
+@preconcurrency import SQLKit
 
-public struct FluentQueue {
+public struct FluentQueue: Queue, Sendable {
     public let context: QueueContext
-    let db: Database
-    let dbType: QueuesFluentDbType
+
+    let db: any Database
+    let sqlDb: any SQLDatabase
     let useSoftDeletes: Bool
-}
 
-extension FluentQueue: Queue {
     public func get(_ id: JobIdentifier) -> EventLoopFuture<JobData> {
-        return db.query(JobModel.self)
+        self.db.query(JobModel.self)
             .filter(\.$id == id.string)
             .first()
             .unwrap(or: QueuesFluentError.missingJob(id))
             .flatMapThrowing { job in
-                return JobData(
+                JobData(
                     payload: Array(job.data.payload),
                     maxRetryCount: job.data.maxRetryCount,
                     jobName: job.data.jobName,
@@ -30,18 +29,17 @@ extension FluentQueue: Queue {
 
     public func set(_ id: JobIdentifier, to jobStorage: JobData) -> EventLoopFuture<Void> {
         let jobModel = JobModel(id: id, queue: queueName.string, jobData: JobDataModel(jobData: jobStorage))
+
         // If the job must run at a later time, ensure it won't be picked earlier since
         // we sort pending jobs by date when querying
         jobModel.runAtOrAfter = jobStorage.delayUntil ?? Date()
-
-        return jobModel.save(on: db).map { metadata in
-            return
-        }
+
+        return jobModel.save(on: self.db)
     }
 
     public func clear(_ id: JobIdentifier) -> EventLoopFuture<Void> {
-        // This does the equivalent of a Fluent Softdelete but sets the `state` to `completed`
-        return db.query(JobModel.self)
+        // This does the equivalent of a Fluent soft delete, but sets the `state` to `completed`
+        self.db.query(JobModel.self)
             .filter(\.$id == id.string)
             .filter(\.$state != .completed)
             .first()
@@ -58,46 +56,34 @@ extension FluentQueue: Queue {
     }
 
     public func push(_ id: JobIdentifier) -> EventLoopFuture<Void> {
-        guard let sqlDb = db as? SQLDatabase else {
-            return self.context.eventLoop.makeFailedFuture(QueuesFluentError.databaseNotFound)
-        }
-        return sqlDb
-            .update(JobModel.schema)
-            .set(SQLColumn("\(FieldKey.state)"), to: SQLBind(QueuesFluentJobState.pending))
-            .where(SQLColumn("\(FieldKey.id)"), .equal, SQLBind(id.string))
+        self.sqlDb
+            .update(JobModel.sqlTable)
+            .set(JobModel.sqlColumn(\.$state), to: SQLBind(QueuesFluentJobState.pending))
+            .where(JobModel.sqlColumn(\.$id), .equal, SQLBind(id.string))
             .run()
     }
 
     /// Currently selects the oldest job pending execution
     public func pop() -> EventLoopFuture<JobIdentifier?> {
-        guard let sqlDb = db as? SQLDatabase else {
-            return self.context.eventLoop.makeFailedFuture(QueuesFluentError.databaseNotFound)
-        }
-
-        var selectQuery = sqlDb
-            .select()
-            .column("\(FieldKey.id)")
-            .from(JobModel.schema)
-            .where(SQLColumn("\(FieldKey.state)"), .equal, SQLBind(QueuesFluentJobState.pending))
-            .where(SQLColumn("\(FieldKey.queue)"), .equal, SQLBind(self.queueName.string))
-            .where(SQLColumn("\(FieldKey.runAt)"), .lessThanOrEqual, SQLBind(Date()))
-            .orderBy("\(FieldKey.runAt)")
-            .limit(1)
-        if self.dbType != .sqlite {
-            selectQuery = selectQuery.lockingClause(SQLSkipLocked.forUpdateSkipLocked)
-        }
-
-        var popProvider: PopQueryProtocol!
-        switch (self.dbType) {
-            case .postgresql:
-                popProvider = PostgresPop()
-            case .mysql:
-                popProvider = MySQLPop()
-            case .sqlite:
-                popProvider = SqlitePop()
-        }
-        return popProvider.pop(db: db, select: selectQuery.query).optionalMap { id in
-            return JobIdentifier(string: id)
+        self.db.eventLoop.makeFutureWithTask {
+            // TODO: Use `SQLSubquery` instead when it becomes available in upstream SQLKit.
+            let select = self.sqlDb
+                .select()
+                .column(JobModel.sqlColumn(\.$id))
+                .from(JobModel.sqlTable)
+                .where(JobModel.sqlColumn(\.$state), .equal, SQLBind(QueuesFluentJobState.pending))
+                .where(JobModel.sqlColumn(\.$queue), .equal, SQLBind(self.queueName.string))
+                .where(JobModel.sqlColumn(\.$runAtOrAfter), .lessThanOrEqual, SQLFunction("now"))
+                .orderBy(JobModel.sqlColumn(\.$runAtOrAfter))
+                .limit(1)
+                .lockingClause(SQLSkipLocked.forUpdateSkipLocked)
+
+            switch self.sqlDb.dialect.name {
+                case "postgresql",
+                     "sqlite":     return try await ReturningClausePopQuery.pop(db: self.db, select: select).map(JobIdentifier.init(string:))
+                case "mysql":      return try await TransactionalPopQuery.pop(db: self.db, select: select).map(JobIdentifier.init(string:))
+                default:           preconditionFailure("This should have already been checked.")
+            }
         }
     }
 }

Sources/QueuesFluentDriver/FluentQueuesDriver.swift

@@ -2,52 +2,57 @@ import Fluent
 import SQLKit
 import Queues
 
-public enum QueuesFluentDbType: String {
-    case postgresql
-    case mysql
-    case sqlite
-}
-
-public struct FluentQueuesDriver {
+public struct FluentQueuesDriver: QueuesDriver {
     let databaseId: DatabaseID?
     let useSoftDeletes: Bool
-    let eventLoopGroup: EventLoopGroup
+    let eventLoopGroup: any EventLoopGroup
 
-    init(on databaseId: DatabaseID? = nil, useSoftDeletes: Bool, on eventLoopGroup: EventLoopGroup) {
+    init(on databaseId: DatabaseID? = nil, useSoftDeletes: Bool, on eventLoopGroup: any EventLoopGroup) {
         self.databaseId = databaseId
         self.useSoftDeletes = useSoftDeletes
         self.eventLoopGroup = eventLoopGroup
     }
-}
 
-extension FluentQueuesDriver: QueuesDriver {
-    public func makeQueue(with context: QueueContext) -> Queue {
-        let db = context
+    public func makeQueue(with context: QueueContext) -> any Queue {
+        let baseDb = context
             .application
             .databases
-            .database(databaseId, logger: context.logger, on: context.eventLoop)
+            .database(self.databaseId, logger: context.logger, on: context.eventLoop)
+
+        // `QueuesDriver` methods cannot throw, so we report errors by returning a fake queue which
+        // always throws errors when used.
+        guard let baseDb else {
+            return FailingQueue(failure: QueuesFluentError.databaseNotFound, context: context)
+        }
 
-        // How do we report that something goes wrong here? Since makeQueue cannot throw.
-        let dialect = (db as? SQLDatabase)?.dialect.name
-        if db == nil || dialect == nil {
-            context.logger.error(
-                "\(Self.self): Database misconfigured or unsupported."
-            )
+        guard let sqlDb = baseDb as? any SQLDatabase else {
+            return FailingQueue(failure: QueuesFluentError.unsupportedDatabase, context: context)
         }
 
-        let dbType = QueuesFluentDbType(rawValue: dialect!)
-        if dbType == nil {
-            context.logger.error("\(Self.self): Unsupported Database type '\(dialect!)'")
+        switch sqlDb.dialect.name {
+        case "sqlite", "mysql", "postgresql": break
+        default: return FailingQueue(failure: QueuesFluentError.unsupportedDatabase, context: context)
         }
 
         return FluentQueue(
             context: context,
-            db: db!,
-            dbType: dbType!,
+            db: baseDb,
+            sqlDb: sqlDb,
             useSoftDeletes: self.useSoftDeletes
         )
     }
 
     public func shutdown() {
     }
 }
+
+struct FailingQueue: Queue {
+    let failure: any Error
+    let context: QueueContext
+
+    func get(_: JobIdentifier) -> EventLoopFuture<JobData> { self.eventLoop.future(error: self.failure) }
+    func set(_: JobIdentifier, to: JobData) -> EventLoopFuture<Void> { self.eventLoop.future(error: self.failure) }
+    func clear(_: JobIdentifier) -> EventLoopFuture<Void> { self.eventLoop.future(error: self.failure) }
+    func push(_: JobIdentifier) -> EventLoopFuture<Void> { self.eventLoop.future(error: self.failure) }
+    func pop() -> EventLoopFuture<JobIdentifier?> { self.eventLoop.future(error: self.failure) }
+}