package jetstream

import (
	"context"
	"sync"
	"sync/atomic"
)

// a crash replays whatever is still pending because the cursor only moves up to the front
type watermark struct {
	mu      sync.Mutex
	front   uint64
	pending []mark
}

type mark struct {
	us   int64
	done bool
}

// a ticket is a held event, the cursor can't pass it until Done.
// skipping Done pins the cursor here forever
type Ticket struct {
	j  *JetstreamClient
	id uint64
	us int64
	// shutdown waits for held tickets before the last cursor save
	holding atomic.Bool
}

func (j *JetstreamClient) ticket(us int64) *Ticket {
	w := &j.marks
	w.mu.Lock()
	defer w.mu.Unlock()
	id := w.front + uint64(len(w.pending))
	w.pending = append(w.pending, mark{us: us})
	return &Ticket{j: j, id: id, us: us}
}

func (t *Ticket) Done() {
	if t == nil {
		return
	}
	defer t.release()
	w := &t.j.marks
	w.mu.Lock()
	defer w.mu.Unlock()
	if t.id < w.front {
		return
	}
	w.pending[t.id-w.front].done = true
	var newest int64
	moved := false
	for len(w.pending) > 0 && w.pending[0].done {
		newest, moved = max(newest, w.pending[0].us), true
		w.pending = w.pending[1:]
		w.front++
	}
	// never move the cursor back, since reconnect replays can be older than it
	if moved && newest+1 > t.j.lastSeenUs.Load() {
		t.j.lastSeenUs.Store(newest + 1)
	}
}

// the cursor stays before this event since dropping doesn't mark it done
func (t *Ticket) Drop() {
	if t != nil {
		t.release()
	}
}

func (t *Ticket) release() {
	if t.holding.CompareAndSwap(true, false) {
		t.j.holding.Add(-1)
	}
}

type holdKey struct{}

type hold struct {
	ticket *Ticket
	held   atomic.Bool
}

// hold before the processor returns, or the cursor moves past it
func Hold(ctx context.Context) *Ticket {
	h, ok := ctx.Value(holdKey{}).(*hold)
	if !ok {
		return nil
	}
	h.held.Store(true)
	if h.ticket.holding.CompareAndSwap(false, true) {
		h.ticket.j.holding.Add(1)
	}
	return h.ticket
}