package engines

import "sort"

// The edge side of the internal mark (spec §5.1 "Setting the mark: the
// serving barrier", "What the edge does with it"; §5.9 box-wide fence; §6.3).
//
// The registry holds the mark (the intent); the edge serves the map its last
// successful reload swapped in, and publishes the registry change it has
// fenced. A launch trusts only the fence.

// loadEdge swaps the registry into the edge's map: rows by value (a later
// registry edit does not leak into the served map) and their marks.
//
// The edge resolves owners at load (spec §5.1, "Platform-minted rows"): a
// row that follows an owner carries the resolved owner in its copy's Owner;
// one whose owner cannot be found, or is found ambiguously, is served NULL
// (refused) until the reconciler records its owner. Any other row's copy has
// no Owner, so edgeGroup is its own host.
func (b *box) loadEdge() {
	b.EdgeRows = make(map[string]row, len(b.Rows))
	b.Serving = make(map[string]mark, len(b.Rows))
	for h, r := range b.Rows {
		c := *r
		c.PathTo = make(map[string]string, len(r.PathTo))
		for p, to := range r.PathTo {
			c.PathTo[p] = to
		}
		b.Serving[h] = b.Mark[h]
		c.Owner = ""
		if followsOwner(r) {
			if o, ok := b.resolveOwner(r); ok {
				c.Owner = o
			} else {
				b.Serving[h] = markNull
			}
		}
		b.EdgeRows[h] = c
	}
}

// groupHosts is every registry row of group g, sorted.
func (w *world) groupHosts(b *box, g string) []string {
	var out []string
	for h := range b.Rows {
		if w.group(b, h) == g {
			out = append(out, h)
		}
	}
	sort.Strings(out)
	return out
}

// groupMark is the registry's mark of group g: internal only if every row of
// the group is, NULL if any row is NULL or the group has no row, else public.
func (w *world) groupMark(b *box, g string) mark {
	hosts := w.groupHosts(b, g)
	if len(hosts) == 0 {
		return markNull
	}
	all := markInternal
	for _, h := range hosts {
		switch b.Mark[h] {
		case markNull:
			return markNull
		case markPublic:
			all = markPublic
		}
	}
	return all
}

// writeMark writes mark m on every registry row of group g: one registry
// change, so Wrote counts it. A change that closes a group the edge has not
// fenced closed (an open group, or a new row's NULL the edge has never
// loaded) is the one a fence must reach: it records the group's fence
// generation (SetAt). 1 written over a NULL whose generation is recorded
// moves nothing: the edge already refuses it.
func (w *world) writeMark(b *box, g string, m mark) {
	wasClosed := w.closed(b, g) && (w.recorded(b, g) || !on("new-group-fenced"))
	b.Wrote++
	for _, h := range w.groupHosts(b, g) {
		b.Mark[h] = m
	}
	if !wasClosed && m != markPublic {
		b.SetAt[g] = b.Wrote
	}
	w.logf("box%d: registry change %d: %s marked %v", b.ID, b.Wrote, g, m)
	if m != markPublic {
		w.dropListeners(b, g) // before any fence of this write (ports-excluded)
	}
}

// recorded: group g has a fence generation, the registry change at which it
// last closed (SetAt has the key). A group whose rows are new, or were
// created NULL and never written, has none.
func (w *world) recorded(b *box, g string) bool {
	_, ok := b.SetAt[g]
	return ok
}

// fenced is the gate's reading of I1's "fenced": the group is closed, it has
// a fence generation, and the edge's fenced counter has reached it (no later
// change has cleared it).
//
// NULL counts as closed here only once the edge has loaded that NULL: the
// group has a recorded generation, set by the change that closed it, and the
// reload that reached it bumped the epoch and drained every older public
// request on the box (the upgrade's restart is such a fence). A newly created
// or unseen row is NULL in the registry but has no generation, so a first
// declaring launch writes 1 and waits for the fence like any other
// (new-group-fenced; final review, finding 1).
func (w *world) fenced(b *box, g string) bool {
	return (w.recorded(b, g) || !on("new-group-fenced")) && w.closed(b, g) && b.Fenced >= b.SetAt[g]
}

