package delivery

import (
	"fmt"
	"maps"
	"slices"
	"sync"
	"time"
)

// Placed is where the placement (the seat mesh-placement's state `placement`, held by the assigner; ADR 0567, ADR 0549 §3) places each module, and since when:
// the one reader of that state for every part that asks "does module stand on node" (issue 774, design 47's third
// milestone), in place of each part reading the controller's assignments (`mesh-controller_assignments`, which goes
// with the controller's lease, ADR 0549 §6).
//
// Its input is the state's revisions in the order the bus wrote them, as a watch with the history (jetstream's
// IncludeHistory) hands them: Put takes each. A module stands since the revision of its node's key that first named it
// in an unbroken run. A module in the first revision heard of a key stands since the zero time: that revision is the
// key's first (the seed of ADR 0549 §4 placed what already ran there) or the oldest the bus still keeps
// (PlacementHistory), and either way the module may have run there before it. A key's value that does not read places
// nothing on its node, and Put says why.
//
// Safe for use from a watch and from a reader at once.
type Placed struct {
	mu    sync.Mutex
	since map[string]map[string]time.Time // node -> module -> since when it stands
	heard map[string]bool                 // the nodes whose key was heard at all
}

// NewPlaced is a Placed that holds no placement yet.
func NewPlaced() *Placed {
	return &Placed{since: map[string]map[string]time.Time{}, heard: map[string]bool{}}
}

// Put takes one revision of a key of the state `placement` (its value, whether it is a delete, and when the bus wrote
// it). It answers the node the key is of and the modules whose standing on it changed (placed or removed), in name
// order, and an error for a key or a value that does not read (the node then places nothing).
func (p *Placed) Put(key string, value []byte, deleted bool, at time.Time) (node string, changed []string, err error) {
	node, err = ParsePlacementKey(key)
	if err != nil {
		return "", nil, err
	}
	now := map[string]bool{}
	if !deleted {
		pl, derr := DecodePlacement(value)
		switch {
		case derr != nil:
			err = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr)
		case pl.Node != node:
			err = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, pl.Node)
		default:
			for _, m := range pl.Assigned() {
				now[m] = true
			}
		}
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if !p.heard[node] {
		p.heard[node] = true
		at = time.Time{}
	}
	was := p.since[node]
	next := map[string]time.Time{}
	for m := range now {
		if s, ok := was[m]; ok {
			next[m] = s
		} else {
			next[m] = at
			changed = append(changed, m)
		}
	}
	for m := range was {
		if !now[m] {
			changed = append(changed, m)
		}
	}
	if len(next) == 0 {
		delete(p.since, node)
	} else {
		p.since[node] = next
	}
	slices.Sort(changed)
	return node, changed, err
}

// Seeded says some node's placement is held: before the assigner's adopt writes the first placement nothing is placed,
// and a reader keeps what it did before (ADR 0549 §5).
func (p *Placed) Seeded() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	return len(p.since) > 0
}

// Since says whether module stands on node, and since when (the zero time: since before the revisions kept).
func (p *Placed) Since(node, module string) (time.Time, bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	s, ok := p.since[node][module]
	return s, ok
}

// PlacedNodes is every node module is placed on, in name order.
func (p *Placed) PlacedNodes(module string) []string {
	p.mu.Lock()
	defer p.mu.Unlock()
	var out []string
	for node, mods := range p.since {
		if _, ok := mods[module]; ok {
			out = append(out, node)
		}
	}
	slices.Sort(out)
	return out
}

// ByNode is, by node, the modules placed on it in name order.
func (p *Placed) ByNode() map[string][]string {
	p.mu.Lock()
	defer p.mu.Unlock()
	out := make(map[string][]string, len(p.since))
	for node, mods := range p.since {
		out[node] = slices.Sorted(maps.Keys(mods))
	}
	return out
}