Merge pull request #16 from m-barthelemy/bugfix/delayed-jobs-block-th…

…e-queues

Ensure delayed jobs are ignored until it's time to run them

Sources/QueuesFluentDriver/FluentQueue.swift

@@ -25,6 +25,9 @@ extension FluentQueue: Queue {
     public func set(_ id: JobIdentifier, to jobStorage: JobData) -> EventLoopFuture<Void> {
         do {
             let jobModel = try JobModel(jobId: id.string, queue: queueName.string, data: jobStorage)
+            // If the job must run at a later time, ensure it won't be picked earlier since
+            // we sort pending jobs by creation date when querying
+            jobModel.createdAt = jobStorage.delayUntil ?? Date()
             return jobModel.save(on: db)
         }
         catch {
@@ -73,6 +76,7 @@ extension FluentQueue: Queue {
             .from(JobModel.schema)
             .where(SQLColumn("\(FieldKey.state)"), .equal, SQLBind(QueuesFluentJobState.pending))
             .where(SQLColumn("\(FieldKey.queue)"), .equal, SQLBind(self.queueName.string))
+            .where(SQLColumn("\(FieldKey.createdAt)"), .lessThanOrEqual, SQLBind(Date()))
             .orderBy("\(FieldKey.createdAt)")
             .limit(1)
         if self.dbType != .sqlite {

Sources/QueuesFluentDriver/JobModel.swift

@@ -39,14 +39,14 @@ class JobModel: Model {
 
     /// The Job data
     @Field(key: .data)
-    //var data: JobData?
     var data: Data
 
     /// The current state of the Job
     @Field(key: .state)
     var state: QueuesFluentJobState
 
-    @Timestamp(key: .createdAt, on: .create)
+    /// Creation date by default; `delayUntil` if it's a delayed job
+    @OptionalField(key: .createdAt)
     var createdAt: Date?
 
     @Timestamp(key: .updatedAt, on: .update)

Sources/QueuesFluentDriver/PopQueries/PostgresPopQuery.swift

@@ -16,7 +16,7 @@ final class PostgresPop: PopQueryProtocol {
             )
             .returning(SQLColumn("\(FieldKey.jobId)"))
             .query
-        
+
         var id: String?
         return db.execute(sql: query) { (row) -> Void in
             id = try? row.decode(column: "\(FieldKey.jobId)", as: String.self)