package main

// The storage meter (#668 slice 1) and its enforcement (#561, insurance-grade):
// how many bytes each cloud-provisioned space occupies, measured server-side
// and shown to members — plus, when caps are configured, per-tier storage
// ceilings pushed into each instance's own LCHAT_BLOB_QUOTA_BYTES guard.
//
// Metering only what ciphertext exposes: an instance's bytes are its DB files
// plus its attachment blobs, all opaque. Nothing here reads content, and
// nothing here ever WAKES a space — measurement is pure filesystem metadata,
// so a sweep over a fleet of hibernating instances stays free.
//
// Split like the reaper: the measuring half runs in controld (root, owns
// /data), the sweeping half in the cloud service (unprivileged, owns the
// ledger), joined by the existing stats endpoint.

import (
	"bytes"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"
)

// ---- measuring (controld-side, root) -------------------------------------------

// storageBytesForDB sums the on-disk footprint of an instance rooted at its DB
// path: the SQLite files (db, -wal, -shm) plus the attachment blob dir
// (<db-sans-ext>-blobs/, the disk blobStore's layout — db.go). Pure but for
// the filesystem, so it's directly testable.
//
// Slice-1 honesty: a hibernating space whose local copy has been cleaned away
// measures as only what remains on disk; measuring the S3 prefix instead is
// follow-up work in #668. Today the activator keeps local copies, so this is
// the fleet's truth in practice.
func storageBytesForDB(dbPath string) int64 {
	var total int64
	for _, suffix := range []string{"", "-wal", "-shm"} {
		if info, err := os.Stat(dbPath + suffix); err == nil {
			total += info.Size()
		}
	}
	blobDir := strings.TrimSuffix(dbPath, filepath.Ext(dbPath)) + "-blobs"
	_ = filepath.WalkDir(blobDir, func(_ string, d os.DirEntry, err error) error {
		if err != nil || d.IsDir() {
			return nil // unreadable entries just don't count; never fail a sweep
		}
		if info, err := d.Info(); err == nil {
			total += info.Size()
		}
		return nil
	})
	return total
}

// instanceStorageBytes measures one hosted instance.
func instanceStorageBytes(host string) int64 {
	return storageBytesForDB(filepath.Join(instanceDataDir, host+".db"))
}

// ---- sweeping (cloud-side) ------------------------------------------------------

// meterLoop refreshes every ledger space's storage_bytes on an interval.
// Enabled by default (interval <= 0 disables): the meter is harmless metadata
// and members should see their usage without an operator opting in.
func (s *cloudServer) meterLoop(interval time.Duration) {
	if interval <= 0 {
		return
	}
	log.Printf("cloud: storage meter on — sweeping every %s", interval)
	for {
		s.meterOnce()
		time.Sleep(interval)
	}
}

