package main

// postgres handed to the provisioner (novox/hq ADR 0373, ADR 0349 clause 12, issue 780): where its consumers file
// is set, postgres runs on the SDK's provider helper instead of the harness reading the controller's grant files.
//
// The helper learns who binds postgres from the plans, composes each consumer's entry and reads its credential
// through ConsumersFile, the file this machine's provisioner client fills. It leaves the rest to the two hooks
// here: Make brings the consumer's database and role into being exactly as the harness's adapter does, and Gone
// retires what was kept for a consumer that no longer binds, the reversible way (retire_pg.go): the login is
// locked and marked to delete, nothing is dropped.
//
// **A consumer not moved is still the grant file's** (issue 937, ADR 0349 clause 17): only a moved consumer's plan
// names its provider, so the controller still gives every other consumer through postgres's grant file
// (MESH_RECEIVES) until R1 (issue 682). Each login it gives is asked for, and one no plan names is made and checked
// by the harness's grant pass exactly as on the grant path; a login a plan names is the helper's alone. A grant file
// that cannot be read is not nobody asking (issue 241): nothing is retired until it is read again.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"

	"git.novox.be/novox/mesh-sdk/go/provider"
)

// ProvisionName is what postgres provides, as its manifest says.
const ProvisionName = "postgres-database"

// servedPort is the port its manifest serves (`serves.postgres-database.port`).
const servedPort = 5432

// movedRun is postgres on the provider helper, with the harness's own memory and retirement behind it: what is held,
// what is counted unasked, what waits for a person (retirement.go). The harness is never Run here; the helper's
// Make and Gone and the loop below drive the same state, so the guards are the harness's, not a copy.
type movedRun struct {
	h *Harness

	// FirstRetry is the first wait before a watch asked too early is asked again (issue 943); a second when zero.
	FirstRetry time.Duration

	mu          sync.Mutex // guards assignments and loaded
	assignments map[string][]byte
	loaded      bool
	verified    map[string]time.Time
}

func newMovedRun(h *Harness) *movedRun {
	h.mu.Lock()
	h.init()
	h.mu.Unlock()
	return &movedRun{h: h, assignments: map[string][]byte{}, verified: map[string]time.Time{}}
}

