package registry

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"reflect"
	"strings"
)

// The import journal (seeded-instances spec §5.7): one row per host this
// box has swapped for an import, holding the complete original route, so
// the box always knows what it replaced, whatever happened to the record.
const (
	JournalSwapped        = "swapped"
	JournalRollingBack    = "rolling-back"
	JournalCleanupPending = "cleanup-pending"
)

// ImportGateClosed is the import gate a rollback installs: it matches no
// token's hash (seeds.ClosedGate holds the same value).
const ImportGateClosed = "closed"

// ErrImportJournal is a journal transaction whose precondition did not
// hold: no row or another seed's, a row already there, a changed route.
var ErrImportJournal = errors.New("registry: the import journal does not allow this")

// What the host's own env file (<host>.env, at the nested path the swap
// gives the import's row) held when the swap ran, before any launch of the
// import: ImportJournal.EnvBefore. The platform mints the instance key into
// that file at the import's first launch, create-only, and a key there
// outranks the app's configuration, so a rollback has to know whether the
// key it finds is the import's own (it retires it) or was there before (it
// never touches it). "" is a journal row written before this was recorded:
// nothing is known, so nothing is retired.
const (
	EnvBeforeAbsent = "absent" // no env file: every byte in it later is the import's
	EnvBeforeNoKey  = "no-key" // an env file with no instance key: a key in it later is the import's
	EnvBeforeKey    = "key"    // an env file that already carried a key: the import minted none
)

// ImportJournal is one import_journal row. Original is the route the swap
// replaced, exactly as it was stored. EnvBefore is one of the EnvBefore
// values, or "" for unknown.
type ImportJournal struct {
	Host      string
	SeedID    string
	Original  Route
	RecordKey string
	State     string
	FenceGen  uint64
	At        int64
	EnvBefore string
}

func (r *Registry) migrateImportJournal() error {
	if _, err := r.sql.Exec(`CREATE TABLE IF NOT EXISTS import_journal (
  host       TEXT PRIMARY KEY,
  seed_id    TEXT NOT NULL,
  original   TEXT NOT NULL,
  record_key TEXT NOT NULL DEFAULT '',
  state      TEXT NOT NULL,
  fence_gen  INTEGER NOT NULL DEFAULT 0,
  at         INTEGER NOT NULL DEFAULT 0,
  env_before TEXT NOT NULL DEFAULT ''
);`); err != nil {
		return err
	}
	// A journal table from before env_before gains it; its rows read "".
	if _, err := r.sql.Exec(`ALTER TABLE import_journal ADD COLUMN env_before TEXT NOT NULL DEFAULT ''`); err != nil &&
		!strings.Contains(err.Error(), "duplicate column") {
		return err
	}
	return nil
}

type queryer interface {
	QueryRow(query string, args ...any) *sql.Row
}

type txer interface {
	queryer
	Exec(query string, args ...any) (sql.Result, error)
}

const journalCols = `host, seed_id, original, record_key, state, fence_gen, at, env_before`

func scanJournalRow(scan func(...any) error) (ImportJournal, error) {
	var j ImportJournal
	var orig string
	if err := scan(&j.Host, &j.SeedID, &orig, &j.RecordKey, &j.State, &j.FenceGen, &j.At, &j.EnvBefore); err != nil {
		return j, err
	}
	if err := json.Unmarshal([]byte(orig), &j.Original); err != nil {
		return j, fmt.Errorf("registry: import journal %s: original row: %w", j.Host, err)
	}
	return j, nil
}

func scanJournal(q queryer, host string) (*ImportJournal, error) {
	j, err := scanJournalRow(q.QueryRow(`SELECT `+journalCols+` FROM import_journal WHERE host = ?`, host).Scan)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &j, nil
}

// ImportJournal returns host's journal row, or nil when there is none.
func (r *Registry) ImportJournal(host string) (*ImportJournal, error) {
	if r.remote != nil {
		return nil, errRemoteMethod
	}
	return scanJournal(r.sql, host)
}

