lnd/actor/mailbox_test.go

package actor

import (
	"context"
	"sync"
	"testing"

	"github.com/stretchr/testify/require"
)

// TestMessage is a test message type that embeds BaseMessage.
type TestMessage struct {
	BaseMessage
	Value int
}

// MessageType returns the type name of the message for routing/filtering.
func (tm TestMessage) MessageType() string {
	return "TestMessage"
}

// TestChannelMailboxSend tests the Send method of ChannelMailbox.
func TestChannelMailboxSend(t *testing.T) {
	t.Run("successful send", func(t *testing.T) {
		mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)
		ctx := context.Background()
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}

		sent := mailbox.Send(ctx, env)
		require.True(t, sent, "Send should succeed")
	})

	t.Run("send with cancelled context", func(t *testing.T) {
		mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 1)
		// Fill the mailbox first.
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}
		mailbox.TrySend(env)

		ctx, cancel := context.WithCancel(context.Background())
		// Cancel immediately.
		cancel()

		env2 := envelope[TestMessage, int]{
			message: TestMessage{Value: 43},
			promise: nil,
		}

		sent := mailbox.Send(ctx, env2)
		require.False(t, sent, "Send should fail with cancelled context")
	})

	t.Run("send to closed mailbox", func(t *testing.T) {
		mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)
		mailbox.Close()

		ctx := context.Background()
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}

		sent := mailbox.Send(ctx, env)
		require.False(t, sent, "Send should fail on closed mailbox")
	})
}

// TestChannelMailboxTrySend tests the TrySend method of ChannelMailbox.
func TestChannelMailboxTrySend(t *testing.T) {
	t.Run("successful try send", func(t *testing.T) {
		mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}

		sent := mailbox.TrySend(env)
		require.True(t, sent, "TrySend should succeed")
	})

	t.Run("try send to full mailbox", func(t *testing.T) {
		mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 1)
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}

		// Fill the mailbox.
		sent := mailbox.TrySend(env)
		require.True(t, sent, "First TrySend should succeed")

		// Try to send again - should fail.
		sent = mailbox.TrySend(env)
		require.False(t, sent, "TrySend should fail on full mailbox")
	})

	t.Run("try send to closed mailbox", func(t *testing.T) {
		mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)
		mailbox.Close()

		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}

		sent := mailbox.TrySend(env)
		require.False(t, sent, "TrySend should fail on closed mailbox")
	})
}

// TestChannelMailboxReceive tests the Receive method of ChannelMailbox.
func TestChannelMailboxReceive(t *testing.T) {
	t.Run("receive messages", func(t *testing.T) {
		mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)
		ctx := context.Background()

		// Send some messages.
		for i := 0; i < 3; i++ {
			env := envelope[TestMessage, int]{
				message: TestMessage{Value: i},
				promise: nil,
			}
			mailbox.Send(ctx, env)
		}

		// Start receiving in a goroutine.
		var received []int
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			for env := range mailbox.Receive(ctx) {
				received = append(received, env.message.Value)
			}
		}()

		// Close the mailbox after sending all messages.
		mailbox.Close()
		wg.Wait()

		require.Len(t, received, 3, "Should receive 3 messages")
		require.Equal(t, []int{0, 1, 2}, received, "Should receive messages in order")
	})

	t.Run("receive with cancelled context", func(t *testing.T) {
		mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)
		ctx, cancel := context.WithCancel(context.Background())

		// Send a message.
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}
		mailbox.Send(context.Background(), env)

		// Start receiving.
		var received int
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			for env := range mailbox.Receive(ctx) {
				received++
				_ = env
			}
		}()

		// Cancel the context.
		cancel()
		wg.Wait()

		// Might receive 0 or 1 message depending on timing.
		require.LessOrEqual(t, received, 1,
			"Should stop receiving after context cancel")
	})
}

// TestChannelMailboxClose tests the Close and IsClosed methods.
func TestChannelMailboxClose(t *testing.T) {
	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)

	require.False(t, mailbox.IsClosed(), "Mailbox should not be closed initially")

	mailbox.Close()
	require.True(t, mailbox.IsClosed(), "Mailbox should be closed after Close()")

	// Closing again should be safe.
	mailbox.Close()
	require.True(t, mailbox.IsClosed(), "Mailbox should remain closed")
}

// TestChannelMailboxDrain tests the Drain method of ChannelMailbox.
func TestChannelMailboxDrain(t *testing.T) {
	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)
	ctx := context.Background()

	// Send some messages.
	for i := 0; i < 3; i++ {
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: i},
			promise: nil,
		}
		mailbox.Send(ctx, env)
	}

	// Close the mailbox.
	mailbox.Close()

	// Drain messages.
	var drained []int
	for env := range mailbox.Drain() {
		drained = append(drained, env.message.Value)
	}

	require.Len(t, drained, 3, "Should drain 3 messages")
	require.Equal(t, []int{0, 1, 2}, drained, "Should drain messages in order")
}

