// Package mqtt provides a bounded MQTT 5 transport without an application outbox.
package mqtt

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"errors"
	"fmt"
	"math/rand/v2"
	"net/url"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"robot-market/test-mqtt-5/internal/config"
	"robot-market/test-mqtt-5/internal/contracts"
	"robot-market/test-mqtt-5/internal/events"

	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
)

var ErrNotReady = errors.New("MQTT subscriptions are not ready")
var ErrUnknown = errors.New("publication outcome unknown; no automatic command retry")

type Client struct {
	Config     config.Config
	Events     *events.Store
	Incoming   chan *paho.Publish
	Down       chan struct{}
	cm         *autopaho.ConnectionManager
	cancel     context.CancelFunc
	ready      atomic.Bool
	dropped    atomic.Uint64
	invalid    atomic.Uint64
	stateMu    sync.Mutex
	generation uint64
	state      connection
	changed    chan struct{}
	firstReady chan struct{}
	firstOnce  sync.Once
	fatal      chan error
	workerDone chan struct{}
}
type connection struct {
	cm         *autopaho.ConnectionManager
	ack        *paho.Connack
	generation uint64
}

func Topic(c config.Config, kind string) string {
	return c.Topics.Namespace + "/" + c.Topics.DeviceID + "/" + kind
}

