package inventory

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/chain"
)

// **The move** (novox/hq ADR 0408 §8, issue 981 note 7 Q2): the serving controller writes the rows of a kind into its
// bucket at its start, holding the lease and after the buckets exist, and records the kind in chain_moved. Idempotent: a
// key that exists is already moved and is never overwritten, so a bucket value written since is not laid over by the row
// it came from. It counts the rows against the keys, logs the largest value, and **refuses** (a Refusal, which the
// controller raises as an urgent condition) on a count that differs or a value over the bus's largest, with nothing
// switched: a kind is recorded moved only once all of it is in the bucket.

// Refusal is a move that stopped without switching a kind, and why. It is permanent until the cause is put right; any
// other error of a move (the bus away, a write not granted) is tried again.
type Refusal struct {
	Kind, Why string
}

func (r *Refusal) Error() string {
	return fmt.Sprintf("the move of %s to the bus is refused: %s", r.Kind, r.Why)
}

// MoveReport is what a move of one kind did.
type MoveReport struct {
	Kind string
	// Rows is how many records the tables held, Written how many keys this move made, and Present how many already
	// held a value (moved by an earlier start).
	Rows, Written, Present int
	// TableRows is how many rows of the tables those records came from: a request and its outcome are two rows and one
	// record.
	TableRows int
	// Largest is the biggest value, in bytes.
	Largest int
	// Already is a kind an earlier start moved: nothing was done.
	Already bool
}

// mover is one kind of the chain: its rows, read from its tables, as the keys and values they become.
type mover struct {
	kind string
	// rows reads the tables: each record's key and its value, whole, and how many rows of the tables they came from.
	rows func(ctx context.Context) ([]moveRow, int, error)
	// create writes one record under its key; chain.ErrExists when the key holds a value.
	create func(ctx context.Context, c *chain.Chain, row moveRow) error
	// keys are the keys the bucket holds.
	keys func(ctx context.Context, c *chain.Chain) ([]string, error)
	// count is how many rows the tables hold now, counted as rows reports them.
	count func(ctx context.Context, tx pgx.Tx) (int, error)
	// lock keeps the tables from taking a row while the switch is recorded.
	lock string
	// settle runs under that lock, with the rows read again: it makes the bucket hold what the rows say now. A row the old
	// controller changed after the first pass is in the bucket before the kind is recorded as moved, for before the switch
	// the table is the only truth.
	settle func(ctx context.Context, c *chain.Chain, rows []moveRow) error
}

type moveRow struct {
	key   string
	value any
	size  int
}