// TestChannelMailboxConcurrent tests concurrent operations on ChannelMailbox.
func TestChannelMailboxConcurrent(t *testing.T) {
	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 100)
	ctx := context.Background()

	const numSenders = 10
	const messagesPerSender = 100

	var wg sync.WaitGroup

	// Start multiple senders.
	for i := 0; i < numSenders; i++ {
		wg.Add(1)
		go func(senderID int) {
			defer wg.Done()
			for j := 0; j < messagesPerSender; j++ {
				env := envelope[TestMessage, int]{
					message: TestMessage{Value: senderID*1000 + j},
					promise: nil,
				}
				mailbox.Send(ctx, env)
			}
		}(i)
	}

	// Start receiver.
	received := make([]int, 0, numSenders*messagesPerSender)
	var receiverWg sync.WaitGroup
	receiverWg.Add(1)
	go func() {
		defer receiverWg.Done()
		for env := range mailbox.Receive(ctx) {
			received = append(received, env.message.Value)
		}
	}()

	// Wait for all senders to complete.
	wg.Wait()

	// Close the mailbox now that all sends are complete.
	mailbox.Close()
	receiverWg.Wait()

	require.Len(t, received, numSenders*messagesPerSender,
		"Should receive all messages")
}

// TestChannelMailboxZeroCapacity tests that zero capacity defaults to 1.
func TestChannelMailboxZeroCapacity(t *testing.T) {
	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 0)

	// Should default to capacity of 1.
	env := envelope[TestMessage, int]{
		message: TestMessage{Value: 42},
		promise: nil,
	}

	sent := mailbox.TrySend(env)
	require.True(t, sent, "Should be able to send one message")

	// Second send should fail (mailbox full).
	sent = mailbox.TrySend(env)
	require.False(t, sent, "Second send should fail on full mailbox")
}

// TestChannelMailboxActorContext tests that the mailbox respects the actor's
// context for cancellation.
func TestChannelMailboxActorContext(t *testing.T) {
	t.Run("send respects actor context", func(t *testing.T) {
		actorCtx, actorCancel := context.WithCancel(context.Background())
		mailbox := NewChannelMailbox[TestMessage, int](actorCtx, 1)

		// Fill the mailbox.
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}
		mailbox.TrySend(env)

		// Cancel the actor context.
		actorCancel()

		// Try to send with a fresh caller context - should fail due to
		// actor context cancellation.
		callerCtx := context.Background()
		env2 := envelope[TestMessage, int]{
			message: TestMessage{Value: 43},
			promise: nil,
		}

		sent := mailbox.Send(callerCtx, env2)
		require.False(t, sent, "Send should fail when actor context is cancelled")
	})

	t.Run("receive respects actor context", func(t *testing.T) {
		actorCtx, actorCancel := context.WithCancel(context.Background())
		mailbox := NewChannelMailbox[TestMessage, int](actorCtx, 10)

		// Send a message.
		env := envelope[TestMessage, int]{
			message: TestMessage{Value: 42},
			promise: nil,
		}
		mailbox.Send(context.Background(), env)

		// Start receiving with a fresh context.
		callerCtx := context.Background()
		var received int
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			for env := range mailbox.Receive(callerCtx) {
				received++
				_ = env
			}
		}()

		// Cancel the actor context.
		actorCancel()
		wg.Wait()

		// Should have stopped receiving due to actor context cancellation.
		require.LessOrEqual(t, received, 1,
			"Should stop receiving when actor context is cancelled")
	})
}

// TestMailboxConcurrentSendAndClose tests concurrent Send and Close operations
// to ensure no race conditions or panics occur.
func TestMailboxConcurrentSendAndClose(t *testing.T) {
	const numSenders = 20
	const sendsPerSender = 100

	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 100)
	ctx := context.Background()

	var wg sync.WaitGroup

	// Start receiver to drain messages.
	var recvWg sync.WaitGroup
	recvWg.Add(1)
	go func() {
		defer recvWg.Done()
		for range mailbox.Receive(ctx) {
			// Just drain.
		}
	}()

	// Start multiple senders.
	for i := 0; i < numSenders; i++ {
		wg.Add(1)
		go func(senderID int) {
			defer wg.Done()
			for j := 0; j < sendsPerSender; j++ {
				env := envelope[TestMessage, int]{
					message: TestMessage{Value: senderID*1000 + j},
					promise: nil,
				}
				// Send may fail if mailbox closes, that's ok.
				mailbox.Send(ctx, env)
			}
		}(i)
	}

	// Concurrently close the mailbox multiple times from different
	// goroutines.
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			mailbox.Close()
		}()
	}

	wg.Wait()
	recvWg.Wait()

	// Mailbox should be closed.
	require.True(t, mailbox.IsClosed(), "Mailbox should be closed")

	// Further sends should fail without panic.
	env := envelope[TestMessage, int]{
		message: TestMessage{Value: 999},
		promise: nil,
	}
	sent := mailbox.Send(ctx, env)
	require.False(t, sent, "Send should fail on closed mailbox")
}

