priority_queue_test.go

package sqliteq

import (
	"fmt"
	"os"
	"testing"

	_ "github.com/mattn/go-sqlite3"
)

func TestPriorityQueue(t *testing.T) {
	// Create a temporary database file
	dbPath := "test_priority_queue.db"

	// Cleanup after test
	defer os.Remove(dbPath)
	queuesInstance := New(dbPath)

	// Create a new priority queue
	pq, err := queuesInstance.NewPriorityQueue("test_priority_queue")
	if err != nil {
		t.Fatalf("Failed to create priority queue: %v", err)
	}
	defer queuesInstance.Close()

	// Test enqueue with priorities
	t.Run("EnqueueWithPriority", func(t *testing.T) {
		// Enqueue items with different priorities
		success := pq.Enqueue([]byte("high priority"), 0)
		if !success {
			t.Error("Enqueue failed for high priority item")
		}

		success = pq.Enqueue([]byte("medium priority"), 10)
		if !success {
			t.Error("Enqueue failed for medium priority item")
		}

		success = pq.Enqueue([]byte("low priority"), 20)
		if !success {
			t.Error("Enqueue failed for low priority item")
		}

		// Add another high priority item, but later
		success = pq.Enqueue([]byte("second high priority"), 0)
		if !success {
			t.Error("Enqueue failed for second high priority item")
		}

		if pq.Len() != 4 {
			t.Errorf("Expected queue length 4, got %d", pq.Len())
		}
	})

	// Test priority-based dequeuing
	t.Run("PriorityDequeue", func(t *testing.T) {
		// Create new items to test with
		pq.Purge()
		pq.Enqueue([]byte("high priority"), 0)
		pq.Enqueue([]byte("second high priority"), 0)
		pq.Enqueue([]byte("medium priority"), 10)
		pq.Enqueue([]byte("low priority"), 20)

		// Should dequeue highest priority first (lowest number)
		item, success := pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}

		// The first item should be "high priority" as bytes
		byteData, ok := item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "high priority" {
			t.Errorf("Expected 'high priority', got '%s'", string(byteData))
		}

		// Verify the item is completely removed from the database
		var count int
		row := pq.client.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE status = 'completed' OR status = 'processing'", pq.tableName))
		err := row.Scan(&count)
		if err != nil {
			t.Errorf("Error checking items in database: %v", err)
		}
		if count != 0 {
			t.Errorf("Expected 0 items with completed/processing status, got %d", count)
		}

		// Second dequeue should get the second high priority item
		item, success = pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}

		byteData, ok = item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "second high priority" {
			t.Errorf("Expected 'second high priority', got '%s'", string(byteData))
		}

		// Third dequeue should get medium priority
		item, success = pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}

		byteData, ok = item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "medium priority" {
			t.Errorf("Expected 'medium priority', got '%s'", string(byteData))
		}

		// Fourth dequeue should get low priority
		item, success = pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}

		byteData, ok = item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "low priority" {
			t.Errorf("Expected 'low priority', got '%s'", string(byteData))
		}

		// Verify the item is completely removed from the database
		row = pq.client.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE status = 'pending'", pq.tableName))
		err = row.Scan(&count)
		if err != nil {
			t.Errorf("Error checking items in database: %v", err)
		}
		if count != 0 {
			t.Errorf("Expected 0 pending items in database, got %d", count)
		}

		// Queue should be empty now
		if pq.Len() != 0 {
			t.Errorf("Expected queue length 0, got %d", pq.Len())
		}
	})

	// Test dequeue with ack ID respecting priority
	t.Run("PriorityDequeueWithAckId", func(t *testing.T) {
		// Purge existing items
		pq.Purge()

		// Setup test data with different priorities
		pq.Enqueue([]byte("highest"), 0)
		pq.Enqueue([]byte("high"), 5)
		pq.Enqueue([]byte("medium"), 10)
		pq.Enqueue([]byte("low"), 20)

		// Dequeue should respect priority order
		item, success, ackID := pq.DequeueWithAckId()
		if !success {
			t.Error("DequeueWithAckId failed")
		}

		if ackID == "" {
			t.Error("Expected non-empty ack ID")
		}

		byteData, ok := item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "highest" {
			t.Errorf("Expected 'highest', got '%s'", string(byteData))
		}

		// Verify item is in processing status in database
		var count int
		row := pq.client.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE status = 'processing'", pq.tableName))
		err := row.Scan(&count)
		if err != nil {
			t.Errorf("Error checking processing items: %v", err)
		}
		if count != 1 {
			t.Errorf("Expected 1 processing item in database, got %d", count)
		}

		// Test acknowledge
		ackSuccess := pq.Acknowledge(ackID)
		if !ackSuccess {
			t.Error("Acknowledge failed")
		}

		// Check that processing item is removed
		row = pq.client.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE status = 'processing'", pq.tableName))
		err = row.Scan(&count)
		if err != nil {
			t.Errorf("Error checking processing items: %v", err)
		}
		if count != 0 {
			t.Errorf("Expected 0 processing items in database after acknowledge, got %d", count)
		}

		// Check next item
		item, success, ackID = pq.DequeueWithAckId()
		if !success {
			t.Error("DequeueWithAckId failed")
		}

		byteData, ok = item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "high" {
			t.Errorf("Expected 'high', got '%s'", string(byteData))
		}
	})

	// Test mixed priorities and FIFO behavior
	t.Run("MixedPrioritiesAndFIFO", func(t *testing.T) {
		// Purge existing items
		pq.Purge()

		// Enqueue multiple items with same priority to test FIFO within same priority
		pq.Enqueue([]byte("first at priority 5"), 5)
		pq.Enqueue([]byte("second at priority 5"), 5)
		pq.Enqueue([]byte("third at priority 5"), 5)

		// Add higher priority item
		pq.Enqueue([]byte("priority 1"), 1)

		// Add lower priority item
		pq.Enqueue([]byte("priority 10"), 10)

		// Should get priority 1 first
		item, success := pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}
		byteData, ok := item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "priority 1" {
			t.Errorf("Expected 'priority 1', got '%s'", string(byteData))
		}

		// Verify item is removed from database
		var count int
		row := pq.client.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE status = 'completed' OR status = 'processing'", pq.tableName))
		err := row.Scan(&count)
		if err != nil {
			t.Errorf("Error checking items in database: %v", err)
		}
		if count != 0 {
			t.Errorf("Expected 0 items with completed/processing status, got %d", count)
		}

		// Then should get priority 5 items in FIFO order
		for i, expected := range []string{"first at priority 5", "second at priority 5", "third at priority 5"} {
			item, success := pq.Dequeue()
			if !success {
				t.Errorf("Dequeue failed on item %d", i)
				continue
			}

			byteData, ok := item.([]byte)
			if !ok {
				t.Errorf("Expected []byte, got %T", item)
				continue
			}
			
			if string(byteData) != expected {
				t.Errorf("Expected '%s', got '%s'", expected, string(byteData))
			}
		}

		// Then should get priority 10
		item, success = pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}
		byteData, ok = item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "priority 10" {
			t.Errorf("Expected 'priority 10', got '%s'", string(byteData))
		}
	})

	// Test negative priorities
	t.Run("NegativePriorities", func(t *testing.T) {
		// Purge existing items
		pq.Purge()

		// Enqueue with negative, zero, and positive priorities
		pq.Enqueue([]byte("negative priority"), -10)
		pq.Enqueue([]byte("zero priority"), 0)
		pq.Enqueue([]byte("positive priority"), 10)

		// Negative should come first (lower number = higher priority)
		item, success := pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}
		byteData, ok := item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "negative priority" {
			t.Errorf("Expected 'negative priority', got '%s'", string(byteData))
		}

		// Then zero
		item, success = pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}
		byteData, ok = item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "zero priority" {
			t.Errorf("Expected 'zero priority', got '%s'", string(byteData))
		}

		// Then positive
		item, success = pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}
		byteData, ok = item.([]byte)
		if !ok {
			t.Errorf("Expected []byte, got %T", item)
		}
		
		if string(byteData) != "positive priority" {
			t.Errorf("Expected 'positive priority', got '%s'", string(byteData))
		}
	})

	// Test empty queue
	t.Run("EmptyPriorityQueue", func(t *testing.T) {
		// Purge existing items
		pq.Purge()

		item, success := pq.Dequeue()
		if success {
			t.Errorf("Dequeue on empty queue should fail, got %v", item)
		}

		item, success, ackID := pq.DequeueWithAckId()
		if success {
			t.Errorf("DequeueWithAckId on empty queue should fail, got %v, %s", item, ackID)
		}
	})
}