// fencedTruth is the oracle's fenced: fenced with every rule in place. I1
// reads it, so a mutation that weakens the gate's reading cannot weaken the
// invariant too.
func (w *world) fencedTruth(b *box, g string) bool {
	return w.recorded(b, g) && w.closed(b, g) && b.Fenced >= b.SetAt[g]
}

// closed: group g has a row and none of its registry rows is 0.
func (w *world) closed(b *box, g string) bool {
	hosts := w.groupHosts(b, g)
	if len(hosts) == 0 {
		return false
	}
	for _, h := range hosts {
		if b.Mark[h] == markPublic {
			return false
		}
	}
	return true
}

// reload is the edge's registry poll. A failed reload swaps nothing and
// advances nothing. A successful one swaps the map; if any row went from not
// internal to internal it bumps the box's admission epoch and cancels every
// public request admitted under an older one (including those waiting in the
// activator's queue); then the fenced counter reaches the registry.
func (w *world) reload(b *box) bool {
	if b.ReloadFails {
		if !on("fence-needs-reload") {
			b.Fenced = b.Wrote
			w.logf("box%d: reload failed; fenced %d anyway", b.ID, b.Fenced)
			return false
		}
		w.logf("box%d: reload failed; nothing swapped, fenced stays %d", b.ID, b.Fenced)
		return false
	}
	old := b.Serving
	b.loadEdge()
	changed := map[string]bool{}
	for h, m := range b.Serving {
		if (m == markInternal && old[h] != markInternal) || (m == markNull && old[h] == markPublic) {
			changed[h] = true
		}
	}
	if len(changed) > 0 {
		b.Epoch++
		if on("fence-drains-admitted") {
			kept := b.Admitted[:0]
			for _, r := range b.Admitted {
				// box-wide: every older public request on any row, deleted
				// rows included (spec §5.9); without it, only the rows the
				// change marked
				if r.Public && r.Epoch < b.Epoch && (on("fence-box-wide") || changed[r.Row]) {
					w.logf("box%d: fence cancels request %d (row %s, epoch %d)", b.ID, r.ID, r.Row, r.Epoch)
					continue
				}
				kept = append(kept, r)
			}
			b.Admitted = kept
		}
	}
	b.Fenced = b.Wrote
	w.logf("box%d: reload: map swapped, epoch %d, fenced %d", b.ID, b.Epoch, b.Fenced)
	return true
}

// admitPublic is the edge's internal check for a request with no caller: it
// resolves the alias, the path map and the minted row to the final row, and
// refuses when that row is internal or NULL (spec §5.1: the check runs on the
// final resolved upstream's route). A static row answers without a proxy, so
// nothing is admitted. Otherwise the request is admitted under the box's
// epoch, holding the final row's backend.
func (w *world) admitPublic(b *box, host, path string) *req {
	r, _ := w.admit(b, host, path)
	return r
}

// The outcomes of admit.
const (
	admitted = "admitted" // proxied: the request holds a backend
	answered = "answered" // a static row: the edge answers it itself
	refused  = "refused"  // the internal refusal
	unknown  = "unknown"  // no such host, alias target or path target
)

