package eventbus

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/nats-io/nats.go"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics"
)

type NATSConfig struct {
	URL            string
	Name           string
	RawSubjects    map[envelope.Protocol]string
	UnifiedSubject string
}

type NATSSink struct {
	conn           *nats.Conn
	publisher      natsPublisher
	rawSubjects    map[envelope.Protocol]string
	unifiedSubject string
}

type NATSPublishOption = nats.PubOpt

type natsPublisher interface {
	Publish(context.Context, string, []byte, ...NATSPublishOption) error
}

func NewNATSSink(cfg NATSConfig) (*NATSSink, error) {
	if cfg.URL == "" {
		return nil, errors.New("nats url is required")
	}
	name := cfg.Name
	if name == "" {
		name = "lingniu-vehicle-gateway"
	}
	conn, err := nats.Connect(cfg.URL, nats.Name(name), nats.Timeout(5*time.Second))
	if err != nil {
		return nil, err
	}
	js, err := conn.JetStream()
	if err != nil {
		conn.Close()
		return nil, err
	}
	sink := newNATSSinkWithPublisher(natsJetStreamPublisher{js: js}, cfg)
	sink.conn = conn
	return sink, nil
}

func newNATSSinkWithPublisher(publisher natsPublisher, cfg NATSConfig) *NATSSink {
	rawSubjects := map[envelope.Protocol]string{
		envelope.ProtocolGB32960:    topics.RawGB32960,
		envelope.ProtocolJT808:      topics.RawJT808,
		envelope.ProtocolYutongMQTT: topics.RawYutongMQTT,
	}
	for protocol, subject := range cfg.RawSubjects {
		if subject != "" {
			rawSubjects[protocol] = subject
		}
	}
	unifiedSubject := cfg.UnifiedSubject
	if unifiedSubject == "" {
		unifiedSubject = topics.Unified
	}
	return &NATSSink{
		publisher:      publisher,
		rawSubjects:    rawSubjects,
		unifiedSubject: unifiedSubject,
	}
}

func (s *NATSSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error {
	subject, ok := s.rawSubjects[env.Protocol]
	if !ok || subject == "" {
		return fmt.Errorf("raw subject not configured for protocol %s", env.Protocol)
	}
	return s.publish(ctx, subject, env)
}

func (s *NATSSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error {
	if s.unifiedSubject == "" {
		return errors.New("unified subject is empty")
	}
	return s.publish(ctx, s.unifiedSubject, env)
}

func (s *NATSSink) Close() error {
	if s == nil || s.conn == nil {
		return nil
	}
	_ = s.conn.Drain()
	s.conn.Close()
	return nil
}

func (s *NATSSink) publish(ctx context.Context, subject string, env envelope.FrameEnvelope) error {
	payload, err := env.MarshalJSONBytes()
	if err != nil {
		return err
	}
	return s.publisher.Publish(ctx, subject, payload, nats.MsgId(env.StableEventID()))
}

type natsJetStreamPublisher struct {
	js nats.JetStreamContext
}

func (p natsJetStreamPublisher) Publish(ctx context.Context, subject string, data []byte, opts ...NATSPublishOption) error {
	_, err := p.js.Publish(subject, data, append(opts, nats.Context(ctx))...)
	return err
}