package driver

import (
	"context"
	"encoding/json"
	"errors"
	"testing"
	"testing/synctest"
	"time"

	"github.com/stretchr/testify/require"

	"onehouse/protocol"
)

func TestPublisherSuppressesEqualRetainedValuesAndKeepsCommandEvidence(t *testing.T) {
	transport := &recordingTransport{}
	publisher := NewPublisher(transport)
	metadata := protocol.Metadata{}
	descriptor := protocol.DescriptorPayload{
		Manufacturer: "example",
		Model:        "switch",
		Firmware:     "1",
		Capabilities: []protocol.Capability{{
			ID:           "relay",
			Type:         "switch",
			ReportsState: true,
		}},
	}

	for range 2 {
		require.NoError(t, publisher.Descriptor(t.Context(), testDeviceID, metadata, descriptor))
		require.NoError(t, publisher.Availability(t.Context(), testDeviceID, metadata, protocol.AvailabilityPayload{Status: "offline"}))
		require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", metadata, protocol.ValuePayload{Value: json.RawMessage("false")}))
	}

	require.Len(t, transport.messages(), 3)
	require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{CorrelationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage("false")}))
	require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{CausationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage("false")}))
	require.Len(t, transport.messages(), 5)

	for _, publication := range transport.messages() {
		require.Equal(t, byte(1), publication.QoS)
		require.True(t, publication.Retain)
	}
}

func TestPublisherRepeatsLatestMeasurementWithoutMovingDeadline(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx, cancel := context.WithCancel(t.Context())
		defer cancel()

		transport := &recordingTransport{}
		publisher := NewPublisher(transport)
		finished := make(chan error, 1)

		require.NoError(t, publisher.Reported(ctx, testDeviceID, "weather", protocol.Metadata{CorrelationID: testCommandID}, protocol.ValuePayload{Value: json.RawMessage(`{"temperature":20,"humidity":50}`)}))

		first := transport.messages()[0]
		go func() {
			finished <- publisher.Run(ctx)
		}()

		for range 299 {
			time.Sleep(time.Second)
			require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "weather", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage(`{ "humidity":50,"temperature":20 }`)}))
			synctest.Wait()
			require.Len(t, transport.messages(), 1)
		}

		time.Sleep(time.Second)
		synctest.Wait()

		packets := transport.messages()
		require.Len(t, packets, 2)

		repeated, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
		require.NoError(t, err)

		original, err := protocol.ParseReported(first.Topic, first.Data, first.Retain)
		require.NoError(t, err)
		require.NotEqual(t, original.Metadata.MessageID, repeated.Metadata.MessageID)
		require.Equal(t, time.Now().Add(-time.Second).UTC(), *repeated.Metadata.SentAt)
		require.Empty(t, repeated.Metadata.CorrelationID)
		require.Empty(t, repeated.Metadata.CausationID)
		require.JSONEq(t, `{"temperature":20,"humidity":50}`, string(repeated.Payload.Value))
		cancel()
		require.ErrorIs(t, <-finished, context.Canceled)
	})
}

func TestPublisherTelemetryKeepsSnapshotOnScheduleAndReconnect(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx, cancel := context.WithCancel(t.Context())
		defer cancel()

		transport := &recordingTransport{}
		publisher := NewPublisher(transport)
		finished := make(chan error, 1)

		require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("10")}))

		go func() {
			finished <- publisher.Run(ctx)
		}()

		time.Sleep(4 * time.Minute)
		require.NoError(t, publisher.Telemetry(ctx, testDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("20")}))
		require.NoError(t, publisher.Telemetry(ctx, testOtherDeviceID, "power", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("30")}))
		synctest.Wait()

		packets := transport.messages()
		require.Len(t, packets, 3)

		for _, packet := range packets {
			require.Equal(t, byte(0), packet.QoS)
			require.False(t, packet.Retain)
		}

		time.Sleep(time.Minute)
		synctest.Wait()

		packets = transport.messages()
		require.Len(t, packets, 4)

		snapshot, err := protocol.ParseReported(packets[3].Topic, packets[3].Data, packets[3].Retain)
		require.NoError(t, err)
		require.JSONEq(t, "20", string(snapshot.Payload.Value))
		require.Equal(t, byte(1), packets[3].QoS)
		require.Equal(t, time.Now().Add(-time.Minute).UTC(), *snapshot.Metadata.SentAt)
		publisher.Refresh()
		synctest.Wait()
		require.Len(t, transport.messages(), 6)
		cancel()
		require.ErrorIs(t, <-finished, context.Canceled)
	})
}

