package hub

import (
	"context"
	"fmt"
	"time"
)

func isTargeted(t string) bool { return t == TypeOffer || t == TypeAnswer || t == TypeCandidate }

// Relay delivers one targeted message. The envelope is rebuilt from the
// binding; the payload is never read.
//
// A member's standing is re-derived first (§15.12): the participation
// authorizes everything after entry (§2.4), but a member who has lost
// the team is dropped by the next thing they do, and this is that thing.
// A member with no standing is ended here and the request answers
// ErrNotYours, which the handler renders as the same not-found body an
// unknown peer gets.
func (h *Hub) Relay(ctx context.Context, sqid, session string, from PeerID, m Message) error {
	if !isTargeted(m.Type) {
		return fmt.Errorf("%w: %q", ErrNotTargeted, m.Type)
	}
	gone, err := h.memberStandingGone(ctx, sqid, session, from, true)
	if err != nil {
		return err
	}
	if gone {
		return ErrNotYours
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	r, ok := h.rooms[sqid]
	if !ok {
		return ErrUnknownPeer
	}
	p, found := h.findLocked(sqid, session, from)
	if !found {
		return ErrUnknownPeer
	}
	if !p.Admitted {
		return ErrNotAdmitted
	}
	if m.To == p.Peer {
		return ErrUnknownPeer // a peer is not its own target
	}
	tp, ok := r.peers[m.To]
	if !ok || !tp.Admitted {
		return ErrUnknownPeer
	}
	if m.Gen != h.genLocked(r, p.Peer, tp.Peer) {
		return ErrStaleGen
	}
	out := Message{Type: m.Type, From: p.Peer, Name: p.Name, To: tp.Peer, Seq: m.Seq, Gen: m.Gen, Payload: m.Payload}
	h.ringLocked(tp, out)
	h.sendLocked(tp, out)
	return nil
}

// ringLocked keeps a copy for the resume window, newest last. What falls
// out of the ring is remembered as a high-water mark per sender, which
// is the only way a resume can tell that a sender's messages were
// evicted wholesale by another sender's burst.
func (h *Hub) ringLocked(p *Participation, m Message) {
	p.ring = append(p.ring, m)
	if len(p.ring) <= RingSize {
		return
	}
	for _, old := range p.ring[:len(p.ring)-RingSize] {
		if !isTargeted(old.Type) {
			continue
		}
		if p.evicted == nil {
			p.evicted = map[PeerID]uint64{}
		}
		if old.Seq > p.evicted[old.From] {
			p.evicted[old.From] = old.Seq
		}
	}
	p.ring = p.ring[len(p.ring)-RingSize:]
}

// Reset bumps the pair's generation and tells both sides. A member's
// standing is re-derived first, exactly as Relay does (§15.12).
func (h *Hub) Reset(ctx context.Context, sqid, session string, from, to PeerID) error {
	gone, err := h.memberStandingGone(ctx, sqid, session, from, true)
	if err != nil {
		return err
	}
	if gone {
		return ErrNotYours
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	r, ok := h.rooms[sqid]
	if !ok {
		return ErrUnknownPeer
	}
	p, found := h.findLocked(sqid, session, from)
	if !found {
		return ErrUnknownPeer
	}
	if !p.Admitted {
		return ErrNotAdmitted
	}
	if to == p.Peer {
		return ErrUnknownPeer // a peer is not its own pair
	}
	tp, ok := r.peers[to]
	if !ok || !tp.Admitted {
		return ErrUnknownPeer
	}
	k := keyFor(p.Peer, tp.Peer)
	h.genLocked(r, p.Peer, tp.Peer) // starts the pair at one if it is new
	r.gens[k]++
	gen := r.gens[k]
	for _, pair := range [][2]*Participation{{p, tp}, {tp, p}} {
		m := Message{Type: TypePair, Peer: pair[1].Peer, Gen: gen, To: pair[0].Peer}
		h.ringLocked(pair[0], m)
		h.sendLocked(pair[0], m)
	}
	return nil
}

// ResumeResult is what the stream handler owes a browser that came back
// inside its window: a fresh subscriber, and, when the participation is
// admitted, the roster, the replay and the senders it must reset with.
type ResumeResult struct {
	Sub      *Sub
	Admitted bool
	Roster   *Roster
	Replay   []Message
	Gap      []PeerID
	// ReportToken is the participation's, for the resumed hello (§15.26).
	ReportToken string
}

// ResumeError is a refused resume in the shape spec §4.5's table asks
// for: a status, a reason on the terminal 410s, and the boot id on the
// 404s so a browser can tell a restart from a lapsed window.
type ResumeError struct {
	Status int
	Reason Reason
	Boot   string
}

func (e *ResumeError) Error() string {
	return fmt.Sprintf("hub: resume refused %d %s", e.Status, e.Reason)
}

// resumeRefusalLocked is spec §4.5's refusal ladder, and the only
// place it is written: Resume applies it before attaching, and
// CanResume applies it and stops there.
func (h *Hub) resumeRefusalLocked(sqid, session string, peer PeerID) *ResumeError {
	if _, ok := h.rooms[sqid]; !ok {
		return &ResumeError{Status: 404, Boot: h.boot}
	}
	// A tombstone is answered only through the room and session it was
	// bound to. Why a participation ended is account state, and a peer
	// id is published in every roster, so an ended peer's room-mate must
	// not be able to read the reason off it (spec §4.2's binding rule).
	if t, ok := h.tombstones[peer]; ok && t.room == sqid && t.session == session {
		return &ResumeError{Status: 410, Reason: t.reason}
	}
	p, found := h.findLocked(sqid, session, peer)
	if !found {
		// Unknown here, unknown in another session, unknown in another
		// room: one answer, so nothing leaks (spec §4.2).
		return &ResumeError{Status: 404, Boot: h.boot}
	}
	// The window is thirty seconds, not thirty plus however long it is
	// until the next tick: a resume that arrives after resume_until is
	// refused now. Removing the participation and announcing the
	// departure is still the tick's, so a refusal never has a room's
	// events as a side effect.
	if p.sub == nil && !p.resumeUntil.IsZero() && h.cfg.Now().After(p.resumeUntil) {
		return &ResumeError{Status: 404, Boot: h.boot}
	}
	return nil
}

// CanResume answers what a resume would answer, and does nothing else:
// it does not attach a subscriber, does not clear the resume window and
// does not read or trim the ring. It is the probe behind
// GET /m/{sqid}/stream?peer=&probe=1, which exists because EventSource
// hides the status code of a refused stream from the page.
// It re-derives a member's standing (§15.12) and answers 404 when it has
// gone — the same body an unknown peer gets — but it does not end the
// participation: that is the next mutating request's, or the tick's. A
// probe changes nothing, and a browser that reads 404 here goes on to
// open a fresh stream, which is what a dropped member should get.
func (h *Hub) CanResume(ctx context.Context, sqid, session string, peer PeerID) *ResumeError {
	gone, err := h.memberStandingGone(ctx, sqid, session, peer, false)
	if err != nil {
		// A standing read that failed is not evidence of anything. The
		// participation answers for itself, as it did before §15.12, and
		// the tick will re-read standing in fifteen seconds.
		h.cfg.Logger.Warn("hub: standing lookup failed on a resume probe", "room", sqid, "peer", peer, "error", err)
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if gone {
		return &ResumeError{Status: 404, Boot: h.boot}
	}
	return h.resumeRefusalLocked(sqid, session, peer)
}

// Resume reattaches a stream to a participation in its window. It is
// never a transition into the admitted state: a resume does not consume
// a cap slot and cannot admit anyone (spec §4.3).
//
// A member's standing is re-derived before anything is attached
// (§15.12). The refusal ladder runs first, so a participation that
// ended for a reason worth naming still answers 410 with that reason
// rather than spending a database read; a member whose standing has
// gone since is ended here and answered 404, which is the answer an
// unknown peer gets, byte for byte.
func (h *Hub) Resume(ctx context.Context, sqid, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) {
	h.mu.Lock()
	if re := h.resumeRefusalLocked(sqid, session, peer); re != nil {
		h.mu.Unlock()
		return ResumeResult{}, re
	}
	h.mu.Unlock()

	gone, err := h.memberStandingGone(ctx, sqid, session, peer, true)
	if err != nil {
		return ResumeResult{}, err
	}
	if gone {
		return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot}
	}

	h.mu.Lock()
	defer h.mu.Unlock()
	// The lock was down for the standing read, so the ladder is walked
	// again: the window may have passed, or the meeting ended, while it
	// was out.
	if re := h.resumeRefusalLocked(sqid, session, peer); re != nil {
		return ResumeResult{}, re
	}
	p, _ := h.findLocked(sqid, session, peer)
	r := h.rooms[sqid]
	if p.sub != nil {
		// A second stream for the same peer replaces the attachment only:
		// no tombstone, no departure, the participation carries on.
		p.sub.terminate(Message{Type: TypeLost, Reason: ReasonReplaced})
	}
	p.sub = newSub()
	p.resumeUntil = time.Time{}
	res := ResumeResult{Sub: p.sub, Admitted: p.Admitted, ReportToken: p.reportToken}
	if !p.Admitted {
		// A waiting guest learns nothing by resuming (ruling 9).
		return res, nil
	}
	res.Roster = h.rosterFor(r, p)
	// Replay is per sender: everything above last, unless the ring has
	// dropped something above last, in which case that sender gets a gap
	// and no partial replay (spec §4.5).
	//
	// Everything here is filtered against the roster that has just been
	// built. The ring and the eviction marks outlive the peers that
	// filled them, and the roster is the authority a browser reconciles
	// against (§4.5): it has already closed its connection to a peer
	// that is gone, so it must not be told to reset with it, replay its
	// signalling, or rebuild its pair.
	gapped := map[PeerID]bool{}
	for from, high := range p.evicted {
		if _, present := rosterGen(res.Roster, from); !present {
			continue
		}
		if high > last[from] {
			gapped[from] = true
			res.Gap = append(res.Gap, from)
		}
	}
	for _, m := range p.ring {
		switch {
		case isTargeted(m.Type):
			if _, present := rosterGen(res.Roster, m.From); !present {
				continue
			}
			if !gapped[m.From] && m.Seq > last[m.From] {
				res.Replay = append(res.Replay, m)
			}
		case m.Type == TypePair:
			// The roster already carries the current generation for every
			// admitted peer, so this never fires for a current roster;
			// the branch is kept because the ring is the authority for
			// what was sent, and the browser reconciles from both (§4.6).
			gen, present := rosterGen(res.Roster, m.Peer)
			if present && m.Gen > gen {
				res.Replay = append(res.Replay, m)
			}
		}
	}
	return res, nil
}

// rosterGen is the recipient's generation with one peer, and whether
// that peer is in the roster at all.
func rosterGen(ro *Roster, peer PeerID) (uint64, bool) {
	for _, rp := range ro.Peers {
		if rp.Peer == peer {
			return rp.Gen, true
		}
	}
	return 0, false
}