package model import ( "bytes" "context" "encoding/json" "sync" "time" "spark-server/internal/module/deliver/entity" ) type MemoryQueue struct { mu sync.Mutex queue [][]byte processing [][]byte wait chan struct{} } func NewMemoryQueue() *MemoryQueue { return &MemoryQueue{wait: make(chan struct{}, 1)} } func (q *MemoryQueue) Push(_ context.Context, job *entity.Job) error { raw, err := json.Marshal(job) if err != nil { return err } q.mu.Lock() q.queue = append(q.queue, raw) q.mu.Unlock() select { case q.wait <- struct{}{}: default: } return nil } func (q *MemoryQueue) Take(ctx context.Context, timeout time.Duration) (*entity.Job, error) { deadline := time.Now().Add(timeout) for { q.mu.Lock() if len(q.queue) > 0 { raw := q.queue[0] q.queue = q.queue[1:] q.processing = append(q.processing, raw) q.mu.Unlock() job := &entity.Job{} if err := json.Unmarshal(raw, job); err != nil { return nil, err } job.SetRaw(raw) return job, nil } q.mu.Unlock() remain := time.Until(deadline) if remain <= 0 { return nil, nil } select { case <-ctx.Done(): return nil, ctx.Err() case <-time.After(remain): return nil, nil case <-q.wait: } } } func (q *MemoryQueue) Ack(_ context.Context, job *entity.Job) error { if job == nil { return nil } raw := job.Raw() q.mu.Lock() defer q.mu.Unlock() for i, p := range q.processing { if bytes.Equal(p, raw) { q.processing = append(q.processing[:i], q.processing[i+1:]...) return nil } } return nil } func (q *MemoryQueue) RequeueStale(context.Context) error { q.mu.Lock() defer q.mu.Unlock() if len(q.processing) == 0 { return nil } q.queue = append(q.processing, q.queue...) q.processing = nil return nil } type MemoryDone struct { mu sync.Mutex seen map[int64]struct{} } func NewMemoryDone() *MemoryDone { return &MemoryDone{seen: make(map[int64]struct{})} } func (d *MemoryDone) Mark(_ context.Context, msgID int64) (bool, error) { if msgID <= 0 { return true, nil } d.mu.Lock() defer d.mu.Unlock() if _, ok := d.seen[msgID]; ok { return false, nil } d.seen[msgID] = struct{}{} return true, nil }