// Make brings one consumer's database and role into being under its login and the vault's credential, as the
// harness does: at once when it is new or changed, and otherwise only when the server no longer holds it, which is
// asked every VerifyEvery.
func (m *movedRun) Make(ctx context.Context, c provider.Consumer) error {
	h, e := m.h, c.Entry
	if e.User == "" {
		return fmt.Errorf("%s has no login to make", e.Key())
	}
	if len(c.Credential) == 0 {
		return fmt.Errorf("%s has no credential to make its login with", e.Key())
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	as, password := e.User, string(c.Credential)
	p := Provision{As: as, Password: password, Values: orEmpty(e.Contributes), Consumer: e.Consumer}
	hash := hashOf(as, password, p.Values, nil)
	if was, ok := h.applied[as]; ok && was.hash == hash {
		if h.Now().Sub(m.verified[as]) < h.VerifyEvery {
			return nil
		}
		m.verified[as] = h.Now()
		hctx, cancel := context.WithTimeout(ctx, h.HoldsTimeout)
		held, err := h.Adapter.Holds(hctx, p)
		cancel()
		if err != nil {
			text := scrub(err, password)
			h.failed(as, e.Consumer, h.classOf(err, text), text)
			return errors.New(text)
		}
		if held {
			h.succeeded(as)
			return nil
		}
		h.say("%s: the backend no longer holds it; applying again", as)
	}
	if err := h.Adapter.Create(ctx, p); err != nil {
		text := scrub(err, password)
		h.failed(as, e.Consumer, h.classOf(err, text), text)
		return errors.New(text)
	}
	h.succeeded(as)
	h.applied[as] = appliedEntry{hash: hash, node: e.Consumer}
	m.verified[as] = h.Now()
	h.reenabled(as, e.Consumer)
	return nil
}

// Gone is the helper asking whether it may forget a consumer that no longer binds. **It never retires**: retiring is
// the harness's rule, the same unasked set over StablePasses passes and StableFor, at most RetireAtOnce at a time
// (retirement.go, ADR 0230), run by retire below. So a consumer missing from one read of the plans loses nothing:
// the helper is told no, and asks again at its next pass, until the login is retired, or asked for again.
func (m *movedRun) Gone(_ context.Context, e provider.Entry) error {
	h := m.h
	as := e.User
	if as == "" {
		return nil // refused: nothing was made for it
	}
	// The plans and the grant file as they are now, not as the last retirement pass read them: with either not
	// read, nothing is let go.
	want, read := m.asked(h.Node)
	h.mu.Lock()
	defer h.mu.Unlock()
	_, granted, ok := m.granted(want)
	read = read && ok
	asked := want[as] || granted[as]
	_, held := h.applied[as]
	_, retired := h.r.retired[as]
	switch {
	case !read:
	case asked:
		return nil // another holder of the same login still binds: nothing to retire
	case retired, !held && h.r.seeded:
		return nil
	}
	return fmt.Errorf("%s is kept until its retirement is stable (%d passes and %s) and within the bound, or a "+
		"person approves it", as, h.StablePasses, h.StableFor)
}

// plan is as much of an assignment as is read to learn who binds postgres.
type plan struct {
	Node         string `json:"node"`
	Module       string `json:"module"`
	Provisioning *struct {
		Provisions []struct {
			Provision      string `json:"provision"`
			Provider       string `json:"provider"`
			ProviderModule string `json:"provider-module"`
			// As is who the consumer is at its provider's end, its slug first (issue 788).
			As string `json:"as"`
		} `json:"provisions"`
	} `json:"provisioning"`
}

// asked is the logins the plans bind postgres for, and whether they could all be read: a plan that is not the
// controller's, or no read yet, is not an answer, so nothing is counted unasked from it (novox/hq issue 241).
func (m *movedRun) asked(node string) (map[string]bool, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if !m.loaded || len(m.assignments) == 0 {
		return nil, false
	}
	want := map[string]bool{}
	for key, raw := range m.assignments {
		var a plan
		if err := json.Unmarshal(raw, &a); err != nil || a.Node+"."+a.Module != key {
			return nil, false
		}
		if a.Provisioning == nil {
			continue
		}
		for _, b := range a.Provisioning.Provisions {
			if b.Provider == node && b.ProviderModule == "postgres" && b.Provision == ProvisionName {
				as := b.As
				if as == "" {
					as = provider.ConsumerIdentity(a.Node, a.Module) // a plan that names none
				}
				want[as] = true
			}
		}
	}
	return want, true
}

// granted is what the grant file gives that no plan names (issue 937), and every login it gives; and whether it
// could be read. The caller holds h.mu. With no grant file named there is nothing given, and that is a read.
func (m *movedRun) granted(planned map[string]bool) ([]contribution, map[string]bool, bool) {
	h := m.h
	if h.Receives == "" {
		return nil, map[string]bool{}, true
	}
	given := h.readContributions()
	if given == nil {
		return nil, nil, false
	}
	var own []contribution
	logins := map[string]bool{}
	for _, g := range given {
		logins[g.As] = true
		if !planned[g.As] {
			own = append(own, g)
		}
	}
	return own, logins, true
}

// retire is one pass of the harness's retirement over what the plans bind and the grant file gives now; the logins
// only the grant file gives are made and checked first, as the grant path does.
func (m *movedRun) retire(ctx context.Context) {
	h := m.h
	want, ok := m.asked(h.Node)
	h.mu.Lock()
	defer h.mu.Unlock()
	var given []contribution
	if ok {
		var granted map[string]bool
		given, granted, ok = m.granted(want)
		for as := range granted {
			want[as] = true
		}
	}
	if !ok {
		// Not a result: the passes that must agree before anything is retired start again.
		h.r.key, h.r.count = "", 0
		return
	}
	h.seed(ctx, want)
	h.apply(ctx, given)
	h.retireUnasked(ctx, want)
	h.r.lastWant = want
	for as := range h.failing {
		if !want[as] {
			delete(h.failing, as)
		}
	}
	for as := range h.trouble {
		if _, held := h.applied[as]; !want[as] && !held {
			h.recovered(as, "no longer asked for")
		}
	}
}

// watch keeps the plans as the bus holds them, from the first read on.
func (m *movedRun) watch(ctx context.Context, bus provider.Bus) error {
	err := bus.Watch(ctx, provider.AssignmentsBucket, func(key string, value []byte) {
		m.mu.Lock()
		defer m.mu.Unlock()
		if value == nil {
			delete(m.assignments, key)
		} else {
			m.assignments[key] = append([]byte(nil), value...)
		}
	})
	if err == nil {
		m.mu.Lock()
		m.loaded = true
		m.mu.Unlock()
	}
	return err
}

// Run answers the consumers through the helper and retires through the harness's rules, until ctx ends.
//
// The bus is reached through the runtime that launched this bundle, which answers only once Serve runs, and Run starts
// beside Serve: a watch asked too early is asked again, after a wait that doubles from FirstRetry up to a minute,
// until it is answered or ctx ends, so a start that loses the race never leaves the consumers unanswered (issue 943).
func (m *movedRun) Run(ctx context.Context, p *provider.Provider, bus provider.Bus) error {
	wait := m.FirstRetry
	if wait <= 0 {
		wait = time.Second
	}
	for tries := 1; ; tries++ {
		err := m.watch(ctx, bus)
		if err == nil {
			break
		}
		if tries == 4 || tries%20 == 0 {
			m.h.say("the plans cannot be watched yet (%d tries), asking again: %v", tries, err)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(wait):
		}
		wait = min(wait*2, time.Minute)
	}
	go func() {
		for {
			m.retire(ctx)
			select {
			case <-ctx.Done():
				return
			case <-time.After(m.h.Every):
			}
		}
	}()
	return p.Run(ctx)
}

// moved is postgres handed to the provisioner: the helper answering its consumers, and the harness (never run) whose
// retirement and tools it shares. consumersFile is where its machine's client fills its consumers.
func moved(consumersFile string, h *Harness) (*movedRun, *provider.Provider, provider.Bus, error) {
	if consumersFile == "" {
		return nil, nil, nil, errors.New("no consumers' file")
	}
	run := newMovedRun(h)
	bus := provider.StateBus("postgres")
	return run, provider.New(provider.Options{
		Node:   h.Node,
		Module: "postgres",
		Offers: map[string]provider.Offer{ProvisionName: {
			Credential: provider.PerConsumer,
			Serves:     map[string]any{"port": servedPort},
		}},
		Bus:   bus,
		Vault: provider.ConsumersFile{Paths: []string{consumersFile}},
		Make:  run.Make,
		Gone:  run.Gone,
		Every: h.Every,
		Logf:  func(f string, a ...any) { h.say(f, a...) },
	}), bus, nil
}