package natscap

import (
	"context"
	"strings"

	"git.sendico.io/tech/sxchnge/shared/messaging/capability"
	basenats "git.sendico.io/tech/sxchnge/shared/messaging/internal/natsb"
	"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
	"github.com/nats-io/nats.go"
)

type DurableAdapter struct {
	broker *basenats.NatsBroker
}

type durableAdminContext interface {
	AddStream(cfg *nats.StreamConfig, opts ...nats.JSOpt) (*nats.StreamInfo, error)
	UpdateStream(cfg *nats.StreamConfig, opts ...nats.JSOpt) (*nats.StreamInfo, error)
	StreamInfo(stream string, opts ...nats.JSOpt) (*nats.StreamInfo, error)
	AddConsumer(stream string, cfg *nats.ConsumerConfig, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
	UpdateConsumer(stream string, cfg *nats.ConsumerConfig, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
	ConsumerInfo(stream, consumer string, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
}

func NewDurableAdapter(broker *basenats.NatsBroker) *DurableAdapter {
	if broker == nil || broker.JetStream() == nil {
		return nil
	}
	return &DurableAdapter{broker: broker}
}

func (a *DurableAdapter) Publish(ctx context.Context, record reliable.Record) error {
	if err := reliable.ValidatePublicRecord(record); err != nil {
		return err
	}
	normalized, err := reliable.NormalizePublicRecord(record)
	if err != nil {
		return err
	}
	return a.publishRecord(ctx, normalized)
}

func (a *DurableAdapter) PublishRuntime(ctx context.Context, record reliable.Record) error {
	normalized, err := reliable.NormalizeRuntimeRecord(record)
	if err != nil {
		return err
	}
	return a.publishRecord(ctx, normalized)
}

func (a *DurableAdapter) publishRecord(ctx context.Context, record reliable.Record) error {
	js := a.jetStream()
	if js == nil {
		return capability.Unavailable("durable transport")
	}
	msg := &nats.Msg{
		Subject: strings.TrimSpace(record.Subject),
		Data:    append([]byte(nil), record.Payload...),
		Header:  nats.Header{},
	}
	for key, values := range record.Headers {
		for _, value := range values {
			msg.Header.Add(key, value)
		}
	}
	if record.MessageID != "" {
		msg.Header.Set(nats.MsgIdHdr, record.MessageID)
	}
	_, err := js.PublishMsg(msg, nats.Context(ctx))
	return err
}

func (a *DurableAdapter) OpenConsumer(ctx context.Context, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) (reliable.Subscription, error) {
	js := a.jetStream()
	if js == nil {
		return nil, capability.Unavailable("durable transport")
	}
	sub, err := js.PullSubscribe(
		consumer.Subject,
		consumer.Durable,
		nats.Bind(settings.Stream.Name, consumer.Durable),
		nats.Context(ctx),
	)
	if err != nil {
		return nil, err
	}
	return &natsSubscription{sub: sub}, nil
}

func (a *DurableAdapter) EnsureProducerResources(ctx context.Context, settings reliable.DurableSettings) error {
	if a == nil || a.broker == nil {
		return capability.Unavailable("durable admin")
	}
	return ensureDurableStream(ctx, a.broker.Logger(), a.jetStream(), settings)
}

func (a *DurableAdapter) EnsureConsumerResources(ctx context.Context, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) error {
	if a == nil || a.broker == nil {
		return capability.Unavailable("durable admin")
	}
	js := a.jetStream()
	if err := ensureDurableStream(ctx, a.broker.Logger(), js, settings); err != nil {
		return err
	}
	return ensureDurableConsumerConfig(ctx, a.broker.Logger(), js, settings, consumer)
}

func (a *DurableAdapter) jetStream() nats.JetStreamContext {
	if a == nil || a.broker == nil {
		return nil
	}
	return a.broker.JetStream()
}