// meterOnce asks controld for each finalized space's byte count and records
// it. A stats error skips that space (its last reading stands, metered_at
// shows the staleness); the sweep never fails as a whole. When storage caps
// are configured, the sweep also RECONCILES quotas: any space whose pushed
// quota doesn't match its tier's cap gets a fresh push — the self-healing
// backstop behind the event-driven pushes (webhooks, creation).
func (s *cloudServer) meterOnce() (metered int) {
	spaces, err := s.db.allFinalizedSpaces()
	if err != nil {
		log.Printf("cloud: meter: ledger query: %v", err)
		return 0
	}
	now := time.Now().Unix()
	if s.usageURL != "" {
		// Applied-report dedupe ids age out with the sweep (usage pipe, #857);
		// the sidecar retry horizon is minutes, so a week is generous.
		if err := s.db.pruneUsageReports(now - 7*24*3600); err != nil {
			log.Printf("cloud: meter: usage-report prune: %v", err)
		}
	}
	for _, sp := range spaces {
		st, err := s.controlStats(sp.Host)
		if err != nil {
			log.Printf("cloud: meter: skip %s (stats unavailable): %v", sp.Host, err)
			continue
		}
		if err := s.db.setStorageBytes(sp.ID, st.StorageBytes, st.SetupComplete, st.Members, now); err != nil {
			log.Printf("cloud: meter: record %s: %v", sp.Host, err)
			continue
		}
		metered++
		if want := s.capFor(sp.Entitlement, sp.Tier, st.People); want > 0 && want != sp.QuotaBytes {
			s.pushQuota(sp.ID, sp.Host, sp.Entitlement, sp.Tier, st.People)
		}
		// The per-seat renewal true-up rides the same sweep: in the runway
		// before a per-seat renewal, re-count and set next year's quantity.
		s.reconcileSeats(sp, st.People, now)
		// The usage pipe (#857): any space without a meter token gets one —
		// how the fleet converges without a per-space backfill step.
		if s.usageURL != "" && sp.MeterTokenHash == "" {
			s.pushMeterToken(sp.ID, sp.Host)
		}
		// Replication class reconcile (cloudreplication.go): no ledger record
		// to compare, so push each pass — controld no-ops on a matching class.
		s.pushReplication(sp.Host, sp.Entitlement, sp.Tier)
	}
	if metered > 0 {
		log.Printf("cloud: meter: refreshed %d space(s)", metered)
	}
	// The free-pool gauge rides the same sweep (pool.go): the storage figures
	// it reads from the ledger were refreshed moments ago.
	if s.pool.enabled() {
		s.poolSweep(time.Now())
	}
	return metered
}

// ---- storage caps (#561 via #668): enforcement as insurance ---------------------
//
// The instance already refuses uploads past LCHAT_BLOB_QUOTA_BYTES (507,
// nothing deleted, text messaging never gated). Enforcement is therefore pure
// wiring: map each tier to a cap, and push the cap into the instance's env
// when its tier changes. Caps are config (LCHAT_CLOUD_STORAGE_CAPS); no
// config = no enforcement anywhere, and self-hosted or operator-run instances
// are never touched (they're not in this ledger).

// parseByteSize reads "25GB" / "2TB" / "500MB" / plain bytes. Decimal units —
// matching how storage is priced and displayed.
func parseByteSize(v string) (int64, error) {
	v = strings.TrimSpace(strings.ToUpper(v))
	mult := int64(1)
	for _, u := range []struct {
		suffix string
		mult   int64
	}{{"TB", 1e12}, {"GB", 1e9}, {"MB", 1e6}, {"KB", 1e3}} {
		if strings.HasSuffix(v, u.suffix) {
			v, mult = strings.TrimSpace(strings.TrimSuffix(v, u.suffix)), u.mult
			break
		}
	}
	n, err := strconv.ParseInt(v, 10, 64)
	if err != nil || n <= 0 {
		return 0, fmt.Errorf("bad size %q", v)
	}
	return n * mult, nil
}

// parseStorageCaps reads LCHAT_CLOUD_STORAGE_CAPS: "free=10GB,family=25GB,…" —
// keys are tier keys from the price book, plus "free" for unpaid spaces. A
// "/seat" suffix on the size ("business=11GB/seat") makes the entry per-seat:
// the pushed cap is that size × the space's live head count (pooled — one
// shared pot, no per-member walls; docs/tier-economics.md § Per-seat).
func parseStorageCaps(v string) (caps, seatCaps map[string]int64, err error) {
	if strings.TrimSpace(v) == "" {
		return nil, nil, nil
	}
	caps, seatCaps = map[string]int64{}, map[string]int64{}
	for _, pair := range strings.Split(v, ",") {
		k, val, ok := strings.Cut(strings.TrimSpace(pair), "=")
		if !ok || strings.TrimSpace(k) == "" {
			return nil, nil, fmt.Errorf("bad entry %q (want key=size)", pair)
		}
		into := caps
		if rest, perSeat := strings.CutSuffix(strings.TrimSpace(val), "/seat"); perSeat {
			val, into = rest, seatCaps
		}
		n, err := parseByteSize(val)
		if err != nil {
			return nil, nil, fmt.Errorf("cap %q: %v", k, err)
		}
		into[strings.TrimSpace(k)] = n
	}
	return caps, seatCaps, nil
}

