package msgport

import (
	"errors"
	"sync"
	"testing"
	"time"

	"git.smesh.lol/nostrino/pkg/chaindata"
	"git.smesh.lol/tiny-spawn"
)

// source is a chain that answers without a network, so the test is about the adapter.
type source struct {
	mu          sync.Mutex
	position    chaindata.Position
	matched     []uint32
	scripts     int
	blockHeight uint32
}

func (s *source) Position(chaindata.Context) (chaindata.Position, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	return s.position, nil
}

func (s *source) Sync(_ chaindata.Context, from uint32) (chaindata.Position, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.position = chaindata.Position{Tip: from + 10, Filters: from + 10}

	return s.position, nil
}

func (s *source) Match(_ chaindata.Context, scripts [][]byte) ([]uint32, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.scripts = len(scripts)

	return append([]uint32{}, s.matched...), nil
}

func (s *source) Block(_ chaindata.Context, height uint32) ([]byte, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.blockHeight = height

	raw := make([]byte, 128)
	raw[0] = byte(height)

	return raw, nil
}

func (s *source) Broadcast(_ chaindata.Context, raw []byte) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.blockHeight = uint32(len(raw))

	return nil
}

func (s *source) Filters(_ chaindata.Context, from, count uint32) ([]chaindata.Filter, error) {
	out := make([]chaindata.Filter, 0, count)

	for offset := uint32(0); offset < count; offset++ {
		hash := make([]byte, 32)
		hash[0] = byte(offset)

		out = append(out, chaindata.Filter{
			Height:    from + offset,
			BlockHash: hash,
			Filter:    []byte{byte(offset)},
		})
	}

	return out, nil
}

// pipePort is one end of a connected pair, standing in for a MessagePort.
type pipePort struct {
	in   chan []byte
	peer *pipePort
	mu   sync.Mutex
	dead bool
}

// Send delivers to the peer's receive channel.
func (p *pipePort) Send(frame []byte) error {
	peer := p.peer

	peer.mu.Lock()
	defer peer.mu.Unlock()

	if peer.dead {
		return ErrClosed
	}

	select {
	case peer.in <- frame:
		return nil
	default:
		return errors.New("the pipe is full")
	}
}

// Receive is the incoming frames.
func (p *pipePort) Receive() <-chan []byte { return p.in }

// Close ends this end and, as a terminated worker does, its peer with it.
func (p *pipePort) Close() error {
	for _, end := range []*pipePort{p, p.peer} {
		end.mu.Lock()

		if !end.dead {
			end.dead = true

			close(end.in)
		}

		end.mu.Unlock()
	}

	return nil
}

// pipe returns two connected ports.
func pipe() (Port, Port) {
	first := &pipePort{in: make(chan []byte, Buffer)}
	second := &pipePort{in: make(chan []byte, Buffer)}

	first.peer = second
	second.peer = first

	return first, second
}

// start serves a chain over one end and returns a client on the other.
func start(t *testing.T, chain *source) (*chaindata.Client, tinyspawn.Breaker, func()) {
	t.Helper()

	serverPort, walletPort := pipe()

	serverStop := tinyspawn.NewBreaker()
	served := make(chan struct{})

	go func() {
		defer close(served)

		chaindata.Serve(chaindata.Data{Source: chain},
			chaindata.Chans{Link: Duplex(serverPort), Stop: serverStop})
	}()

	clientStop := tinyspawn.NewBreaker()
	client := chaindata.NewClient("btc", Duplex(walletPort), clientStop)

	shutdown := func() {
		client.Close()

		if !serverStop.Closed() {
			serverStop.Close()
		}

		select {
		case <-served:
		case <-time.After(5 * time.Second):
			t.Error("the domain did not stop")
		}
	}

	return client, clientStop, shutdown
}

