package tests

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"os"
	"reflect"
	"testing"
	"time"

	"github.com/amenzhinsky/iothub/iotdevice"
	"github.com/amenzhinsky/iothub/iotdevice/transport"
	"github.com/amenzhinsky/iothub/iotdevice/transport/mqtt"
	"github.com/amenzhinsky/iothub/iotservice"
)

func TestEnd2End(t *testing.T) {
	cs := os.Getenv("TEST_IOTHUB_SERVICE_CONNECTION_STRING")
	if cs == "" {
		t.Fatal("$TEST_IOTHUB_SERVICE_CONNECTION_STRING is empty")
	}
	sc, err := iotservice.NewFromConnectionString(cs) // iotservice.WithLogger(logger.New(logger.LevelDebug, nil)),
	if err != nil {
		t.Fatal(err)
	}
	defer sc.Close()

	// create devices with all possible authentication types
	_, err = sc.DeleteDevices(context.Background(), []*iotservice.Device{
		{DeviceID: "golang-iothub-sas"},
		{DeviceID: "golang-iothub-self-signed"},
		{DeviceID: "golang-iothub-ca"},
	}, true)
	if err != nil {
		t.Fatal(err)
	}
	result, err := sc.CreateDevices(context.Background(), []*iotservice.Device{{
		DeviceID: "golang-iothub-sas",
		Authentication: &iotservice.Authentication{
			Type: iotservice.AuthSAS,
		},
	}, {
		DeviceID: "golang-iothub-self-signed",
		Authentication: &iotservice.Authentication{
			Type: iotservice.AuthSelfSigned,
			X509Thumbprint: &iotservice.X509Thumbprint{
				PrimaryThumbprint:   "443ABB6DEA8F93D5987D31D2607BE2931217752C",
				SecondaryThumbprint: "443ABB6DEA8F93D5987D31D2607BE2931217752C",
			},
		},
	}, {
		DeviceID: "golang-iothub-ca",
		Authentication: &iotservice.Authentication{
			Type: iotservice.AuthCA,
		},
	}})
	if err != nil {
		t.Fatal(err)
	}
	if !result.IsSuccessful {
		t.Fatalf("couldn't create devices: %v", result.Errors)
	}

	for name, mktransport := range map[string]func() transport.Transport{
		"mqtt":    func() transport.Transport { return mqtt.New() },
		"mqtt-ws": func() transport.Transport { return mqtt.New(mqtt.WithWebSocket(true)) },
		// TODO: "amqp": func() transport.Transport { return amqp.New() },
		// TODO: "http": func() transport.Transport { return http.New() },
	} {
		mktransport := mktransport
		t.Run(name, func(t *testing.T) {
			for auth, suite := range map[string]struct {
				init func(transport transport.Transport) (*iotdevice.Client, error)
				only string
			}{
				// TODO: ca authentication
				"x509": {
					func(transport transport.Transport) (*iotdevice.Client, error) {
						return iotdevice.NewFromX509FromFile(
							transport,
							"golang-iothub-self-signed",
							sc.HostName(),
							"testdata/device.crt",
							"testdata/device.key",
						)
					},
					"DeviceToCloud", // just need to check access
				},
				"sas": {
					func(transport transport.Transport) (*iotdevice.Client, error) {
						device, err := sc.GetDevice(context.Background(), "golang-iothub-sas")
						if err != nil {
							return nil, err
						}
						dcs, err := sc.DeviceConnectionString(device, false)
						if err != nil {
							t.Fatal(err)
						}
						return iotdevice.NewFromConnectionString(transport, dcs)
					},
					"*",
				},
			} {
				for name, test := range map[string]func(*testing.T, *iotservice.Client, *iotdevice.Client){
					"DeviceToCloud":    testDeviceToCloud,
					"CloudToDevice":    testCloudToDevice,
					"DirectMethod":     testDirectMethod,
					"UpdateDeviceTwin": testUpdateTwin,
					"SubscribeTwin":    testSubscribeTwin,
				} {
					if suite.only != "*" && suite.only != name {
						continue
					}
					test, suite, mktransport := test, suite, mktransport
					t.Run(auth+"/"+name, func(t *testing.T) {
						dc, err := suite.init(mktransport())
						if err != nil {
							t.Fatal(err)
						}
						defer dc.Close()
						if err := dc.Connect(context.Background()); err != nil {
							t.Fatal(err)
						}
						test(t, sc, dc)
					})
				}
			}
		})
	}
}