// capFor maps a space's billing state to its storage cap. people is the live
// non-agent member count, which only per-seat entries use. 0 = no cap to push
// (caps unconfigured, or a paid tier with no entry — never guess a limit).
func (s *cloudServer) capFor(entitlement, tier string, people int) int64 {
	if len(s.storageCaps) == 0 && len(s.seatCaps) == 0 {
		return 0
	}
	if entitlement == entPaid {
		if per, ok := s.seatCaps[tier]; ok {
			return per * int64(max(people, 1))
		}
		return s.storageCaps[tier]
	}
	// Demos get their own (tiny) cap when configured, else the free one —
	// falling through beats leaving a demo uncapped.
	if entitlement == entDemo {
		if cap, ok := s.storageCaps[entDemo]; ok {
			return cap
		}
	}
	return s.storageCaps["free"]
}

// controlSetQuota pushes one instance's storage ceiling to controld.
func (s *cloudServer) controlSetQuota(host string, quota int64, entitlement string) error {
	payload, _ := json.Marshal(map[string]any{"bytes": quota, "entitlement": entitlement})
	req, err := http.NewRequest("POST", "http://control/instances/"+host+"/quota", bytes.NewReader(payload))
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+s.controlToken)
	req.Header.Set("Content-Type", "application/json")
	resp, err := s.control.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("control quota %s: HTTP %d", host, resp.StatusCode)
	}
	return nil
}

// pushQuota applies a space's tier cap to its instance, recording what was
// pushed. Best-effort by design: a failed push logs and stands down — the
// meter sweep reconciles it on the next pass.
func (s *cloudServer) pushQuota(id int64, host, entitlement, tier string, people int) {
	cap := s.capFor(entitlement, tier, people)
	if cap <= 0 {
		return
	}
	if err := s.controlSetQuota(host, cap, entitlement); err != nil {
		log.Printf("cloud: quota push %s: %v (sweep will retry)", host, err)
		return
	}
	if err := s.db.setQuotaBytes(id, cap); err != nil {
		log.Printf("cloud: quota record %s: %v", host, err)
		return
	}
	log.Printf("cloud: quota %s → %d bytes (%s)", host, cap, entitlementLabel(entitlement, tier))
}

func entitlementLabel(entitlement, tier string) string {
	if entitlement == entPaid {
		return "tier " + tier
	}
	return "free"
}

// ---- quota apply (controld-side, root) -------------------------------------------

// instanceEnvMu serializes every read-modify-write of instance env files.
// controld's quota and meter pushes arrive on separate request goroutines
// (a Stripe webhook and the sweep can land together) and write DIFFERENT
// keys to the SAME file — unserialized, the loser's read-before-write
// snapshot would silently drop the winner's keys, and because both callers
// record success in the ledger, nothing would ever retry. One process-wide
// mutex is plenty at this scale; per-host granularity would buy nothing.
var instanceEnvMu sync.Mutex

// updateInstanceEnv applies one mutation to an env file under the lock —
// the only correct way to touch an instance env after the daemon is up.
func updateInstanceEnv(envPath string, mutate func(env map[string]string)) error {
	instanceEnvMu.Lock()
	defer instanceEnvMu.Unlock()
	env, err := readEnvFile(envPath)
	if err != nil {
		return err
	}
	mutate(env)
	return writeEnvFile(envPath, env)
}

