package mq

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"strconv"
	"strings"
	"sync"
	"time"

	"kra/internal/integration/runtimeconfig"
	platformmq "kra/pkg/mq"
)

const (
	ProviderEMQX     = "emqx"
	ProviderRabbitMQ = "rabbitmq"
)

// Reloadable owns the process-wide message clients. Configuration comes only
// from sys_integration_configs through runtimeconfig.Store.
type Reloadable struct {
	mu            sync.RWMutex
	opMu          sync.Mutex
	clients       map[string]platformmq.Client
	subscriptions map[string]map[string]subscription
	stop          []func()
	logger        *slog.Logger
	closed        bool
}

type subscription struct {
	qos     byte
	handler platformmq.Handler
}

type namedClient struct {
	owner    *Reloadable
	provider string
}

func New(store *runtimeconfig.Store, logger *slog.Logger) (*Reloadable, func(), error) {
	if logger == nil {
		logger = slog.Default()
	}
	r := &Reloadable{
		clients:       make(map[string]platformmq.Client),
		subscriptions: make(map[string]map[string]subscription),
		logger:        logger,
	}
	if store != nil {
		r.apply(ProviderEMQX, storeConfig(store, ProviderEMQX))
		r.apply(ProviderRabbitMQ, storeConfig(store, ProviderRabbitMQ))
		r.stop = append(r.stop,
			store.Subscribe("mq", ProviderEMQX, func(config runtimeconfig.Config) { r.apply(ProviderEMQX, config) }),
			store.Subscribe("mq", ProviderRabbitMQ, func(config runtimeconfig.Config) { r.apply(ProviderRabbitMQ, config) }),
		)
	}
	cleanup := func() {
		for _, stop := range r.stop {
			stop()
		}
		_ = r.Close()
	}
	return r, cleanup, nil
}

func storeConfig(store *runtimeconfig.Store, provider string) runtimeconfig.Config {
	config, _ := store.Get("mq", provider)
	return config
}

func (r *Reloadable) apply(provider string, config runtimeconfig.Config) {
	r.opMu.Lock()
	defer r.opMu.Unlock()
	if r.closed {
		return
	}
	if !config.Enabled {
		r.replaceClientLocked(provider, nil)
		return
	}
	client, err := newProviderClient(provider, config.Values)
	if err != nil {
		r.logger.Warn("message integration unavailable", "mod", "mq", "provider", provider, "error", err)
		return
	}
	if err = r.restoreSubscriptionsLocked(provider, client); err != nil {
		_ = client.Close()
		r.logger.Warn("restore message subscriptions failed", "mod", "mq", "provider", provider, "error", err)
		return
	}
	r.replaceClientLocked(provider, client)
}

func newProviderClient(provider string, raw json.RawMessage) (platformmq.Client, error) {
	values := map[string]any{}
	if err := json.Unmarshal(raw, &values); err != nil {
		return nil, fmt.Errorf("decode %s configuration: %w", provider, err)
	}
	switch provider {
	case ProviderEMQX:
		return platformmq.NewMQTT(platformmq.Config{
			Enabled:        true,
			Broker:         configText(values, "broker"),
			ClientID:       configText(values, "client_id"),
			Username:       configText(values, "username"),
			Password:       configText(values, "password"),
			KeepAlive:      configSeconds(values, "keep_alive"),
			CleanSession:   configBool(values, "clean_session"),
			ConnectTimeout: configSeconds(values, "connect_timeout"),
		})
	case ProviderRabbitMQ:
		return platformmq.NewRabbitMQ(platformmq.RabbitMQConfig{
			Enabled:        true,
			Host:           configText(values, "host"),
			Port:           configInt(values, "port"),
			Username:       configText(values, "username"),
			Password:       configText(values, "password"),
			VHost:          configText(values, "vhost"),
			Exchange:       configText(values, "exchange"),
			ExchangeType:   configText(values, "exchange_type"),
			Queue:          configText(values, "queue"),
			RoutingKey:     configText(values, "routing_key"),
			Durable:        configBool(values, "durable"),
			AutoDelete:     configBool(values, "auto_delete"),
			PrefetchCount:  configInt(values, "prefetch_count"),
			Heartbeat:      configSeconds(values, "heartbeat"),
			ConnectTimeout: configSeconds(values, "connect_timeout"),
			TLS:            configBool(values, "tls"),
		})
	default:
		return nil, fmt.Errorf("unsupported message provider %q", provider)
	}
}

func configText(values map[string]any, key string) string {
	value, ok := values[key]
	if !ok || value == nil {
		return ""
	}
	return strings.TrimSpace(fmt.Sprint(value))
}

func configInt(values map[string]any, key string) int {
	switch value := values[key].(type) {
	case float64:
		return int(value)
	case int:
		return value
	case json.Number:
		parsed, _ := strconv.Atoi(string(value))
		return parsed
	default:
		parsed, _ := strconv.Atoi(configText(values, key))
		return parsed
	}
}