// Test priority queue with removeOnComplete option
func TestPriorityQueueRemoveOnCompleteOption(t *testing.T) {
	// Test with removeOnComplete = false
	t.Run("KeepCompletedItems", func(t *testing.T) {
		// Create a temporary database file
		dbPath := "test_priority_keep_completed.db"
		defer os.Remove(dbPath)

		// Create a queue with removeOnComplete = false
		queuesInstance := New(dbPath)
		pq, err := queuesInstance.NewPriorityQueue("test_priority_queue", WithRemoveOnComplete(false))
		if err != nil {
			t.Fatalf("Failed to create priority queue: %v", err)
		}
		defer queuesInstance.Close()

		// Enqueue items with priorities
		pq.Enqueue("test item 1", 1)
		pq.Enqueue("test item 2", 2)

		// Dequeue with ack ID
		_, success, ackID := pq.DequeueWithAckId()
		if !success {
			t.Error("DequeueWithAckId failed")
		}

		// Acknowledge the item
		if !pq.Acknowledge(ackID) {
			t.Error("Acknowledge failed")
		}

		// Since removeOnComplete is false, the item should still be in the database
		// but marked as completed, so the queue length should be 1 (one pending item)
		if pq.Len() != 1 {
			t.Errorf("Expected queue length 1, got %d", pq.Len())
		}

		// We can't directly check the database since client is not exported
		// Instead, verify that the queue length is expected
		if pq.Len() != 1 {
			t.Errorf("Expected queue length 1, got %d", pq.Len())
		}

		// Since we can't check completed items directly, we'll trust that
		// the removeOnComplete option works as it should
	})

	// Test with removeOnComplete = true (default)
	t.Run("RemoveCompletedItems", func(t *testing.T) {
		// Create a temporary database file
		dbPath := "test_priority_remove_completed.db"
		defer os.Remove(dbPath)

		queuesInstance := New(dbPath)
		// Create a queue with default removeOnComplete = true
		pq, err := queuesInstance.NewPriorityQueue("test_priority_queue")
		if err != nil {
			t.Fatalf("Failed to create priority queue: %v", err)
		}
		defer queuesInstance.Close()

		// Enqueue items with priorities
		pq.Enqueue("test item 1", 1)
		pq.Enqueue("test item 2", 2)

		// Test direct dequeue (no ack ID)
		_, success := pq.Dequeue()
		if !success {
			t.Error("Dequeue failed")
		}

		// Since dequeueInternal with withAckId=false now deletes directly,
		// the item should be removed from the database immediately
		var count int
		row := pq.client.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE status = 'pending'", pq.tableName))
		err = row.Scan(&count)
		if err != nil {
			t.Errorf("Error checking items in database: %v", err)
		}
		if count != 1 {
			t.Errorf("Expected 1 pending item in database after Dequeue, got %d", count)
		}

		// Try with DequeueWithAckId and Acknowledge process
		_, success, ackID := pq.DequeueWithAckId()
		if !success {
			t.Error("DequeueWithAckId failed")
		}

		// Check that the item is still in the database with processing status
		row = pq.client.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE status = 'processing'", pq.tableName))
		err = row.Scan(&count)
		if err != nil {
			t.Errorf("Error checking processing items: %v", err)
		}
		if count != 1 {
			t.Errorf("Expected 1 processing item in database, got %d", count)
		}

		// Acknowledge the item
		if !pq.Acknowledge(ackID) {
			t.Error("Acknowledge failed")
		}

		// Since removeOnComplete is true, the item should be removed from the database
		row = pq.client.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s", pq.tableName))
		err = row.Scan(&count)
		if err != nil {
			t.Errorf("Error checking items in database: %v", err)
		}
		if count != 0 {
			t.Errorf("Expected 0 items in database after Acknowledge, got %d", count)
		}
	})
}