// ImportJournals returns every journal row, ordered by host.
func (r *Registry) ImportJournals() ([]ImportJournal, error) {
	if r.remote != nil {
		return nil, errRemoteMethod
	}
	rows, err := r.sql.Query(`SELECT ` + journalCols + ` FROM import_journal ORDER BY host`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []ImportJournal
	for rows.Next() {
		j, err := scanJournalRow(rows.Scan)
		if err != nil {
			return nil, err
		}
		out = append(out, j)
	}
	return out, rows.Err()
}

// ImportJournalHosts returns the hosts the journal holds a row for, in any
// state, ordered. It is what a box says about its imports in the claim
// activation census (hostops.CollectCensusSeeds). Unlike the journal's
// other methods it answers on a handle that proxies to root's registry
// child too, so a collector that runs as root can ask.
func (r *Registry) ImportJournalHosts() ([]string, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "ImportJournalHosts"}, nil)
		if err != nil {
			return nil, err
		}
		// The transport drops an empty list; "none" is not "did not say".
		return append([]string{}, out.Hosts...), nil
	}
	rows, err := r.sql.Query(`SELECT host FROM import_journal ORDER BY host`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := []string{}
	for rows.Next() {
		var h string
		if err := rows.Scan(&h); err != nil {
			return nil, err
		}
		out = append(out, h)
	}
	return out, rows.Err()
}

// hashImportJournal folds the journal into SnapshotFingerprint's digest:
// the journal is box-local state nothing in the bucket can rebuild, and a
// snapshot that kept a deleted cleanup-pending row would, restored cold,
// resume a finished rollback's bucket cleanup over a later instance on the
// same host. Every column, in host order, under its own section marker. An
// empty journal writes nothing, so a box with no import in flight digests
// exactly as it did before the journal existed.
func hashImportJournal(db *sql.DB, h io.Writer) error {
	rows, err := db.Query(`SELECT ` + journalCols + ` FROM import_journal ORDER BY host`)
	if err != nil {
		return err
	}
	defer rows.Close()
	first := true
	for rows.Next() {
		var host, seedID, orig, recordKey, state, envBefore string
		var fenceGen uint64
		var at int64
		if err := rows.Scan(&host, &seedID, &orig, &recordKey, &state, &fenceGen, &at, &envBefore); err != nil {
			return err
		}
		if first {
			fmt.Fprint(h, "\x1dimport_journal\x1e")
			first = false
		}
		fmt.Fprintf(h, "%q\x1f%q\x1f%q\x1f%q\x1f%q\x1f%d\x1f%d", host, seedID, orig, recordKey, state, fenceGen, at)
		// env_before is hashed only when set, tagged, so a row written
		// before the column existed digests as it did.
		if envBefore != "" {
			fmt.Fprintf(h, "\x1fenv_before=%q", envBefore)
		}
		fmt.Fprint(h, "\x1e")
	}
	return rows.Err()
}

// inTx runs fn in one BEGIN IMMEDIATE transaction at synchronous=FULL (the
// claim transactions' form). IMMEDIATE takes the write lock before the first
// read, so another process's write waits on busy_timeout instead of failing
// the upgrade; FULL makes the commit durable before the caller acts on it
// (the swap's record CAS follows it). The registry's pool holds one
// connection (Open), so in-process writers serialise on it.
//
// The import journal, the fence generation source and the seed column are
// the agent's own: the agent opens the registry itself (Open), and no root
// pass reads or writes them. A handle that proxies to root's registry child
// (OpenRemote) has no dispatch for them, so each refuses there
// (errRemoteMethod) rather than touch a database it does not hold.
func (r *Registry) inTx(fn func(tx txer) error) error {
	if r.remote != nil {
		return errRemoteMethod
	}
	tx, err := r.beginClaimTransaction(context.Background(), true)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if err := fn(tx); err != nil {
		return err
	}
	return tx.Commit()
}