// ObservedDevices is an explicit list: broker ACLs can authorize each exact topic.
func ObservedDevices(c config.Config) []string {
	devices := []string{c.Topics.DeviceID}
	seen := map[string]bool{c.Topics.DeviceID: true}
	if c.Topics.ObserveDevices != "" {
		for _, device := range strings.Split(c.Topics.ObserveDevices, ",") {
			device = strings.TrimSpace(device)
			if !seen[device] {
				devices = append(devices, device)
				seen[device] = true
			}
		}
	}
	return devices
}
func InNamespace(c config.Config, topic string) bool {
	return topic == c.Topics.Namespace || strings.HasPrefix(topic, c.Topics.Namespace+"/")
}
func Identity(c config.Config, topic string) (device, kind string) {
	if c.Topics.ObserveAll && InNamespace(c, topic) {
		parts := strings.Split(strings.TrimPrefix(topic, c.Topics.Namespace+"/"), "/")
		if len(parts) >= 2 {
			return parts[0], strings.Join(parts[1:], "/")
		}
		return "", ""
	}
	for _, device := range ObservedDevices(c) {
		for _, kind := range []string{"telemetry", "status", "commands", "acks"} {
			if topic == c.Topics.Namespace+"/"+device+"/"+kind {
				return device, kind
			}
		}
	}
	return "", ""
}
func Kind(c config.Config, topic string) string {
	_, kind := Identity(c, topic)
	return kind
}
func Backoff(min, max time.Duration) func(int) time.Duration {
	return func(attempt int) time.Duration {
		upper := min
		for i := 1; i < attempt && upper < max; i++ {
			if upper > max/2 {
				upper = max
			} else {
				upper *= 2
			}
		}
		if upper > max {
			upper = max
		}
		if upper <= min {
			return min
		}
		return min + time.Duration(rand.Int64N(int64(upper-min)+1))
	}
}
func TLSConfig(c config.Config) (*tls.Config, error) {
	tc := &tls.Config{MinVersion: tls.VersionTLS12}
	if c.TLS.CAFile != "" {
		data, err := os.ReadFile(c.TLS.CAFile)
		if err != nil {
			return nil, errors.New("cannot read CA file")
		}
		roots, err := x509.SystemCertPool()
		if err != nil {
			roots = x509.NewCertPool()
		}
		if !roots.AppendCertsFromPEM(data) {
			return nil, errors.New("CA file contains no valid certificates")
		}
		tc.RootCAs = roots
	}
	if c.TLS.ClientCert != "" {
		pair, err := tls.LoadX509KeyPair(c.TLS.ClientCert, c.TLS.ClientKey)
		if err != nil {
			return nil, errors.New("invalid client certificate/key pair")
		}
		tc.Certificates = []tls.Certificate{pair}
	}
	return tc, nil
}
func New(c config.Config, store *events.Store) *Client {
	return &Client{Config: c, Events: store, Incoming: make(chan *paho.Publish, c.Limits.InboundQueue), Down: make(chan struct{}, 1), changed: make(chan struct{}, 1), firstReady: make(chan struct{}), fatal: make(chan error, 1), workerDone: make(chan struct{})}
}
func (c *Client) Ready() bool         { return c.ready.Load() }
func (c *Client) Fatal() <-chan error { return c.fatal }
func (c *Client) change(cm *autopaho.ConnectionManager, ack *paho.Connack) {
	c.ready.Store(false)
	c.stateMu.Lock()
	c.generation++
	c.state = connection{cm, ack, c.generation}
	c.stateMu.Unlock()
	select {
	case c.changed <- struct{}{}:
	default:
	}
	if ack == nil {
		select {
		case c.Down <- struct{}{}:
		default:
		}
	}
}
func (c *Client) receive(r paho.PublishReceived) (bool, error) {
	p := r.Packet
	if p == nil {
		return true, nil
	}
	admitted := Kind(c.Config, p.Topic) != ""
	if c.Config.Topics.ObserveAll {
		admitted = InNamespace(c.Config, p.Topic)
	}
	if !admitted || len(p.Payload) > c.Config.Limits.PayloadBytes || (!c.Config.Topics.ObserveAll && p.Properties != nil && (len(p.Properties.CorrelationData) > 64 || len(p.Properties.ResponseTopic) > 256)) {
		c.invalid.Add(1)
		return true, nil
	}
	cp := *p
	cp.Payload = append([]byte(nil), p.Payload...)
	if p.Properties != nil {
		props := *p.Properties
		props.CorrelationData = append([]byte(nil), props.CorrelationData...)
		props.User = nil
		if props.MessageExpiry != nil {
			v := *props.MessageExpiry
			props.MessageExpiry = &v
		}
		cp.Properties = &props
	}
	select {
	case c.Incoming <- &cp:
	default:
		c.dropped.Add(1)
	}
	// Paho acknowledges receipt after callbacks. Receipt is not admission/execution.
	return true, nil
}
func (c *Client) Start(ctx context.Context, kinds []string, will *paho.WillMessage) error {
	if err := c.Config.Validate(); err != nil {
		return err
	}
	tc, err := TLSConfig(c.Config)
	if err != nil {
		return err
	}
	password := []byte(c.Config.MQTT.Password)
	if c.Config.MQTT.PasswordFile != "" {
		password, err = os.ReadFile(c.Config.MQTT.PasswordFile)
		if err != nil {
			return errors.New("cannot read password file")
		}
		password = []byte(strings.TrimRight(string(password), "\r\n"))
	}
	u, _ := url.Parse(c.Config.MQTT.URL)
	// Transport remains alive until explicit Close, so graceful offline publication can finish.
	life, cancel := context.WithCancel(context.Background())
	c.cancel = cancel
	go c.supervise(life, kinds)
	ac := autopaho.ClientConfig{
		ServerUrls: []*url.URL{u}, TlsCfg: tc, KeepAlive: uint16(c.Config.MQTT.KeepAlive / time.Second),
		CleanStartOnInitialConnection: c.Config.MQTT.CleanStart, SessionExpiryInterval: uint32(c.Config.MQTT.SessionExpiry / time.Second),
		ReconnectBackoff: Backoff(c.Config.MQTT.ReconnectMin, c.Config.MQTT.ReconnectMax), ConnectTimeout: c.Config.Deadlines.Connect,
		ConnectUsername: c.Config.MQTT.Username, ConnectPassword: password, WillMessage: will,
		OnConnectionUp:   func(cm *autopaho.ConnectionManager, a *paho.Connack) { c.change(cm, a) },
		OnConnectionDown: func() bool { c.change(nil, nil); return true },
		OnConnectError:   func(error) { c.invalid.Add(1) },
		ConnectPacketBuilder: func(p *paho.Connect, _ *url.URL) (*paho.Connect, error) {
			if p.Properties == nil {
				p.Properties = &paho.ConnectProperties{}
			}
			size := uint32(32768)
			receive := uint16(min(20, c.Config.Limits.InboundQueue))
			p.Properties.MaximumPacketSize = &size
			p.Properties.ReceiveMaximum = &receive
			return p, nil
		},
		ClientConfig: paho.ClientConfig{ClientID: c.Config.MQTT.ClientID, PacketTimeout: c.Config.Deadlines.Operation, OnPublishReceived: []func(paho.PublishReceived) (bool, error){c.receive}},
	}
	if will != nil {
		ac.WillProperties = &paho.WillProperties{ContentType: "application/json", PayloadFormat: paho.Byte(1), WillDelayInterval: paho.Uint32(0)}
	}
	c.cm, err = autopaho.NewConnection(life, ac)
	if err != nil {
		cancel()
		return errors.New("MQTT manager initialization failed")
	}
	wait, stop := context.WithTimeout(ctx, c.Config.Deadlines.Connect)
	defer stop()
	select {
	case <-c.firstReady:
		return nil
	case err := <-c.fatal:
		c.Close()
		return err
	case <-wait.Done():
		c.Close()
		return errors.New("MQTT readiness deadline exceeded or interrupted")
	}
}
func (c *Client) supervise(ctx context.Context, kinds []string) {
	defer close(c.workerDone)
	subscribed := false
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if n := c.dropped.Swap(0); n > 0 {
				c.Events.Add(events.Event{Kind: events.EventGap, Direction: events.Internal, Outcome: fmt.Sprintf("inbound_dropped=%d; MQTT receipt was not application admission", n)})
			}
			if n := c.invalid.Swap(0); n > 0 {
				c.Events.Add(events.Event{Kind: events.ValidationFailed, Direction: events.Internal, Outcome: fmt.Sprintf("invalid_input_or_connection_attempts=%d", n)})
			}
		case <-c.changed:
			c.stateMu.Lock()
			state := c.state
			c.stateMu.Unlock()
			if state.ack == nil {
				c.Events.Add(events.Event{Kind: events.ConnectionDown, Direction: events.Internal, Outcome: "disconnected; pending execution unknown"})
				continue
			}
			c.Events.Add(events.Event{Kind: events.ConnectionUp, Direction: events.Internal, ClientID: c.Config.MQTT.ClientID, Outcome: fmt.Sprintf("session_present=%t; protocol=5", state.ack.SessionPresent)})
			if state.ack.Properties != nil && state.ack.Properties.MaximumQoS != nil && *state.ack.Properties.MaximumQoS < 1 {
				c.fail("broker does not support required QoS 1")
				continue
			}
			if !state.ack.SessionPresent || !subscribed {
				subs := make([]paho.SubscribeOptions, 0, len(kinds))
				for _, device := range ObservedDevices(c.Config) {
					for _, kind := range kinds {
						subs = append(subs, paho.SubscribeOptions{Topic: c.Config.Topics.Namespace + "/" + device + "/" + kind, QoS: 1})
					}
				}
				if c.Config.Topics.ObserveAll {
					subs = []paho.SubscribeOptions{{Topic: c.Config.Topics.Namespace + "/#", QoS: 1}}
				}
				op, stop := context.WithTimeout(ctx, c.Config.Deadlines.Operation)
				ack, err := state.cm.Subscribe(op, &paho.Subscribe{Subscriptions: subs})
				stop()
				c.stateMu.Lock()
				current := c.generation == state.generation
				c.stateMu.Unlock()
				// A lost connection invalidates this SUBACK attempt. Let the
				// newer connection restore readiness instead of failing it.
				if !current {
					continue
				}
				if err != nil || ack == nil || len(ack.Reasons) != len(subs) {
					c.fail("subscription setup failed")
					continue
				}
				valid := true
				for _, reason := range ack.Reasons {
					if reason >= 128 {
						valid = false
					}
				}
				if !valid {
					c.fail("broker rejected required subscription")
					continue
				}
				subscribed = true
			}
			c.stateMu.Lock()
			current := c.generation == state.generation
			if current {
				c.ready.Store(true)
			}
			c.stateMu.Unlock()
			if current {
				c.Events.Add(events.Event{Kind: events.Kind("subscriptions_ready"), Direction: events.Internal, ClientID: c.Config.MQTT.ClientID, Outcome: c.SubscriptionSummary(kinds)})
				c.firstOnce.Do(func() { close(c.firstReady) })
			}
		}
	}
}
func (c *Client) SubscriptionSummary(kinds []string) string {
	if c.Config.Topics.ObserveAll {
		return "topic=" + c.Config.Topics.Namespace + "/#; raw observation; contract validation disabled"
	}
	return "devices=" + strings.Join(ObservedDevices(c.Config), ",") + "; categories=" + strings.Join(kinds, ",")
}
func (c *Client) fail(message string) {
	c.ready.Store(false)
	select {
	case c.fatal <- errors.New(message):
	default:
	}
}
func (c *Client) Close() error {
	ctx, stop := context.WithTimeout(context.Background(), c.Config.Deadlines.Shutdown)
	defer stop()
	return c.CloseContext(ctx)
}

