// The `carlos db` replica door (spec 2026-09-02 §6). It serves an
// instance's replicated LTX files BY COORDINATE (level plus TXID span) from
// the region data bucket, to a member of the account and nobody else, so the
// CLI can restore a private copy on the member's own machine. The console
// never executes tenant SQL and never holds a restored database.
//
// resolveReplica is the half that makes no S3 call: resolving an instance
// record to one replica prefix, and refusing everything that is not this
// console's to read. Every refusal there happens BEFORE the bucket is
// touched. That is deliberate: a door that tried and mapped AccessDenied
// would leak the attempt, the latency, and eventually a bucket name into a
// log a member can provoke.

package console

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"sync"
	"time"

	"github.com/aws/smithy-go"
	"github.com/superfly/ltx"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

var (
	// ErrResidency: the data is in another region's bucket, or on a
	// customer's own fleet box. IAM makes those reads impossible by design,
	// so this is the design working rather than a fault to retry.
	ErrResidency = errors.New("db: the instance's data is outside this region")
	// ErrPooled: a pooled record's data lives on the pool box, the same
	// class the instance-move worker refuses.
	ErrPooled = errors.New("db: pooled instances have no replica of their own")
	// ErrBadHost: path.Join cleans "../x" to "x" rather than refusing it,
	// so a host becomes a key segment only after ValidHostShape, and only
	// when it is the host the record itself names.
	ErrBadHost = errors.New("db: invalid host")
)

func (s *Server) resolveReplica(rec provision.Record, host string) (string, error) {
	if !provision.ValidHostShape(host) || rec.Host != host {
		return "", fmt.Errorf("%w: %q", ErrBadHost, host)
	}
	if rec.Pool != "" {
		return "", fmt.Errorf("%w: %s is in pool %q", ErrPooled, host, rec.Pool)
	}
	// A fleet placement is checked on its own, not left to the region
	// compare: the data is on the customer's box whatever this console's
	// CARLOS_REGION says.
	if _, fleet := registry.FleetPlacement(rec.Region); fleet {
		return "", fmt.Errorf("%w: %s is placed on fleet %q", ErrResidency, host, rec.Region)
	}
	if rec.Region != s.cfg.Region {
		return "", fmt.Errorf("%w: %s is homed in %q, this console serves %q",
			ErrResidency, host, rec.Region, s.cfg.Region)
	}
	return activator.HibernatePrefix(s.cfg.HibernatePrefix, host), nil
}

// replicaReader is the only part of a litestream replica client the door
// may hold: the two read methods. Holding the narrow interface means no
// handler here can reach WriteLTXFile, DeleteLTXFiles or DeleteAll by
// mistake, whatever the concrete client underneath can do.
type replicaReader interface {
	LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)
	OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)
}

// cliDBBurst bounds the db door per TOKEN within the ordinary minute window.
// Its own budget, not the instance verbs' 60 (spec §13.6): one restore is
// ten listings plus one request per planned file, and an app with a long L0
// tail plans dozens. The number is a ceiling on a stolen token, not a
// policing of honest pulls.
const cliDBBurst = 600

// dbAuditAction names the audit row every served file writes.
const dbAuditAction = "db-read"

// dbRequest is what dbGate resolves for both handlers.
type dbRequest struct {
	claim  hostclaim.Observed
	acct   *Account
	actor  string
	app    string
	host   string
	prefix string
	rd     replicaReader
}