func testDeviceToCloud(t *testing.T, sc *iotservice.Client, dc *iotdevice.Client) {
	evsc := make(chan *iotservice.Event, 1)
	errc := make(chan error, 2)
	go func() {
		errc <- sc.SubscribeEvents(context.Background(), func(ev *iotservice.Event) error {
			if ev.ConnectionDeviceID == dc.DeviceID() {
				evsc <- ev
			}
			return nil
		})
	}()

	payload := []byte("hello")
	props := map[string]string{"a": "a", "b": "b"}
	done := make(chan struct{})
	defer close(done)

	// send events until one of them is received
	go func() {
		for {
			if err := dc.SendEvent(context.Background(), payload,
				iotdevice.WithSendMessageID(genID()),
				iotdevice.WithSendCorrelationID(genID()),
				iotdevice.WithSendProperties(props),
				iotdevice.WithSendCreationTime(time.Now().Add(-24*time.Hour)),
			); err != nil {
				errc <- err
				break
			}
			select {
			case <-done:
			case <-time.After(500 * time.Millisecond):
			}
		}
	}()

	select {
	case msg := <-evsc:
		if msg.MessageID == "" {
			t.Error("MessageID is empty")
		}
		if msg.CorrelationID == "" {
			t.Error("CorrelationID is empty")
		}
		if msg.ConnectionDeviceID != dc.DeviceID() {
			t.Errorf("ConnectionDeviceID = %q, want %q", msg.ConnectionDeviceID, dc.DeviceID())
		}
		if msg.ConnectionDeviceGenerationID == "" {
			t.Error("ConnectionDeviceGenerationID is empty")
		}
		if msg.ConnectionAuthMethod == nil {
			t.Error("ConnectionAuthMethod is nil")
		}
		if msg.MessageSource == "" {
			t.Error("MessageSource is empty")
		}
		if msg.EnqueuedTime.IsZero() {
			t.Error("EnqueuedTime is zero")
		}
		if !bytes.Equal(msg.Payload, payload) {
			t.Errorf("Payload = %v, want %v", msg.Payload, payload)
		}
		if msg.Properties["iothub-creation-time-utc"] == "" {
			t.Error("iothub-creation-time-utc missing")
		}
		testProperties(t, msg.Properties, props)
	case err := <-errc:
		t.Fatal(err)
	case <-time.After(10 * time.Second):
		t.Fatal("d2c timed out")
	}
}

func testProperties(t *testing.T, got, want map[string]string) {
	t.Helper()
	for k, v := range want {
		x, ok := got[k]
		if !ok || x != v {
			t.Errorf("Properties = %v, want %v", got, want)
			return
		}
	}
}

func testCloudToDevice(t *testing.T, sc *iotservice.Client, dc *iotdevice.Client) {
	fbsc := make(chan *iotservice.Feedback, 1)
	errc := make(chan error, 3)

	sub, err := dc.SubscribeEvents(context.Background())
	if err != nil {
		t.Fatal(err)
	}

	// subscribe to feedback and report first registered message id
	go func() {
		errc <- sc.SubscribeFeedback(context.Background(), func(fb *iotservice.Feedback) error {
			fbsc <- fb
			return nil
		})
	}()

	payload := []byte("hello")
	props := map[string]string{"a": "a", "b": "b"}
	uid := "golang-iothub"

	mid := genID()
	if err := sc.SendEvent(context.Background(), dc.DeviceID(), payload,
		iotservice.WithSendAck(iotservice.AckFull),
		iotservice.WithSendProperties(props),
		iotservice.WithSendUserID(uid),
		iotservice.WithSendMessageID(mid),
		iotservice.WithSendCorrelationID(genID()),
		iotservice.WithSendExpiryTime(time.Now().Add(5*time.Second)),
	); err != nil {
		errc <- err
		return
	}

	for {
		select {
		case msg := <-sub.C():
			if msg.MessageID != mid {
				continue
			}

			// validate event feedback
		Outer:
			for {
				select {
				case fb := <-fbsc:
					if fb.OriginalMessageID != mid {
						continue
					}
					if fb.StatusCode != "Success" {
						t.Errorf("feedback status = %q, want %q", fb.StatusCode, "Success")
					}
					break Outer
				case <-time.After(15 * time.Second):
					t.Log("feedback timed out, probably due to eventhub batching")
					break Outer
				}
			}

			// validate message content
			if msg.To == "" {
				t.Error("To is empty")
			}
			if msg.UserID != uid {
				t.Errorf("UserID = %q, want %q", msg.UserID, uid)
			}
			if !bytes.Equal(msg.Payload, payload) {
				t.Errorf("Payload = %v, want %v", msg.Payload, payload)
			}
			if msg.MessageID == "" {
				t.Error("MessageID is empty")
			}
			if msg.CorrelationID == "" {
				t.Error("CorrelationID is empty")
			}
			if msg.ExpiryTime.IsZero() {
				t.Error("ExpiryTime is zero")
			}
			testProperties(t, msg.Properties, props)
			return
		case err := <-errc:
			t.Fatal(err)
		case <-time.After(10 * time.Second):
			t.Fatal("c2d timed out")
		}
	}
}

