package engines

import (
	"fmt"
	"sort"
	"strings"
)

// Classification, minted groups and the operator clear (spec §5.1 "The
// column", "Platform-minted rows", "Becoming public is never automatic",
// "Classification before serving"; §6.3).

// ---------- minted groups ----------

// followsOwner: the row is a platform-minted row that takes its execution
// owner's mark and barrier. Only minted instance rows do: a minted static or
// service row is 0 by construction (static-minted-public).
func followsOwner(r *row) bool {
	if !r.Platform || !on("minted-follow-owner") {
		return false
	}
	return r.Kind == "instance" || (r.Kind == "static" && !on("static-minted-public"))
}

// resolveOwner is the edge's (and the agent's) owner for a minted row, under
// the rules in force.
func (b *box) resolveOwner(r *row) (string, bool) {
	return b.resolveOwnerWith(r, on("owner-nonempty-unique"))
}

// resolveOwnerWith: the recorded owner, if its row exists; else the
// non-platform instance row with the same account and app and the same
// backend. strict is the spec's fallback: empty values never match, and a
// match that is not unique counts as no match. Without it, empty values
// match and the first match wins.
func (b *box) resolveOwnerWith(r *row, strict bool) (string, bool) {
	if r.Owner != "" {
		if o := b.Rows[r.Owner]; o != nil && !o.Platform {
			return r.Owner, true
		}
		return "", false
	}
	var found []string
	for _, h := range hostsOf(b) {
		c := b.Rows[h]
		if c.Platform || c.Kind != "instance" || c.App != r.App {
			continue
		}
		cs, ca, rs, ra := socketOf(c), addrOf(c), socketOf(r), addrOf(r)
		if strict && ((cs != "" && cs == rs) || (ca != "" && ca == ra)) ||
			!strict && (cs == rs || ca == ra) {
			found = append(found, h)
		}
	}
	if len(found) == 1 || (!strict && len(found) > 0) {
		return found[0], true
	}
	return "", false
}

// socketOf and addrOf split a backend into the registry's Socket and Addr.
func socketOf(r *row) string {
	if strings.HasPrefix(r.Backend, "/") {
		return r.Backend
	}
	return ""
}

func addrOf(r *row) string {
	if r.Backend != "" && !strings.HasPrefix(r.Backend, "/") {
		return r.Backend
	}
	return ""
}

func hostsOf(b *box) []string {
	hosts := make([]string, 0, len(b.Rows))
	for h := range b.Rows {
		hosts = append(hosts, h)
	}
	sort.Strings(hosts)
	return hosts
}

// truthGroup is the group a row really serves: a minted row's source, else
// its own host. The oracle reads it; the protocol never does.
func truthGroup(b *box, host string) string {
	if r := b.Rows[host]; r != nil && r.Platform && r.Source != "" {
		return r.Source
	}
	return host
}

// mint is the domains reconciler minting a platform row from src: it copies
// the source's kind, backend and app (no version), and, being a row of the
// same group, the group's mark. recordOwner is whether it writes
// routes.owner (rows minted before the column was added have none until the
// reconciler's next pass).
func (w *world) mint(b *box, src, host string, recordOwner bool) {
	s := b.Rows[src]
	m := w.groupMark(b, w.group(b, src))
	r := &row{Host: host, Kind: s.Kind, App: s.App, Backend: s.Backend, Platform: true, Source: src, PathTo: map[string]string{}}
	if recordOwner {
		r.Owner = src
	}
	b.Rows[host] = r
	b.Mark[host] = m
	b.Wrote++
	w.logf("box%d: registry change %d: minted %s %s from %s (owner %q, mark %v)", b.ID, b.Wrote, r.Kind, host, src, r.Owner, m)
}

// reconcileOwners is the reconciler's backfill pass: it records the owner of
// every minted instance row that has none. The row takes its owner group's
// mark (one group, one mark). It reports whether it recorded any.
func (w *world) reconcileOwners(b *box) bool {
	did := false
	for _, h := range hostsOf(b) {
		r := b.Rows[h]
		if !r.Platform || r.Kind != "instance" || r.Owner != "" || b.Rows[r.Source] == nil {
			continue
		}
		m := w.groupMark(b, w.group(b, r.Source))
		r.Owner = r.Source
		b.Mark[h] = m
		b.Wrote++
		did = true
		w.logf("box%d: registry change %d: owner of %s recorded as %s (mark %v)", b.ID, b.Wrote, h, r.Owner, m)
	}
	return did
}