func getRouteTx(tx txer, host string) (*Route, error) {
	t, err := scanRoute(tx.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host = ?`, host).Scan)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &t, nil
}

// insertRouteTx inserts t exactly as given (no Upsert defaults).
func insertRouteTx(tx txer, t Route) error {
	_, err := tx.Exec(`INSERT INTO routes (`+routeCols+`) VALUES (`+placeholders(routeColCount)+`)`, routeArgs(t)...)
	return err
}

// IsImportRow reports whether rt is the row j's swap put in place: the
// journal's host, an instance row a record provisioned, still gated (by
// the verify hash, or the closed sentinel once a rollback began), and
// whose seed column names j's seed. The rollback acts only on this row
// (spec §5.6): any other row at the host belongs to someone else, such as
// a later instance when a stale journal row came back from a snapshot.
func IsImportRow(rt Route, j ImportJournal) bool {
	return j.SeedID != "" && rt.Host == j.Host && rt.Kind == "instance" && rt.Provisioned &&
		rt.ImportGate != "" && seedColumnID(rt.SeedSource) == j.SeedID
}

// seedColumnID is the seed id a seed_source column names, "" for none or
// one that does not parse. The column is seeds.RowSeed's JSON; the
// registry reads only its id, since seeds imports this package.
func seedColumnID(col string) string {
	if col == "" {
		return ""
	}
	var v struct {
		ID string `json:"id"`
	}
	if err := json.Unmarshal([]byte(col), &v); err != nil {
		return ""
	}
	return v.ID
}

// SwapForImport is the swap's one transaction (spec §5.4, codex 14): with
// no journal row for the host and the stored route still exactly
// j.Original, it replaces the route with next (Upsert's defaults applied)
// and writes the journal row at swapped. next must be the import's row as
// IsImportRow reads it (the same host, gated, naming the journal's seed),
// and the journal must name its record: a journal row with no record key
// can never be decided, so it is never written. The original must be one
// the journal can put back: not claim-owned (claims are off for imports,
// spec §3.4), not held by an operation, not itself gated, and unchanged by
// its JSON round trip. j.EnvBefore is stored as given: one of the EnvBefore
// values, or "" when the caller could not say.
func (r *Registry) SwapForImport(j ImportJournal, next Route) error {
	if j.Host == "" || j.SeedID == "" || j.RecordKey == "" || j.Original.Host != j.Host || !IsImportRow(normalizeRoute(next), j) {
		return fmt.Errorf("%w: the swap for %s is not well formed", ErrImportJournal, j.Host)
	}
	switch j.EnvBefore {
	case "", EnvBeforeAbsent, EnvBeforeNoKey, EnvBeforeKey:
	default:
		return fmt.Errorf("%w: the swap for %s is not well formed", ErrImportJournal, j.Host)
	}
	if j.Original.ClaimOwnership != (ClaimOwnership{}) || j.Original.HoldOp != "" || j.Original.ImportGate != "" {
		return fmt.Errorf("%w: the route at %s is not one an import replaces", ErrImportJournal, j.Host)
	}
	orig, err := json.Marshal(j.Original)
	if err != nil {
		return err
	}
	var back Route
	if err := json.Unmarshal(orig, &back); err != nil || !reflect.DeepEqual(back, j.Original) {
		return fmt.Errorf("%w: the route at %s would not restore exactly", ErrImportJournal, j.Host)
	}
	return r.inTx(func(tx txer) error {
		if cur, err := scanJournal(tx, j.Host); err != nil {
			return err
		} else if cur != nil {
			return fmt.Errorf("%w: %s already has a journal row for seed %s", ErrImportJournal, j.Host, cur.SeedID)
		}
		stored, err := getRouteTx(tx, j.Host)
		if err != nil {
			return err
		}
		if stored == nil || !reflect.DeepEqual(*stored, j.Original) {
			return fmt.Errorf("%w: the route at %s changed since it was checked", ErrImportJournal, j.Host)
		}
		if _, err := tx.Exec(`DELETE FROM routes WHERE host = ?`, j.Host); err != nil {
			return err
		}
		if err := insertRouteTx(tx, normalizeRoute(next)); err != nil {
			return err
		}
		_, err = tx.Exec(`INSERT INTO import_journal (`+journalCols+`) VALUES (?,?,?,?,?,0,?,?)`,
			j.Host, j.SeedID, string(orig), j.RecordKey, JournalSwapped, j.At, j.EnvBefore)
		return err
	})
}

// journalFor is host's journal row, which must name seedID and be in one of
// states.
func journalFor(tx txer, host, seedID string, states ...string) (*ImportJournal, error) {
	j, err := scanJournal(tx, host)
	if err != nil {
		return nil, err
	}
	if j == nil || j.SeedID != seedID {
		return nil, fmt.Errorf("%w: no journal row for %s seed %s", ErrImportJournal, host, seedID)
	}
	for _, s := range states {
		if j.State == s {
			return j, nil
		}
	}
	return nil, fmt.Errorf("%w: %s's journal is %s", ErrImportJournal, host, j.State)
}

// importRowTx is host's row when it is j's import row, nil when the host
// has no row, and ErrImportJournal when the row there is anyone else's.
func importRowTx(tx txer, j *ImportJournal) (*Route, error) {
	cur, err := getRouteTx(tx, j.Host)
	if err != nil {
		return nil, err
	}
	if cur != nil && !IsImportRow(*cur, *j) {
		return nil, fmt.Errorf("%w: the route at %s is not import %s's", ErrImportJournal, j.Host, j.SeedID)
	}
	return cur, nil
}

// CloseImportGate is the rollback's step 1 (spec §5.6): the journal says
// rolling-back with the fence generation, and the gate refuses even verify
// requests, in one transaction. Run again on a row already rolling-back, it
// re-closes the gate and keeps the stored generation (fenceGen is ignored):
// step 2 recognises this rollback's own fence by that generation, so the
// caller reads FenceGen back from the journal rather than trusting its own.
//
// It refuses when the host's row is not the import's (IsImportRow), so it
// never gates a row the import does not own. A host with no row is closed
// all the same: the journal moves on, and there is no gate to set.
func (r *Registry) CloseImportGate(host, seedID string, fenceGen uint64) error {
	return r.inTx(func(tx txer) error {
		j, err := journalFor(tx, host, seedID, JournalSwapped, JournalRollingBack)
		if err != nil {
			return err
		}
		if _, err := importRowTx(tx, j); err != nil {
			return err
		}
		if j.State == JournalSwapped {
			if _, err := tx.Exec(`UPDATE import_journal SET state = ?, fence_gen = ? WHERE host = ?`, JournalRollingBack, fenceGen, host); err != nil {
				return err
			}
		}
		_, err = tx.Exec(`UPDATE routes SET import_gate = ? WHERE host = ?`, ImportGateClosed, host)
		return err
	})
}

// RestoreImportRoute is the rollback's step 4 (spec §5.6, codex R3-1): the
// original route back exactly as it was stored, and the journal at
// cleanup-pending.
//
// Only the import's own row is replaced (IsImportRow). Any other row at
// the host is refused, and nothing changes: it is not the import's to
// remove. A host with no row at all (something removed the import's row)
// gets no route back, since whatever removed it decided the host should
// not be served; the journal still moves to cleanup-pending, so the
// rollback carries on with its cleanup.
func (r *Registry) RestoreImportRoute(host, seedID string) error {
	return r.inTx(func(tx txer) error {
		j, err := journalFor(tx, host, seedID, JournalRollingBack)
		if err != nil {
			return err
		}
		cur, err := importRowTx(tx, j)
		if err != nil {
			return err
		}
		if cur != nil {
			if _, err := tx.Exec(`DELETE FROM routes WHERE host = ?`, host); err != nil {
				return err
			}
			if err := insertRouteTx(tx, j.Original); err != nil {
				return err
			}
		}
		_, err = tx.Exec(`UPDATE import_journal SET state = ? WHERE host = ?`, JournalCleanupPending, host)
		return err
	})
}

// CommitImport clears the gate and retires the journal in one transaction
// (spec §5.4 "Commit").
func (r *Registry) CommitImport(host, seedID string) error {
	return r.inTx(func(tx txer) error {
		if _, err := journalFor(tx, host, seedID, JournalSwapped); err != nil {
			return err
		}
		if _, err := tx.Exec(`UPDATE routes SET import_gate = '' WHERE host = ?`, host); err != nil {
			return err
		}
		_, err := tx.Exec(`DELETE FROM import_journal WHERE host = ?`, host)
		return err
	})
}

// DeleteImportJournal is the rollback's step 7.
func (r *Registry) DeleteImportJournal(host, seedID string) error {
	return r.inTx(func(tx txer) error {
		if _, err := journalFor(tx, host, seedID, JournalCleanupPending); err != nil {
			return err
		}
		_, err := tx.Exec(`DELETE FROM import_journal WHERE host = ?`, host)
		return err
	})
}