func configSeconds(values map[string]any, key string) time.Duration {
	seconds := configInt(values, key)
	if seconds <= 0 {
		return 0
	}
	return time.Duration(seconds) * time.Second
}

func configBool(values map[string]any, key string) bool {
	value, _ := values[key].(bool)
	return value
}

func (r *Reloadable) replaceClientLocked(provider string, next platformmq.Client) {
	r.mu.Lock()
	old := r.clients[provider]
	if next == nil {
		delete(r.clients, provider)
	} else {
		r.clients[provider] = next
	}
	r.mu.Unlock()
	if old != nil {
		_ = old.Close()
	}
}

func (r *Reloadable) restoreSubscriptionsLocked(provider string, client platformmq.Client) error {
	for topic, item := range r.subscriptions[provider] {
		if err := client.Subscribe(context.Background(), topic, item.qos, item.handler); err != nil {
			return err
		}
	}
	return nil
}

func (r *Reloadable) client(provider string) platformmq.Client {
	provider = strings.ToLower(strings.TrimSpace(provider))
	r.mu.RLock()
	defer r.mu.RUnlock()
	return r.clients[provider]
}

func (r *Reloadable) Client(provider string) platformmq.Client {
	provider = strings.ToLower(strings.TrimSpace(provider))
	if provider != ProviderEMQX && provider != ProviderRabbitMQ {
		return nil
	}
	return &namedClient{owner: r, provider: provider}
}

func (c *namedClient) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error {
	return c.owner.PublishTo(ctx, c.provider, topic, payload, qos, retain)
}
func (c *namedClient) Subscribe(ctx context.Context, topic string, qos byte, handler platformmq.Handler) error {
	return c.owner.SubscribeTo(ctx, c.provider, topic, qos, handler)
}
func (c *namedClient) Unsubscribe(ctx context.Context, topics ...string) error {
	return c.owner.UnsubscribeFrom(ctx, c.provider, topics...)
}
func (c *namedClient) Connected() bool { return c.owner.ConnectedTo(c.provider) }
func (*namedClient) Close() error      { return nil }

func (r *Reloadable) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error {
	return r.PublishTo(ctx, ProviderEMQX, topic, payload, qos, retain)
}

func (r *Reloadable) PublishTo(ctx context.Context, provider, topic string, payload []byte, qos byte, retain bool) error {
	provider = strings.ToLower(strings.TrimSpace(provider))
	r.mu.RLock()
	defer r.mu.RUnlock()
	client := r.clients[provider]
	if client == nil {
		return platformmq.ErrUnavailable
	}
	return client.Publish(ctx, topic, payload, qos, retain)
}

func (r *Reloadable) Subscribe(ctx context.Context, topic string, qos byte, handler platformmq.Handler) error {
	return r.SubscribeTo(ctx, ProviderEMQX, topic, qos, handler)
}

func (r *Reloadable) SubscribeTo(ctx context.Context, provider, topic string, qos byte, handler platformmq.Handler) error {
	provider = strings.ToLower(strings.TrimSpace(provider))
	r.opMu.Lock()
	defer r.opMu.Unlock()
	client := r.client(provider)
	if client == nil {
		return platformmq.ErrUnavailable
	}
	if err := client.Subscribe(ctx, topic, qos, handler); err != nil {
		return err
	}
	if r.subscriptions[provider] == nil {
		r.subscriptions[provider] = make(map[string]subscription)
	}
	r.subscriptions[provider][topic] = subscription{qos: qos, handler: handler}
	return nil
}

func (r *Reloadable) Unsubscribe(ctx context.Context, topics ...string) error {
	return r.UnsubscribeFrom(ctx, ProviderEMQX, topics...)
}

func (r *Reloadable) UnsubscribeFrom(ctx context.Context, provider string, topics ...string) error {
	provider = strings.ToLower(strings.TrimSpace(provider))
	r.opMu.Lock()
	defer r.opMu.Unlock()
	client := r.client(provider)
	if client == nil {
		return platformmq.ErrUnavailable
	}
	if err := client.Unsubscribe(ctx, topics...); err != nil {
		return err
	}
	for _, topic := range topics {
		delete(r.subscriptions[provider], topic)
	}
	return nil
}

func (r *Reloadable) Connected() bool { return r.ConnectedTo(ProviderEMQX) }

func (r *Reloadable) ConnectedTo(provider string) bool {
	provider = strings.ToLower(strings.TrimSpace(provider))
	r.mu.RLock()
	defer r.mu.RUnlock()
	client := r.clients[provider]
	return client != nil && client.Connected()
}

func (r *Reloadable) Close() error {
	r.opMu.Lock()
	defer r.opMu.Unlock()
	if r.closed {
		return nil
	}
	r.closed = true
	r.mu.Lock()
	clients := make([]platformmq.Client, 0, len(r.clients))
	for provider, client := range r.clients {
		clients = append(clients, client)
		delete(r.clients, provider)
	}
	r.mu.Unlock()
	for _, client := range clients {
		if client != nil {
			_ = client.Close()
		}
	}
	return nil
}
