package main

import (
	"database/sql"
	"time"
)

// cloudDB is the provisioner's state: which callers (home servers) may
// provision, and which spaces each caller's users already have (the quota
// ledger). It never holds message data, keys, or anything about a space beyond
// its hostname — the requester ids are the home server's opaque random user
// ids, meaningless outside it.
type cloudDB struct{ sql *sql.DB }

func openCloudDB(path string) (*cloudDB, error) {
	d, err := sql.Open("sqlite", path)
	if err != nil {
		return nil, err
	}
	// One writer at a time keeps WAL happy (mirrors openDB).
	d.SetMaxOpenConns(1)
	for _, p := range []string{
		"PRAGMA journal_mode=WAL",
		"PRAGMA busy_timeout=5000",
		"PRAGMA synchronous=NORMAL",
	} {
		if _, err := d.Exec(p); err != nil {
			return nil, err
		}
	}
	db := &cloudDB{sql: d}
	if err := db.migrate(); err != nil {
		return nil, err
	}
	return db, nil
}

// migrate is additive, like every other schema in this repo, so deploying new
// code over an existing DB is safe.
func (db *cloudDB) migrate() error {
	if _, err := db.sql.Exec(`
-- Who may provision: bearer tokens minted for home servers (cf. the push
-- relay's relay_tokens). Stored hashed — a stolen DB yields no live credential.
CREATE TABLE IF NOT EXISTS caller_tokens (
  id           TEXT PRIMARY KEY,
  label        TEXT NOT NULL,
  token_hash   TEXT NOT NULL UNIQUE,
  created_at   INTEGER NOT NULL,
  last_used_at INTEGER
);
-- Provisioned spaces: the quota ledger AND the id source for a space's hostname.
-- The integer id is the primary key; the subdomain label is sqids(id) (see
-- cloud.go), so the hostname encodes the id reversibly. host is assigned only
-- after the id is known (it's NULL while a create is in flight), so it can't be
-- the PK. requester is the caller's opaque user id (a home server's random uid).
CREATE TABLE IF NOT EXISTS spaces (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  host       TEXT UNIQUE,
  caller_id  TEXT NOT NULL,
  requester  TEXT NOT NULL,
  created_at INTEGER NOT NULL
);
-- Every allocation, forever (rows here survive space deletion): the global
-- creation budget counts THIS, so create-and-delete can't reset the throttle.
CREATE TABLE IF NOT EXISTS creates_log (
  created_at INTEGER NOT NULL
);`); err != nil {
		return err
	}
	if err := db.migrateSpacesID(); err != nil {
		return err
	}
	if err := db.migrateBilling(); err != nil {
		return err
	}
	if err := db.migrateMeter(); err != nil {
		return err
	}
	if err := db.migrateUsage(); err != nil {
		return err
	}
	return db.migratePool()
}

// migrateUsage (#857): the per-space usage ledger — quantities only (token
// counts, request counts, run counts), never content — plus the meter token
// each instance authenticates its reports with. The token itself is pushed
// into the instance env (LCHAT_METER_URL/_TOKEN) by the sweep; only its hash
// lives here, same posture as caller tokens.
func (db *cloudDB) migrateUsage() error {
	if has, err := db.spacesHasColumn("meter_token_hash"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`ALTER TABLE spaces ADD COLUMN meter_token_hash TEXT NOT NULL DEFAULT ''`); err != nil {
			return err
		}
	}
	_, err := db.sql.Exec(`CREATE TABLE IF NOT EXISTS usage (
  space_id  INTEGER NOT NULL,
  day       TEXT NOT NULL,
  model     TEXT NOT NULL DEFAULT '',
  dimension TEXT NOT NULL,
  quantity  INTEGER NOT NULL DEFAULT 0,
  PRIMARY KEY (space_id, day, model, dimension)
);
-- Applied report ids — the dedupe ledger that makes ingest idempotent. A
-- sidecar resends a report verbatim when an ack is lost; the id tells a
-- replay from new usage. Pruned after a retention window (the sidecar
-- retries on a minutes scale, so days of memory is generous).
CREATE TABLE IF NOT EXISTS usage_reports (
  space_id   INTEGER NOT NULL,
  report_id  TEXT NOT NULL,
  applied_at INTEGER NOT NULL,
  PRIMARY KEY (space_id, report_id)
)`)
	return err
}

// setMeterTokenHash records the hash of the meter token last pushed to a
// space's instance. Overwriting is the rotation path: last push wins.
func (db *cloudDB) setMeterTokenHash(id int64, hash string) error {
	_, err := db.sql.Exec(`UPDATE spaces SET meter_token_hash = ? WHERE id = ?`, hash, id)
	return err
}

// spaceByMeterToken resolves a usage report's bearer token to its space. An
// empty stored hash never matches (hashToken of any presented token is
// non-empty), so unmetered spaces can't be reported against.
func (db *cloudDB) spaceByMeterToken(token string) (int64, bool) {
	var id int64
	err := db.sql.QueryRow(
		`SELECT id FROM spaces WHERE meter_token_hash = ? AND meter_token_hash != ''`,
		hashToken(token)).Scan(&id)
	return id, err == nil
}

