package console

import (
	"context"
	"encoding/json"
	"log"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/release"
)

// MaterializeAccounts teaches the console about accounts it did not mint.
//
// The bucket is the control plane, and the console is NOT the only minter:
// `carlos accounts create` writes accounts/<sqid>/account.json and a registry
// row without ever opening this database — deliberately, because the carlos
// binary has no business inside console state. The consequence, before this,
// was a console that would refuse `member -add … <sqid>` for an account that
// demonstrably exists on the very box that created it. The agent's reconcile
// pass (internal/domains §1) already materializes exactly these manifests into
// the registry; this is the same pass for the console's own tables, reading
// the same objects through the same key parser.
//
// The id in the manifest is TAKEN AS GIVEN. Nothing here ever mints or
// reassigns one: the id is what the sqid encodes, the sqid is in every
// platform domain and every bucket key, and inventing one would re-home an
// account's whole world. A manifest that disagrees with itself — path sqid,
// body sqid and Sqid(id) must all be the same string — is corruption, not a
// new account, and is skipped loudly.
//
// INSERT-only: an account the console already knows keeps its row, name and
// members. Returns how many were newly materialized.
//
// Best-effort by design. A nil store (no CARLOS_DEPLOYMENT_BUCKET/DIR on this
// box) and a failed listing are both "nothing to catch up on", not errors: an
// unreachable bucket must never be what stops an operator removing a
// membership, which is the emergency this whole verb family exists for.
func MaterializeAccounts(ctx context.Context, s release.Store, st *Store) (int, error) {
	if s == nil {
		return 0, nil
	}
	keys, err := s.List(ctx, "accounts/")
	if err != nil {
		return 0, err
	}
	n := 0
	for _, key := range keys {
		sqid, ok := domains.AccountManifestSqid(key)
		if !ok {
			continue
		}
		// The UNFILTERED lookup, deliberately: this pass is the one place in
		// the console that must tell "never seen" from "seen, and deleted".
		// Every other door asks AccountBySqid and gets nothing for a marked
		// row; here, a marked row is the console's MEMORY of a tombstone, and
		// forgetting it is a takeover (see the switch below).
		a, deleted, err := st.AccountAnyBySqid(sqid)
		if err != nil {
			return n, err
		}
		// Read UNCONDITIONALLY now, including for accounts we already know.
		// This is the accepted cost of reaping: tombstone state lives in the
		// manifest BODY, so the old "known, stop here before the GET"
		// short-circuit would make a deletion performed elsewhere invisible
		// here forever. One small GET per account, at startup.
		body, _, err := s.Get(ctx, key)
		if err != nil {
			log.Printf("console: skipping %s: read failed: %v", key, err)
			continue
		}
		var m accountManifest
		if err := json.Unmarshal(body, &m); err != nil || m.ID <= 0 || m.Sqid == "" {
			log.Printf("console: skipping %s: malformed account manifest: %v", key, err)
			continue
		}
		if m.Sqid != sqid || domains.Sqid(m.ID) != sqid {
			log.Printf("console: skipping %s: sqid %q/%q is not Sqid(%d)=%q — that is corruption, "+
				"not a new account", key, sqid, m.Sqid, m.ID, domains.Sqid(m.ID))
			continue
		}
		// A live account holding a reserved sqid (#285). The allocator never
		// mints one, so this fires only if the list grows over an existing
		// account or a manifest was written by hand. The account is left alone,
		// as the ruling says; this line is what makes it visible on every
		// console start instead of surfacing as broken DNS.
		if !m.Deleted() && domains.IsReservedSqid(m.Sqid) {
			log.Printf("console: account %s (id %d) has a reserved sqid; its hostnames under the apps domain "+
				"collide with platform infrastructure (#285)", m.Sqid, m.ID)
		}
		// Everything the row and the manifest can say to each other. Only the
		// last case — never seen here, and live — falls through to birth.
		switch {
		case deleted && m.Deleted():
			continue // both sides agree the account is retired; nothing to do

		case deleted && !m.Deleted():
			// A marked row is tombstone MEMORY, and memory does not argue with
			// a rewritten manifest. Re-materializing here — let alone
			// re-adopting the owner the manifest now names — would hand the
			// account, its sqid, its platform hostnames and its whole apps
			// prefix to whoever holds bucket write access, with no signing key
			// involved. The manifest is not write-once (see materializeOwner),
			// so "reap, rewrite, re-birth" would be a complete account
			// takeover available to anyone who can PUT one object.
			//
			// An account id is spent once. This is corruption, not an
			// un-deletion, and it is logged as such.
			log.Printf("console: manifest %s is live again but the account was deleted here — "+
				"refusing to resurrect it; an account id is spent once", key)
			continue

		case a != nil && m.Deleted():
			// The control plane has ruled; converge on it. This is what makes a
			// deletion performed anywhere — another box, `carlos accounts
			// delete`, an operator with only bucket access — become real here
			// without hand-run SQL (the Kass gap).
			if err := st.ReapAccount(a.ID, time.Now()); err != nil {
				log.Printf("console: reap deleted account %s: %v", sqid, err)
			} else {
				log.Printf("console: reaped account %s (%s) — its manifest is tombstoned", sqid, a.Name)
			}
			continue

		case a != nil:
			// Known and live: settled. INSERT-only, and — the security
			// boundary — owner adoption is never revisited for an account the
			// console has seen before.
			continue

		case m.Deleted():
			// Never seen here and already retired: it stays unseen. Without
			// this, deletion would be a lie that lasts until the next console
			// start on a fresh box — the manifest would be read, no row found,
			// and the account re-created together with the owner membership it
			// names, which after an ownership transfer is not even the person
			// who deleted it.
			continue
		}

		// FIRST-EVER materialization from here down. Nothing above may reach
		// EnsureAccount or materializeOwner: birth happens once per account id,
		// for as long as this database lives.
		created := m.CreatedAt
		if created.IsZero() {
			created = time.Now()
		}
		added, err := st.EnsureAccount(m.ID, m.Sqid, m.Name, created)
		if err != nil {
			return n, err
		}
		if !added {
			// The sqid was free but the id was not: some other row already
			// holds it. Loud, because two accounts cannot share an id and one
			// of these two is wrong.
			log.Printf("console: did not materialize %s: account id %d is already taken here", key, m.ID)
			continue
		}
		// The manifest's owner, when it names one, becomes the account's first
		// membership. This is what makes `carlos accounts create -owner <addr>`
		// (and a logged-in CLI's create) reach the console at all: without it
		// the account materializes with no members, and an account with no
		// members is one nobody can open — the exact failure this pass used to
		// leave behind. Only on the FIRST materialization, and only for an
		// address the sign-in form would accept: a typo'd owner is a grant
		// nobody could ever claim, and inventing it would just be a stranded
		// row with a confusing name on it.
		if m.Owner != "" {
			if !ValidAddress(m.Owner) {
				log.Printf("console: account %s names owner %q, which is not a keymail address — "+
					"nobody could sign in as it, so no membership was created. Fix it with: "+
					"console member --add <address> %s --role owner", m.Sqid, m.Owner, m.Sqid)
			} else if err := st.AddMember(m.ID, strings.ToLower(m.Owner), RoleOwner, created); err != nil {
				// Loud but not fatal: the account itself materialized, and an
				// operator can still add the owner box-side. Failing the whole
				// pass here would hide every other account behind one bad row.
				log.Printf("console: account %s materialized but its owner %s could not be recorded: %v — "+
					"add them with: console member --add %s %s --role owner", m.Sqid, m.Owner, err, m.Owner, m.Sqid)
			}
		}
		log.Printf("console: materialized account %s (%s) from the bucket", m.Sqid, m.Name)
		n++
		// ONLY here — on the pass that created the row, never on a later one.
		materializeOwner(st, key, m, created)
	}
	return n, nil
}