// CloseContext shares the caller's complete shutdown budget, including offline publication.
func (c *Client) CloseContext(ctx context.Context) error {
	if c.cancel == nil {
		return nil
	}
	c.change(nil, nil)
	var err error
	if c.cm != nil {
		err = c.cm.Disconnect(ctx)
	}
	c.cancel()
	select {
	case <-c.workerDone:
	case <-ctx.Done():
		return errors.New("MQTT shutdown deadline exceeded")
	}
	if err != nil {
		return errors.New("MQTT shutdown failed")
	}
	return nil
}

func (c *Client) Record(kind events.Kind, direction events.Direction, p *paho.Publish, request, outcome string) {
	e := events.Event{Kind: kind, Direction: direction, ClientID: c.Config.MQTT.ClientID, DeviceID: c.Config.Topics.DeviceID, RequestID: request, Outcome: outcome}
	if p != nil {
		e.Topic = p.Topic
		e.QoS = p.QoS
		e.Retain = p.Retain
		duplicate := p.Duplicate()
		e.Duplicate = &duplicate
		device, category := Identity(c.Config, p.Topic)
		e.DeviceID = device
		e.Payload = events.SafePreview(category, device, p.Payload, c.Config.Limits.PreviewBytes)
		if c.Config.Logging.RawPayloads || c.Config.Topics.ObserveAll {
			e.Payload = events.RawPreview(p.Payload, c.Config.Limits.PreviewBytes)
		}
		// Only native values belonging to our exact test contract are visible.
		if p.Properties != nil {
			if p.Properties.ResponseTopic == Topic(c.Config, "acks") || (c.Config.Topics.ObserveAll && len(p.Properties.ResponseTopic) <= 256) {
				e.Properties.ResponseTopic = p.Properties.ResponseTopic
			}
			if len(p.Properties.CorrelationData) <= 64 && (c.Config.Topics.ObserveAll || (request != "" && string(p.Properties.CorrelationData) == request)) {
				e.Properties.CorrelationData = append([]byte(nil), p.Properties.CorrelationData...)
			}
			e.Properties.MessageExpirySeconds = p.Properties.MessageExpiry
		}
	}
	c.Events.Add(e)
}
func (c *Client) Publish(ctx context.Context, p *paho.Publish) error {
	kind := Kind(c.Config, p.Topic)
	if kind == "" || len(p.Payload) > c.Config.Limits.PayloadBytes {
		return errors.New("publication outside test contract")
	}
	msg, err := contracts.Decode(kind, p.Payload, c.Config.Topics.DeviceID)
	if err != nil {
		return err
	}
	request := ""
	switch v := msg.(type) {
	case *contracts.Command:
		request = v.RequestID
	case *contracts.Ack:
		request = v.RequestID
	}
	return c.publishPacket(ctx, p, request)
}