// addUsage folds one report into the ledger — additive upserts, so repeated
// partial-day reports accumulate instead of clobbering. Idempotent on
// (space, report id): the report row and the quantity upserts share one
// transaction, so a replay of an already-applied report (ack lost after
// commit) is recognized and dropped whole.
func (db *cloudDB) addUsage(spaceID int64, reportID string, rows []botUsageRow) error {
	tx, err := db.sql.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	res, err := tx.Exec(`INSERT INTO usage_reports(space_id, report_id, applied_at)
VALUES(?, ?, ?) ON CONFLICT(space_id, report_id) DO NOTHING`,
		spaceID, reportID, time.Now().Unix())
	if err != nil {
		return err
	}
	if n, err := res.RowsAffected(); err != nil {
		return err
	} else if n == 0 {
		return nil // replay of an applied report — already counted
	}
	for _, r := range rows {
		if _, err := tx.Exec(`INSERT INTO usage(space_id, day, model, dimension, quantity)
VALUES(?, ?, ?, ?, ?)
ON CONFLICT(space_id, day, model, dimension) DO UPDATE SET quantity = quantity + excluded.quantity`,
			spaceID, r.Day, r.Model, r.Dimension, r.Quantity); err != nil {
			return err
		}
	}
	return tx.Commit()
}

// pruneUsageReports drops applied-report ids older than the cutoff. The
// sidecar's retry horizon is minutes, so days of dedupe memory is generous.
func (db *cloudDB) pruneUsageReports(cutoff int64) error {
	_, err := db.sql.Exec(`DELETE FROM usage_reports WHERE applied_at < ?`, cutoff)
	return err
}

// usageForMonth reads one space's ledger rows for a UTC month ("2006-01") —
// the budget's allowance window (cloud_budget.go).
func (db *cloudDB) usageForMonth(spaceID int64, month string) ([]botUsageRow, error) {
	rows, err := db.sql.Query(
		`SELECT day, model, dimension, quantity FROM usage WHERE space_id = ? AND day LIKE ?`,
		spaceID, month+"-%")
	if err != nil {
		return nil, err
	}
	return scanUsageRows(rows)
}

// fleetUsageForMonth sums every space's rows for a month — the pool's view.
func (db *cloudDB) fleetUsageForMonth(month string) ([]botUsageRow, error) {
	rows, err := db.sql.Query(
		`SELECT day, model, dimension, SUM(quantity) FROM usage WHERE day LIKE ?
		 GROUP BY day, model, dimension`, month+"-%")
	if err != nil {
		return nil, err
	}
	return scanUsageRows(rows)
}

// usageFor reads a space's ledger rows (tests + the future usage surface).
func (db *cloudDB) usageFor(spaceID int64) ([]botUsageRow, error) {
	rows, err := db.sql.Query(
		`SELECT day, model, dimension, quantity FROM usage WHERE space_id = ? ORDER BY day, model, dimension`, spaceID)
	if err != nil {
		return nil, err
	}
	return scanUsageRows(rows)
}