// materializeOwner turns the manifest's `owner` into the account's first owner
// membership — the console half of the fix for `carlos accounts create` making
// unreachable accounts (see Store.AdoptManifestOwner for the production failure
// this comes from).
//
// IT RUNS EXACTLY ONCE PER ACCOUNT: on the pass that first creates the console's
// row, and never again. That is the whole security argument for the field, and
// it is deliberately structural rather than a check inside this function.
//
// The reason is that the manifest is NOT write-once. MintAccount claims the
// object with a create-only Put, but that is a convention of one code path, not
// a property of the object: release.Store.Put takes its ifMatch from the caller,
// the deployment bucket has no Object Lock, and the host IAM role holds
// unconditional s3:PutObject across the bucket with nothing excluding
// accounts/*/account.json. Anyone who can write to the bucket can rewrite a live
// account's manifest.
//
// If adoption happened on any pass, that rewrite would BE an authority grant:
// name yourself owner of a vacant account, wait for the next console start or
// `console member` invocation — both run this pass — and you own a live account
// and everything deployed in it. That is a real widening of the trust model, not
// something bucket write access already implied: forging a channel pointer needs
// CARLOS_RELEASE_KEY, which most bucket-writers do not hold, and this path needs
// no key at all. AdoptManifestOwner's transactional vacancy check does not cover
// it either — it is TOCTOU-safe, but the account was LEGITIMATELY vacant and it
// is the manifest that changed afterwards.
//
// Adopting only at birth is safe precisely because `-owner` is now REQUIRED on
// `carlos accounts create`: manifests are born carrying their owner, so there is
// no legitimate case of one arriving later.
//
// THE TRADE-OFF, stated plainly: an owner added to a manifest AFTER the account
// first materialized is deliberately NOT adopted, however well-meant the edit.
// Editing the bucket is not how you grant console authority; `console member
// -add <address> <sqid> -role owner` is, and it is box-side for exactly this
// reason. The same applies to an account that materialized ownerless from a
// pre-`owner` manifest — it stays ownerless until an operator says otherwise.
//
// Identity and authority stay SEPARATE decisions, which is why this runs after
// the account row rather than inside EnsureAccount: a manifest with an owner
// keymail could never authenticate still yields a perfectly good account. One
// unusable string must not hide an account from the console entirely.
//
// Best-effort and never fatal, like the rest of this pass: an account that
// materialized but could not be given its owner is recoverable with `console
// member -add`, whereas a console that refuses to boot is not.
func materializeOwner(st *Store, key string, m accountManifest, created time.Time) {
	owner := domains.NormalizeAddress(m.Owner)
	if owner == "" {
		return // ownerless by design: every manifest written before the field existed
	}
	if !domains.ValidAddress(owner) {
		log.Printf("console: %s names owner %q, which is not a keymail address — "+
			"materializing the account without it; give it one with `console member --add <address> %s --role owner`",
			key, m.Owner, m.Sqid)
		return
	}
	adopted, err := st.AdoptManifestOwner(m.ID, owner, created)
	if err != nil {
		log.Printf("console: could not adopt %s as owner of %s: %v", owner, m.Sqid, err)
		return
	}
	if adopted {
		log.Printf("console: %s is the owner of %s (%s), from the account manifest", owner, m.Sqid, m.Name)
	}
}