// provision creates instance route host (a new route: NULL, unclassified),
// then its first adoption classifies it before it is served: a declaring
// build through the gate, a readable plain one as 0 (spec §5.1, "A route the
// agent creates later ... is classified by its first adoption").
func (w *world) provision(b *box, host, app, backend string, ver int) bool {
	b.Rows[host] = &row{Host: host, Kind: "instance", App: app, Backend: backend, PathTo: map[string]string{}}
	b.Wrote++
	w.logf("box%d: registry change %d: provisioned %s (%s, %s), NULL until its first adoption", b.ID, b.Wrote, host, app, backend)
	ok := w.adopt(b, host, ver, nil)
	if w.groupMark(b, host) == markNull {
		if m := w.classify(b, host); m != markNull {
			w.writeMark(b, host, m)
		}
	}
	return ok
}

// addStaticApp creates static app name with its minted platform row and a
// static canary on its own channel (domains/reconcile.go:368).
func (w *world) addStaticApp(b *box, name string) (minted, canary string) {
	b.Rows[name] = &row{Host: name, Kind: "static", App: "acct/" + name, PathTo: map[string]string{}}
	b.Mark[name] = markPublic // 0 by construction
	b.Wrote++
	w.logf("box%d: registry change %d: static app %s", b.ID, b.Wrote, name)
	minted, canary = name+".acct.apps.example", "canary-"+name+".acct.apps.example"
	w.mint(b, name, minted, true)
	w.mint(b, name, canary, true)
	return minted, canary
}

// ---------- grants ----------

// groupHasGrant: a row of group g holds the doc-engine grant. With
// grant-marks-group, any row of the group; without it, only the owner's own
// row (a grant on a minted row is ignored, R6-1).
func (w *world) groupHasGrant(b *box, g string) bool {
	if !on("grant-marks-group") {
		r := b.Rows[g]
		return r != nil && r.Grant
	}
	for _, h := range w.groupHosts(b, g) {
		if b.Rows[h].Grant {
			return true
		}
	}
	return false
}

// grantLoaded is the oracle's grant: a row whose true group is g holds the
// grant and the edge has loaded the change that recorded it. It returns that
// row, or "".
func (w *world) grantLoaded(b *box, g string) string {
	for _, h := range hostsOf(b) {
		r := b.Rows[h]
		if r.Grant && truthGroup(b, h) == g && b.Fenced >= r.GrantAt {
			return h
		}
	}
	return ""
}

// grant is `carlos route --host <host> --grant doc-engine`: it records the
// grant, writes 1 on the group if the group now holds it, and returns once
// the edge has loaded the change (or says it timed out). On a minted row
// whose owner cannot be resolved it is refused: the CLI cannot find the
// group to mark, and the row itself stays NULL and refused.
func (w *world) grant(b *box, host string) bool {
	r := b.Rows[host]
	if r == nil || r.Kind != "instance" || r.Grant {
		return false
	}
	if followsOwner(r) {
		if _, ok := b.resolveOwner(r); !ok {
			w.logf("box%d: --grant doc-engine on %s: refused: the row's owner is not recorded and cannot be found uniquely; the row stays NULL and refused", b.ID, host)
			return false
		}
	}
	b.Wrote++
	r.Grant, r.GrantAt = true, b.Wrote
	g := w.group(b, host)
	w.logf("box%d: registry change %d: --grant doc-engine on %s (group %s)", b.ID, b.Wrote, host, g)
	if w.groupHasGrant(b, g) && w.groupMark(b, g) != markInternal {
		w.writeMark(b, g, markInternal)
	}
	want := b.Wrote
	for i := 0; i < fenceWaitTicks && b.Fenced < want; i++ {
		w.reload(b)
	}
	if b.Fenced < want {
		w.logf("box%d: --grant on %s: the edge has not loaded it; timed out", b.ID, host)
		return false
	}
	return true
}

