// db.go is `carlos db`: a member reads an app's database from its replica,
// with no box access and no AWS credentials (spec 2026-09-02).
//
// It never connects to the running database. The console serves the
// replica's LTX files by coordinate (internal/console/apidb.go); this side
// restores them, with litestream's own restore and a client that has no
// write methods (internal/dbdoor), into a private 0700 directory, proves the
// result with lease.CheckDB, and then either copies it to --out (pull) or
// opens it read-only and runs one query (query). The console never executes
// tenant SQL and never holds a restored database.
//
// Read-only is enforced, not assumed (spec §4): mode=ro and query_only. A
// copy is harmless to write to; the point is that nobody, person or agent,
// can come away believing a write reached production.
//
// Unit 1 of three (spec §13): ls and its freshness fields, then shell,
// follow.
package main

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"os"
	"os/signal"
	"slices"
	"sort"
	"strings"
	"syscall"
	"time"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbquery"
	"github.com/carlosframework/platform/internal/dbreads"
)

const dbUsage = `usage: carlos db <verb> --app <a> [--instance <host>] [flags]

  pull   --out <path>   restore a copy of the instance's database and save it
                        at <path>, which must not exist yet (saved 0600)
  query  --sql <text>   restore a copy and run one read-only query against it;
                        rows go to stdout, tab-separated, with a header row

Add --at <time> to either verb to read the database as it was at that
time, for example --at 2026-10-07T14:00:00Z. The copy holds every change
replicated before then, so it can be a little older than the time you give.
How far back you can go depends on how much history the platform keeps.

Both read the platform's replica of the database, never the live one, and
nothing you do to the copy reaches the instance. --instance names which of
the app's instances to read; it may be left out when the app has one. The
usual --account / --account-id / --console / --destination apply.
`

func runDB(args []string) {
	if len(args) == 0 {
		fmt.Fprint(os.Stderr, dbUsage)
		os.Exit(2)
	}
	switch args[0] {
	case "pull":
		runDBPull(args[1:])
	case "query":
		runDBQuery(args[1:])
	case "-h", "--help", "help":
		fmt.Fprint(stdout, dbUsage)
	default:
		fmt.Fprintf(os.Stderr, "carlos db: unknown verb %q\n%s", args[0], dbUsage)
		os.Exit(2)
	}
}

// dbTarget is what both verbs resolve first: the session, the account, the
// app and the one instance the verb reads.
type dbTarget struct {
	sess *apiSession
	acct apiAcct
	app  string
	host string
	// at is --at: the copy holds what was replicated before it. Zero is
	// the newest copy.
	at time.Time
}

// dbFlags registers the flags both verbs share and returns the resolver.
// Status lines go to stderr: query's stdout is its rows, and must pipe
// cleanly into whatever reads them.
func dbFlags(fs *flag.FlagSet, verb string) func() dbTarget {
	app := appFlag(fs, "app whose database to read")
	instance := fs.String("instance", "", "the instance whose database to read (may be left out when the app has one)")
	acctRaw, acctIDRaw, _ := accountFlags(fs)
	consoleFlag := consoleOption(fs, "console to read through (default: the only logged-in one)")
	atRaw := fs.String("at", "", "read the database as it was at this time (RFC 3339, for example 2026-10-07T14:00:00Z)")
	return func() dbTarget {
		// Before any network: a time that can't be read is the person's
		// to fix, and costs nothing to refuse.
		at, err := parseDBAt(*atRaw, time.Now())
		if err != nil {
			log.Fatalf("carlos db %s: %v", verb, err)
		}
		sess := requireDBAPISession(*consoleFlag)
		printUsing(os.Stderr, sess)
		acct := resolvedAccount(sess, acctRaw, acctIDRaw)
		name := resolvedApp(sess, acct, strings.TrimSpace(*app), dbUsage)
		host, err := resolveDBInstance(context.Background(), sess, acct, name, strings.TrimSpace(*instance))
		if err != nil {
			log.Fatalf("carlos db %s: %v", verb, err)
		}
		return dbTarget{sess: sess, acct: acct, app: name, host: host, at: at}
	}
}