func (i *Inventory) movers() []mover {
	return []mover{{
		kind: KindBuilds,
		rows: func(ctx context.Context) ([]moveRow, int, error) {
			builds, err := i.allBuildRows(ctx)
			if err != nil {
				return nil, 0, err
			}
			requests, err := i.allRequestRows(ctx)
			if err != nil {
				return nil, 0, err
			}
			// A build request is the build's record in state asked, and the request stays beside the outcome that
			// answered it: one record under one id (issue 1091).
			byID := make(map[string]*chain.Build, len(builds))
			var order []string
			for _, b := range builds {
				v := toChainBuild(b)
				byID[b.ID] = &v
				order = append(order, b.ID)
			}
			for _, r := range requests {
				if v, answered := byID[r.ID]; answered {
					v.Request = toRequest(r)
					continue
				}
				v := askedRecord(r)
				if r.NotAsked != "" {
					// Never handed over: asked again once, after the move (askAgain).
					v.Request.Reask = chain.ReaskPending
				}
				byID[r.ID] = &v
				order = append(order, r.ID)
			}
			out := make([]moveRow, 0, len(order))
			for _, id := range order {
				body, err := json.Marshal(byID[id])
				if err != nil {
					return nil, 0, err
				}
				out = append(out, moveRow{key: chainBuildKey(id), value: *byID[id], size: len(body)})
			}
			return out, len(builds) + len(requests), nil
		},
		create: func(ctx context.Context, c *chain.Chain, row moveRow) error {
			_, err := c.Builds.Create(ctx, row.key, row.value.(chain.Build))
			return err
		},
		keys: func(ctx context.Context, c *chain.Chain) ([]string, error) { return c.Builds.Keys(ctx) },
		count: func(ctx context.Context, tx pgx.Tx) (int, error) {
			var n int
			err := tx.QueryRow(ctx, `select (select count(*) from build) + (select count(*) from build_request)`).Scan(&n)
			return n, err
		},
		lock: `lock table build, build_request in share mode`,
		settle: func(ctx context.Context, c *chain.Chain, rows []moveRow) error {
			return settleBucket(ctx, c.Builds, rows)
		},
	}, {
		kind: KindRollouts,
		rows: func(ctx context.Context) ([]moveRow, int, error) {
			plans, err := i.plans(ctx, `order by created`)
			if err != nil {
				return nil, 0, err
			}
			merges, err := i.merges(ctx, ``)
			if err != nil {
				return nil, 0, err
			}
			tablePlans := len(plans)
			if plans, err = holdMerges(plans, merges); err != nil {
				return nil, 0, err
			}
			out := make([]moveRow, 0, len(plans))
			for _, p := range plans {
				body, err := json.Marshal(p)
				if err != nil {
					return nil, 0, err
				}
				out = append(out, moveRow{key: p.ID, value: p, size: len(body)})
			}
			return out, tablePlans + len(merges), nil
		},
		create: func(ctx context.Context, c *chain.Chain, row moveRow) error {
			_, err := chain.RolloutsOf[Plan](c).Create(ctx, row.key, row.value.(Plan))
			return err
		},
		keys: func(ctx context.Context, c *chain.Chain) ([]string, error) {
			return chain.RolloutsOf[Plan](c).Keys(ctx)
		},
		count: func(ctx context.Context, tx pgx.Tx) (int, error) {
			var n int
			err := tx.QueryRow(ctx, `select (select count(*) from release_plan) + (select count(*) from batched_merge)`).Scan(&n)
			return n, err
		},
		lock: `lock table release_plan, batched_merge in share mode`,
		settle: func(ctx context.Context, c *chain.Chain, rows []moveRow) error {
			return settleBucket(ctx, chain.RolloutsOf[Plan](c), rows)
		},
	}, {
		kind: KindAssignments,
		rows: func(ctx context.Context) ([]moveRow, int, error) {
			assigned, err := i.assignedWithSends(ctx)
			if err != nil {
				return nil, 0, err
			}
			out := make([]moveRow, 0, len(assigned))
			for _, a := range assigned {
				body, err := json.Marshal(a)
				if err != nil {
					return nil, 0, err
				}
				out = append(out, moveRow{key: assignmentKey(a.Node, a.Module), value: a, size: len(body)})
			}
			return out, len(out), nil
		},
		create: createAssignment,
		keys:   func(ctx context.Context, c *chain.Chain) ([]string, error) { return c.Assignments.Keys(ctx) },
		count: func(ctx context.Context, tx pgx.Tx) (int, error) {
			var n int
			err := tx.QueryRow(ctx, `select count(*) from assignment`).Scan(&n)
			return n, err
		},
		lock: `lock table assignment in share mode`,
		settle: func(ctx context.Context, c *chain.Chain, rows []moveRow) error {
			have := make(map[string]bool, len(rows))
			for _, r := range rows {
				have[r.key] = true
				if err := settleAssignment(ctx, c, r); err != nil {
					return err
				}
			}
			// An assignment the store no longer holds ended since the first pass: the bucket says so.
			all, err := c.Assignments.All(ctx)
			if err != nil {
				return err
			}
			for key, k := range all {
				if have[key] || k.Value.Assigned.IsZero() || !k.Value.Stands() {
					continue
				}
				k.Value.Ended = time.Now().UTC()
				if _, err := c.Assignments.Update(ctx, key, k.Value, k.Revision); err != nil && !errors.Is(err, chain.ErrMoved) {
					return err
				}
			}
			return nil
		},
	}}
}