// MaterializeApps converges the console's app rows onto the bucket, which is
// the control plane for trash state exactly as it is for account identity.
//
// Trash and purge both happen where this console is not: `carlos` running
// headless on a box, a second console, the auto-purge pass sweeping cooled-off
// markers. Each writes the bucket; none opens this database. So the console
// catches up the same way it catches up on accounts — by reading the objects.
//
// Four cases, decided per row:
//
//	(a) live row + valid marker      → mark it trashed (trashed elsewhere)
//	(b) marked row + no marker + keys → un-mark it (restored elsewhere)
//	(c) marked row + no marker + no keys → DELETE the row (purged elsewhere);
//	    the row IS the name reservation, so this is where the name frees
//	(d) live row + no marker + no keys → NOTHING.
//
// Case (d) is the guard the whole design turns on: an app claimed but never
// shipped has no objects, which makes it byte-for-byte indistinguishable from
// a purged one in the bucket. The only thing that tells them apart is the row's
// own deleted_at — so convergence may only ever destroy a row that was ALREADY
// trashed here. PurgeAppRow enforces the same rule in SQL.
//
// Row-driven, not bucket-driven: a purged app leaves no keys to list, so the
// listing alone can never raise case (c). Keys under an apps/<sqid>/ prefix
// whose sqid this console does not know contribute nothing to anything —
// another deployment's tenants are not this console's business.
//
// Case (c) is the only one that reasons from absence, and it is held back
// entirely when the apps/ listing comes back EMPTY — see the tripwire below,
// which is what stops a mistyped bucket name from releasing every held name.
//
// Best-effort like the rest of this file: a nil store or a failed listing is
// "nothing to catch up on", never a reason to stop the console.
func MaterializeApps(ctx context.Context, s release.Store, st *Store, claimEnabled ...bool) error {
	if s == nil {
		return nil
	}
	rows, err := st.AppRows()
	if err != nil {
		return err
	}
	if len(rows) == 0 {
		return nil
	}
	keys, err := s.List(ctx, "apps/")
	if err != nil {
		return err
	}
	// THE EMPTY-BUCKET TRIPWIRE. Case (c) reasons from ABSENCE, and absence is
	// exactly what a mistyped CARLOS_DEPLOYMENT_BUCKET produces: a listing that
	// succeeds and returns nothing. Without this, one wrong environment
	// variable purges every trashed row on the first pass and frees every held
	// name — unrecoverably, because this pass is row-driven and a marker with
	// no row is invisible to it forever.
	//
	// So a completely empty apps/ listing is treated as no evidence rather
	// than as evidence of destruction: cases (a), (b) and (d) act on PRESENCE
	// and are unaffected, and only the purges are held back.
	//
	// The accepted corner: a deployment whose every app really was purged
	// keeps its trashed names reserved until anything ships again. A
	// reservation outliving its need is a small annoyance; names released
	// because the console was pointed at the wrong bucket are gone.
	//
	// Trash markers do not count as evidence (bucketEvidence, settle.go):
	// they are the one thing the temporary-app pass writes into a bucket.
	purgesHeld := !bucketEvidence(keys)
	if purgesHeld {
		trashed := 0
		for _, r := range rows {
			if r.Deleted {
				trashed++
			}
		}
		if trashed == 0 {
			purgesHeld = false // nothing to hold back; stay quiet
		} else {
			log.Printf("console: the apps/ listing is empty but %d trashed row(s) exist here — "+
				"refusing to release their names; is CARLOS_DEPLOYMENT_BUCKET pointing at the right bucket?", trashed)
		}
	}
	// One listing, indexed by app prefix. The prefixes come from the ROWS, so
	// an unknown account's keys simply never find a home here.
	type bucketState struct{ anyKeys, marker bool }
	state := make(map[string]*bucketState, len(rows))
	for _, r := range rows {
		state["apps/"+r.Sqid+"/"+r.App+"/"] = &bucketState{}
	}
	for _, key := range keys {
		parts := strings.SplitN(key, "/", 4)
		if len(parts) < 4 || parts[0] != "apps" {
			continue // not an app-scoped key: apps/<sqid>/<app>/<something>
		}
		bs := state["apps/"+parts[1]+"/"+parts[2]+"/"]
		if bs == nil {
			continue
		}
		// Any object under the prefix counts, including a bare directory
		// placeholder: convergence fails toward KEEPING the row.
		bs.anyKeys = true
		if key == release.AppTombstoneKey(parts[1], parts[2]) {
			bs.marker = true
		}
	}
	for _, r := range rows {
		if len(claimEnabled) > 0 && claimEnabled[0] {
			handled, claimErr := materializeClaimApp(ctx, s, st, r.AccountID, r.Sqid, r.App, time.Now())
			if claimErr != nil {
				log.Printf("console: hold app materialization %s/%s: %v", r.Sqid, r.App, claimErr)
			}
			if handled {
				continue
			}
		}
		bs := state["apps/"+r.Sqid+"/"+r.App+"/"]
		switch {
		case bs.marker:
			// Re-read rather than trusting the listing: it is one GET, and a
			// marker is an instruction to hide an app. ReadAppTombstone
			// VALIDATES — a body that disagrees with its path, or carries no
			// timestamp, is not something to act on either way.
			tomb, err := release.ReadAppTombstone(ctx, s, r.Sqid, r.App)
			if err != nil {
				log.Printf("console: leaving %s/%s alone: %v", r.Sqid, r.App, err)
				continue
			}
			if tomb == nil {
				continue // restored between the listing and the read
			}
			if r.Deleted {
				// A restore that crashed before deleting the marker: release
				// its cap slot; the app stays in the Trash (spec §6).
				if r.Restoring {
					if _, err := st.SetRestoringContext(ctx, r.AccountID, r.App, false); err != nil {
						log.Printf("console: clear restoring on %s/%s: %v", r.Sqid, r.App, err)
					}
				}
				continue
			}
			// The clock comes from the MARKER. TrashAppRow also sets the
			// 3-day window when the row is temporary (spec §5 "Materialize").
			if err := st.TrashAppRowContext(ctx, r.AccountID, r.App, tomb.DeletedAt); err != nil {
				log.Printf("console: mark %s/%s trashed: %v", r.Sqid, r.App, err)
				continue
			}
			log.Printf("console: %s/%s is in the trash in the bucket — marking it trashed here", r.Sqid, r.App)

		case !r.Deleted:
			continue // (d), and the ordinary live app: no marker, nothing to do

		case bs.anyKeys:
			// Restored elsewhere (a headless restore), or a console restore
			// that failed its last step. A temporary row gets restore step 2
			// here too (spec §6).
			if err := st.FinishRestoreRow(r.AccountID, r.App, time.Now()); err != nil {
				log.Printf("console: restore %s/%s: %v", r.Sqid, r.App, err)
				continue
			}
			log.Printf("console: %s/%s has no trash marker and still has objects — restoring it here", r.Sqid, r.App)

		default:
			// Purged: marker gone, objects gone, and the row was already
			// trashed here — so this frees the name rather than destroying
			// anything. PurgeAppRow is guarded to refuse a live row even if
			// this reasoning is ever wrong.
			if purgesHeld {
				continue // an empty listing is no evidence; see the tripwire above
			}
			if err := st.PurgeAppRow(r.AccountID, r.App); err != nil {
				log.Printf("console: purge %s/%s: %v", r.Sqid, r.App, err)
				continue
			}
			log.Printf("console: %s/%s is gone from the bucket — releasing the name here", r.Sqid, r.App)
		}
	}
	return nil
}

// materializeQuietly is the call site every entry point actually wants: catch
// up if we can, log and carry on if we cannot. Separated from
// MaterializeAccounts so the failure policy lives in ONE place rather than
// being re-decided by the server, the member verb and whatever comes next.
func materializeQuietly(ctx context.Context, s release.Store, st *Store, claimEnabled ...bool) {
	if _, err := MaterializeAccounts(ctx, s, st); err != nil {
		log.Printf("console: could not materialize accounts from the bucket: %v", err)
	}
	// After accounts, always: a reaped account's app rows are gone with it, so
	// this pass then has nothing to say about them.
	if err := MaterializeApps(ctx, s, st, claimEnabled...); err != nil {
		log.Printf("console: could not converge app rows with the bucket: %v", err)
	}
}