package upstream

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"cmpp-platform/gateway/internal/queue"

	cmpp "github.com/bigwhite/gocmpp"
)

func TestFailedSupplierConnectionReconnectsWhenEndpointRecovers(t *testing.T) {
	address := reserveSupplierAddress(t)
	states := make(chan ConnectionState, 16)
	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var state ConnectionState
		if err := json.NewDecoder(r.Body).Decode(&state); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		states <- state
		w.WriteHeader(http.StatusNoContent)
	}))
	defer api.Close()

	host, portText, err := net.SplitHostPort(address)
	if err != nil {
		t.Fatalf("split address: %v", err)
	}
	var port int
	if _, err := fmt.Sscanf(portText, "%d", &port); err != nil {
		t.Fatalf("parse port: %v", err)
	}
	manager := &Manager{APIBaseURL: api.URL}
	command := queue.ConnectChannelCommand{
		SchemaVersion:      queue.SchemaVersion,
		MessageType:        queue.MessageTypeConnectChannel,
		ChannelID:          "channel-reconnect",
		ConnectionID:       "channel-reconnect:primary",
		DesiredConnections: 1,
		Channel: queue.ConnectChannelConfig{
			GatewayHost:              host,
			GatewayPort:              port,
			Account:                  "sp",
			PasswordCipher:           "secret",
			CMPPVersion:              "3.0",
			HeartbeatIntervalSeconds: 1,
			HeartbeatMissThreshold:   3,
		},
	}
	initial, err := manager.ConnectChannel(context.Background(), command)
	if err != nil {
		t.Fatalf("initial connect command: %v", err)
	}
	if initial.Status != "failed" {
		t.Fatalf("initial state = %s, want failed", initial.Status)
	}

	go func() {
		_ = cmpp.ListenAndServe(address, cmpp.V30, time.Hour, 3, nil,
			cmpp.HandlerFunc(func(response *cmpp.Response, packet *cmpp.Packet, _ *log.Logger) (bool, error) {
				if _, ok := packet.Packer.(*cmpp.CmppConnReqPkt); ok {
					response.Packer.(*cmpp.Cmpp3ConnRspPkt).Version = 0x30
				}
				return false, nil
			}),
		)
	}()
	time.Sleep(100 * time.Millisecond)

	manager.mu.Lock()
	pool := manager.conns[command.ChannelID]
	manager.mu.Unlock()
	pool.mu.Lock()
	pool.nextReconnectAt = time.Now()
	pool.mu.Unlock()
	pool.signalReconnect()

	deadline := time.After(4 * time.Second)
	connected := false
	for {
		select {
		case state := <-states:
			if state.Status == "connected" && state.CurrentConnections == 1 {
				connected = true
			}
			if connected && state.Status == "heartbeat" && state.LastHeartbeatAt != "" {
				_, _ = manager.DisconnectChannel(context.Background(), queue.DisconnectChannelCommand{
					MessageType:  queue.MessageTypeDisconnectChannel,
					ChannelID:    command.ChannelID,
					ConnectionID: command.ConnectionID,
				})
				return
			}
		case <-deadline:
			t.Fatal("timed out waiting for automatic supplier reconnection and active heartbeat")
		}
	}
}

func reserveSupplierAddress(t *testing.T) string {
	t.Helper()
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("reserve address: %v", err)
	}
	address := listener.Addr().String()
	if err := listener.Close(); err != nil {
		t.Fatalf("close reserved listener: %v", err)
	}
	return address
}