// assignedWithSends is every assignment of the store as the chain keeps it: with the commit its machine was last sent for
// the module (node.sent_builds), the build that is of it and what the build published, and since when (when the machine was
// last sent). An assignment of a module the machine was never sent has none of those.
func (i *Inventory) assignedWithSends(ctx context.Context) ([]chain.Assignment, error) {
	rows, err := i.store.Pool().Query(ctx,
		`select n.name, a.module, a.assigned, n.sent_at, n.sent_builds
		   from assignment a join node n on n.id = a.node order by n.name, a.module`)
	if err != nil {
		return nil, err
	}
	var out []chain.Assignment
	for rows.Next() {
		var a chain.Assignment
		var sentAt *time.Time
		var raw []byte
		if err := rows.Scan(&a.Node, &a.Module, &a.Assigned, &sentAt, &raw); err != nil {
			rows.Close()
			return nil, err
		}
		a.Assigned = a.Assigned.UTC()
		var sent map[string]string
		if len(raw) > 0 {
			_ = json.Unmarshal(raw, &sent)
		}
		if commit := sent[a.Module]; commit != "" && sentAt != nil {
			a.Commit, a.Since = commit, sentAt.UTC()
		}
		out = append(out, a)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return nil, err
	}
	// The build of each commit, read once: the newest successful build of the module at it.
	type at struct{ module, commit string }
	var all []Build
	if _, moved, err := i.ChainReads(ctx, KindBuilds); err != nil {
		return nil, err
	} else if moved {
		if all, err = chainBuilds(ctx, i.attached()); err != nil {
			return nil, err
		}
	} else if all, err = i.allBuildRows(ctx); err != nil {
		return nil, err
	}
	newest := map[at]Build{}
	for _, b := range all {
		k := at{b.Module, b.Commit}
		if b.Failed != "" || b.Module == "" {
			continue
		}
		if had, has := newest[k]; !has || b.At.After(had.At) || (b.At.Equal(had.At) && b.ID > had.ID) {
			newest[k] = b
		}
	}
	for k := range out {
		if out[k].Commit == "" {
			continue
		}
		b, found := newest[at{out[k].Module, out[k].Commit}]
		if !found {
			continue
		}
		out[k].Build = b.ID
		if published := publishedOf(b.Made); published != nil {
			out[k].Artifact, out[k].Package = published.Artifact, published.Package
		}
	}
	return out, nil
}

// holdMerges puts each batched merge into the record of the plan that answers it. A merge no plan answers — the plan_id
// of a merge a failed walk carried is null, or names a plan the store never kept — is held, flagged waiting, by the
// newest plan that lists it among its merges, else the record kept for the merges no plan answers: the record that held it
// last.
func holdMerges(plans []Plan, merges []BatchedMerge) ([]Plan, error) {
	waitingAt := -1
	at := make(map[string]int, len(plans))
	for k, p := range plans {
		at[p.ID] = k
	}
	newest := func(match func(Plan) bool) int {
		best := -1
		for k, p := range plans {
			if match(p) && (best < 0 || p.Created.After(plans[best].Created)) {
				best = k
			}
		}
		return best
	}
	for _, m := range merges {
		held := heldOf(m, m.Heard)
		holder, answered := at[m.Plan]
		if m.Plan == "" || !answered {
			held.Waiting = true
			holder = newest(func(p Plan) bool {
				if p.Delivery == nil {
					return false
				}
				for _, listed := range p.Delivery.Merges {
					if strings.EqualFold(listed.Repository, m.Repository) && listed.Commit == m.Commit {
						return true
					}
				}
				return false
			})
			if holder < 0 {
				// Held by the record kept for the merges no plan answers, made with the others.
				if waitingAt < 0 {
					plans = append(plans, Plan{ID: WaitingRecord, State: "merges", Created: time.Unix(0, 0).UTC(), Revision: 1,
						Tiers: [][]string{}, Modules: map[string]*PlanModule{}, Note: "the merges no plan answers"})
					waitingAt = len(plans) - 1
				}
				holder = waitingAt
			}
		}
		plans[holder].Held = append(plans[holder].Held, held)
	}
	return plans, nil
}