// dbGate is the door's whole gate, in the repo's order: principal (cliRoute
// has already refused every service credential at the zero Set) ->
// membership (byte-identical 404) -> budget (429) -> configuration (503) ->
// the record (404 when absent) -> the class rail and residency (409), none
// of which touches the bucket -> the replica client. ok=false means a
// response has been written.
func (s *Server) dbGate(w http.ResponseWriter, r *http.Request) (dbRequest, bool) {
	// Belt to cliRoute's braces: the zero Set already stops every service
	// principal before this runs, so this line is only reached by a request
	// that did not come through cliRoute at all.
	if p, ok := principalFrom(r.Context()); !ok || p.Kind != PrincipalUser {
		http.NotFound(w, r)
		return dbRequest{}, false
	}
	actor, acct, app, ok := s.apiAccountApp(w, r)
	if !ok {
		return dbRequest{}, false
	}
	if !s.cli.db.allow(hashToken(bearer(r)), time.Now()) {
		jsonError(w, http.StatusTooManyRequests, "too many database requests. Wait a minute and try again.")
		return dbRequest{}, false
	}
	if s.cfg.Releases == nil || s.cfg.HibernateBucket == "" {
		jsonError(w, http.StatusServiceUnavailable, "this console has no data bucket set up, so it cannot read databases.")
		return dbRequest{}, false
	}
	host := r.PathValue("host")
	ctx := r.Context()
	src, err := s.resolveDBSource(ctx, acct, actor, app, host)
	switch {
	case errors.Is(err, os.ErrNotExist), errors.Is(err, ErrBadHost):
		http.NotFound(w, r)
		return dbRequest{}, false
	case errors.Is(err, errSourceClass):
		jsonError(w, http.StatusConflict, dbClassRefusal(src.rec))
		return dbRequest{}, false
	case errors.Is(err, ErrPooled):
		jsonError(w, http.StatusConflict, fmt.Sprintf(
			"%s runs in the pool %s, so its data is on the pool's own server. carlos db cannot read it.", host, src.rec.Pool))
		return dbRequest{}, false
	case errors.Is(err, ErrResidency):
		// Main's sentence unless this console forwards (db region forward
		// spec §5.5 step 1): then DBF-1, DBF-2 or the forward marker.
		s.dbResidency(w, r, src.rec)
		return dbRequest{}, false
	case errors.Is(err, errSourceClaim):
		g := dbRequest{acct: acct, actor: actor, app: app, host: host}
		if s.cfg.DBVersions == nil {
			jsonError(w, http.StatusServiceUnavailable, "this console has no versioned database reader")
			return dbRequest{}, false
		}
		s.dbClaimFailure(w, r, g, src.claimErr)
		return dbRequest{}, false
	case errors.Is(err, errSourceResolve):
		log.Printf("console: api db %s/%s/%s: resolve: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
		return dbRequest{}, false
	case err != nil:
		log.Printf("console: api db %s/%s/%s: read record: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, "could not read the instance. Try again.")
		return dbRequest{}, false
	}
	if s.cfg.HostClaim {
		return dbRequest{claim: src.claim, acct: acct, actor: actor, app: app, host: host, prefix: src.prefix}, true
	}
	rd, err := s.dbReplica(ctx, src.prefix)
	if err != nil {
		log.Printf("console: api db %s/%s/%s: replica client: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
		return dbRequest{}, false
	}
	return dbRequest{acct: acct, actor: actor, app: app, host: host, prefix: src.prefix, rd: rd}, true
}

var (
	errSourceClass = errors.New("db: not an instance the platform keeps a replica of")
	errSourceClaim = errors.New("db: the instance's claim could not be read")
	// errSourceResolve marks a resolveReplica failure that is none of the
	// named refusals, so the db door keeps its own log line and answer for
	// it apart from a failed record read.
	errSourceResolve = errors.New("db: the replica could not be resolved")
)

// dbSource is a replica source both the db door and the --from door may
// read (seeded-instances spec §4, codex 17): a record of this account and
// app, an instance with a replica of its own, homed in this console's
// region, and, with claims on, its versioned claim. claimErr is the claim
// read's own error, set beside the errSourceClaim wrap so a caller can hand
// it to dbClaimFailure unwrapped.
type dbSource struct {
	rec      provision.Record
	prefix   string
	claim    hostclaim.Observed
	claimErr error
}

// resolveDBSource holds the source checks dbGate made inline. It does no
// authorization: dbGate keeps its person-only gate, and the --from door
// authorizes through cliRoute's instance role.
func (s *Server) resolveDBSource(ctx context.Context, acct *Account, actor, app, host string) (dbSource, error) {
	if !provision.ValidHostShape(host) {
		return dbSource{}, fmt.Errorf("%w: %q", ErrBadHost, host)
	}
	rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
	if err != nil {
		return dbSource{}, err // os.ErrNotExist passes through as itself
	}
	if msg := dbClassRefusal(*rec); msg != "" {
		return dbSource{rec: *rec}, fmt.Errorf("%w: %s", errSourceClass, msg)
	}
	prefix, err := s.resolveReplica(*rec, host)
	if err != nil {
		switch {
		case errors.Is(err, ErrBadHost), errors.Is(err, ErrPooled), errors.Is(err, ErrResidency):
			return dbSource{rec: *rec}, err
		}
		return dbSource{rec: *rec}, fmt.Errorf("%w: %w", errSourceResolve, err)
	}
	src := dbSource{rec: *rec, prefix: prefix}
	if s.cfg.HostClaim {
		if s.cfg.DBVersions == nil {
			return src, fmt.Errorf("%w: this console has no versioned database reader", errSourceClaim)
		}
		g := dbRequest{acct: acct, actor: actor, app: app, host: host}
		if src.claim, err = s.dbClaim(ctx, g, *rec); err != nil {
			src.claimErr = err
			return src, fmt.Errorf("%w: %w", errSourceClaim, err)
		}
		src.prefix = src.claim.Claim.Holder.Location
	}
	return src, nil
}

// dbClassRefusal is the class rail as a sentence, or "" for an
// activator-managed exec instance: the only class whose database the
// platform replicates (spec §1 scope). Same three facts the move doors
// refuse on; the fleet and pool cases are resolveReplica's. The rule lives
// in dbreads.ClassRefusal (db region forward spec §8.2).
func dbClassRefusal(rec provision.Record) string { return dbreads.ClassRefusal(rec) }

// dbResidencyRefusal names where the data is, in the vocabulary the
// regions door uses (the home region by its CONSOLE_HOME_REGION name).
func (s *Server) dbResidencyRefusal(rec provision.Record) string {
	if fleet, ok := registry.FleetPlacement(rec.Region); ok {
		return fmt.Sprintf("%s runs on your fleet %s, so its data stays on that fleet's own server. carlos db cannot read it.", rec.Host, fleet)
	}
	return fmt.Sprintf("%s keeps its data in %s. This console can only read data kept in %s.",
		rec.Host, s.dbRegionName(rec.Region), s.dbRegionName(s.cfg.Region))
}

func (s *Server) dbRegionName(region string) string {
	if r := s.homeRegionOut(region); r != "" {
		return "region " + r
	}
	return "the home region"
}

// dbStoreFailure answers a failed bucket read. AccessDenied is the
// residency refusal as a backstop (spec §6): the role this console runs as
// cannot read that data, which is the design working, not a fault to
// retry. Everything else is logged with its detail (bucket names, request
// ids) and answered with none of it.
func (s *Server) dbStoreFailure(w http.ResponseWriter, g dbRequest, what string, err error) {
	log.Printf("console: api db %s/%s/%s: %s: %v", g.acct.Sqid, g.app, g.host, what, err)
	var api smithy.APIError
	if errors.As(err, &api) && api.ErrorCode() == "AccessDenied" {
		jsonError(w, http.StatusConflict, fmt.Sprintf(
			"this console is not allowed to read the data for %s. It is kept outside the region this console serves.", g.host))
		return
	}
	jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
}

// parseDBLevel accepts a bare decimal level in 0..MaxLevel. Anything else
// is not a coordinate.
func parseDBLevel(s string) (int, bool) {
	n, err := strconv.ParseUint(s, 10, 8)
	if err != nil || n > dbdoor.MaxLevel {
		return 0, false
	}
	return int(n), true
}

// handleAPIDBList answers one level's listing. An out-of-range level is a
// 404 (it is not a coordinate); a level with nothing in it is 200 and an
// empty list (it is a coordinate that happens to be empty, spec §6.1).
func (s *Server) handleAPIDBList(w http.ResponseWriter, r *http.Request) {
	level, ok := parseDBLevel(r.PathValue("level"))
	if !ok {
		http.NotFound(w, r)
		return
	}
	g, ok := s.dbGate(w, r)
	if !ok {
		return
	}
	if s.cfg.HostClaim {
		out, err := s.listDBVersions(r.Context(), g, level)
		if err != nil {
			s.dbClaimFailure(w, r, g, err)
			return
		}
		w.Header().Set("Cache-Control", "no-store")
		writeJSON(w, http.StatusOK, out)
		return
	}
	out, err := listReplicaLevel(r.Context(), g.rd, level)
	if err != nil {
		s.dbStoreFailure(w, g, fmt.Sprintf("list level %d", level), err)
		return
	}
	writeJSON(w, http.StatusOK, out)
}

// listReplicaLevel lists one level of rd in the door's wire shape, ordered
// by MinTXID, with an empty level as an empty list. The db door and the
// console's own in-process door (replicaDoer, for --from) both list
// through it, so a seeded copy restores from the listing carlos db pull
// reads. The body moved to dbdoor.ListLevel so the box's planner lists
// through it too (db region forward spec §6).
func listReplicaLevel(ctx context.Context, rd replicaReader, level int) (dbdoor.ListResponse, error) {
	return dbdoor.ListLevel(ctx, rd, level)
}

// handleAPIDBGet streams one LTX file by coordinate. The bytes pass through
// the console, never a presigned URL (the edge store door's F8 precedent):
// nothing the request supplies picks the bucket or the key, only a level
// and a span inside the prefix resolveReplica chose.
func (s *Server) handleAPIDBGet(w http.ResponseWriter, r *http.Request) {
	level, ok := parseDBLevel(r.PathValue("level"))
	if !ok {
		http.NotFound(w, r)
		return
	}
	minTXID, maxTXID, err := dbdoor.ParseSpan(r.PathValue("span"))
	if err != nil {
		http.NotFound(w, r)
		return
	}
	g, ok := s.dbGate(w, r)
	if !ok {
		return
	}
	var pinned *dbVersionReader
	if s.cfg.HostClaim {
		pinned, err = s.pinnedDBReader(r, g, level, minTXID, maxTXID)
		if err != nil {
			s.dbClaimFailure(w, r, g, err)
			return
		}
		g.rd = pinned
	}
	rangeHdr := r.Header.Get("Range")
	offset, size, err := dbdoor.ParseRange(rangeHdr)
	if err != nil {
		jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
		return
	}
	if pinned != nil {
		if _, err := dbVersionRange(pinned.handle.Size, offset, size); err != nil {
			jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
			return
		}
	}
	if !s.dbStreams.acquire(g.actor) {
		w.Header().Set("Retry-After", dbStreamRetryAfter)
		jsonError(w, http.StatusTooManyRequests, "too many database downloads at once. Wait a few seconds and try again.")
		return
	}
	defer s.dbStreams.release(g.actor)
	// The idle watchdog: any stretch of dbStreamIdle with no byte moved
	// cancels the bucket read. The write side has its own deadline below,
	// renewed per chunk, because a cancelled context does not unblock a
	// write to a client that has stopped reading.
	idle := s.dbStreamIdle
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()
	watchdog := time.AfterFunc(idle, cancel)
	defer watchdog.Stop()
	rc, err := g.rd.OpenLTXFile(ctx, level, ltx.TXID(minTXID), ltx.TXID(maxTXID), offset, size)
	if errors.Is(err, os.ErrNotExist) {
		http.NotFound(w, r)
		return
	}
	if err != nil {
		s.dbStoreFailure(w, g, fmt.Sprintf("open level %d %s", level, dbdoor.Span(minTXID, maxTXID)), err)
		return
	}
	defer rc.Close()
	// One row per successful GET, ranged or not. A ranged read is still a
	// read of the file: skipping it would let a caller list the coordinates
	// and fetch every file from byte 1 with no row written. A resumed read
	// says where it started.
	detail := fmt.Sprintf("read database replica file level %d, transactions %d to %d (cli)", level, minTXID, maxTXID)
	if offset > 0 {
		detail = fmt.Sprintf("read database replica file level %d, transactions %d to %d, from byte %d (cli)", level, minTXID, maxTXID, offset)
	}
	s.auditActivity(g.acct, g.app, g.actor, Activity{
		Action: dbAuditAction, TargetType: "instance", Target: g.host, Detail: detail,
	})
	if pinned != nil {
		if err := dbVersionHeaders(w, r, pinned.handle.Size, offset, size); err != nil {
			jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
			return
		}
	}
	w.Header().Set("Content-Type", "application/octet-stream")
	w.Header().Set("Cache-Control", "no-store")
	status := http.StatusOK
	if rangeHdr != "" {
		status = http.StatusPartialContent
	}
	rw := http.NewResponseController(w)
	// The deadline is never cleared here. net/http flushes what is still
	// buffered (the last bytes, the chunked terminator) in finishRequest,
	// after this handler returns, and clears the write deadline itself
	// only after that flush. Clearing it on the way out would leave that
	// flush unbounded, holding the connection for as long as a client that
	// stopped reading near the end cares to wait.
	_ = rw.SetWriteDeadline(time.Now().Add(idle))
	w.WriteHeader(status)
	err = dbStream(w, rw, rc, watchdog, idle)
	// Flush the tail now, under a fresh deadline, so it is done (or cut)
	// before the deferred release gives this stream's slot back. What
	// finishRequest has left after this is the chunked terminator, still
	// under the same deadline. On a failed read the deadline is renewed
	// too, so the client gets its truncated body promptly (and resumes)
	// rather than a connection cut by a deadline the stall used up.
	_ = rw.SetWriteDeadline(time.Now().Add(idle))
	if ferr := rw.Flush(); err == nil {
		err = ferr
	}
	if err != nil {
		// The status is already on the wire. The client's resumable reader
		// sees a short body against the listed size and reopens at the
		// offset it reached.
		log.Printf("console: api db %s/%s/%s: stream level %d %s: %v",
			g.acct.Sqid, g.app, g.host, level, dbdoor.Span(minTXID, maxTXID), err)
	}
}

// dbStream copies rc to w a chunk at a time, renewing the write deadline
// before each chunk and the read watchdog after it, so a stream ends when
// either side stops moving for idle.
func dbStream(w io.Writer, rw *http.ResponseController, rc io.Reader, watchdog *time.Timer, idle time.Duration) error {
	buf := make([]byte, 32<<10)
	for {
		n, rerr := rc.Read(buf)
		if n > 0 {
			if err := rw.SetWriteDeadline(time.Now().Add(idle)); err != nil && !errors.Is(err, http.ErrNotSupported) {
				return err
			}
			if _, err := w.Write(buf[:n]); err != nil {
				return err
			}
			watchdog.Reset(idle)
		}
		if rerr == io.EOF {
			return nil
		}
		if rerr != nil {
			return rerr
		}
	}
}

// dbReplicaCacheMax bounds the replica-client cache. Each client owns an
// HTTP transport, so one per request would cost a TLS handshake per file and
// leave an idle connection behind each; one per prefix lets a restore's
// dozens of requests share one. Over the bound the map is dropped
// wholesale, the rateLimiter's rule: forgetting clients costs a handshake,
// a walk would cost a lock held across every door.
const dbReplicaCacheMax = 64

type dbReplicaCache struct {
	mu      sync.Mutex
	clients map[string]replicaReader
}

// cachedReadReplica is the production dbReplica: lease.ReadReplicaClient on
// the configured bucket, one per prefix.
func (s *Server) cachedReadReplica(ctx context.Context, prefix string) (replicaReader, error) {
	c := &s.dbClients
	c.mu.Lock()
	defer c.mu.Unlock()
	if rd, ok := c.clients[prefix]; ok {
		return rd, nil
	}
	rc, err := lease.ReadReplicaClient(ctx, s.cfg.HibernateBucket, prefix, s.cfg.HibernateRegion)
	if err != nil {
		return nil, err
	}
	if c.clients == nil || len(c.clients) >= dbReplicaCacheMax {
		c.clients = make(map[string]replicaReader, 8)
	}
	c.clients[prefix] = rc
	return rc, nil
}

// The db door's stream bounds. The per-minute token budget bounds how many
// requests start, not how many are still running, and a slow or stalled
// download holds a goroutine, an S3 connection and a client connection for
// as long as it lasts. So active file streams are bounded per principal and
// across the console, and refused with 429 and Retry-After when full; and
// each stream must move a byte at least every dbStreamIdle, in both
// directions, or it is cut. The CLI's restore downloads one file at a time
// (dbdoor.RestoreTo), so a person's restore needs one slot.
const (
	dbMaxStreams             = 32
	dbMaxStreamsPerPrincipal = 4
	dbStreamIdleDefault      = 30 * time.Second
	dbStreamRetryAfter       = "5"
)

type dbStreamLimits struct {
	mu               sync.Mutex
	maxTotal, maxPer int
	total            int
	per              map[string]int
}

func newDBStreamLimits(maxTotal, maxPer int) *dbStreamLimits {
	return &dbStreamLimits{maxTotal: maxTotal, maxPer: maxPer, per: map[string]int{}}
}

// acquire takes a slot for key, or reports that the console or key is full.
func (l *dbStreamLimits) acquire(key string) bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.total >= l.maxTotal || l.per[key] >= l.maxPer {
		return false
	}
	l.total++
	l.per[key]++
	return true
}

func (l *dbStreamLimits) release(key string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.total--
	if l.per[key]--; l.per[key] <= 0 {
		delete(l.per, key)
	}
}