initial

app/storage/mongo/client.go

@@ -0,0 +1,25 @@
+package client
+
+import (
+	"context"
+	"fmt"
+	"queues-go/app/utils"
+
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+var Database mongo.Database
+
+func Connect() {
+	connectionString := fmt.Sprintf("mongodb://mongo:%s@%s:27017/", utils.GetEnv("MONGO_PASSWORD", "password"), utils.GetEnv("MONGO_HOST", "localhost"))
+	client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(connectionString))
+	if err != nil {
+		panic(err)
+	}
+	db := client.Database("queues")
+	if db == nil {
+		panic(fmt.Errorf("DATABASE DOES NOT EXIST"))
+	}
+	Database = *db
+}

app/storage/mongo/collections/tasks.go

@@ -0,0 +1,105 @@
+package tasks
+
+import (
+	"context"
+	client "queues-go/app/storage/mongo"
+	errors "queues-go/app/utils"
+	"time"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+type Task struct {
+	Queue            string             `bson:"queue"`
+	Payload          bson.M             `bson:"payload"`
+	PutAt            time.Time          `bson:"put_at"`
+	AvailableFrom    time.Time          `bson:"available_from"`
+	SecondsToExecute int                `bson:"seconds_to_execute"`
+	Id               primitive.ObjectID `bson:"_id"`
+	TakenAt          *time.Time         `bson:"taken_at"`
+	Attempts         int                `bson:"attempts"`
+}
+
+type InsertedTask struct {
+	Queue            string      `bson:"queue"`
+	Payload          interface{} `bson:"payload"`
+	PutAt            time.Time   `bson:"put_at"`
+	AvailableFrom    time.Time   `bson:"available_from"`
+	SecondsToExecute int         `bson:"seconds_to_execute"`
+	TakenAt          *time.Time  `bson:"taken_at"`
+	Attempts         int         `bson:"attempts"`
+}
+
+func Add(task InsertedTask) error {
+	_, err := collection().InsertOne(context.TODO(), task)
+	if err != nil {
+		return errors.ErrInternalError
+	}
+	return nil
+}
+
+func Finish(id string) error {
+	objectId, err := primitive.ObjectIDFromHex(id)
+	if err != nil {
+		return errors.ErrIncorrectFormat
+	}
+
+	_, err = collection().DeleteOne(context.TODO(), bson.M{"_id": objectId})
+	if err != nil {
+		return errors.ErrInternalError
+	}
+	return nil
+}
+
+func Take(queue string) (*Task, error) {
+	now := time.Now()
+	task, err := findTask(queue, now)
+	if err != nil {
+		return nil, errors.ErrInternalError
+	}
+	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}})
+	if err != nil {
+		return nil, errors.ErrInternalError
+	}
+	return task, nil
+}
+
+func findTask(queue string, now time.Time) (*Task, error) {
+	cursor, err := collection().Find(
+		context.TODO(),
+		bson.M{
+			"queue":          queue,
+			"available_from": bson.M{"$lte": now},
+		},
+	)
+	if err != nil {
+		return nil, errors.ErrInternalError
+	}
+
+	var results []Task
+	err = cursor.All(context.TODO(), &results)
+	if err != nil {
+		return nil, errors.ErrInternalError
+	}
+
+	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 nil, nil
+}
+
+func collection() *mongo.Collection {
+	return client.Database.Collection("tasks")
+}