// MoveChain moves every kind not moved yet, in order, and says what it did. It needs the chain attached and the lease
// held (the chain's writes ask). It stops at the first kind that fails.
func (i *Inventory) MoveChain(ctx context.Context, say func(format string, args ...any)) ([]MoveReport, error) {
	if say == nil {
		say = func(string, ...any) {}
	}
	if i.attached() == nil {
		return nil, ErrNoChain
	}
	var out []MoveReport
	for _, m := range i.movers() {
		r, err := i.moveKind(ctx, m, say)
		if err != nil {
			return out, err
		}
		out = append(out, r)
	}
	return out, nil
}

// movePasses is how often a move reads the tables again when they took a row while it wrote: the controller before this
// one may still have been writing as the lease changed hands.
const movePasses = 5

func (i *Inventory) moveKind(ctx context.Context, m mover, say func(string, ...any)) (MoveReport, error) {
	report := MoveReport{Kind: m.kind}
	moved, err := i.ChainMoved(ctx)
	if err != nil {
		return report, err
	}
	if _, done := moved[m.kind]; done {
		report.Already = true
		return report, nil
	}
	for pass := 1; pass <= movePasses; pass++ {
		rows, tableRows, err := m.rows(ctx)
		if err != nil {
			return report, fmt.Errorf("the %s cannot be read from the tables: %w", m.kind, err)
		}
		report.Rows, report.TableRows, report.Largest = len(rows), tableRows, 0
		// Every value measured before any is written: a value over the bus's largest stops the move with nothing
		// written (ADR 0408 §6), never cut.
		for _, r := range rows {
			report.Largest = max(report.Largest, r.size)
			if r.size > broker.ChainValueBytes {
				return report, &Refusal{m.kind, fmt.Sprintf("the value of %s is %d bytes and the bus holds at most %d; "+
					"nothing is cut, so the record must be made smaller first", r.key, r.size, broker.ChainValueBytes)}
			}
		}
		report.Written, report.Present = 0, 0
		for _, r := range rows {
			switch err := m.create(ctx, i.attached(), r); {
			case err == nil:
				report.Written++
			case errors.Is(err, chain.ErrExists):
				report.Present++
			case errors.Is(err, chain.ErrBadKey):
				return report, &Refusal{m.kind, fmt.Sprintf("%s cannot be a key of the bus: %v", r.key, err)}
			default:
				return report, fmt.Errorf("%s %s is not written to the bus: %w", m.kind, r.key, err)
			}
		}
		// Rows against keys: every row's key is in the bucket.
		keys, err := m.keys(ctx, i.attached())
		if err != nil {
			return report, fmt.Errorf("the keys of the %s cannot be read back: %w", m.kind, err)
		}
		have := make(map[string]bool, len(keys))
		for _, k := range keys {
			have[k] = true
		}
		for _, r := range rows {
			if !have[r.key] {
				return report, &Refusal{m.kind, fmt.Sprintf("the tables hold %d records and the bucket does not hold %s", len(rows), r.key)}
			}
		}
		settled, err := i.switchKind(ctx, m, report)
		if err != nil {
			return report, err
		}
		if settled {
			say("moved the %s to the bus: %d records, %d written now and %d already there, the largest %d bytes",
				m.kind, report.Rows, report.Written, report.Present, report.Largest)
			return report, nil
		}
		say("the %s took a row while they were moved, so they are read again (pass %d of %d)", m.kind, pass, movePasses)
	}
	return report, &Refusal{m.kind, fmt.Sprintf("the tables kept taking rows through %d passes; another controller is still writing them", movePasses)}
}