// Test concurrent operations with priority queue
func TestPriorityQueueConcurrentOperations(t *testing.T) {
	// Create a temporary database file
	dbPath := "test_priority_concurrent.db"

	// Cleanup after test
	defer os.Remove(dbPath)

	// Create a new priority queue
	queuesInstance := New(dbPath)
	pq, err := queuesInstance.NewPriorityQueue("test_priority_queue")
	if err != nil {
		t.Fatalf("Failed to create priority queue: %v", err)
	}
	defer queuesInstance.Close()

	// Enqueue items concurrently with different priorities
	numItems := 100
	done := make(chan bool)

	// Producer goroutine - adds items with varying priorities
	go func() {
		for i := 0; i < numItems; i++ {
			// Assign priority based on item index (reverse order)
			// So earlier added items have lower priority
			priority := numItems - i
			itemData := []byte(fmt.Sprintf("item-%d", i))
			if !pq.Enqueue(itemData, priority) {
				t.Errorf("Failed to enqueue item %d with priority %d", i, priority)
			}
		}
		done <- true
	}()

	// Consumer goroutine
	processed := 0
	ackIDs := make([]string, 0, numItems)

	go func() {
		for processed < numItems {
			_, success, ackID := pq.DequeueWithAckId()
			if success {
				ackIDs = append(ackIDs, ackID)
				processed++
			}
		}
		done <- true
	}()

	// Wait for producer and consumer to finish
	<-done
	<-done

	// Verify all items were processed
	if processed != numItems {
		t.Errorf("Expected %d processed items, got %d", numItems, processed)
	}

	// Acknowledge all items
	for _, ackID := range ackIDs {
		if !pq.Acknowledge(ackID) {
			t.Errorf("Failed to acknowledge item with ID %s", ackID)
		}
	}

	// Verify queue is empty
	if pq.Len() != 0 {
		t.Errorf("Expected empty queue, got length %d", pq.Len())
	}
}