add producer wrapper with timestamp behavior test

Author: 1mpossible-code

pkg/producer/producer.go

@@ -0,0 +1,31 @@
+package producer
+
+import (
+	"context"
+	"time"
+
+	"distributed_job_queue/pkg/metrics"
+	"distributed_job_queue/pkg/queue"
+)
+
+type Publisher struct {
+	broker  queue.Broker
+	metrics *metrics.Collector
+}
+
+func New(broker queue.Broker, m *metrics.Collector) *Publisher {
+	return &Publisher{broker: broker, metrics: m}
+}
+
+func (p *Publisher) Enqueue(ctx context.Context, job queue.Job) error {
+	if job.EnqueuedAt.IsZero() {
+		job.EnqueuedAt = time.Now().UTC()
+	}
+	if err := p.broker.Enqueue(ctx, job); err != nil {
+		return err
+	}
+	if p.metrics != nil {
+		p.metrics.Enqueued.Inc()
+	}
+	return nil
+}

pkg/producer/producer_test.go

@@ -0,0 +1,36 @@
+package producer
+
+import (
+	"context"
+	"testing"
+
+	"distributed_job_queue/pkg/queue"
+)
+
+type fakeBroker struct{ job queue.Job }
+
+func (f *fakeBroker) Enqueue(_ context.Context, job queue.Job) error { f.job = job; return nil }
+func (f *fakeBroker) Reserve(context.Context, string) (queue.Job, queue.Lease, error) {
+	return queue.Job{}, queue.Lease{}, nil
+}
+func (f *fakeBroker) Ack(context.Context, queue.Lease) error { return nil }
+func (f *fakeBroker) Nack(context.Context, queue.Lease, queue.RetryDecision) error { return nil }
+func (f *fakeBroker) RequeueExpired(context.Context) (int64, error) { return 0, nil }
+
+func TestEnqueueSetsTimestamp(t *testing.T) {
+	fb := &fakeBroker{}
+	p := New(fb, nil)
+	err := p.Enqueue(context.Background(), queue.Job{
+		ID:             "job-1",
+		Type:           "t",
+		IdempotencyKey: "idem-1",
+		Priority:       queue.PriorityHigh,
+		MaxAttempts:    3,
+	})
+	if err != nil {
+		t.Fatalf("enqueue failed: %v", err)
+	}
+	if fb.job.EnqueuedAt.IsZero() {
+		t.Fatal("expected enqueued timestamp")
+	}
+}