func scanUsageRows(rows *sql.Rows) ([]botUsageRow, error) {
	defer rows.Close()
	var out []botUsageRow
	for rows.Next() {
		var r botUsageRow
		if err := rows.Scan(&r.Day, &r.Model, &r.Dimension, &r.Quantity); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// migrateSpacesID rebuilds a spaces table created under the original host-PK
// schema (no integer id) into the id-keyed one above. AUTOINCREMENT means a
// torn-down space's id is never reused, so its sqid subdomain can't be handed to
// a different space later. A no-op once the id column exists.
func (db *cloudDB) migrateSpacesID() error {
	rows, err := db.sql.Query(`PRAGMA table_info(spaces)`)
	if err != nil {
		return err
	}
	hasID := false
	for rows.Next() {
		var cid, notnull, pk int
		var name, ctype string
		var dflt sql.NullString
		if err := rows.Scan(&cid, &name, &ctype, &notnull, &dflt, &pk); err != nil {
			rows.Close()
			return err
		}
		if name == "id" {
			hasID = true
		}
	}
	rows.Close()
	if hasID {
		return nil
	}
	// Preserve any existing rows (they get fresh sequential ids in created_at order).
	_, err = db.sql.Exec(`
ALTER TABLE spaces RENAME TO spaces_old;
CREATE TABLE spaces (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  host       TEXT UNIQUE,
  caller_id  TEXT NOT NULL,
  requester  TEXT NOT NULL,
  created_at INTEGER NOT NULL
);
INSERT INTO spaces (host, caller_id, requester, created_at)
  SELECT host, caller_id, requester, created_at FROM spaces_old ORDER BY created_at;
DROP TABLE spaces_old;`)
	return err
}

// spacesHasColumn reports whether the spaces table already carries a column —
// the guard that keeps the billing migrations additive and re-runnable.
func (db *cloudDB) spacesHasColumn(column string) (bool, error) {
	rows, err := db.sql.Query(`PRAGMA table_info(spaces)`)
	if err != nil {
		return false, err
	}
	defer rows.Close()
	for rows.Next() {
		var cid, notnull, pk int
		var name, ctype string
		var dflt sql.NullString
		if err := rows.Scan(&cid, &name, &ctype, &notnull, &dflt, &pk); err != nil {
			return false, err
		}
		if name == column {
			return true, nil
		}
	}
	return false, rows.Err()
}

// migrateBilling adds the billing columns to spaces (see billing.go). This is
// the ENTIRE billing data footprint, by design: two opaque Stripe ids, a flag,
// a tier key, and a period end — never an email, name, or card detail (Stripe
// holds those; see docs/cloud-provisioner.md § Billing).
func (db *cloudDB) migrateBilling() error {
	if has, err := db.spacesHasColumn("entitlement"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`
ALTER TABLE spaces ADD COLUMN stripe_customer_id TEXT;
ALTER TABLE spaces ADD COLUMN stripe_subscription_id TEXT;
ALTER TABLE spaces ADD COLUMN entitlement TEXT NOT NULL DEFAULT 'free';
ALTER TABLE spaces ADD COLUMN paid_until INTEGER NOT NULL DEFAULT 0;`); err != nil {
			return err
		}
	}
	// tier landed one change after the columns above, so it gets its own guard
	// (a DB migrated in between has entitlement but not tier).
	if has, err := db.spacesHasColumn("tier"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`ALTER TABLE spaces ADD COLUMN tier TEXT NOT NULL DEFAULT ''`); err != nil {
			return err
		}
	}
	// cancel_at_period_end: the payer has cancelled and the subscription runs
	// out at paid_until — home shows "ends <date>" on the row.
	if has, err := db.spacesHasColumn("cancel_at_period_end"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`ALTER TABLE spaces ADD COLUMN cancel_at_period_end INTEGER NOT NULL DEFAULT 0`); err != nil {
			return err
		}
	}
	// band: which honor price the subscription is on (docs/pricing-bands.md) —
	// base/supporter/reduced/solidarity. '' on pre-band rows reads as base.
	if has, err := db.spacesHasColumn("band"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`ALTER TABLE spaces ADD COLUMN band TEXT NOT NULL DEFAULT ''`); err != nil {
			return err
		}
	}
	// seats: the head count a per-seat subscription currently bills, mirrored
	// from Stripe's own subscription item (docs/tier-economics.md § Per-seat).
	// 0 on flat tiers and pre-seat rows; the renewal true-up compares it
	// against the live member count.
	if has, err := db.spacesHasColumn("seats"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`ALTER TABLE spaces ADD COLUMN seats INTEGER NOT NULL DEFAULT 0`); err != nil {
			return err
		}
	}
	// Pay-before-create (the tier picker in the create flow): the Checkout
	// Session bound to a pending allocation, and the name chosen at create —
	// held here (never in Stripe metadata) until the claim provisions it.
	if has, err := db.spacesHasColumn("checkout_session_id"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`
ALTER TABLE spaces ADD COLUMN checkout_session_id TEXT;
ALTER TABLE spaces ADD COLUMN pending_name TEXT NOT NULL DEFAULT ''`); err != nil {
			return err
		}
	}
	return nil
}

// migrateMeter adds the storage meter's columns (#668 slice 1): the byte count
// controld last measured and when. Bytes of ciphertext — content-blind, like
// everything in this ledger.
func (db *cloudDB) migrateMeter() error {
	if has, err := db.spacesHasColumn("storage_bytes"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`
ALTER TABLE spaces ADD COLUMN storage_bytes INTEGER NOT NULL DEFAULT 0;
ALTER TABLE spaces ADD COLUMN metered_at INTEGER NOT NULL DEFAULT 0;`); err != nil {
			return err
		}
	}
	// quota_bytes landed one change later (enforcement): what the cloud last
	// pushed into the instance's LCHAT_BLOB_QUOTA_BYTES. 0 = never pushed.
	if has, err := db.spacesHasColumn("quota_bytes"); err != nil {
		return err
	} else if !has {
		if _, err := db.sql.Exec(`ALTER TABLE spaces ADD COLUMN quota_bytes INTEGER NOT NULL DEFAULT 0`); err != nil {
			return err
		}
	}
	// The sweep also records each space's claimed-ness (setup_complete /
	// members), so clients can tell an unusable shell from a live space with
	// server truth instead of client-held credentials. Defaults match a fresh
	// create exactly: never set up, one pre-seeded admin.
	if has, err := db.spacesHasColumn("setup_complete"); err != nil || has {
		return err
	}
	_, err := db.sql.Exec(`
ALTER TABLE spaces ADD COLUMN setup_complete INTEGER NOT NULL DEFAULT 0;
ALTER TABLE spaces ADD COLUMN members INTEGER NOT NULL DEFAULT 1;`)
	return err
}

// mintCallerToken creates a caller token, stores only its hash, and returns the
// plaintext (shown once).
func (db *cloudDB) mintCallerToken(label string) (id, token string, err error) {
	if token, err = randToken(); err != nil {
		return "", "", err
	}
	if id, err = randToken(); err != nil {
		return "", "", err
	}
	_, err = db.sql.Exec(
		`INSERT INTO caller_tokens(id, label, token_hash, created_at) VALUES(?, ?, ?, ?)`,
		id, label, hashToken(token), time.Now().Unix(),
	)
	return id, token, err
}

