98 lines
2 KiB
Go
98 lines
2 KiB
Go
package jobs
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.nakama.town/fmartingr/hako/internal/model"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// MemoryQueue implements an in-memory job queue
|
|
type MemoryQueue struct {
|
|
jobs chan *model.Job
|
|
mu sync.Mutex
|
|
jobMap map[string]*model.Job
|
|
}
|
|
|
|
// NewMemoryQueue creates a new in-memory queue
|
|
func NewMemoryQueue() *MemoryQueue {
|
|
return &MemoryQueue{
|
|
jobs: make(chan *model.Job, 100), // Buffer of 100 jobs
|
|
jobMap: make(map[string]*model.Job),
|
|
}
|
|
}
|
|
|
|
// Enqueue adds a new job to the queue
|
|
func (q *MemoryQueue) Enqueue(ctx context.Context, jobType model.JobType, payload any) error {
|
|
// Marshal payload to JSON
|
|
payloadBytes, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to marshal payload: %w", err)
|
|
}
|
|
|
|
job := &model.Job{
|
|
ID: uuid.New().String(),
|
|
Type: jobType,
|
|
Payload: string(payloadBytes),
|
|
Status: "pending",
|
|
CreatedAt: time.Now(),
|
|
}
|
|
|
|
q.mu.Lock()
|
|
q.jobMap[job.ID] = job
|
|
q.mu.Unlock()
|
|
|
|
// Send to channel
|
|
select {
|
|
case q.jobs <- job:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
|
|
// Dequeue retrieves the next job from the queue
|
|
func (q *MemoryQueue) Dequeue(ctx context.Context) (*model.Job, error) {
|
|
select {
|
|
case job := <-q.jobs:
|
|
q.mu.Lock()
|
|
job.Status = "processing"
|
|
q.mu.Unlock()
|
|
return job, nil
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
default:
|
|
return nil, nil // No job available
|
|
}
|
|
}
|
|
|
|
// Complete marks a job as completed
|
|
func (q *MemoryQueue) Complete(ctx context.Context, jobID string) error {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
|
|
if job, ok := q.jobMap[jobID]; ok {
|
|
job.Status = "completed"
|
|
return nil
|
|
}
|
|
|
|
return fmt.Errorf("job not found: %s", jobID)
|
|
}
|
|
|
|
// Fail marks a job as failed
|
|
func (q *MemoryQueue) Fail(ctx context.Context, jobID string, errorMsg string) error {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
|
|
if job, ok := q.jobMap[jobID]; ok {
|
|
job.Status = "failed"
|
|
// Store error message in payload for now
|
|
return nil
|
|
}
|
|
|
|
return fmt.Errorf("job not found: %s", jobID)
|
|
}
|