// TestAFrameTravelsBothWays: status, sync and match work through ports, which is the whole
// claim of this package.
func TestAFrameTravelsBothWays(t *testing.T) {
	chain := &source{position: chaindata.Position{Tip: 42, Filters: 40}, matched: []uint32{11, 12}}
	client, _, shutdown := start(t, chain)

	defer shutdown()

	position, err := client.Position()

	if err != nil {
		t.Fatal(err)
	}

	if position.Tip != 42 || position.Filters != 40 {
		t.Fatalf("the position is %+v", position)
	}

	position, err = client.Sync(7)

	if err != nil {
		t.Fatal(err)
	}

	if position.Tip != 17 || position.Filters != 17 {
		t.Fatalf("the synced position is %+v", position)
	}

	matched, err := client.Match([][]byte{{0x00, 0x14}, {0x00, 0x14, 0x01}})

	if err != nil {
		t.Fatal(err)
	}

	if len(matched) != 2 || matched[0] != 11 || matched[1] != 12 {
		t.Fatalf("the matches are %v", matched)
	}

	if chain.scripts != 2 {
		t.Fatalf("the source was asked about %d scripts", chain.scripts)
	}
}

// TestAFilterRangeArrivesAsOneFrame: the payload the browser transfers is the payload the
// wallet reads, with no per-filter message in between.
func TestAFilterRangeArrivesAsOneFrame(t *testing.T) {
	chain := &source{}
	client, _, shutdown := start(t, chain)

	defer shutdown()

	filters, err := client.Filters(100, 5)

	if err != nil {
		t.Fatal(err)
	}

	if len(filters) != 5 {
		t.Fatalf("%d filters", len(filters))
	}

	for index, filter := range filters {
		if filter.Height != uint32(100+index) {
			t.Fatalf("filter %d is at height %d", index, filter.Height)
		}
	}
}

// TestManyRequestsInOrder: ids are matched to answers, not to arrival order.
func TestManyRequestsInOrder(t *testing.T) {
	chain := &source{position: chaindata.Position{Tip: 3, Filters: 3}}
	client, _, shutdown := start(t, chain)

	defer shutdown()

	for round := 0; round < 20; round++ {
		position, err := client.Position()

		if err != nil {
			t.Fatalf("round %d: %v", round, err)
		}

		if position.Tip != 3 {
			t.Fatalf("round %d: the position is %+v", round, position)
		}
	}
}

// TestAPortThatDiesFailsTheRequest: a worker termination has to arrive as an error, not as an
// answer that never comes.
func TestAPortThatDiesFailsTheRequest(t *testing.T) {
	serverPort, walletPort := pipe()

	chain := &source{position: chaindata.Position{Tip: 9, Filters: 9}}
	serverStop := tinyspawn.NewBreaker()

	go chaindata.Serve(chaindata.Data{Source: chain},
		chaindata.Chans{Link: Duplex(serverPort), Stop: serverStop})

	defer func() {
		_ = serverPort.Close()

		if !serverStop.Closed() {
			serverStop.Close()
		}
	}()

	clientStop := tinyspawn.NewBreaker()
	client := chaindata.NewClient("btc", Duplex(walletPort), clientStop)

	defer client.Close()

	if _, err := client.Position(); err != nil {
		t.Fatal(err)
	}

	// The worker is terminated: both ends of the port pair go away.
	_ = walletPort.Close()

	deadline := time.After(5 * time.Second)

	for {
		select {
		case <-deadline:
			t.Fatal("the client never noticed the port had died")
		default:
		}

		if _, err := client.Position(); err != nil {
			return
		}

		time.Sleep(5 * time.Millisecond)
	}
}

// TestADeadServerStopsTheDomain: a closed port ends the serve loop instead of leaving it
// waiting for frames that cannot come.
func TestADeadServerStopsTheDomain(t *testing.T) {
	serverPort, _ := pipe()

	serverStop := tinyspawn.NewBreaker()
	defer serverStop.Close()

	served := make(chan struct{})

	go func() {
		defer close(served)

		chaindata.Serve(chaindata.Data{Source: &source{}},
			chaindata.Chans{Link: Duplex(serverPort), Stop: serverStop})
	}()

	_ = serverPort.Close()

	select {
	case <-served:
	case <-time.After(5 * time.Second):
		t.Fatal("the domain outlived its port")
	}
}