// setInstanceQuota upserts LCHAT_BLOB_QUOTA_BYTES in an instance's env file
// (0 removes it — back to the compiled default) and, when entitlement is
// non-empty, mirrors LCHAT_ENTITLEMENT ("free"/"paid"/"demo" — the UI's
// billing-affordance / demo-band gate) in the same write, preserving every
// other var.
func setInstanceQuota(envPath string, quota int64, entitlement string) error {
	return updateInstanceEnv(envPath, func(env map[string]string) {
		if quota <= 0 {
			delete(env, "LCHAT_BLOB_QUOTA_BYTES")
		} else {
			env["LCHAT_BLOB_QUOTA_BYTES"] = strconv.FormatInt(quota, 10)
		}
		if entitlement != "" {
			env["LCHAT_ENTITLEMENT"] = entitlement
		}
	})
}

// setInstanceMeter upserts LCHAT_METER_URL + LCHAT_METER_TOKEN in an
// instance's env file (both empty removes the pair — metering off),
// preserving every other var. The controld half of the meter-token push,
// next to its quota sibling.
func setInstanceMeter(envPath, url, token string) error {
	return updateInstanceEnv(envPath, func(env map[string]string) {
		if url == "" {
			delete(env, "LCHAT_METER_URL")
			delete(env, "LCHAT_METER_TOKEN")
		} else {
			env["LCHAT_METER_URL"] = url
			env["LCHAT_METER_TOKEN"] = token
		}
	})
}

// pushQuotaByID looks a space's billing state up and applies its cap —
// the webhook-side entry point (events carry ids, not rows).
func (s *cloudServer) pushQuotaByID(id int64) {
	if len(s.storageCaps) == 0 && len(s.seatCaps) == 0 {
		return
	}
	m, err := s.db.spaceBillingByID(id)
	if err != nil || m.Host == "" {
		return // pending create or gone — the sweep reconciles survivors
	}
	// Per-seat caps want the live head count; the last metered member count
	// (agents included — slightly generous, never tight) covers a stats
	// outage, and the sweep reconciles either way.
	people := m.Members
	if _, perSeat := s.seatCaps[m.Tier]; perSeat && m.Entitlement == entPaid {
		if st, err := s.controlStats(m.Host); err == nil {
			people = st.People
		}
	}
	s.pushQuota(id, m.Host, m.Entitlement, m.Tier, people)
}

// seatTrueUpWindow is how far before a per-seat renewal the sweep starts
// reconciling the subscription's quantity with the live head count. A week of
// hourly passes absorbs a missed webhook or a Stripe blip comfortably.
const seatTrueUpWindow = 7 * 24 * time.Hour

// reconcileSeats is the renewal true-up (docs/tier-economics.md § Per-seat):
// inside the window before a per-seat space renews, set the subscription's
// quantity to the live non-agent head count so the next invoice bills the
// real team size. proration_behavior=none inside trueUpSeats means nothing is
// charged or credited mid-cycle — this is the entire dropoff/growth policy.
// The ledger's seats update when Stripe's subscription.updated webhook lands;
// until then a repeat pass just re-reads the (now matching) quantity and
// no-ops.
func (s *cloudServer) reconcileSeats(sp meterRow, people int, now int64) {
	if s.billing == nil || sp.Entitlement != entPaid || sp.SubscriptionID == "" {
		return
	}
	t, ok := s.billing.tier(sp.Tier)
	if !ok || !t.PerSeat {
		return
	}
	if sp.PaidUntil <= now || sp.PaidUntil-now > int64(seatTrueUpWindow/time.Second) {
		return
	}
	want := max(people, t.MinSeats, 1)
	if want == sp.Seats {
		return
	}
	changed, err := s.billing.trueUpSeats(sp.SubscriptionID, want)
	if err != nil {
		log.Printf("cloud: seat true-up %s: %v (sweep will retry)", sp.Host, err)
		return
	}
	if changed {
		log.Printf("cloud: seats %s → %d at renewal (was %d)", sp.Host, want, sp.Seats)
	}
}