add locker job

examples/config/advanced/locker.yaml

@@ -0,0 +1,23 @@
+jobs:
+  - type: lock
+    args:
+      key: test
+      job:
+        type: loop
+        args:
+          interval: 1s
+          job:
+            type: log
+            args:
+              text: test
+  - type: lock
+    args:
+      key: test
+      job:
+        type: loop
+        args:
+          interval: 1s
+          job:
+            type: log
+            args:
+              text: test2

src/job/base.go

@@ -129,6 +129,8 @@ func Get(t string) Job {
 		return timeoutJob
 	case "loop":
 		return loopJob
+	case "lock":
+		return lockJob
 	case "js":
 		return jsJob
 	case "encrypted":

src/job/utils.go

@@ -38,6 +38,8 @@ import (
 	"github.com/Arriven/db1000n/src/utils/templates"
 )
 
+var locker utils.Locker
+
 // "log" in config
 func logJob(ctx context.Context, args config.Args, globalConfig *GlobalConfig, a *metrics.Accumulator, logger *zap.Logger) (
 	data any, err error, //nolint:unparam // data is here to match Job
@@ -190,6 +192,32 @@ func loopJob(ctx context.Context, args config.Args, globalConfig *GlobalConfig,
 	return nil, nil
 }
 
+func lockJob(ctx context.Context, args config.Args, globalConfig *GlobalConfig, a *metrics.Accumulator, logger *zap.Logger) (data any, err error) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	var jobConfig struct {
+		BasicJobConfig
+
+		Key string
+		Job config.Config
+	}
+
+	if err := mapstructure.Decode(templates.ParseAndExecuteMapStruct(logger, args, ctx), &jobConfig); err != nil {
+		return nil, fmt.Errorf("error parsing job config: %w", err)
+	}
+
+	unlock := locker.Lock(jobConfig.Key)
+	defer unlock()
+
+	job := Get(jobConfig.Job.Type)
+	if job == nil {
+		return nil, fmt.Errorf("unknown job %q", jobConfig.Job.Type)
+	}
+
+	return job(ctx, jobConfig.Job.Args, globalConfig, a, logger)
+}
+
 // "js" in config
 func jsJob(ctx context.Context, args config.Args, globalConfig *GlobalConfig, a *metrics.Accumulator, logger *zap.Logger) (
 	data any, err error,

src/utils/locker.go

@@ -0,0 +1,15 @@
+package utils
+
+import "sync"
+
+type Locker struct {
+	mutexes sync.Map // Zero value is empty and ready for use
+}
+
+func (m *Locker) Lock(key string) func() {
+	value, _ := m.mutexes.LoadOrStore(key, &sync.Mutex{})
+	mtx := value.(*sync.Mutex)
+	mtx.Lock()
+
+	return func() { mtx.Unlock() }
+}