// switchKind records the kind as moved, in one transaction that holds the tables against writes and finds their count
// as the move read it: false when they took a row since.
func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool, error) {
	tx, err := i.store.Pool().Begin(ctx)
	if err != nil {
		return false, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	if _, err := tx.Exec(ctx, m.lock); err != nil {
		return false, err
	}
	n, err := m.count(ctx, tx)
	if err != nil {
		return false, err
	}
	if n != r.TableRows {
		return false, nil
	}
	// The rows read again under the lock: nothing writes them now, so what the bucket holds is made what they say.
	rows, tableRows, err := m.rows(ctx)
	if err != nil {
		return false, err
	}
	if tableRows != r.TableRows {
		return false, nil
	}
	if err := m.settle(ctx, i.attached(), rows); err != nil {
		return false, fmt.Errorf("the %s changed since they were first written, and the bucket cannot be brought to them: %w", m.kind, err)
	}
	counts, _ := json.Marshal(map[string]int{m.kind: r.Rows})
	largest, _ := json.Marshal(map[string]int{m.kind: r.Largest})
	if _, err := tx.Exec(ctx,
		`insert into chain_moved (counts, largest) values ($1, $2)
		 on conflict (only_one) do update
		    set counts = chain_moved.counts || excluded.counts, largest = chain_moved.largest || excluded.largest,
		        moved_at = now()`, counts, largest); err != nil {
		return false, err
	}
	if err := tx.Commit(ctx); err != nil {
		return false, err
	}
	i.moved.mu.Lock()
	if i.moved.kind == nil {
		i.moved.kind = map[string]bool{}
	}
	i.moved.kind[m.kind] = true
	i.moved.mu.Unlock()
	return true, nil
}

// allBuildRows is every build of the table, whole.
func (i *Inventory) allBuildRows(ctx context.Context) ([]Build, error) {
	rows, err := i.store.Pool().Query(ctx,
		`select id, repository, ref, coalesce(module, ''), commit_hash, built_on, failed, made, source_path, manifest,
		        built_against, built_contexts, build_sources, source_fingerprint, asked, at
		   from build order by at, id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Build
	for rows.Next() {
		var b Build
		var made, manifest, against, read, sources []byte
		var asked *time.Time
		if err := rows.Scan(&b.ID, &b.Repository, &b.Ref, &b.Module, &b.Commit, &b.On, &b.Failed, &made, &b.Path,
			&manifest, &against, &read, &sources, &b.SourceFingerprint, &asked, &b.At); err != nil {
			return nil, err
		}
		if asked != nil {
			b.Asked = *asked
		}
		b.Manifest = manifest
		for _, part := range []struct {
			raw []byte
			to  any
			who string
		}{{made, &b.Made, "made"}, {against, &b.Against, "built_against"}, {read, &b.Read, "built_contexts"},
			{sources, &b.Sources, "build_sources"}} {
			if len(part.raw) == 0 {
				continue
			}
			if err := json.Unmarshal(part.raw, part.to); err != nil {
				return nil, fmt.Errorf("build %s: %s is not readable: %w", b.ID, part.who, err)
			}
		}
		out = append(out, b)
	}
	return out, rows.Err()
}

// allRequestRows is every build request of the table.
func (i *Inventory) allRequestRows(ctx context.Context) ([]BuildRequest, error) {
	rows, err := i.store.Pool().Query(ctx,
		`select id, repository, seat, source_path, ref, commit_hash, asked_for, coalesce(not_asked, ''),
		        coalesce(outcome_unknown, ''), asked_at, at_terminal
		   from build_request order by asked_at, id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []BuildRequest
	for rows.Next() {
		var a BuildRequest
		if err := rows.Scan(&a.ID, &a.Repository, &a.Seat, &a.Path, &a.Ref, &a.Commit, &a.For, &a.NotAsked,
			&a.OutcomeUnknown, &a.At, &a.AtTerminal); err != nil {
			return nil, err
		}
		out = append(out, a)
	}
	return out, rows.Err()
}

// createAssignment writes an assignment's row under its key; chain.ErrExists when the key holds a value that stays.
func createAssignment(ctx context.Context, c *chain.Chain, row moveRow) error {
	want := row.value.(chain.Assignment)
	// The mirror of the store kept a value under this key until now, with the provisioner's plan beside it: the
	// plan stays, the record's links are laid in. A value that names its build already is a later word, and stays.
	for try := 0; try < casTries; try++ {
		cur, found, err := c.Assignments.Get(ctx, row.key)
		if err != nil {
			return err
		}
		if !found {
			if _, err := c.Assignments.Create(ctx, row.key, want); !errors.Is(err, chain.ErrExists) {
				return err
			}
			continue
		}
		if cur.Value.Build != "" || cur.Value.Commit != "" || !cur.Value.Since.IsZero() || (want.Build == "" && want.Commit == "") {
			return chain.ErrExists
		}
		merged := want
		merged.RequiredBy, merged.Provisioning = cur.Value.RequiredBy, cur.Value.Provisioning
		if _, err := c.Assignments.Update(ctx, row.key, merged, cur.Revision); errors.Is(err, chain.ErrMoved) {
			continue
		} else {
			return err
		}
	}
	return fmt.Errorf("the assignment %s kept moving under the write", row.key)
}

// settleBucket makes a bucket of whole values hold each row's value, by compare-and-set: a key that is absent is made, one
// whose bytes differ from the row's is written over, and one that is the same is left.
func settleBucket[T any](ctx context.Context, b chain.Bucket[T], rows []moveRow) error {
	for _, r := range rows {
		want, err := json.Marshal(r.value)
		if err != nil {
			return err
		}
		settled := false
		for try := 0; try < casTries && !settled; try++ {
			e, found, err := b.GetRaw(ctx, r.key)
			if err != nil {
				return err
			}
			switch {
			case !found:
				_, err = b.CreateRaw(ctx, r.key, want)
				if errors.Is(err, chain.ErrExists) {
					continue
				}
			case bytes.Equal(e.Value, want):
				settled = true
				continue
			default:
				_, err = b.UpdateRaw(ctx, r.key, want, e.Revision)
				if errors.Is(err, chain.ErrMoved) {
					continue
				}
			}
			if err != nil {
				return err
			}
			settled = true
		}
		if !settled {
			return fmt.Errorf("%s kept moving under the write", r.key)
		}
	}
	return nil
}

// settleAssignment makes the value under an assignment's key what its row says, keeping the plan the mirror put beside it:
// before the switch the store is the only truth of the links, so a commit it changed since the first pass is written over.
func settleAssignment(ctx context.Context, c *chain.Chain, row moveRow) error {
	want := row.value.(chain.Assignment)
	for try := 0; try < casTries; try++ {
		cur, found, err := c.Assignments.Get(ctx, row.key)
		if err != nil {
			return err
		}
		if !found {
			if _, err := c.Assignments.Create(ctx, row.key, want); errors.Is(err, chain.ErrExists) {
				continue
			} else {
				return err
			}
		}
		merged := want
		merged.RequiredBy, merged.Provisioning = cur.Value.RequiredBy, cur.Value.Provisioning
		a, _ := json.Marshal(merged)
		b, _ := json.Marshal(cur.Value)
		if bytes.Equal(a, b) {
			return nil
		}
		if _, err := c.Assignments.Update(ctx, row.key, merged, cur.Revision); errors.Is(err, chain.ErrMoved) {
			continue
		} else {
			return err
		}
	}
	return fmt.Errorf("the assignment %s kept moving under the write", row.key)
}