func TestPublisherRepeatsOnlineButNotBooleanOrOffline(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		ctx, cancel := context.WithCancel(t.Context())
		defer cancel()

		transport := &recordingTransport{}
		publisher := NewPublisher(transport)
		finished := make(chan error, 1)

		require.NoError(t, publisher.Availability(ctx, testDeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "online"}))
		require.NoError(t, publisher.Availability(ctx, testOtherDeviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"}))
		require.NoError(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))

		go func() {
			finished <- publisher.Run(ctx)
		}()

		time.Sleep(5 * time.Minute)
		synctest.Wait()

		packets := transport.messages()
		require.Len(t, packets, 4)

		online, err := protocol.ParseAvailability(packets[3].Topic, packets[3].Data, packets[3].Retain)
		require.NoError(t, err)
		require.Equal(t, "online", online.Payload.Status)
		publisher.Refresh()
		synctest.Wait()
		require.Len(t, transport.messages(), 7)
		cancel()
		require.ErrorIs(t, <-finished, context.Canceled)
	})
}

func TestPublisherDoesNotSuppressEventsResultsOrExpiringValues(t *testing.T) {
	transport := &recordingTransport{}
	publisher := NewPublisher(transport)
	expiry := time.Now().UTC().Add(time.Minute)

	for range 2 {
		require.NoError(t, publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{"action": json.RawMessage(`"pressed"`)}))
		require.NoError(t, publisher.Result(t.Context(), testDeviceID, testCommandID, protocol.Metadata{CorrelationID: testCommandID}, protocol.ResultPayload{Status: "completed"}))
		require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{ExpiresAt: &expiry}, protocol.ValuePayload{Value: json.RawMessage("true")}))
	}

	require.Len(t, transport.messages(), 6)
	require.NoError(t, publisher.repeat(t.Context(), true))
	require.Len(t, transport.messages(), 6)
}

func TestPublisherKeepsFailedLatestSampleAndReturnsRepeatFailures(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		transport := &recordingTransport{}
		publisher := NewPublisher(transport)
		require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("20")}))

		failure := errors.New("broker unavailable")
		transport.setError(failure)
		require.ErrorIs(t, publisher.Telemetry(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("21")}), failure)
		require.ErrorIs(t, publisher.Telemetry(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("22")}), failure)
		publisher.Refresh()
		require.ErrorIs(t, publisher.Run(t.Context()), failure)
		transport.setError(nil)
		require.NoError(t, publisher.repeat(t.Context(), false))

		packets := transport.messages()
		last := packets[len(packets)-1]
		snapshot, err := protocol.ParseReported(last.Topic, last.Data, last.Retain)
		require.NoError(t, err)
		require.JSONEq(t, "22", string(snapshot.Payload.Value))
	})
}

func TestPublisherValidatesBeforeSendingAndHonorsCancellation(t *testing.T) {
	transport := &recordingTransport{}
	publisher := NewPublisher(transport)
	require.Error(t, publisher.Reported(t.Context(), "invalid", "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}))
	require.Error(t, publisher.Reported(t.Context(), testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("invalid")}))

	past := time.Now().UTC().Add(-time.Second)
	require.ErrorContains(t, publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{ExpiresAt: &past}, protocol.EventPayload{}), "expired")

	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()

	cancel()
	require.ErrorIs(t, publisher.Reported(ctx, testDeviceID, "relay", protocol.Metadata{}, protocol.ValuePayload{Value: json.RawMessage("true")}), context.Canceled)
	require.Empty(t, transport.messages())
}

