// Package ops is the box-side operations surface — verbs a systemd timer (or
// an operator's shell on the box) runs to keep replication honest. Inspired by
// titogo internal/ops: every verb encodes knowledge that otherwise lives only
// in prose, and prose drifts silently where a command that runs drifts loudly.
package ops

import (
	"github.com/carlosframework/platform/internal/cliflag"

	"context"
	"errors"
	"flag"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/dbverify"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

const (
	defaultConfigPath = "/etc/litestream.yml"
	defaultDataDir    = "/data"
)

// Run dispatches `carlos ops <verb>`.
func Run(args []string) {
	verb := ""
	if len(args) > 0 {
		verb = args[0]
		args = args[1:]
	}
	switch verb {
	case "litestream":
		runLitestream(args)
	case "restore-verify":
		runRestoreVerify(args)
	case "hibernate-verify":
		runHibernateVerify(args)
	case "", "-h", "--help", "help":
		usage()
		if verb == "" {
			os.Exit(2)
		}
	default:
		fmt.Fprintf(os.Stderr, "ops: unknown verb %q\n\n", verb)
		usage()
		os.Exit(2)
	}
}

func usage() {
	fmt.Fprintf(os.Stderr, `carlos ops — box-side operations verbs

  carlos ops litestream      (re)generate the fleet litestream config from the
                             registry + data dir; prints "changed"/"unchanged"
                             so the caller knows whether to restart replication
  carlos ops restore-verify  restore every configured replica from S3 and
                             integrity-check each (run by the daily timer)
  carlos ops hibernate-verify
                             the same drill for the replicas the config does
                             NOT name: hibernating instances, whose
                             replication the activator owns under its lease.
                             Publishes a per-host verdict to the deployment
                             bucket so the result is readable off-box.

No verb for your task? Extend ops in a small PR — never a hand-run sequence
reconstructed from docs.
`)
}

// dataDir resolves the data directory: flag wins, then CARLOS_DATA_DIR, then
// the fleet default.
func dataDir(flagVal string) string {
	if flagVal != "" {
		return flagVal
	}
	if env := os.Getenv("CARLOS_DATA_DIR"); env != "" {
		return env
	}
	return defaultDataDir
}

// runLitestream — regenerate the fleet replication config. Idempotent: run it
// after every `carlos add`/`remove` (or on a timer; a no-op re-run is cheap
// and prints "unchanged"). Databases of hibernating routes are excluded — see
// FleetDBs for Eleven's two-replicators lesson.
func runLitestream(args []string) {
	fs := flag.NewFlagSet("ops litestream", flag.ExitOnError)
	regPath := fs.String("registry", "", "registry database (default <data>/registry.db)")
	data := fs.String("data", "", "data directory holding the fleet's *.db files (default $CARLOS_DATA_DIR or /data)")
	bucket := fs.String("bucket", "", "S3 bucket receiving the replicas (required)")
	prefix := fs.String("prefix", "", "key prefix inside the bucket for this fleet (required)")
	region := fs.String("region", "", "S3 region for the replica stanza")
	out := fs.String("out", defaultConfigPath, "config file to write")
	lsBin := fs.String("litestream-bin", "litestream", "litestream binary to check the snapshot policy against (pinned release "+litestreamVersion+")")
	allowShrink := fs.Bool("allow-shrink", false, "permit a config that replicates FEWER databases than the one it replaces")
	fs.Usage = cliflag.Usage(fs, "")
	_ = fs.Parse(args)

	if *bucket == "" || *prefix == "" {
		log.Fatal("litestream: --bucket and --prefix are required")
	}
	dir := dataDir(*data)
	rp := *regPath
	if rp == "" {
		rp = dir + "/registry.db"
	}
	reg, err := registry.Open(rp)
	if err != nil {
		log.Fatalf("litestream: open registry %s: %v", rp, err)
	}
	dbs, skipped, err := FleetDBsWithSkipped(reg, dir, LayoutJournalPresent(dir))
	if err != nil {
		log.Fatalf("litestream: %v", err)
	}
	if len(dbs) == 0 {
		log.Fatalf("litestream: no non-empty *.db files under %s — wrong box?", dir)
	}
	if len(skipped) > 0 {
		// Named, never silent: these are files under the data dir that no route
		// claims. Usually leftovers (a purge backup, a restore scratch); if one
		// is a database someone still wants, the fix is a route that says so.
		fmt.Fprintf(os.Stderr, "litestream: %d file(s) under %s are claimed by no route and will NOT be replicated:\n  %s\n",
			len(skipped), dir, strings.Join(skipped, "\n  "))
	}

	// Never quietly replicate less than the file being replaced — see
	// CoverageShortfall. This is what makes a cutover from a hand-maintained
	// config to a generated one safe to do unattended.
	if prior, err := os.ReadFile(*out); err == nil {
		if missing := CoverageShortfall(string(prior), dbs); len(missing) > 0 && !*allowShrink {
			log.Fatalf(`litestream: REFUSING to write %s — it would stop replicating %d database(s) that %s names today:
  %s
Nothing has been changed. Either make the registry describe them (a route with
--db, or a unit tenant's state directory), or pass --allow-shrink if they really
should no longer be replicated.`,
				*out, len(missing), *out, strings.Join(missing, "\n  "))
		} else if len(missing) > 0 {
			fmt.Fprintf(os.Stderr, "litestream: --allow-shrink: dropping %d database(s) from replication:\n  %s\n",
				len(missing), strings.Join(missing, "\n  "))
		}
	}

	cfg := LitestreamConfig(dbs, *bucket, *prefix, *region)
	wrote, err := writeIfChanged(*out, cfg)
	if err != nil {
		log.Fatalf("litestream: write %s: %v", *out, err)
	}
	fmt.Fprintf(os.Stderr, "litestream: %d databases -> s3://%s/%s/ (%s)\n", len(dbs), *bucket, *prefix, *out)

	// The config now STATES a recovery window, so say plainly when the binary
	// on this box will ignore it. Loud, not fatal: a box on an old litestream
	// still wants its database SET kept current (that is what carries
	// console.db), and refusing here would withhold the fix that does work in
	// order to protest the one that does not.
	if v := litestreamVersionOutput(*lsBin); !SnapshotPolicySupported(v) {
		fmt.Fprintf(os.Stderr, `litestream: WARNING — this box's litestream (%s) predates the snapshot
  policy (0.5+), so the "snapshot: retention: %s" this file now states is
  IGNORED and the real recovery window stays at litestream's 24h default.
  The database set below is still correct and still replicated. Upgrade the
  binary to %s to make the stated window real.
`, strings.TrimSpace(firstLine(v)), SnapshotRetention, litestreamVersion)
	}
	// The one contract line on stdout: the calling unit restarts the
	// replicator on "changed" and leaves it alone on "unchanged".
	if wrote {
		fmt.Println("changed")
	} else {
		fmt.Println("unchanged")
	}
}

// runRestoreVerify — restore every database in the generated config from S3
// into a temp dir and prove each opens clean. Exits non-zero on any failure so
// the oneshot unit lands in 'failed' where the fleet sweeps see it.
func runRestoreVerify(args []string) {
	fs := flag.NewFlagSet("ops restore-verify", flag.ExitOnError)
	cfgPath := fs.String("config", defaultConfigPath, "generated litestream config to read the database set from")
	lsBin := fs.String("litestream-bin", "litestream", "litestream binary (pinned release "+litestreamVersion+")")
	fs.Usage = cliflag.Usage(fs, "")
	_ = fs.Parse(args)

	tmp, err := os.MkdirTemp("", "carlos-restore")
	if err != nil {
		log.Fatalf("restore-verify: %v", err)
	}
	defer os.RemoveAll(tmp)

	if err := RestoreVerify(*lsBin, *cfgPath, tmp); err != nil {
		os.RemoveAll(tmp) // log.Fatal skips the defer
		log.Fatal(err)
	}
}

// firstLine is the version banner's first line, or a stand-in when the binary
// answered nothing at all.
func firstLine(s string) string {
	if s = strings.TrimSpace(s); s == "" {
		return "version unknown — binary missing or unreadable"
	}
	line, _, _ := strings.Cut(s, "\n")
	return line
}

// defaultHibernateMax caps one drill run. Each verification is a full restore
// from S3, so an uncapped nightly pass over a growing fleet is a bill that
// grows with it (the request-cost discipline internal/lease's LeaseTTL comment
// records). 25 covers today's fleets in a single run and keeps a large one
// fully covered within days, oldest-first.
const defaultHibernateMax = 25

// runHibernateVerify — restore-drill the hibernating instances' replicas, the
// set /etc/litestream.yml deliberately omits, and publish the verdicts.
//
// Exit status carries corruption only: a damaged replica fails the run (the
// oneshot unit lands in 'failed' where fleet sweeps see it), while replicas
// the pass could not decide are recorded as unverified and do NOT fail it.
// Conflating the two would either cry wolf on every S3 blip or — worse —
// train an operator to ignore a red run.
func runHibernateVerify(args []string) {
	fs := flag.NewFlagSet("ops hibernate-verify", flag.ExitOnError)
	regPath := fs.String("registry", "", "registry database (default <data>/registry.db)")
	data := fs.String("data", "", "data directory (default $CARLOS_DATA_DIR or "+defaultDataDir+")")
	max := fs.Int("max", defaultHibernateMax, "most replicas to restore in one run (0 = no cap)")
	fs.Usage = cliflag.Usage(fs, "")
	_ = fs.Parse(args)

	ctx := context.Background()
	cfg, err := activator.ConfigFromEnv()
	if err != nil {
		log.Fatalf("hibernate-verify: %v", err)
	}
	if cfg == nil {
		// Hibernation disabled on this box: there is no such replica set, so
		// there is nothing to drill and nothing is wrong.
		fmt.Println("hibernate-verify: CARLOS_HIBERNATE_BUCKET unset — no hibernating replicas on this box")
		return
	}

	dir := dataDir(*data)
	rp := *regPath
	if rp == "" {
		rp = dir + "/registry.db"
	}
	reg, err := registry.Open(rp)
	if err != nil {
		log.Fatalf("hibernate-verify: open registry %s: %v", rp, err)
	}
	hosts, err := HibernatingHosts(reg)
	if err != nil {
		log.Fatalf("hibernate-verify: %v", err)
	}
	if len(hosts) == 0 {
		fmt.Println("hibernate-verify: no hibernating instances registered on this box")
		return
	}

	// The prior record does two jobs: it orders this run (stalest first) and
	// it carries the verdicts this run's cap will not revisit.
	store, storeErr := openDeployStore(ctx)
	var prior *dbverify.Record
	if storeErr == nil {
		if prior, err = dbverify.Load(ctx, store); err != nil {
			log.Printf("hibernate-verify: reading %s: %v (continuing; this run starts the record over)", dbverify.Key, err)
			prior = nil
		}
	}

	tmp, err := os.MkdirTemp("", "carlos-hibernate-verify")
	if err != nil {
		log.Fatalf("hibernate-verify: %v", err)
	}
	defer os.RemoveAll(tmp)

	now := time.Now()
	results := VerifyHibernating(ctx, cfg, OldestFirst(hosts, prior), *max, now, tmp)
	fleet := make([]string, 0, len(hosts))
	for _, h := range hosts {
		fleet = append(fleet, h.Host)
	}
	rec := dbverify.Merge(prior, boxLabel(), now, fleet, results)

	// Publication is best-effort and LOUD when it fails: the drill's value is
	// that someone can read the result without a session on this box, so a
	// silent failure to publish would leave exactly the blind spot this verb
	// exists to remove.
	switch {
	case storeErr != nil:
		log.Printf("hibernate-verify: NOT PUBLISHED (%v) — the verdicts below exist only in this log", storeErr)
	default:
		if err := dbverify.Publish(ctx, store, rec); err != nil {
			log.Printf("hibernate-verify: NOT PUBLISHED: %v — the verdicts below exist only in this log", err)
		} else {
			fmt.Printf("hibernate-verify: published %s (%d hosts, %d unverified)\n", dbverify.Key, len(rec.Hosts), rec.Unverified())
		}
	}

	if n := rec.Failures(); n > 0 {
		os.RemoveAll(tmp) // log.Fatal skips the defer
		log.Fatalf("hibernate-verify: %d replica(s) did not restore clean", n)
	}
	// Count what was PROVEN, not what was visited: an unverified host is not
	// coverage, and a summary that said otherwise would be the same comfortable
	// lie this verb exists to remove.
	proven := 0
	for _, v := range results {
		if v.OK {
			proven++
		}
	}
	fmt.Printf("hibernate-verify: %d of %d hibernating replicas proven this run (%d visited, %d in the fleet)\n",
		proven, len(hosts), len(results), len(hosts))
}

// boxLabel names the writing box in the published record. The hostname is
// enough — the record is informational, unlike the cold-standby snapshot
// sidecar, whose Box field guards against a standby clobbering the dead box's
// registry.
func boxLabel() string {
	h, err := os.Hostname()
	if err != nil || h == "" {
		return "unknown"
	}
	return h
}

// openDeployStore resolves the deployment bucket the same way the agent's
// bucket/dir env vars do. A fleet box that reaches its deployment through the
// console door instead (openDeployStoreEnv's third case in internal/host) is
// NOT covered here and is told so out loud rather than publishing nothing
// quietly.
func openDeployStore(ctx context.Context) (release.Store, error) {
	if b := os.Getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" {
		return release.NewDeploymentS3Store(ctx, b)
	}
	if d := os.Getenv("CARLOS_DEPLOYMENT_DIR"); d != "" {
		ds := release.NewDirStore(d)
		ds.EnableChangeIndex()
		return ds, nil
	}
	return nil, errors.New("neither CARLOS_DEPLOYMENT_BUCKET nor CARLOS_DEPLOYMENT_DIR is set")
}