// revoke is --revoke doc-engine on any row of a group: the grant goes and
// the group's mark stays (becoming public is never automatic; only --public
// writes 0). Like --grant it runs the barrier for the group: it returns once
// the edge has loaded the change, so the group's mark is fenced, or says it
// timed out. Refused on a minted row whose owner cannot be resolved.
func (w *world) revoke(b *box, host string) bool {
	r := b.Rows[host]
	if r == nil || !r.Grant {
		return false
	}
	if followsOwner(r) {
		if _, ok := b.resolveOwner(r); !ok {
			w.logf("box%d: --revoke doc-engine on %s: refused: the row's owner cannot be resolved", b.ID, host)
			return false
		}
	}
	g := w.group(b, host)
	b.Wrote++
	r.Grant = false
	want := b.Wrote
	w.logf("box%d: registry change %d: --revoke doc-engine on %s (group %s); the mark stays %v", b.ID, b.Wrote, host, g, w.groupMark(b, g))
	for i := 0; i < fenceWaitTicks && b.Fenced < want; i++ {
		w.reload(b)
	}
	if b.Fenced < want {
		w.logf("box%d: --revoke on %s: the edge has not loaded it; timed out", b.ID, host)
		return false
	}
	return true
}

// ---------- classification before serving ----------

// classify is the upgrade's classification of group g: 1 if a row of the
// group holds the grant, the live manifest declares, or the manifest of any
// other version of the app staged on the box declares, whichever instance of
// the app staged it (classify-staged); 0 if the live manifest is readable and
// none of those holds; else NULL.
func (w *world) classify(b *box, g string) mark {
	if w.groupHasGrant(b, g) {
		return markInternal
	}
	live, liveOK := w.builds[b.Live[g]]
	liveOK = liveOK && b.Live[g] != 0 && live.ManifestOK
	if liveOK && live.Declares {
		return markInternal
	}
	if on("classify-staged") {
		for _, v := range w.stagedForApp(b, g) {
			if bd := w.builds[v]; bd.ManifestOK && bd.Declares {
				return markInternal
			}
		}
	}
	if liveOK {
		return markPublic
	}
	return markNull
}

// classifiedGroups is every group with a row the classification covers: an
// instance row, or a row that follows an owner.
func (w *world) classifiedGroups(b *box) []string {
	seen := map[string]bool{}
	var out []string
	for _, h := range hostsOf(b) {
		r := b.Rows[h]
		if r.Kind != "instance" && !followsOwner(r) {
			continue
		}
		if g := w.group(b, h); !seen[g] {
			seen[g] = true
			out = append(out, g)
		}
	}
	return out
}

// upgrade is the agent's first start on the build that adds routes.internal.
// The column arrives NULL on every instance route; before any edge listener
// opens, the agent classifies every NULL group. The edge then starts on the
// classified map: in-flight requests did not survive the restart, and the
// fence covers everything written. Running processes are untouched.
func (w *world) upgrade(b *box) {
	if w.unwrittenManifestAcrossUpgrade {
		if why, procs := w.upgradeResidual(b); why != "" {
			for _, p := range procs {
				p.Residual = "unwrittenManifestAcrossUpgrade"
			}
			w.logf("box%d: upgrade: the stated residual (%s); its processes' breaches are counted as residual", b.ID, why)
		}
	}
	gs := w.classifiedGroups(b)
	w.logf("box%d: upgrade: routes.internal arrives NULL on %v", b.ID, gs)
	for _, g := range gs {
		w.writeMark(b, g, markNull)
	}
	for _, g := range gs {
		if m := w.classify(b, g); m != markNull {
			w.writeMark(b, g, m)
		} else {
			w.logf("box%d: upgrade: %s stays NULL (live manifest unreadable): refused publicly, named every pass", b.ID, g)
		}
	}
	b.loadEdge()
	b.Epoch++
	if len(b.Admitted) > 0 {
		w.logf("box%d: upgrade: %d in-flight requests ended with the restart", b.ID, len(b.Admitted))
	}
	b.Admitted = nil
	b.Fenced = b.Wrote
	// the restart loaded every group's mark and drained the whole box: a
	// fence for every closed group, NULL included
	for _, g := range gs {
		if w.closed(b, g) && !w.recorded(b, g) {
			b.SetAt[g] = b.Wrote
		}
	}
	w.logf("box%d: upgrade: listeners open on the classified map, epoch %d, fenced %d", b.ID, b.Epoch, b.Fenced)
}