// authCaller resolves a bearer token to its caller id, bumping last_used_at.
func (db *cloudDB) authCaller(token string) (id string, ok bool) {
	if token == "" {
		return "", false
	}
	if db.sql.QueryRow(
		`SELECT id FROM caller_tokens WHERE token_hash = ?`, hashToken(token),
	).Scan(&id) != nil {
		return "", false
	}
	// Best-effort; a failed bump doesn't deny an otherwise-valid token.
	_, _ = db.sql.Exec(`UPDATE caller_tokens SET last_used_at = ? WHERE id = ?`, time.Now().Unix(), id)
	return id, true
}

// CallerToken is a token row for the admin listing — never the token itself.
type CallerToken struct {
	ID         string `json:"id"`
	Label      string `json:"label"`
	CreatedAt  int64  `json:"created_at"`
	LastUsedAt *int64 `json:"last_used_at"`
}

func (db *cloudDB) listCallerTokens() ([]CallerToken, error) {
	rows, err := db.sql.Query(`SELECT id, label, created_at, last_used_at FROM caller_tokens ORDER BY created_at`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := []CallerToken{}
	for rows.Next() {
		var t CallerToken
		if err := rows.Scan(&t.ID, &t.Label, &t.CreatedAt, &t.LastUsedAt); err != nil {
			return nil, err
		}
		out = append(out, t)
	}
	return out, rows.Err()
}

func (db *cloudDB) revokeCallerToken(id string) (bool, error) {
	res, err := db.sql.Exec(`DELETE FROM caller_tokens WHERE id = ?`, id)
	if err != nil {
		return false, err
	}
	n, _ := res.RowsAffected()
	return n > 0, nil
}

// allocateSpace reserves the next id for a (caller, requester) if they're under
// quota, atomically: the quota count and the insert share one transaction (and
// SetMaxOpenConns(1) serializes writers), so two concurrent creates can't both
// slip past the limit. The row is inserted with a NULL host — the caller turns
// the returned id into a hostname via sqids, then finalizeSpace records it.
// overQuota=true (with a zero id) means the request is at/over the limit.
//
// Only FREE spaces count against the quota: a paid space carries itself
// (per-space pricing, #562), so upgrading one frees its slot. Pending creates
// (NULL host) are free by default and still count — they hold a slot until
// they finalize or roll back.
func (db *cloudDB) allocateSpace(callerID, requester string, quota int) (id int64, overQuota bool, err error) {
	tx, err := db.sql.Begin()
	if err != nil {
		return 0, false, err
	}
	defer tx.Rollback()
	var n int
	if err := tx.QueryRow(
		`SELECT COUNT(*) FROM spaces WHERE caller_id = ? AND requester = ? AND entitlement = 'free'`,
		callerID, requester,
	).Scan(&n); err != nil {
		return 0, false, err
	}
	if n >= quota {
		return 0, true, nil
	}
	now := time.Now().Unix()
	res, err := tx.Exec(
		`INSERT INTO spaces(host, caller_id, requester, created_at) VALUES(NULL, ?, ?, ?)`,
		callerID, requester, now,
	)
	if err != nil {
		return 0, false, err
	}
	if _, err := tx.Exec(`INSERT INTO creates_log(created_at) VALUES(?)`, now); err != nil {
		return 0, false, err
	}
	if id, err = res.LastInsertId(); err != nil {
		return 0, false, err
	}
	return id, false, tx.Commit()
}

// allocateDemoSpace reserves a demo row (docs/demo-spaces.md). Demos never
// count against — and are never counted by — the free-space quota; their own
// bounds are the live-demo cap (counted in cloud.go before this runs) and the
// shared creates_log budget, which they do feed: a demo create is still a
// create.
func (db *cloudDB) allocateDemoSpace(callerID, requester string) (int64, error) {
	tx, err := db.sql.Begin()
	if err != nil {
		return 0, err
	}
	defer tx.Rollback()
	now := time.Now().Unix()
	res, err := tx.Exec(
		`INSERT INTO spaces(host, caller_id, requester, created_at, entitlement) VALUES(NULL, ?, ?, ?, ?)`,
		callerID, requester, now, entDemo,
	)
	if err != nil {
		return 0, err
	}
	if _, err := tx.Exec(`INSERT INTO creates_log(created_at) VALUES(?)`, now); err != nil {
		return 0, err
	}
	id, err := res.LastInsertId()
	if err != nil {
		return 0, err
	}
	return id, tx.Commit()
}

// allocatePendingPaidSpace reserves a pay-first row (the tier picker's paid
// branch): entitlement 'pending' keeps it outside the free quota and every
// sweep, and the chosen display name waits here for the claim. It still feeds
// creates_log — minting checkouts is a create for the budget breaker's
// purposes, or session spam would sidestep it.
func (db *cloudDB) allocatePendingPaidSpace(callerID, requester, name string) (int64, error) {
	tx, err := db.sql.Begin()
	if err != nil {
		return 0, err
	}
	defer tx.Rollback()
	now := time.Now().Unix()
	res, err := tx.Exec(
		`INSERT INTO spaces(host, caller_id, requester, created_at, entitlement, pending_name) VALUES(NULL, ?, ?, ?, ?, ?)`,
		callerID, requester, now, entPending, name,
	)
	if err != nil {
		return 0, err
	}
	if _, err := tx.Exec(`INSERT INTO creates_log(created_at) VALUES(?)`, now); err != nil {
		return 0, err
	}
	id, err := res.LastInsertId()
	if err != nil {
		return 0, err
	}
	return id, tx.Commit()
}

// setCheckoutSession binds the Stripe Checkout Session to its pending row —
// audit and debugging; the claim's authority is the session Stripe returns.
func (db *cloudDB) setCheckoutSession(id int64, session string) error {
	_, err := db.sql.Exec(`UPDATE spaces SET checkout_session_id = ? WHERE id = ?`, session, id)
	return err
}

// pendingSpace reads the claim path's view of one allocation: who it belongs
// to, whether it's provisioned yet, and the name chosen at create.
type pendingSpace struct {
	ID          int64
	CallerID    string
	Requester   string
	Host        string // "" until provisioned
	Entitlement string
	PendingName string
}

func (db *cloudDB) pendingSpaceByID(id int64) (pendingSpace, bool, error) {
	p := pendingSpace{ID: id}
	err := db.sql.QueryRow(
		`SELECT caller_id, requester, COALESCE(host, ''), entitlement, pending_name FROM spaces WHERE id = ?`, id,
	).Scan(&p.CallerID, &p.Requester, &p.Host, &p.Entitlement, &p.PendingName)
	if err == sql.ErrNoRows {
		return pendingSpace{}, false, nil
	}
	if err != nil {
		return pendingSpace{}, false, err
	}
	return p, true, nil
}

// prunePendingCreates drops pending allocations whose checkout can no longer
// complete (Stripe sessions expire after 24h; the cutoff arrives well past
// that). A PAID row is never pending, so money is never pruned.
func (db *cloudDB) prunePendingCreates(cutoff int64) (int64, error) {
	res, err := db.sql.Exec(
		`DELETE FROM spaces WHERE entitlement = ? AND host IS NULL AND created_at <= ?`, entPending, cutoff)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// countDemos counts every live demo row, pending creates included — the
// live-demo cap must see a create the moment it's reserved, not once it's up.
func (db *cloudDB) countDemos() (int, error) {
	var n int
	err := db.sql.QueryRow(`SELECT COUNT(*) FROM spaces WHERE entitlement = ?`, entDemo).Scan(&n)
	return n, err
}

// demosPastTTL lists every demo past its lifetime — finalized or pending (a
// pending row is an allocation whose create died mid-flight; it holds a demo
// slot and gets cleaned ledger-only).
func (db *cloudDB) demosPastTTL(cutoff int64) ([]reapCandidate, error) {
	rows, err := db.sql.Query(
		`SELECT id, COALESCE(host, ''), created_at FROM spaces
		  WHERE entitlement = ? AND created_at <= ? ORDER BY created_at`, entDemo, cutoff)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []reapCandidate
	for rows.Next() {
		var c reapCandidate
		if err := rows.Scan(&c.ID, &c.Host, &c.CreatedAt); err != nil {
			return nil, err
		}
		out = append(out, c)
	}
	return out, rows.Err()
}

// createsSince counts allocations at/after cutoff — the global budget's meter.
func (db *cloudDB) createsSince(cutoff int64) (int, error) {
	var n int
	err := db.sql.QueryRow(`SELECT COUNT(*) FROM creates_log WHERE created_at >= ?`, cutoff).Scan(&n)
	return n, err
}

// finalizeSpace records the hostname once the instance is created.
func (db *cloudDB) finalizeSpace(id int64, host string) error {
	_, err := db.sql.Exec(`UPDATE spaces SET host = ? WHERE id = ?`, host, id)
	return err
}

// deleteSpaceByID rolls back an allocation whose instance never came up, so a
// failed create doesn't hold a quota slot. (The id is still spent — AUTOINCREMENT
// never reuses it — which is fine; ids are cheap and gaps are harmless.)
func (db *cloudDB) deleteSpaceByID(id int64) error {
	_, err := db.sql.Exec(`DELETE FROM spaces WHERE id = ?`, id)
	return err
}

// reapCandidate is a finalized space old enough for the reaper to check.
type reapCandidate struct {
	ID        int64
	Host      string
	CreatedAt int64
}

// reclaimCandidates returns one account's own free-entitlement rows created
// at/before cutoff, oldest first — the pool the create path may self-reclaim
// from when the quota is hit. Unlike spacesOlderThan this INCLUDES pending
// rows (NULL host, returned as ""): a create that died between allocate and
// finalize holds a quota slot forever otherwise, which at quota 1 would brick
// the account. Paid rows are excluded at the source, same as the reaper.
func (db *cloudDB) reclaimCandidates(callerID, requester string, cutoff int64) ([]reapCandidate, error) {
	rows, err := db.sql.Query(
		`SELECT id, COALESCE(host, ''), created_at FROM spaces
		 WHERE caller_id = ? AND requester = ? AND entitlement = 'free' AND created_at <= ?
		 ORDER BY created_at`, callerID, requester, cutoff)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []reapCandidate
	for rows.Next() {
		var c reapCandidate
		if err := rows.Scan(&c.ID, &c.Host, &c.CreatedAt); err != nil {
			return nil, err
		}
		out = append(out, c)
	}
	return out, rows.Err()
}

// spacesOlderThan returns finalized spaces created at/before cutoff (unix secs),
// oldest first — the reaper's candidate set. A NULL host (create in flight or
// rolled back) is never a candidate, and neither is a paid space: someone paying
// for a space is the strongest possible "claimed" signal, so it's excluded here
// at the source rather than trusted to a downstream guard.
func (db *cloudDB) spacesOlderThan(cutoff int64) ([]reapCandidate, error) {
	rows, err := db.sql.Query(
		`SELECT id, host, created_at FROM spaces
		 WHERE host IS NOT NULL AND created_at <= ? AND entitlement = 'free'
		 ORDER BY created_at`, cutoff)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []reapCandidate
	for rows.Next() {
		var c reapCandidate
		if err := rows.Scan(&c.ID, &c.Host, &c.CreatedAt); err != nil {
			return nil, err
		}
		out = append(out, c)
	}
	return out, rows.Err()
}

// spaceOwned reports whether this exact (caller, requester) provisioned host —
// the ownership check in front of deletion. Nobody can delete a space through
// the provisioner that the provisioner didn't create for them.
func (db *cloudDB) spaceOwned(callerID, requester, host string) (int64, bool, error) {
	var id int64
	err := db.sql.QueryRow(
		`SELECT id FROM spaces WHERE caller_id = ? AND requester = ? AND host = ?`,
		callerID, requester, host,
	).Scan(&id)
	if err == sql.ErrNoRows {
		return 0, false, nil
	}
	if err != nil {
		return 0, false, err
	}
	return id, true, nil
}

// Space is one provisioned instance in a requester's listing.
type Space struct {
	Host        string `json:"host"`
	URL         string `json:"url"`
	CreatedAt   int64  `json:"created_at"`
	Entitlement string `json:"entitlement"`
	Tier        string `json:"tier,omitempty"`
	PaidUntil   int64  `json:"paid_until,omitempty"`
	// The payer cancelled; the subscription (and tier) runs out at PaidUntil.
	CancelAtPeriodEnd bool `json:"cancel_at_period_end,omitempty"`
	// The storage meter's last reading (bytes of ciphertext; 0 = not yet
	// metered), when it was taken, and the cap last pushed to the instance
	// (0 = none — the instance's compiled default applies).
	StorageBytes int64 `json:"storage_bytes,omitempty"`
	MeteredAt    int64 `json:"metered_at,omitempty"`
	StorageCap   int64 `json:"storage_cap,omitempty"`
	// Claimed-ness, as of the last sweep: whether the admin ever finished
	// setup, and the member count (1 = just the pre-seeded admin). Lets the
	// owner's UI recognize an unusable shell with server truth.
	SetupComplete bool `json:"setup_complete"`
	Members       int  `json:"members,omitempty"`
}

// listSpaces returns a requester's finalized spaces (a NULL host is a create
// still in flight or one that failed before rollback — not shown).
func (db *cloudDB) listSpaces(callerID, requester string) ([]Space, error) {
	rows, err := db.sql.Query(
		`SELECT host, created_at, entitlement, tier, paid_until, cancel_at_period_end, storage_bytes, metered_at, quota_bytes, setup_complete, members FROM spaces
		 WHERE caller_id = ? AND requester = ? AND host IS NOT NULL ORDER BY id`,
		callerID, requester,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := []Space{}
	for rows.Next() {
		var s Space
		if err := rows.Scan(&s.Host, &s.CreatedAt, &s.Entitlement, &s.Tier, &s.PaidUntil, &s.CancelAtPeriodEnd, &s.StorageBytes, &s.MeteredAt, &s.StorageCap, &s.SetupComplete, &s.Members); err != nil {
			return nil, err
		}
		s.URL = "https://" + s.Host
		out = append(out, s)
	}
	return out, rows.Err()
}

// ---- billing state (see billing.go) -------------------------------------------

const (
	entFree = "free"
	entPaid = "paid"
	entDemo = "demo" // 24h try-it spaces (docs/demo-spaces.md, demo.go)
	// entPending marks a pay-first allocation awaiting its checkout: not a
	// space yet (host NULL, never listed, never quota-counted, invisible to
	// the reaper's free-only sweep) — it either becomes paid at claim time or
	// is pruned once its checkout can no longer complete.
	entPending = "pending"
)

// spaceEntitlement reads one space's entitlement flag.
func (db *cloudDB) spaceEntitlement(id int64) (string, error) {
	var ent string
	err := db.sql.QueryRow(`SELECT entitlement FROM spaces WHERE id = ?`, id).Scan(&ent)
	return ent, err
}

// markSpacePaid records a completed checkout: the opaque Stripe ids, the tier
// bought, plus the paid flag (and a fresh subscription is by definition not
// cancelling). paid_until is left alone — the subscription event that follows
// the checkout carries the period end (see handleSubscriptionEvent).
func (db *cloudDB) markSpacePaid(id int64, customerID, subscriptionID, tier, band string) error {
	res, err := db.sql.Exec(
		`UPDATE spaces SET stripe_customer_id = ?, stripe_subscription_id = ?, tier = ?, band = ?, entitlement = ?, cancel_at_period_end = 0 WHERE id = ?`,
		customerID, subscriptionID, tier, band, entPaid, id,
	)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return sql.ErrNoRows
	}
	return nil
}

// setEntitlement flips a space's entitlement + subscription period end.
func (db *cloudDB) setEntitlement(id int64, ent string, paidUntil int64) error {
	return db.setEntitlementTier(id, ent, "", "", 0, paidUntil, false)
}

// setEntitlementTier additionally moves the tier + band when known ("" keeps
// the recorded value — e.g. an event without price data), mirrors the billed
// seat count (0 keeps the recorded value, same rationale), and records
// whether the subscription is set to run out at paid_until.
func (db *cloudDB) setEntitlementTier(id int64, ent, tier, band string, seats int, paidUntil int64, cancelAtEnd bool) error {
	res, err := db.sql.Exec(
		`UPDATE spaces SET entitlement = ?, paid_until = ?, tier = COALESCE(NULLIF(?, ''), tier), band = COALESCE(NULLIF(?, ''), band), seats = COALESCE(NULLIF(?, 0), seats), cancel_at_period_end = ? WHERE id = ?`,
		ent, paidUntil, tier, band, seats, cancelAtEnd, id,
	)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return sql.ErrNoRows
	}
	return nil
}

// meterRow is one space in the meter's sweep set — enough billing state to
// reconcile its quota alongside the byte reading, plus what the per-seat
// renewal true-up needs (subscription id, billed seats, renewal time).
type meterRow struct {
	ID             int64
	Host           string
	Entitlement    string
	Tier           string
	QuotaBytes     int64
	MeterTokenHash string
	Seats          int
	PaidUntil      int64
	SubscriptionID string
	Members        int // last metered COUNT(users) — the stats-down fallback for per-seat caps
}

// allFinalizedSpaces is the meter's sweep set: every space with a hostname.
func (db *cloudDB) allFinalizedSpaces() ([]meterRow, error) {
	rows, err := db.sql.Query(
		`SELECT id, host, entitlement, tier, quota_bytes, meter_token_hash, seats, paid_until, COALESCE(stripe_subscription_id, ''), members FROM spaces WHERE host IS NOT NULL ORDER BY id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []meterRow
	for rows.Next() {
		var m meterRow
		if err := rows.Scan(&m.ID, &m.Host, &m.Entitlement, &m.Tier, &m.QuotaBytes, &m.MeterTokenHash, &m.Seats, &m.PaidUntil, &m.SubscriptionID, &m.Members); err != nil {
			return nil, err
		}
		out = append(out, m)
	}
	return out, rows.Err()
}

// migratePool adds the free-pool gauge's history table (docs/free-pool.md):
// one row per meter sweep, enough for the control loop, the admin API, and
// eventually the status page. Additive like every other migration here.
func (db *cloudDB) migratePool() error {
	_, err := db.sql.Exec(`
CREATE TABLE IF NOT EXISTS pool_meter (
  ts              INTEGER PRIMARY KEY,
  free_count      INTEGER NOT NULL,
  demo_count      INTEGER NOT NULL,
  paid_count      INTEGER NOT NULL,
  free_cost_day   REAL NOT NULL,
  total_cost_day  REAL NOT NULL,
  net_revenue_day REAL NOT NULL,
  budget_day      REAL NOT NULL,
  pressure        REAL NOT NULL
);`)
	return err
}

// poolRows is the pool gauge's sweep set: what the budget math needs to know
// about every finalized space. NetEURYear stays zero here — the DB layer
// doesn't know the price book; the sweep resolves it.
func (db *cloudDB) poolRows() ([]poolRow, error) {
	rows, err := db.sql.Query(
		`SELECT entitlement, tier, band, seats, storage_bytes FROM spaces WHERE host IS NOT NULL`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []poolRow
	for rows.Next() {
		var r poolRow
		if err := rows.Scan(&r.Entitlement, &r.Tier, &r.Band, &r.Seats, &r.StorageBytes); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// insertPoolMeter records one gauge reading and prunes history past ~400 days
// (a year of context plus slack; the table stays a few thousand rows).
func (db *cloudDB) insertPoolMeter(ts int64, r poolReading) error {
	if _, err := db.sql.Exec(
		`INSERT OR REPLACE INTO pool_meter
		   (ts, free_count, demo_count, paid_count, free_cost_day, total_cost_day, net_revenue_day, budget_day, pressure)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		ts, r.FreeCount, r.DemoCount, r.PaidCount, r.FreeCostDay, r.TotalCostDay, r.NetRevenueDay, r.BudgetDay, r.Pressure,
	); err != nil {
		return err
	}
	_, err := db.sql.Exec(`DELETE FROM pool_meter WHERE ts < ?`, ts-400*24*3600)
	return err
}

// poolRevenueHistory reads trailing sweeps' instantaneous revenue readings —
// the smoothing window's raw material (pool.go smoothRevenue).
func (db *cloudDB) poolRevenueHistory(since int64) ([]float64, error) {
	rows, err := db.sql.Query(`SELECT net_revenue_day FROM pool_meter WHERE ts >= ? ORDER BY ts`, since)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []float64
	for rows.Next() {
		var v float64
		if err := rows.Scan(&v); err != nil {
			return nil, err
		}
		out = append(out, v)
	}
	return out, rows.Err()
}

// latestPoolMeter reads the newest gauge reading (ok=false: never swept).
func (db *cloudDB) latestPoolMeter() (ts int64, r poolReading, ok bool, err error) {
	err = db.sql.QueryRow(
		`SELECT ts, free_count, demo_count, paid_count, free_cost_day, total_cost_day, net_revenue_day, budget_day, pressure
		   FROM pool_meter ORDER BY ts DESC LIMIT 1`,
	).Scan(&ts, &r.FreeCount, &r.DemoCount, &r.PaidCount, &r.FreeCostDay, &r.TotalCostDay, &r.NetRevenueDay, &r.BudgetDay, &r.Pressure)
	if err == sql.ErrNoRows {
		return 0, r, false, nil
	}
	if err != nil {
		return 0, r, false, err
	}
	return ts, r, true, nil
}

// spaceBillingByID reads the billing state a quota push needs.
func (db *cloudDB) spaceBillingByID(id int64) (m meterRow, err error) {
	m.ID = id
	err = db.sql.QueryRow(
		`SELECT COALESCE(host,''), entitlement, tier, quota_bytes, seats, members FROM spaces WHERE id = ?`, id,
	).Scan(&m.Host, &m.Entitlement, &m.Tier, &m.QuotaBytes, &m.Seats, &m.Members)
	return m, err
}

// spaceCustomer reads a space's opaque Stripe customer id ("" = never paid)
// plus its recorded band — the portal path picks a per-band configuration.
func (db *cloudDB) spaceCustomer(id int64) (customer, band string, err error) {
	var c *string
	if err := db.sql.QueryRow(`SELECT stripe_customer_id, band FROM spaces WHERE id = ?`, id).Scan(&c, &band); err != nil {
		return "", "", err
	}
	if c == nil {
		return "", band, nil
	}
	return *c, band, nil
}

// spaceSubscription reads a space's entitlement plus its opaque Stripe
// subscription id ("" = never subscribed) — what destroy consults to decide
// whether a live subscription must be cancelled first.
func (db *cloudDB) spaceSubscription(id int64) (entitlement, subscriptionID string, err error) {
	var sub *string
	if err := db.sql.QueryRow(`SELECT entitlement, stripe_subscription_id FROM spaces WHERE id = ?`, id).Scan(&entitlement, &sub); err != nil {
		return "", "", err
	}
	if sub == nil {
		return entitlement, "", nil
	}
	return entitlement, *sub, nil
}

// setQuotaBytes records what the cloud last pushed into an instance's env.
func (db *cloudDB) setQuotaBytes(id, quota int64) error {
	_, err := db.sql.Exec(`UPDATE spaces SET quota_bytes = ? WHERE id = ?`, quota, id)
	return err
}

// setStorageBytes records one meter reading plus the space's claimed-ness.
func (db *cloudDB) setStorageBytes(id, bytes int64, setupComplete bool, members int, at int64) error {
	_, err := db.sql.Exec(
		`UPDATE spaces SET storage_bytes = ?, setup_complete = ?, members = ?, metered_at = ? WHERE id = ?`,
		bytes, setupComplete, members, at, id)
	return err
}

// findSpaceBySubscription maps a Stripe subscription id back to its space — the
// fallback for subscription events that arrive without space_id metadata.
func (db *cloudDB) findSpaceBySubscription(subscriptionID string) (int64, bool, error) {
	var id int64
	err := db.sql.QueryRow(
		`SELECT id FROM spaces WHERE stripe_subscription_id = ?`, subscriptionID,
	).Scan(&id)
	if err == sql.ErrNoRows {
		return 0, false, nil
	}
	if err != nil {
		return 0, false, err
	}
	return id, true, nil
}

// spaceByHost reads the claim path's view of one space (demo.go). ok=false:
// no such finalized host.
func (db *cloudDB) spaceByHost(host string) (id int64, entitlement string, createdAt int64, ok bool, err error) {
	err = db.sql.QueryRow(
		`SELECT id, entitlement, created_at FROM spaces WHERE host = ?`, host,
	).Scan(&id, &entitlement, &createdAt)
	if err == sql.ErrNoRows {
		return 0, "", 0, false, nil
	}
	if err != nil {
		return 0, "", 0, false, err
	}
	return id, entitlement, createdAt, true, nil
}

// claimDemoSpace converts one demo into the claiming account's free space —
// ownership transfer + entitlement flip in one guarded UPDATE, so a racing
// claim (or the reaper's ledger delete) loses cleanly: rows affected 0 means
// the row is no longer a claimable demo.
func (db *cloudDB) claimDemoSpace(id int64, callerID, requester string) (bool, error) {
	res, err := db.sql.Exec(
		`UPDATE spaces SET caller_id = ?, requester = ?, entitlement = ? WHERE id = ? AND entitlement = ?`,
		callerID, requester, entFree, id, entDemo)
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	return n == 1, err
}

// countFreeSpaces counts an account's free spaces — the claim path's quota
// check (allocateSpace runs the same count inside its own transaction).
func (db *cloudDB) countFreeSpaces(callerID, requester string) (int, error) {
	var n int
	err := db.sql.QueryRow(
		`SELECT COUNT(*) FROM spaces WHERE caller_id = ? AND requester = ? AND entitlement = 'free'`,
		callerID, requester).Scan(&n)
	return n, err
}