// parseDBAt reads --at. Empty is the zero time, the newest copy. A time
// must carry its zone: a bare clock time would be read in whichever zone
// this shell is in, and the copy would silently be hours off. A time after
// now is refused rather than quietly answered with the newest copy, so
// nobody holds a copy labelled with a moment that hasn't happened.
func parseDBAt(raw string, now time.Time) (time.Time, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return time.Time{}, nil
	}
	at, err := time.Parse(time.RFC3339, raw)
	if err != nil {
		return time.Time{}, fmt.Errorf("--at %q is not a time. Write it in RFC 3339 with a time zone, for example 2026-10-07T14:00:00Z for UTC", raw)
	}
	if at.After(now) {
		return time.Time{}, fmt.Errorf("--at %s is in the future. Leave --at out to read the newest copy", raw)
	}
	// 0001-01-01T00:00:00Z parses to Go's zero time, which means "no
	// --at". It is before any history, so it is asked as the earliest
	// time instead, and gets the before-history answer.
	if at.Before(dbdoor.EarliestAsOf) {
		at = dbdoor.EarliestAsOf
	}
	return at, nil
}

// requireDBAPISession resolves signed-in mode, refusing direct bucket mode
// by name: the door is the console's, and an operator holding the bucket
// already has a box.
func requireDBAPISession(consoleFlag string) *apiSession {
	sess, err := resolveMode(consoleFlag)
	if err != nil {
		log.Fatal(err)
	}
	if sess == nil {
		log.Fatal("carlos db works through a console, and this shell is in direct bucket mode " +
			"(CARLOS_DEPLOYMENT_BUCKET or CARLOS_DEPLOYMENT_DIR is set). Unset it and run: carlos auth login --console <url>.")
	}
	return sess
}

// resolveDBInstance is --instance, or the app's only instance when it has
// exactly one. Several is a refusal naming them: guessing would hand
// someone the wrong data while telling them it is the right data (spec §7).
func resolveDBInstance(ctx context.Context, sess *apiSession, acct apiAcct, app, host string) (string, error) {
	if host != "" {
		return host, nil
	}
	recs, truncated, _, err := sess.instanceList(ctx, acct, app, "", "")
	if err != nil {
		return "", fmt.Errorf("could not list the app's instances to pick one: %w", err)
	}
	if len(recs) == 1 && !truncated {
		return recs[0].Host, nil
	}
	if len(recs) == 0 {
		return "", fmt.Errorf("%s has no instances, so there is no database to read", app)
	}
	hosts := make([]string, 0, len(recs))
	for _, r := range recs {
		hosts = append(hosts, r.Host)
	}
	sort.Strings(hosts)
	more := ""
	if truncated {
		more = ", and more"
	}
	return "", fmt.Errorf("%s has %d instances. Name one with --instance: %s%s", app, len(recs), strings.Join(hosts, ", "), more)
}

// dbRestore restores the target's replica into a private directory under
// the system temp dir. The caller removes Restored.Dir.
//
// A door refusal carrying the db region forward marker comes back as an
// *errDBForward, so a verb can switch to the forwarded read (dbforward.go).
func dbRestore(ctx context.Context, t dbTarget) (*dbdoor.Restored, error) {
	r, err := dbdoor.RestoreAsOf(ctx, t.sess.dbClient(t.acct, t.app, t.host), "", t.at)
	if err != nil {
		var door *dbdoor.DoorError
		if errors.As(err, &door) && door.Forward != nil {
			return nil, &errDBForward{door: door}
		}
		return nil, dbRestoreError(t.host, err)
	}
	return r, nil
}

// dbRestoreError turns a failed restore into the sentence a member needs.
// A refusal from the door carries the console's own words; the others are
// named here so none of them reads as a transient "no replica found".
func dbRestoreError(host string, err error) error {
	var door *dbdoor.DoorError
	var before *dbdoor.BeforeHistoryError
	switch {
	case errors.As(err, &before):
		return fmt.Errorf("%s's replica does not go back that far. The oldest point it keeps is %s", host, dbTime(before.Oldest))
	case errors.Is(err, dbdoor.ErrNoTimes):
		return fmt.Errorf("%s's replica can't be read as of a time yet. Leave --at out to read the newest copy", host)
	case errors.As(err, &door) && door.Status == 404:
		return fmt.Errorf("no instance %s that you can read", host)
	case errors.As(err, &door):
		return errors.New(door.Error())
	case errors.Is(err, dbdoor.ErrNothingReplicated):
		return fmt.Errorf("the platform has no copy of %s's database yet. Nothing has been replicated. "+
			"If the app replicates its own database (replicate: \"app\"), the platform never will", host)
	case errors.Is(err, dbdoor.ErrCheckFailed):
		return fmt.Errorf("the copy of %s's database failed its integrity check, so it was not kept: %v", host, err)
	case errors.Is(err, os.ErrNotExist):
		return fmt.Errorf("the replica of %s changed while it was being read, twice. Try again", host)
	case errors.Is(err, context.Canceled):
		return errors.New("stopped")
	}
	return fmt.Errorf("could not restore %s's database: %w", host, err)
}