// residualAtUpgrade names the spec's stated residual if an upgrade of b now
// would produce it: a running declaring build whose manifest cannot be read,
// in a group the (correct) classification would mark 0 because the live
// manifest reads and does not declare. The default world does not generate
// it; Task 5's flag (unwrittenManifestAcrossUpgrade) does.
func (w *world) residualAtUpgrade(b *box) string {
	why, _ := w.upgradeResidual(b)
	return why
}

// upgradeResidual is residualAtUpgrade with the processes it names: every
// running unreadable declaring process of each group it applies to.
func (w *world) upgradeResidual(b *box) (string, []*proc) {
	first := ""
	var procs []*proc
	for _, g := range w.groups(b) {
		var unreadable *proc
		var group []*proc
		for _, p := range b.Procs {
			if bd := w.builds[p.Ver]; p.Group == g && bd.Declares && !bd.ManifestOK {
				unreadable = p
				group = append(group, p)
			}
		}
		if unreadable == nil {
			continue
		}
		granted := false
		for _, h := range hostsOf(b) {
			granted = granted || (b.Rows[h].Grant && truthGroup(b, h) == g)
		}
		live, ok := w.builds[b.Live[g]]
		if granted || b.Live[g] == 0 || !ok || !live.ManifestOK || live.Declares {
			continue
		}
		staged := false
		for _, v := range w.stagedForApp(b, g) {
			if bd := w.builds[v]; bd.ManifestOK && bd.Declares {
				staged = true
			}
		}
		if !staged {
			if first == "" {
				first = fmt.Sprintf("%s runs v%d (declares, manifest unreadable) while live v%d reads as plain", g, unreadable.Ver, b.Live[g])
			}
			procs = append(procs, group...)
		}
	}
	return first, procs
}

// ---------- the operator clear ----------

// operatorPublic is `carlos route --host <host> --public` on group g (spec
// §5.1, "Becoming public is never automatic"). Holding the group's lock for
// its whole run (adoption's repoint honours it, public-holds-lock; so does
// every launch through the gate, clear-launch-lock):
//  1. refuse unless the live manifest is readable and does not declare
//     (public-checks-live), and no row of the group holds the grant;
//  2. stop the whole group (public-stops-group);
//  3. refuse unless every cgroup of the group is empty (public-confirms-empty);
//  4. write 0 and release the lock.
//
// Step 3 is what catches a stop that did not complete (w.clearStopMisses).
//
// w.clearHook, if set, runs after step 1 ("locked"), after step 2
// ("stopped") and after step 3 ("confirmed").
func (w *world) operatorPublic(b *box, g string) bool {
	w.logf("box%d: --public %s: taking the group's lock", b.ID, g)
	b.Locked[g] = true
	defer delete(b.Locked, g)
	if on("public-checks-live") {
		live, ok := w.builds[b.Live[g]]
		if b.Live[g] == 0 || !ok || !live.ManifestOK || live.Declares {
			w.logf("box%d: --public %s: refused: the live release v%d is unreadable or declares doc_engine", b.ID, g, b.Live[g])
			return false
		}
	}
	if w.groupHasGrant(b, g) {
		w.logf("box%d: --public %s: refused: a row of the group holds the doc-engine grant", b.ID, g)
		return false
	}
	w.hook(b, g, "locked")
	if on("public-stops-group") {
		w.clearStop(b, g)
	}
	w.hook(b, g, "stopped")
	if on("public-confirms-empty") && w.running(b, g) {
		w.logf("box%d: --public %s: refused: a cgroup of the group is not empty", b.ID, g)
		return false
	}
	w.hook(b, g, "confirmed")
	w.writeMark(b, g, markPublic)
	w.logf("box%d: --public %s: 0 written; the lock is released", b.ID, g)
	return true
}

// clearStop is step 2: stop every process of group g, except, when
// w.clearStopMisses is set, the first one, whose stop did not complete.
func (w *world) clearStop(b *box, g string) {
	var left *proc
	if w.clearStopMisses {
		for _, p := range b.Procs {
			if p.Group == g {
				left = p
				break
			}
		}
	}
	w.stopGroup(b, g)
	if left != nil {
		b.Procs = append(b.Procs, left)
		w.logf("box%d: --public %s: the stop did not complete: process %d (v%d) still runs", b.ID, g, left.ID, left.Ver)
		return
	}
	w.logf("box%d: --public %s: the group is stopped", b.ID, g)
}

