package natscap

import (
	"context"
	"errors"
	"strings"
	"sync"
	"time"

	merrors "git.sendico.io/tech/sxchnge/shared/errors"
	"git.sendico.io/tech/sxchnge/shared/messaging/capability"
	"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
	"github.com/nats-io/nats.go"
)

type natsSubscription struct {
	mu     sync.Mutex
	sub    *nats.Subscription
	closed bool
}

func fetchPollContext(ctx context.Context, wait time.Duration) (context.Context, context.CancelFunc, bool) {
	if ctx == nil {
		ctx = context.Background()
	}
	if wait <= 0 {
		return ctx, nil, false
	}
	if deadline, hasDeadline := ctx.Deadline(); hasDeadline && time.Until(deadline) <= wait {
		return ctx, nil, false
	}
	nextCtx, cancel := context.WithTimeout(ctx, wait)
	return nextCtx, cancel, true
}

func (s *natsSubscription) Fetch(ctx context.Context, limit int, wait time.Duration) ([]reliable.Delivery, error) {
	if limit <= 0 {
		return nil, merrors.InvalidArgument("fetch limit must be greater than zero", "limit")
	}
	s.mu.Lock()
	closed := s.closed
	sub := s.sub
	s.mu.Unlock()
	if closed || sub == nil {
		return nil, capability.ErrClosed
	}
	fetchCtx, cancel, derivedDeadline := fetchPollContext(ctx, wait)
	if cancel != nil {
		defer cancel()
	}
	msgs, err := sub.Fetch(limit, nats.Context(fetchCtx))
	if err != nil {
		switch {
		case errors.Is(err, nats.ErrTimeout):
			return nil, nil
		case derivedDeadline && errors.Is(err, context.DeadlineExceeded):
			return nil, nil
		default:
			return nil, err
		}
	}
	s.mu.Lock()
	closed = s.closed
	s.mu.Unlock()
	if closed {
		return nil, capability.ErrClosed
	}
	result := make([]reliable.Delivery, 0, len(msgs))
	for _, msg := range msgs {
		if msg == nil {
			continue
		}
		result = append(result, &natsDelivery{sub: s, msg: msg, record: messageRecord(msg), attempt: deliveryAttempt(msg)})
	}
	return result, nil
}

func (s *natsSubscription) Close() error {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return nil
	}
	s.closed = true
	sub := s.sub
	s.sub = nil
	s.mu.Unlock()
	if sub == nil {
		return nil
	}
	if err := sub.Unsubscribe(); err != nil && !errors.Is(err, nats.ErrBadSubscription) {
		return err
	}
	return nil
}

func (s *natsSubscription) isClosed() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.closed
}

type natsDelivery struct {
	sub     *natsSubscription
	msg     *nats.Msg
	record  reliable.Record
	attempt int

	mu    sync.Mutex
	acked bool
}

func (d *natsDelivery) Record() reliable.Record {
	return d.record
}

func (d *natsDelivery) Attempt() int {
	if d.attempt <= 0 {
		return 1
	}
	return d.attempt
}

func (d *natsDelivery) Ack(ctx context.Context) error {
	d.mu.Lock()
	if d.acked {
		d.mu.Unlock()
		return nil
	}
	if d.sub != nil && d.sub.isClosed() {
		d.mu.Unlock()
		return capability.ErrClosed
	}
	msg := d.msg
	d.mu.Unlock()
	if msg == nil {
		return capability.ErrClosed
	}
	if err := msg.AckSync(nats.Context(ctx)); err != nil {
		return err
	}
	d.mu.Lock()
	d.acked = true
	d.mu.Unlock()
	return nil
}

func messageRecord(msg *nats.Msg) reliable.Record {
	headers := make(map[string][]string, len(msg.Header)+1)
	for key, values := range msg.Header {
		headers[strings.ToLower(strings.TrimSpace(key))] = append([]string(nil), values...)
	}
	messageID := strings.TrimSpace(msg.Header.Get(nats.MsgIdHdr))
	if businessID := reliable.HeaderValue(headers, reliable.HeaderOriginalBusinessMsgID); businessID != "" {
		messageID = businessID
	}
	headers[reliable.HeaderTransportMessageID] = []string{strings.TrimSpace(msg.Header.Get(nats.MsgIdHdr))}
	return reliable.Record{
		Subject:   strings.TrimSpace(msg.Subject),
		Payload:   append([]byte(nil), msg.Data...),
		Headers:   headers,
		MessageID: strings.TrimSpace(messageID),
	}
}

func deliveryAttempt(msg *nats.Msg) int {
	if msg == nil {
		return 1
	}
	meta, err := msg.Metadata()
	if err != nil || meta == nil || meta.NumDelivered == 0 {
		return 1
	}
	if meta.NumDelivered > uint64(^uint(0)>>1) {
		return int(^uint(0) >> 1)
	}
	return int(meta.NumDelivered)
}