// dbAsOf renders the copy's newest transaction and its time (R1).
func dbAsOf(r *dbdoor.Restored) string {
	if r.At.IsZero() {
		return fmt.Sprintf("transaction %d", r.TXID)
	}
	return fmt.Sprintf("transaction %d, written %s", r.TXID, dbTime(r.At))
}

func dbTime(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05 UTC") }

// dbPullHolds is pull's line saying what the copy holds; with --at it also
// names the requested time, since the copy can be older than it.
func dbPullHolds(r *dbdoor.Restored, at time.Time) string {
	if at.IsZero() {
		return fmt.Sprintf("It holds every change up to %s.\n", dbAsOf(r))
	}
	return fmt.Sprintf("It holds every change up to %s, the last one replicated before %s.\n", dbAsOf(r), dbTime(at))
}

// dbQueryStatus is query's stderr line before the rows.
func dbQueryStatus(host string, r *dbdoor.Restored, at time.Time) string {
	if at.IsZero() {
		return fmt.Sprintf("Querying a copy of %s's database, up to %s. Nothing you run here reaches %s.\n", host, dbAsOf(r), host)
	}
	return fmt.Sprintf("Querying a copy of %s's database as it was before %s, up to %s. Nothing you run here reaches %s.\n", host, dbTime(at), dbAsOf(r), host)
}

func runDBPull(args []string) {
	fs := flag.NewFlagSet("db pull", flag.ExitOnError)
	target := dbFlags(fs, "pull")
	out := fs.String("out", "", "where to save the copy; must not exist yet")
	fs.Usage = flagUsage(fs, dbUsage)
	fs.Parse(args)
	if strings.TrimSpace(*out) == "" {
		fmt.Fprint(os.Stderr, "carlos db pull: --out is required: name a path for the copy\n\n"+dbUsage)
		os.Exit(2)
	}
	if _, err := os.Lstat(*out); err == nil {
		fmt.Fprintf(os.Stderr, "carlos db pull: %s already exists. Choose a new path for --out\n", *out)
		os.Exit(2)
	}
	t := target()
	ctx, stop := dbSignalContext()
	defer stop()
	if err := dbPull(ctx, t, *out); err != nil {
		stop()
		log.Fatalf("carlos db pull: %v", err)
	}
}

// dbPull restores, copies to out, and removes the private directory on
// every path. It returns rather than exits so that cleanup always runs.
func dbPull(ctx context.Context, t dbTarget, out string) error {
	r, err := dbRestore(ctx, t)
	var fwd *errDBForward
	if errors.As(err, &fwd) {
		// The copy's files come from the region that holds them (db region
		// forward spec §6), when the console forwards pulls there.
		if !slices.Contains(fwd.door.Forward.Ops, dbreads.OpPull) {
			// A console that forwards queries and not pulls (plan 2's): DBF-8.
			return fmt.Errorf(dbfNoPull, t.host, fwd.door.Forward.Region)
		}
		r, err = dbPullForward(ctx, t, *fwd.door.Forward)
	}
	if err != nil {
		return err
	}
	defer os.RemoveAll(r.Dir)
	if err := copyOut(ctx, r.Path, out); err != nil {
		return err
	}
	fmt.Fprintf(stdout, "Saved a copy of %s's database to %s.\n", t.host, out)
	fmt.Fprint(stdout, dbPullHolds(r, t.at))
	fmt.Fprintf(stdout, "It is a copy. Nothing you do to it reaches %s.\n", t.host)
	return nil
}

// copyOut copies the restored database to out, 0600, refusing an existing
// out. A copy and never os.Rename: the temp directory and the user's
// destination are routinely on different filesystems, where a rename
// fails. A failed copy leaves nothing behind at out.
func copyOut(ctx context.Context, src, out string) (err error) {
	// Nothing is published once the person has asked to stop.
	if err := ctx.Err(); err != nil {
		return err
	}
	in, err := os.Open(src)
	if err != nil {
		return err
	}
	defer in.Close()
	f, err := os.OpenFile(out, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
	if err != nil {
		if errors.Is(err, os.ErrExist) {
			return fmt.Errorf("%s already exists. Choose a new path for --out", out)
		}
		return err
	}
	defer func() {
		if err != nil {
			f.Close()
			os.Remove(out)
		}
	}()
	// The mode passed to OpenFile is filtered by the umask; 0600 is what
	// the spec promises, so it is set, not hoped for.
	if err = f.Chmod(0o600); err != nil {
		return err
	}
	if _, err = io.Copy(f, ctxReader{ctx, in}); err != nil {
		return err
	}
	if err = f.Sync(); err != nil {
		return err
	}
	if err = f.Close(); err != nil {
		return err
	}
	// A stop that lands as the copy finishes still withdraws it: the
	// caller reports success only for a copy that was not interrupted.
	if err = ctx.Err(); err != nil {
		os.Remove(out)
		return err
	}
	return nil
}

// ctxReader stops a copy at the next read once ctx is done.
type ctxReader struct {
	ctx context.Context
	r   io.Reader
}

func (c ctxReader) Read(p []byte) (int, error) {
	if err := c.ctx.Err(); err != nil {
		return 0, err
	}
	return c.r.Read(p)
}

func runDBQuery(args []string) {
	fs := flag.NewFlagSet("db query", flag.ExitOnError)
	target := dbFlags(fs, "query")
	sqlText := fs.String("sql", "", "the query to run")
	fs.Usage = flagUsage(fs, dbUsage)
	fs.Parse(args)
	if strings.TrimSpace(*sqlText) == "" {
		fmt.Fprint(os.Stderr, "carlos db query: --sql is required\n\n"+dbUsage)
		os.Exit(2)
	}
	t := target()
	ctx, stop := dbSignalContext()
	defer stop()
	if err := dbQuery(ctx, t, *sqlText); err != nil {
		stop()
		log.Fatalf("carlos db query: %v", err)
	}
}

func dbQuery(ctx context.Context, t dbTarget, sqlText string) error {
	r, err := dbRestore(ctx, t)
	var fwd *errDBForward
	if errors.As(err, &fwd) {
		// The instance is read in its own region (db region forward spec
		// §5.5), when the console forwards queries there.
		if !slices.Contains(fwd.door.Forward.Ops, dbreads.OpQuery) {
			return errors.New(fwd.door.Error())
		}
		return dbQueryForward(ctx, t, sqlText, *fwd.door.Forward)
	}
	if err != nil {
		return err
	}
	defer os.RemoveAll(r.Dir)
	fmt.Fprint(os.Stderr, dbQueryStatus(t.host, r, t.at))
	return runQueryContext(ctx, stdout, r.Path, sqlText)
}

// The query runner and its guards live in internal/dbquery (spec 2026-10-22
// §8.3), so the CLI and the db-read worker run one copy of them. These
// names stay for db.go's callers and tests.
var (
	errOneStatement = dbquery.ErrOneStatement
	errPragmaChange = dbquery.ErrPragmaChange
)

func checkQuerySQL(q string) error { return dbquery.CheckSQL(q) }

func dbQueryCell(v any) string { return dbquery.Cell(v) }

func runQuery(w io.Writer, path, sqlText string) error {
	return runQueryContext(context.Background(), w, path, sqlText)
}

// runQueryContext reads nothing itself: the open and the query are
// dbquery.RunContext's.
func runQueryContext(ctx context.Context, w io.Writer, path, sqlText string) error {
	return dbquery.RunContext(ctx, w, path, sqlText, dbquery.Options{})
}

// dbStopSignals end a db verb early. Each one cancels the context instead of
// killing the process, so the deferred removal of the private directory
// always runs: Ctrl-C, a closed terminal (SIGHUP), a supervisor's SIGTERM,
// and SIGPIPE. The last is the one that matters most: `carlos db query ...
// | head` closes stdout early, and with SIGPIPE left to the runtime Go
// kills the process on the next write, which would leave a plaintext copy
// of production in the temp directory. Caught, the write returns EPIPE
// instead and the query loop stops on it.
var dbStopSignals = []os.Signal{os.Interrupt, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGPIPE}

func dbSignalContext() (context.Context, context.CancelFunc) {
	return signal.NotifyContext(context.Background(), dbStopSignals...)
}