package messaging

import (
	"sync"
	"sync/atomic"
	"testing"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

func TestEveryJobRuns(t *testing.T) {
	pool := newPosPool("test", 4, 16)

	var done int64
	var wg sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg.Add(1)
		pool.submit(func() {
			defer wg.Done()
			atomic.AddInt64(&done, 1)
		})
	}

	wg.Wait()
	pool.stop()

	if got := atomic.LoadInt64(&done); got != 100 {
		t.Errorf("ran %d jobs, want 100", got)
	}
}

func TestConcurrencyIsBounded(t *testing.T) {
	// The reason the pool exists. Unbounded concurrency would open a database
	// transaction per message and exhaust the connection pool under a storm,
	// stalling every one of them at once.
	const workers = 4
	pool := newPosPool("test", workers, 64)

	var inFlight, peak int64
	var wg sync.WaitGroup

	for i := 0; i < 200; i++ {
		wg.Add(1)
		pool.submit(func() {
			defer wg.Done()

			now := atomic.AddInt64(&inFlight, 1)
			for {
				was := atomic.LoadInt64(&peak)
				if now <= was || atomic.CompareAndSwapInt64(&peak, was, now) {
					break
				}
			}

			time.Sleep(time.Millisecond)
			atomic.AddInt64(&inFlight, -1)
		})
	}

	wg.Wait()
	pool.stop()

	if got := atomic.LoadInt64(&peak); got > workers {
		t.Errorf("peak concurrency %d exceeded the %d workers", got, workers)
	}
}

func TestSubmitBlocksRatherThanDroppingWork(t *testing.T) {
	// A full queue must slow the caller down, not discard a bill. Dropping
	// would throw away work already accepted from the broker, and the terminal
	// would only find out at its ack timeout.
	pool := newPosPool("test", 1, 1)

	release := make(chan struct{})
	var ran int64

	// Occupy the single worker.
	pool.submit(func() {
		<-release
		atomic.AddInt64(&ran, 1)
	})

	// Fill the queue, then a third submit must block until the worker frees up.
	pool.submit(func() { atomic.AddInt64(&ran, 1) })

	blocked := make(chan struct{})
	go func() {
		pool.submit(func() { atomic.AddInt64(&ran, 1) })
		close(blocked)
	}()

	select {
	case <-blocked:
		t.Fatal("submit returned while the queue was full; work would be dropped under load")
	case <-time.After(100 * time.Millisecond):
		// Correctly blocked.
	}

	close(release)

	select {
	case <-blocked:
	case <-time.After(3 * time.Second):
		t.Fatal("submit never unblocked after the worker freed up")
	}

	pool.stop()

	if got := atomic.LoadInt64(&ran); got != 3 {
		t.Errorf("ran %d jobs, want 3 — none may be lost", got)
	}
}

func TestStopDrainsAcceptedWork(t *testing.T) {
	// A bill mid-commit must still get its ack. Without one the terminal holds
	// it and sends it again on restart — harmless, but avoidable.
	pool := newPosPool("test", 2, 64)

	var done int64
	for i := 0; i < 50; i++ {
		pool.submit(func() {
			time.Sleep(time.Millisecond)
			atomic.AddInt64(&done, 1)
		})
	}

	pool.stop()

	if got := atomic.LoadInt64(&done); got != 50 {
		t.Errorf("only %d of 50 jobs completed before shutdown finished", got)
	}
}

func TestSubmitAfterStopStillRunsTheWork(t *testing.T) {
	// A message that arrived during shutdown has already been taken from the
	// broker. Discarding it would lose a bill we accepted responsibility for.
	pool := newPosPool("test", 2, 8)
	pool.stop()

	var ran int64
	pool.submit(func() { atomic.AddInt64(&ran, 1) })

	if got := atomic.LoadInt64(&ran); got != 1 {
		t.Error("work submitted during shutdown was discarded")
	}
}

func TestStopIsIdempotent(t *testing.T) {
	// Close() may be reached twice on a shutdown path; a second close of the
	// jobs channel would panic and take the process down mid-drain.
	pool := newPosPool("test", 2, 8)
	pool.stop()
	pool.stop()
	pool.stop()
}

func TestPoolSizeFallsBackAndClamps(t *testing.T) {
	t.Setenv("POS_TEST_WORKERS", "")
	if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 {
		t.Errorf("unset = %d, want the fallback 8", got)
	}

	t.Setenv("POS_TEST_WORKERS", "not a number")
	if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 {
		t.Errorf("garbage = %d, want the fallback 8", got)
	}

	t.Setenv("POS_TEST_WORKERS", "0")
	if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 {
		t.Errorf("zero = %d, want the fallback 8", got)
	}

	t.Setenv("POS_TEST_WORKERS", "-4")
	if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 {
		t.Errorf("negative = %d, want the fallback 8", got)
	}

	t.Setenv("POS_TEST_WORKERS", "24")
	if got := posPoolSize("POS_TEST_WORKERS", 8); got != 24 {
		t.Errorf("explicit = %d, want 24", got)
	}

	// Clamped: more workers than the database can serve just moves the queue
	// inside the driver, where there is no backpressure to feel.
	t.Setenv("POS_TEST_WORKERS", "100000")
	if got := posPoolSize("POS_TEST_WORKERS", 8); got != 128 {
		t.Errorf("absurd = %d, want the 128 clamp", got)
	}
}

func TestAWrappedHandlerCopiesThePayload(t *testing.T) {
	// paho reuses its buffer once a handler returns, and with a pool the work
	// now happens *after* that. Without a copy a queued bill would be read as
	// whatever message happened to arrive next — silently, and as valid JSON
	// often enough to commit the wrong sale.
	pool := newPosPool("test", 1, 4)

	seen := make(chan string, 1)
	wrapped := wrapHandler(pool, func(_ mqtt.Client, msg mqtt.Message) {
		seen <- string(msg.Payload())
	})

	// A buffer paho would reuse.
	buffer := []byte(`{"batch_id":"original"}`)
	wrapped(nil, fakeMessage{topic: "nearle/pos/12/T4A9/order", payload: buffer})

	// Overwrite it the instant the handler returns, exactly as paho would.
	for i := range buffer {
		buffer[i] = 'X'
	}

	select {
	case got := <-seen:
		if got != `{"batch_id":"original"}` {
			t.Errorf("handler saw %q — the payload was not copied before queueing", got)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("the wrapped handler never ran")
	}

	pool.stop()
}

func TestAWrappedHandlerKeepsTheTopic(t *testing.T) {
	// Store and terminal are read from the topic, never the body. Losing it in
	// the hand-off would leave the ack with nowhere to go.
	pool := newPosPool("test", 1, 4)

	seen := make(chan string, 1)
	wrapped := wrapHandler(pool, func(_ mqtt.Client, msg mqtt.Message) {
		seen <- msg.Topic()
	})

	wrapped(nil, fakeMessage{topic: "nearle/pos/1135/T4A9/order", payload: []byte("{}")})

	select {
	case got := <-seen:
		if got != "nearle/pos/1135/T4A9/order" {
			t.Errorf("topic = %q, want nearle/pos/1135/T4A9/order", got)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("the wrapped handler never ran")
	}

	pool.stop()
}