queue.go

package main

import (
	"sync"
	"time"

	"github.com/rs/xid"
)

type Message struct {
	ID        string
	Body      string
	VisibleAt time.Time
	Receipt   string
	CreatedAt time.Time
	Locked    bool
}

type Queue struct {
	config   QueueConfig
	mu       sync.Mutex
	messages []*Message
}

func NewQueue(config QueueConfig) *Queue {
	return &Queue{config: config, messages: []*Message{}}
}

func (q *Queue) sendMessage(body string) *Message {
	q.mu.Lock()
	defer q.mu.Unlock()
	msg := &Message{
		ID:        xid.New().String(),
		Body:      body,
		VisibleAt: time.Now(),
		Receipt:   xid.New().String(),
		CreatedAt: time.Now(),
	}
	q.messages = append(q.messages, msg)
	return msg
}

func (q *Queue) receiveMessages(max int) []*Message {
	q.mu.Lock()
	defer q.mu.Unlock()
	var received []*Message
	now := time.Now()
	for _, msg := range q.messages {
		if len(received) >= max {
			break
		}
		if !msg.Locked && (msg.VisibleAt.Before(now) || msg.VisibleAt.Equal(now)) {
			msg.VisibleAt = now.Add(time.Duration(q.config.VisibilityTimeout) * time.Second)
			msg.Locked = true
			received = append(received, msg)
		}
	}
	return received
}

func (q *Queue) deleteMessage(receipt string) bool {
	q.mu.Lock()
	defer q.mu.Unlock()
	for i, msg := range q.messages {
		if msg.Receipt == receipt {
			q.messages = append(q.messages[:i], q.messages[i+1:]...)
			return true
		}
	}
	return false
}

func (q *Queue) requeueExpiredMessages() {
	q.mu.Lock()
	defer q.mu.Unlock()
	now := time.Now()
	for _, msg := range q.messages {
		if msg.Locked && now.After(msg.VisibleAt) {
			msg.Locked = false
		}
	}
}

func (q *Queue) purge() {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.messages = []*Message{}
}

func (q *Queue) stats() map[string]any {
	q.mu.Lock()
	defer q.mu.Unlock()
	available := 0
	inFlight := 0
	now := time.Now()
	for _, msg := range q.messages {
		if msg.Locked && now.Before(msg.VisibleAt) {
			inFlight++
		} else {
			available++
		}
	}
	return map[string]any{
		"messagesAvailable": available,
		"messagesInFlight":  inFlight,
	}
}

func (q *Queue) extendVisibilityTimeout(receipt string, extraSeconds int) bool {
	q.mu.Lock()
	defer q.mu.Unlock()
	for _, msg := range q.messages {
		if msg.Receipt == receipt && msg.Locked {
			msg.VisibleAt = msg.VisibleAt.Add(time.Duration(extraSeconds) * time.Second)
			return true
		}
	}
	return false
}