// PublishRaw sends test receipt replies without requiring a device JSON contract.
func (c *Client) PublishRaw(ctx context.Context, p *paho.Publish) error {
	if p == nil || !InNamespace(c.Config, p.Topic) || len(p.Payload) > c.Config.Limits.PayloadBytes {
		return errors.New("receipt publication outside test namespace or size limit")
	}
	return c.publishPacket(ctx, p, "")
}
func (c *Client) publishPacket(ctx context.Context, p *paho.Publish, request string) error {
	c.Record(events.PublishAttempt, events.Out, p, request, "")
	if !c.Ready() {
		return ErrNotReady
	}
	op, stop := context.WithTimeout(ctx, c.Config.Deadlines.Operation)
	defer stop()
	ack, err := c.cm.Publish(op, p)
	if err != nil {
		c.Record(events.Kind("transport_unknown"), events.Internal, p, request, "publication not confirmed; execution unknown")
		return ErrUnknown
	}
	if ack != nil && ack.ReasonCode >= 128 {
		c.Record(events.Kind("transport_rejected"), events.Internal, p, request, fmt.Sprintf("reason_code=%d", ack.ReasonCode))
		return errors.New("broker rejected publication")
	}
	outcome := "QoS0 write completed; no broker acknowledgement"
	if p.QoS == 1 {
		if ack == nil {
			return ErrUnknown
		}
		outcome = fmt.Sprintf("reason_code=%d; device execution not proven", ack.ReasonCode)
	}
	c.Record(events.TransportAck, events.Internal, p, request, outcome)
	return nil
}
func Packet(topic string, qos byte, retained bool, value any) *paho.Publish {
	data, _ := json.Marshal(value)
	return &paho.Publish{Topic: topic, QoS: qos, Retain: retained, Payload: data, Properties: &paho.PublishProperties{ContentType: "application/json", PayloadFormat: paho.Byte(1)}}
}