Fully overhaul package for modern Swift/Vapor/Fluent

.github/CODEOWNERS

@@ -0,0 +1 @@
+* @gwynne

.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,35 @@ 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"),
+        .package(url: "https://github.com/vapor/fluent-sqlite-driver.git", from: "4.6.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: [
+                .product(name: "XCTVapor", package: "vapor"),
+                .product(name: "FluentSQLiteDriver", package: "fluent-sqlite-driver"),
+                .target(name: "QueuesFluentDriver"),
+            ]
         ),
     ]
 )

Package@swift-5.9.swift

@@ -0,0 +1,55 @@
+// swift-tools-version:5.9
+import PackageDescription
+
+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"),
+        .package(url: "https://github.com/vapor/fluent-sqlite-driver.git", from: "4.6.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: [
+                .product(name: "XCTVapor", package: "vapor"),
+                .product(name: "FluentSQLiteDriver", package: "fluent-sqlite-driver"),
+                .target(name: "QueuesFluentDriver"),
+            ],
+            swiftSettings: swiftSettings
+        ),
+    ]
+)
+
+var swiftSettings: [SwiftSetting] { [
+    .enableUpcomingFeature("ForwardTrailingClosures"),
+    .enableUpcomingFeature("ExistentialAny"),
+    .enableUpcomingFeature("ConciseMagicFile"),
+    .enableUpcomingFeature("DisableOutwardActorInference"),
+    .enableUpcomingFeature("StrictConcurrency"),
+    .enableExperimentalFeature("StrictConcurrency=complete"),
+] }

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,103 +1,97 @@
-import Foundation
-import Queues
-import Fluent
-import SQLKit
+@preconcurrency import Queues
+@preconcurrency import SQLKit
+@preconcurrency import FluentKit
 
-public struct FluentQueue {
+/// An implementation of `Queue` which stores job data and metadata in a Fluent database.
+public struct FluentQueue: Queue, Sendable {
+    // See `Queue.context`.
     public let context: QueueContext
-    let db: Database
-    let dbType: QueuesFluentDbType
-    let useSoftDeletes: Bool
-}
 
-extension FluentQueue: Queue {
+    let db: any Database
+    let sqlDb: any SQLDatabase
+
+    // See `Queue.get(_:)`.
     public func get(_ id: JobIdentifier) -> EventLoopFuture<JobData> {
-        return db.query(JobModel.self)
-            .filter(\.$id == id.string)
+        self.sqlDb.select()
+            .columns(\JobModel.$payload, \JobModel.$maxRetryCount, \JobModel.$jobName,
+                     \JobModel.$delayUntil, \JobModel.$queuedAt, \JobModel.$attempts)
+            .from(JobModel.self)
+            .where(\JobModel.$id, .equal, id.string)
             .first()
             .unwrap(or: QueuesFluentError.missingJob(id))
-            .flatMapThrowing { job in
-                return JobData(
-                    payload: Array(job.data.payload),
-                    maxRetryCount: job.data.maxRetryCount,
-                    jobName: job.data.jobName,
-                    delayUntil: job.data.delayUntil,
-                    queuedAt: job.data.queuedAt,
-                    attempts: job.data.attempts ?? 0
-                )
+            .flatMapThrowing {
+                try $0.decode(model: JobData.self, keyDecodingStrategy: .convertFromSnakeCase)
             }
     }
 
+    // See `Queue.get(_:to:)`.
     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
-        }
+        JobModel(id: id, queue: queueName.string, jobData: jobStorage).save(on: self.db)
     }
 
+    // See `Queue.clear(_:)`.
     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)
+        self.db.query(JobModel.self)
             .filter(\.$id == id.string)
             .filter(\.$state != .completed)
             .first()
             .unwrap(or: QueuesFluentError.missingJob(id))
-            .flatMap { job in
-                if self.useSoftDeletes {
-                    job.state = .completed
-                    job.deletedAt = Date()
-                    return job.update(on: self.db)
-                } else {
-                    return job.delete(force: true, on: self.db)
-                }
-        }
+            .flatMap { $0.delete(force: true, on: self.db) }
     }
 
+    // See `Queue.push(_:)`.
     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.$state, to: QueuesFluentJobState.pending)
+            .where(\JobModel.$id, .equal, id.string)
             .run()
     }
 
-    /// Currently selects the oldest job pending execution
+    // See `Queue.pop()`.
     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` when it becomes available in upstream SQLKit.
+            let select = self.sqlDb
+                .select()
+                .column(\JobModel.$id)
+                .from(JobModel.self)
+                .where(\JobModel.$state, .equal, QueuesFluentJobState.pending)
+                .where(\JobModel.$queue, .equal, self.queueName.string)
+                .where(.dateValue(.function("coalesce", JobModel.sqlColumn(\.$delayUntil), SQLNow())), .lessThanOrEqual, .now())
+                .orderBy(\JobModel.$delayUntil)
+                .limit(1)
+                .lockingClause(SQLLockingClauseWithSkipLocked.updateSkippingLocked)
+
+            if self.sqlDb.dialect.supportsReturning {
+                return try await self.sqlDb.update(JobModel.sqlTable)
+                    .set(\JobModel.$state, to: QueuesFluentJobState.processing)
+                    .set(\JobModel.$updatedAt, to: .now())
+                    .where(\JobModel.$id, .equal, .group(select.query))
+                    .returning(JobModel.sqlColumn(\.$id))
+                    .first(decodingColumn: JobModel.key(for: \.$id), as: String.self)
+                    .map(JobIdentifier.init(string:))
+            } else {
+                return try await self.db.transaction { transaction in
+                    let database = transaction as! any SQLDatabase
+
+                    guard let id = try await database.raw("\(select.query)") // using raw() to make sure we run on the transaction connection
+                        .first(decodingColumn: JobModel.key(for: \.$id), as: String.self)
+                    else {
+                        return nil
+                    }
+
+                    try await database
+                        .update(JobModel.sqlTable)
+                        .set(\JobModel.$state, to: QueuesFluentJobState.processing)
+                        .set(\JobModel.$updatedAt, to: .now())
+                        .where(\JobModel.$id, .equal, id)
+                        .where(\JobModel.$state, .equal, QueuesFluentJobState.pending)
+                        .run()
+
+                    return JobIdentifier(string: id)
+                }
+            }
         }
     }
 }