func TestPublisherSendTimeoutAndCallerCancellation(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		publisher := NewPublisher(&waitingTransport{})
		err := publisher.Event(t.Context(), testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{})
		require.ErrorIs(t, err, context.DeadlineExceeded)

		ctx, cancel := context.WithCancel(t.Context())
		defer cancel()

		finished := make(chan error, 1)
		go func() {
			finished <- publisher.Event(ctx, testDeviceID, "button", protocol.Metadata{}, protocol.EventPayload{})
		}()

		synctest.Wait()
		cancel()
		require.ErrorIs(t, <-finished, context.Canceled)
	})
}

func TestPublisherCopiesSnapshotsAndAcceptsTypedNull(t *testing.T) {
	transport := &recordingTransport{}
	publisher := NewPublisher(transport)
	now := time.Now().UTC()
	metadata := protocol.Metadata{SentAt: &now}
	value := json.RawMessage(`{"power":10}`)
	require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "power", metadata, protocol.ValuePayload{Value: value}))

	observation := now
	now = now.Add(time.Hour)

	copy(value, `{"power":99}`)
	publisher.Refresh()
	publisher.Refresh()
	require.NoError(t, publisher.repeat(t.Context(), true))

	packets := transport.messages()
	require.Len(t, packets, 2)

	snapshot, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
	require.NoError(t, err)
	require.JSONEq(t, `{"power":10}`, string(snapshot.Payload.Value))
	require.Equal(t, observation, *snapshot.Metadata.SentAt)
	require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "power", metadata, protocol.ValuePayload{}))
}

func TestPublisherEqualTelemetryAtRepeatTimeSendsRetainedSnapshot(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		transport := &recordingTransport{}
		publisher := NewPublisher(transport)
		value := protocol.ValuePayload{Value: json.RawMessage("null")}
		require.NoError(t, publisher.Telemetry(t.Context(), testDeviceID, "power", protocol.Metadata{}, value))
		time.Sleep(protocol.MeasurementRepeatInterval)
		require.NoError(t, publisher.Telemetry(t.Context(), testDeviceID, "power", protocol.Metadata{}, value))

		packets := transport.messages()
		require.Len(t, packets, 2)
		require.True(t, packets[1].Retain)

		snapshot, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
		require.NoError(t, err)
		require.JSONEq(t, "null", string(snapshot.Payload.Value))
	})
}

func TestPublisherPreservesDirectMeasurementEvidenceAtRepeatDeadline(t *testing.T) {
	testCases := []struct {
		name          string
		correlationID string
		causationID   string
	}{
		{
			name:          "correlation keeps caller identity",
			correlationID: testCommandID,
		},
		{
			name:        "causation keeps caller identity",
			causationID: testCommandID,
		},
		{
			name:          "correlation and causation keep caller identity",
			correlationID: testCommandID,
			causationID:   testOtherDeviceID,
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			synctest.Test(t, func(t *testing.T) {
				transport := &recordingTransport{}
				publisher := NewPublisher(transport)
				value := protocol.ValuePayload{Value: json.RawMessage("20")}
				require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", protocol.Metadata{}, value))
				time.Sleep(protocol.MeasurementRepeatInterval)

				observedAt := time.Now().UTC()
				metadata := protocol.Metadata{
					MessageID:     testOtherDeviceID,
					SentAt:        &observedAt,
					CorrelationID: tc.correlationID,
					CausationID:   tc.causationID,
				}
				require.NoError(t, publisher.Reported(t.Context(), testDeviceID, "temperature", metadata, value))

				packets := transport.messages()
				require.Len(t, packets, 2)

				message, err := protocol.ParseReported(packets[1].Topic, packets[1].Data, packets[1].Retain)
				require.NoError(t, err)
				require.Equal(t, metadata, message.Metadata)
				require.Equal(t, value, message.Payload)
			})
		})
	}
}

type waitingTransport struct{}

func (p *waitingTransport) Publish(ctx context.Context, _ protocol.Publication) error {
	<-ctx.Done()

	return ctx.Err()
}