func testUpdateTwin(t *testing.T, sc *iotservice.Client, dc *iotdevice.Client) {
	// update state and keep track of version
	s := fmt.Sprintf("%d", time.Now().UnixNano())
	v, err := dc.UpdateTwinState(context.Background(), map[string]interface{}{
		"ts": s,
	})
	if err != nil {
		t.Fatal(err)
	}

	_, r, err := dc.RetrieveTwinState(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if v != r.Version() {
		t.Errorf("update-twin version = %d, want %d", r.Version(), v)
	}
	if r["ts"] != s {
		t.Errorf("update-twin parameter = %q, want %q", r["ts"], s)
	}
}

// TODO: very flaky
func testSubscribeTwin(t *testing.T, sc *iotservice.Client, dc *iotdevice.Client) {
	sub, err := dc.SubscribeTwinUpdates(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	defer dc.UnsubscribeTwinUpdates(sub)

	// TODO: hacky, but reduces flakiness
	time.Sleep(time.Second)

	twin, err := sc.UpdateDeviceTwin(context.Background(), &iotservice.Twin{
		DeviceID: dc.DeviceID(),
		Tags: map[string]interface{}{
			"test-device": true,
		},
		Properties: &iotservice.Properties{
			Desired: map[string]interface{}{
				"test-prop": time.Now().UnixNano() / 1000,
			},
		},
	})
	if err != nil {
		t.Fatal(err)
	}

	select {
	case state := <-sub.C():
		if state["$version"] != twin.Properties.Desired["$version"] {
			t.Errorf("version = %d, want %d", state["$version"], twin.Properties.Desired["$version"])
		}
		if state["test-prop"] != twin.Properties.Desired["test-prop"] {
			t.Errorf("test-prop = %q, want %q", state["test-prop"], twin.Properties.Desired["test-prop"])
		}
	case <-time.After(10 * time.Second):
		t.Fatal("SubscribeTwinUpdates timed out")
	}
}

func testDirectMethod(t *testing.T, sc *iotservice.Client, dc *iotdevice.Client) {
	if err := dc.RegisterMethod(
		context.Background(),
		"sum",
		func(v map[string]interface{}) (int, map[string]interface{}, error) {
			return 222, map[string]interface{}{
				"result": v["a"].(float64) + v["b"].(float64),
			}, nil
		},
	); err != nil {
		t.Fatal(err)
	}

	resc := make(chan *iotservice.MethodResult, 1)
	errc := make(chan error, 2)
	go func() {
		v, err := sc.CallDeviceMethod(context.Background(), dc.DeviceID(), &iotservice.MethodCall{
			MethodName:      "sum",
			ConnectTimeout:  5,
			ResponseTimeout: 5,
			Payload: map[string]interface{}{
				"a": 1.5,
				"b": 3,
			},
		})
		if err != nil {
			errc <- err
		}
		resc <- v
	}()

	select {
	case v := <-resc:
		w := &iotservice.MethodResult{
			Status: 222,
			Payload: map[string]interface{}{
				"result": 4.5,
			},
		}
		if !reflect.DeepEqual(v, w) {
			t.Errorf("direct-method result = %v, want %v", v, w)
		}
	case err := <-errc:
		t.Fatal(err)
	}
}

func genID() string {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}