// TestMailboxConcurrentTrySendAndClose tests concurrent TrySend and Close
// operations to ensure no race conditions or panics occur.
func TestMailboxConcurrentTrySendAndClose(t *testing.T) {
	const numSenders = 20
	const sendsPerSender = 100

	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)

	var wg sync.WaitGroup

	// Start multiple senders using TrySend.
	for i := 0; i < numSenders; i++ {
		wg.Add(1)
		go func(senderID int) {
			defer wg.Done()
			for j := 0; j < sendsPerSender; j++ {
				env := envelope[TestMessage, int]{
					message: TestMessage{Value: senderID*1000 + j},
					promise: nil,
				}
				// TrySend may fail if mailbox is full or closed.
				mailbox.TrySend(env)
			}
		}(i)
	}

	// Concurrently close the mailbox.
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			mailbox.Close()
		}()
	}

	wg.Wait()

	// Mailbox should be closed.
	require.True(t, mailbox.IsClosed(), "Mailbox should be closed")

	// Further sends should fail without panic.
	env := envelope[TestMessage, int]{
		message: TestMessage{Value: 999},
		promise: nil,
	}
	sent := mailbox.TrySend(env)
	require.False(t, sent, "TrySend should fail on closed mailbox")
}

// TestMailboxMultipleCloseCallers tests that multiple goroutines calling
// Close() simultaneously don't cause panics or issues.
func TestMailboxMultipleCloseCallers(t *testing.T) {
	const numClosers = 100

	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 10)

	var wg sync.WaitGroup

	// Start many goroutines all trying to close the mailbox.
	for i := 0; i < numClosers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			mailbox.Close()
		}()
	}

	wg.Wait()

	// Mailbox should be closed exactly once.
	require.True(t, mailbox.IsClosed(), "Mailbox should be closed")

	// Calling Close again should be safe.
	mailbox.Close()
	require.True(t, mailbox.IsClosed(), "Mailbox should remain closed")
}

// TestMailboxCloseWhileSending tests closing the mailbox while multiple
// senders are actively sending messages.
func TestMailboxCloseWhileSending(t *testing.T) {
	const numSenders = 10
	const sendsPerSender = 1000

	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 100)
	ctx := context.Background()

	var sendWg sync.WaitGroup

	// Start multiple senders.
	for i := 0; i < numSenders; i++ {
		sendWg.Add(1)
		go func(senderID int) {
			defer sendWg.Done()
			for j := 0; j < sendsPerSender; j++ {
				env := envelope[TestMessage, int]{
					message: TestMessage{Value: senderID*1000 + j},
					promise: nil,
				}
				// Send may fail after close, that's expected.
				mailbox.Send(ctx, env)
			}
		}(i)
	}

	// Start receiver to drain messages.
	var recvWg sync.WaitGroup
	recvWg.Add(1)
	receivedCount := 0
	go func() {
		defer recvWg.Done()
		for range mailbox.Receive(ctx) {
			receivedCount++
		}
	}()

	// Close mailbox while sends are happening.
	mailbox.Close()

	sendWg.Wait()
	recvWg.Wait()

	// Should have received at least some messages (exact count depends on
	// timing).
	t.Logf("Received %d messages before close", receivedCount)

	// Mailbox should be closed.
	require.True(t, mailbox.IsClosed(), "Mailbox should be closed")
}

// TestMailboxStressTest performs a high-concurrency stress test with multiple
// senders, receivers, and close operations.
func TestMailboxStressTest(t *testing.T) {
	const numSenders = 50
	const numReceivers = 5
	const sendsPerSender = 200

	mailbox := NewChannelMailbox[TestMessage, int](context.Background(), 200)
	ctx := context.Background()

	var sendWg sync.WaitGroup

	// Start multiple senders.
	for i := 0; i < numSenders; i++ {
		sendWg.Add(1)
		go func(senderID int) {
			defer sendWg.Done()
			for j := 0; j < sendsPerSender; j++ {
				env := envelope[TestMessage, int]{
					message: TestMessage{Value: senderID*1000 + j},
					promise: nil,
				}
				mailbox.Send(ctx, env)
			}
		}(i)
	}

	// Start multiple receivers.
	var recvWg sync.WaitGroup
	for i := 0; i < numReceivers; i++ {
		recvWg.Add(1)
		go func() {
			defer recvWg.Done()
			for range mailbox.Receive(ctx) {
				// Just drain messages.
			}
		}()
	}

	// Wait for all sends to complete.
	sendWg.Wait()

	// Close mailbox.
	mailbox.Close()

	// Wait for all receivers to finish.
	recvWg.Wait()

	// Mailbox should be closed.
	require.True(t, mailbox.IsClosed(), "Mailbox should be closed")
}