package main

import (
	"context"
	"errors"
	"sort"
	"sync"
)

// **One composer, one pass at a time, each node once per pass** (issue 1945). A done key's take-in registers its module
// and asks the composer for its nodes; the composer gathers every node asked for while a pass runs and composes them in
// the next pass, in one composition of each node, then writes the grants once. Before, each take-in of a burst composed
// every node of its module itself, at once: they collided on the compositions' revisions ("wrong last sequence … key
// revision mismatch"), retried every 30 s, and the take-in that mattered waited behind them (2026-10-25 at 11:18 and
// 12:32). A caller waits for the pass that began after it asked, and hears that pass's error.
type composer struct {
	composeNodes func(ctx context.Context, nodes []string) map[string]error
	writeGrants  func(ctx context.Context) error

	mu      sync.Mutex
	pending map[string]bool
	grants  bool
	waiters []waiter
	wake    chan struct{}
}

// waiter is a caller waiting for its pass: the nodes it asked for, whether it asked for the grants, and where its answer
// goes.
type waiter struct {
	nodes  []string
	grants bool
	done   chan error
}

func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
	return &composer{composeNodes: compose, writeGrants: grants, pending: map[string]bool{}, wake: make(chan struct{}, 1)}
}

// compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass
// that does it, with the errors of its own nodes alone.
func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
	done := make(chan error, 1)
	c.mu.Lock()
	for _, n := range nodes {
		c.pending[n] = true
	}
	c.grants = c.grants || grants
	c.waiters = append(c.waiters, waiter{nodes: nodes, grants: grants, done: done})
	c.mu.Unlock()
	select {
	case c.wake <- struct{}{}:
	default:
	}
	select {
	case err := <-done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// run makes the passes until ctx ends: each takes everything asked for since the last began.
func (c *composer) run(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-c.wake:
		}
		c.mu.Lock()
		nodes := make([]string, 0, len(c.pending))
		for n := range c.pending {
			nodes = append(nodes, n)
		}
		grants, waiters := c.grants, c.waiters
		c.pending, c.grants, c.waiters = map[string]bool{}, false, nil
		c.mu.Unlock()
		if len(waiters) == 0 {
			continue
		}
		sort.Strings(nodes)
		var failed map[string]error
		if len(nodes) > 0 {
			failed = c.composeNodes(ctx, nodes)
		}
		// The grants whenever a caller asked for them, whatever a node's composition said: a node that failed fails its
		// own callers alone (issue 1945).
		var grantsErr error
		if grants {
			grantsErr = c.writeGrants(ctx)
		}
		for _, w := range waiters {
			var errs []error
			for _, n := range w.nodes {
				if err := failed[n]; err != nil {
					errs = append(errs, err)
				}
			}
			if w.grants && grantsErr != nil {
				errs = append(errs, grantsErr)
			}
			w.done <- errors.Join(errs...)
		}
	}
}