// admit is admitPublic with its outcome, for INV-AV. The edge treats NULL
// like 1 (classify-null-refused).
func (w *world) admit(b *box, host, path string) (*req, string) {
	asked, ok := b.EdgeRows[host]
	if !ok {
		w.logf("box%d: public %s%s: unknown host", b.ID, host, path)
		return nil, unknown
	}
	final := asked
	if asked.Alias != "" {
		if final, ok = b.EdgeRows[asked.Alias]; !ok {
			w.logf("box%d: public %s%s: alias target %s unknown", b.ID, host, path, asked.Alias)
			return nil, unknown
		}
	}
	if to, ok := final.PathTo[path]; ok {
		if final, ok = b.EdgeRows[to]; !ok {
			w.logf("box%d: public %s%s: path target %s unknown", b.ID, host, path, to)
			return nil, unknown
		}
	}
	checked := final
	if !on("mark-on-final-upstream") {
		checked = asked
	}
	m := b.Serving[checked.Host]
	if m == markNull && !on("classify-null-refused") {
		w.logf("box%d: public %s%s: %s is NULL, served like 0", b.ID, host, path, checked.Host)
		m = markPublic
	}
	if m != markPublic {
		w.logf("box%d: public %s%s: internal refusal (%s is %v)", b.ID, host, path, checked.Host, m)
		return nil, refused
	}
	if final.Backend == "" {
		w.logf("box%d: public %s%s: answered by the edge (%s)", b.ID, host, path, final.Host)
		return nil, answered
	}
	if why := w.dedicatedRefusal(b, final); why != "" {
		w.logf("box%d: public %s%s: internal refusal: %s", b.ID, host, path, why)
		return nil, refused
	}
	w.nextID++
	r := &req{ID: w.nextID, Epoch: b.Epoch, Row: checked.Host, Backend: final.Backend, Public: true, Box: b.ID}
	b.Admitted = append(b.Admitted, r)
	w.logf("box%d: public %s%s admitted as request %d (row %s, backend %s, epoch %d)", b.ID, host, path, r.ID, r.Row, r.Backend, r.Epoch)
	return r, admitted
}

// backendGroup is the group whose process listens on backend: the group of
// the instance row that carries it, whatever row the request came in by. A
// minted row copies its source's backend; the process is the source's, so
// the source's own row decides (the truth, whatever the edge resolved).
func (w *world) backendGroup(b *box, backend string) string {
	hosts := make([]string, 0, len(b.Rows))
	for h := range b.Rows {
		hosts = append(hosts, h)
	}
	sort.Strings(hosts)
	for _, h := range hosts {
		if r := b.Rows[h]; r.Kind == "instance" && !r.Platform && r.Backend == backend {
			return w.group(b, h)
		}
	}
	return ""
}

// deliver resumes an admitted request: unless the fence cancelled it, its
// proxy reaches the newest process of the group listening on its backend.
// INV-P: a public request never reaches a process whose build declares
// doc_engine, nor one of a group that holds the doc-engine grant once the
// edge has loaded the grant (the grant sets 1 whatever the manifest says,
// spec §5.1 "The column"; a grant whose barrier timed out protects nothing
// until the edge loads it, and the load drains every request admitted before).
func (w *world) deliver(r *req) {
	b := w.boxes[r.Box]
	idx := -1
	for i, x := range b.Admitted {
		if x == r {
			idx = i
		}
	}
	if idx < 0 {
		w.logf("box%d: request %d was cancelled; nothing delivered", b.ID, r.ID)
		return
	}
	b.Admitted = append(b.Admitted[:idx], b.Admitted[idx+1:]...)
	r.Held = false
	g := w.backendGroup(b, r.Backend)
	if g == "" && w.deliverDirect(b, r) {
		return
	}
	var p *proc
	for _, x := range b.Procs {
		if x.Group == g && g != "" {
			p = x
		}
	}
	if p == nil {
		w.logf("box%d: request %d: no process on %s; wake failure answer", b.ID, r.ID, r.Backend)
		return
	}
	r.Reached = p.ID
	w.events["delivered"]++
	w.logf("box%d: request %d delivered to process %d (%s v%d)", b.ID, r.ID, p.ID, p.Group, p.Ver)
	if r.Public && w.builds[p.Ver].Declares {
		w.failProc(b, p, "INV-P", true, "box%d: public request %d (row %s) delivered to process %d of %s running v%d, which declares doc_engine", b.ID, r.ID, r.Row, p.ID, p.Group, p.Ver)
	} else if r.Public && w.grantLoaded(b, g) != "" {
		w.failProc(b, p, "INV-P", true, "box%d: public request %d (row %s) delivered to process %d of %s running v%d, whose group holds the doc-engine grant (on %s)", b.ID, r.ID, r.Row, p.ID, p.Group, p.Ver, w.grantLoaded(b, g))
	}
	if p.Ver == b.Live[g] {
		b.Good[g] = p.Ver // the live build answered: it is the good version now
	}
}