func (w *world) hook(b *box, g, stage string) {
	if w.clearHook != nil {
		w.clearHook(b, g, stage)
	}
}

// ---------- the random walk ----------

// pickInstanceRow is any instance row of b, minted included, or "".
func (w *world) pickInstanceRow(b *box) string {
	var hs []string
	for _, h := range hostsOf(b) {
		if b.Rows[h].Kind == "instance" {
			hs = append(hs, h)
		}
	}
	if len(hs) == 0 {
		return ""
	}
	return hs[w.rng.Intn(len(hs))]
}

// maxRows bounds a box's table in the walk.
const maxRows = 24

func init() {
	actions = append(actions,
		action{name: "upgrade", weight: 1, run: func(w *world) bool {
			b := w.pickBox()
			if why := w.residualAtUpgrade(b); why != "" && !w.unwrittenManifestAcrossUpgrade {
				w.logf("box%d: upgrade not generated: the stated residual (%s)", b.ID, why)
				return false
			}
			w.upgrade(b)
			return true
		}},
		action{name: "grant", weight: 1, run: func(w *world) bool {
			b := w.pickBox()
			h := w.pickInstanceRow(b)
			if h == "" || b.Rows[h].Grant {
				return false
			}
			w.grant(b, h)
			return true
		}},
		action{name: "revoke", weight: 1, run: func(w *world) bool {
			b := w.pickBox()
			h := w.pickInstanceRow(b)
			if h == "" || !b.Rows[h].Grant {
				return false
			}
			w.revoke(b, h)
			return true
		}},
		action{name: "operator-public", weight: 2, run: func(w *world) bool {
			b := w.pickBox()
			g := w.pickGroup(b)
			if g == "" {
				return false
			}
			switch w.rng.Intn(5) {
			case 1: // an adoption tries to run during the clear
				w.clearHook = func(hb *box, hg, stage string) {
					if stage == "locked" {
						w.adopt(hb, hg, w.pickVer(), nil)
					}
				}
			case 2: // an awake repoint tries to run during the clear
				w.clearHook = func(hb *box, hg, stage string) {
					if stage == "locked" {
						w.repointAwake(hb, hg, w.pickVer())
					}
				}
			case 3: // a swap's incoming generation races the stop
				w.clearHook = func(hb *box, hg, stage string) {
					if stage == "stopped" {
						w.launch(hb, hg, w.pickVer(), "swap")
					}
				}
			case 4: // a launch races the write of 0, after the empty check
				w.clearHook = func(hb *box, hg, stage string) {
					if stage == "confirmed" {
						w.launch(hb, hg, w.pickVer(), []string{"cmd", "recovery", "seed"}[w.rng.Intn(3)])
					}
				}
			}
			w.clearStopMisses = w.rng.Intn(4) == 0
			w.operatorPublic(b, g)
			w.clearHook, w.clearStopMisses = nil, false
			return true
		}},
		action{name: "adopt-awake", weight: 2, run: func(w *world) bool {
			b := w.pickBox()
			g := w.pickGroup(b)
			if g == "" {
				return false
			}
			w.repointAwake(b, g, w.pickVer())
			return true
		}},
		action{name: "add-tcp-instance", weight: 1, run: func(w *world) bool {
			b := w.pickBox()
			if len(b.Rows) >= maxRows {
				return false
			}
			w.nextID++
			host := fmt.Sprintf("t%d", w.nextID)
			app := []string{"acct/shop", "acct/tcp"}[w.rng.Intn(2)]
			addr := fmt.Sprintf("10.0.%d.1:%d", b.ID, 9100+w.nextID) // unique: no ambiguous owner in the walk
			w.provision(b, host, app, addr, w.pickVer())
			w.mint(b, host, host+".acct.apps.example", w.rng.Intn(2) == 0)
			return true
		}},
		action{name: "add-static-app", weight: 1, run: func(w *world) bool {
			b := w.pickBox()
			if len(b.Rows) >= maxRows {
				return false
			}
			w.nextID++
			w.addStaticApp(b, fmt.Sprintf("s%d", w.nextID))
			return true
		}},
		action{name: "reconcile-owners", weight: 1, run: func(w *world) bool {
			return w.reconcileOwners(w.pickBox())
		}},
	)
}