.gitea/workflows/deploy-dev.yaml

@@ -9,7 +9,7 @@ on:
 jobs:
   build:
     name: Build
-    runs-on: [ dev ]
+    runs-on: [dev]
     steps:
       - name: login
         run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
@@ -21,18 +21,11 @@ jobs:
         run: docker build -t mathwave/sprint-repo:queues .
   push:
     name: Push
-    runs-on: [ dev ]
+    runs-on: [dev]
     needs: build
     steps:
       - name: push
         run: docker push mathwave/sprint-repo:queues
-  create_dir:
-    name: Create dir
-    runs-on: [ dev ]
-    needs: build
-    steps:
-      - name: create_dir
-        run: mkdir /sprint-data/queues-mongo || true
   deploy-dev:
     name: Deploy dev
     runs-on: [prod]

.gitea/workflows/deploy-prod.yaml

@@ -9,7 +9,7 @@ on:
 jobs:
   build:
     name: Build
-    runs-on: [ dev ]
+    runs-on: [dev]
     steps:
       - name: login
         run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
@@ -21,18 +21,11 @@ jobs:
         run: docker build -t mathwave/sprint-repo:queues .
   push:
     name: Push
-    runs-on: [ dev ]
+    runs-on: [dev]
     needs: build
     steps:
       - name: push
         run: docker push mathwave/sprint-repo:queues
-  create_dir:
-    name: Create dir
-    runs-on: [ prod ]
-    needs: build
-    steps:
-      - name: create_dir
-        run: mkdir /sprint-data/queues-mongo || true
   deploy-prod:
     name: Deploy prod
     runs-on: [prod]

app/storage/mongo/collections/tasks.go

@@ -9,6 +9,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
 type Task struct {
@@ -62,7 +63,19 @@ func Take(queue string) (*Task, error) {
 	if task == nil {
 		return nil, nil
 	}
-	_, err = collection().UpdateByID(context.TODO(), task.Id, bson.M{"$set": bson.M{"taken_at": now, "attempts": task.Attempts + 1}})
+	_, err = collection().UpdateByID(
+		context.TODO(),
+		task.Id,
+		bson.M{
+			"$set": bson.M{
+				"taken_at": now,
+				"attempts": task.Attempts + 1,
+				"available_from": now.Add(
+					time.Duration(task.SecondsToExecute) * time.Second,
+				),
+			},
+		},
+	)
 	if err != nil {
 		println("Error updaing")
 		println(err.Error())
@@ -78,6 +91,7 @@ func findTask(queue string, now time.Time) (*Task, error) {
 			"queue":          queue,
 			"available_from": bson.M{"$lte": now},
 		},
+		options.Find().SetLimit(1),
 	)
 	if err != nil {
 		println("Error find")
@@ -94,14 +108,7 @@ func findTask(queue string, now time.Time) (*Task, error) {
 	}
 
 	for _, task := range results {
-		if task.TakenAt == nil {
-			return &task, nil
-		}
-		takenAt := *task.TakenAt
-
-		if takenAt.Add(time.Second * time.Duration(task.SecondsToExecute)).Before(now) {
-			return &task, nil
-		}
+		return &task, nil
 	}
 	return nil, nil
 }