package process

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

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

// Message is one piece of work this process's queue consumer delivered, with how many times the bus has
// delivered it. Nothing acts on its behalf: the handler decides its fate exactly once, by calling Ack,
// NakWithDelay or Term (mesh-tools node-tools/internal/bus/seat.go's seat work Consume, with the same
// three outcomes, offered here explicitly instead of by a handler's return value, since the caller
// chooses its own delay before a message comes back).
type Message struct {
	Subject string
	Body    []byte
	Headers map[string]string
	// Delivered is how many times the bus has handed this message over, this time included.
	Delivered uint64

	ack  func() error
	nak  func(time.Duration) error
	term func() error
}

// Ack acknowledges the message: the bus never delivers it again.
func (m *Message) Ack() error { return m.ack() }

// NakWithDelay asks for the message to be delivered again no sooner than delay, its delivery count
// raised by one.
func (m *Message) NakWithDelay(delay time.Duration) error { return m.nak(delay) }

// Term terminates the message: the bus never delivers it again, and does not count it a dead letter the
// way running out of deliveries does.
func (m *Message) Term() error { return m.term() }

// ProcessID is this account's own process id, the last segment of its user name <node>.<module>.<id>
// (mesh-controller internal/broker, Principal.Username for KindProcess).
func (a Account) ProcessID() string {
	prefix := a.Node + "." + a.Module + "."
	if !strings.HasPrefix(a.User, prefix) {
		return ""
	}
	return strings.TrimPrefix(a.User, prefix)
}

// QueueConsumerName is the one durable consumer a process's bus.queues names it on its work-queue stream
// (novox/hq ADR 0436 clauses 1, 2, 6; mesh-controller internal/broker/modstreams.go QueueConsumer):
// <node>_<module>_<process id>.
func (a Account) QueueConsumerName() string { return a.Node + "_" + a.Module + "_" + a.ProcessID() }

// errNoMessage is one pull that fetched nothing within its wait: not an error to the caller, just a
// reason to pull again.
var errNoMessage = errors.New("no message this pull")

// Consume binds this process's one durable consumer on stream (QueueConsumerName) and pulls from it, one
// message at a time, handing each to handle until ctx ends: it then stops pulling and returns ctx.Err(),
// never acknowledging a message still in handle's hands — only handle decides a message's fate, by
// calling Ack, NakWithDelay or Term on it.
func (c *Conn) Consume(ctx context.Context, stream string, handle func(*Message)) error {
	js, err := jetstream.New(c.nc)
	if err != nil {
		return err
	}
	name := c.account.QueueConsumerName()
	cons, err := js.Consumer(ctx, stream, name)
	if err != nil {
		return fmt.Errorf("binding %s's queue consumer %s on %s: %w", c.account.Module, name, stream, err)
	}
	return runQueue(ctx, fetchOne(cons), handle)
}

// fetchOne is one pull of at most one message, non-blocking past ctx: Fetch itself does not take a
// context, so this selects on its message channel and ctx together, and returns promptly once ctx ends
// even while a fetch is still outstanding.
func fetchOne(cons jetstream.Consumer) func(ctx context.Context) (*Message, error) {
	return func(ctx context.Context) (*Message, error) {
		batch, err := cons.Fetch(1, jetstream.FetchMaxWait(5*time.Second))
		if err != nil {
			return nil, err
		}
		select {
		case msg, ok := <-batch.Messages():
			if !ok {
				return nil, errNoMessage
			}
			headers := map[string]string{}
			for k := range msg.Headers() {
				headers[k] = msg.Headers().Get(k)
			}
			var delivered uint64 = 1
			if meta, err := msg.Metadata(); err == nil && meta != nil {
				delivered = meta.NumDelivered
			}
			return &Message{Subject: msg.Subject(), Body: append([]byte(nil), msg.Data()...), Headers: headers,
				Delivered: delivered, ack: msg.Ack, nak: msg.NakWithDelay, term: msg.Term}, nil
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
}

// runQueue is Consume's loop, apart from the bus so a unit test can drive it with a fake next
// (novox/hq issue 1190 AC4): it checks ctx before every pull, and stops as soon as next or the check
// says ctx has ended, without touching the message past handing it to handle.
func runQueue(ctx context.Context, next func(context.Context) (*Message, error), handle func(*Message)) error {
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		msg, err := next(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			if errors.Is(err, errNoMessage) {
				continue
			}
			return err
		}
		handle(msg)
	}
}