// The leftover-replica rail on the instance and sidecar create doors
// (discussion #199). This is a stopgap, not the fix.
//
// A hibernating record's replica lives at instances/<host> and a residence
// move stages one at moves/<host>/. Both are keyed by HOST alone, and
// deleting the record leaves both behind (deliberately: retiring a record is
// not destroying its data). So whoever next creates a record on the same host
// inherits the old database, on its first wake or through `carlos db pull`.
// The full fix is being designed separately. Until it lands, the create
// doors refuse a new record on a host whose replica or staging prefix in this
// console's data bucket still holds an LTX file, and refuse on uncertainty
// when the bucket cannot be read.
//
// The gap, stated plainly: the check only sees THIS console's region. A
// record homed in another region, placed on a customer fleet, or run in a
// pool is created unchecked, because its data lives somewhere this console's
// role cannot read (resolveReplica's residency rules). Leftovers in this region under a
// host created elsewhere are not seen either, and a later move of that
// instance into this region would land on top of them. A console with no
// data bucket configured checks nothing.

package console

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"strings"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/registry"
)

// leftoverCheckApplies reports whether a record homed in region, in pool,
// can be checked from here: the data bucket is configured, and the record's
// data would live in it. The residency rules are resolveReplica's: a pooled
// record's data is on the pool's own box; a fleet placement is never this
// console's to read, whatever its own region string says; and otherwise the
// record's region must equal the console's ("" == "" on the home region and
// on a single-region deployment).
func (s *Server) leftoverCheckApplies(region, pool string) bool {
	if s.cfg.HibernateBucket == "" || pool != "" {
		return false
	}
	if _, fleet := registry.FleetPlacement(region); fleet {
		return false
	}
	return region == s.cfg.Region
}

// hostHoldsReplicaData reports whether host's replica prefix or its move
// staging prefix holds any LTX file, at any level from 0 to the snapshot
// level. It stops at the first file. Every error is returned, never treated
// as "empty": the caller refuses on uncertainty.
//
// The listing goes through s.dbReplica, the db door's cached
// lease.ReadReplicaClient seam, so the reader here and the writer on the box
// cannot disagree about the bucket's endpoint, region or layout.
//
// Neither prefix may end in "/". litestream's S3 client lists
// <path>/<level>/, so MovesPrefix's own trailing slash would ask for
// moves/<host>//0000/, which matches nothing in S3 and would read every
// staged replica as absent. (The file client joins paths and forgives it,
// which is why TestLeftoverCheckListsTheExactS3Prefixes goes through the
// S3 client.) The slash litestream adds is also what keeps app.example.co
// from seeing app.example.com's files.
func (s *Server) hostHoldsReplicaData(ctx context.Context, host string) (bool, error) {
	for _, prefix := range []string{
		activator.HibernatePrefix(s.cfg.HibernatePrefix, host),
		strings.TrimSuffix(activator.MovesPrefix(host), "/"),
	} {
		rd, err := s.dbReplica(ctx, prefix)
		if err != nil {
			return false, fmt.Errorf("replica client for %s: %w", prefix, err)
		}
		for level := 0; level <= dbdoor.MaxLevel; level++ {
			found, err := anyLTXFile(ctx, rd, level)
			if err != nil {
				return false, fmt.Errorf("list %s level %d: %w", prefix, level, err)
			}
			if found {
				return true, nil
			}
		}
	}
	return false, nil
}

// anyLTXFile reports whether one level lists at least one file.
func anyLTXFile(ctx context.Context, rd replicaReader, level int) (bool, error) {
	itr, err := rd.LTXFiles(ctx, level, 0, false)
	if err != nil {
		return false, err
	}
	defer itr.Close()
	if itr.Next() {
		return true, nil
	}
	return false, itr.Err()
}

// leftoverCreate is one create as the rail needs it.
type leftoverCreate struct {
	logTag             string // "api instances create <sqid>/<app>"
	host, region, pool string
	// exists reads the record this create would write.
	exists func(ctx context.Context) (bool, error)
	// duplicate and readFailed are the door's OWN answers to a record that
	// already exists and to a store read that failed, byte for byte.
	duplicate, readFailed doorRefusal
	// taken and unsure are the rail's 409 and 503 sentences.
	taken, unsure string
}

// leftoverReplicaRefusal is the rail as the doors call it. It answers nil
// when the create may go ahead: the record is out of this console's sight
// (leftoverCheckApplies), or the host holds no replica data. Otherwise it
// answers 409 with taken, or 503 with unsure when the bucket could not be
// read. The store's own error is logged and never returned: it carries the
// bucket name.
//
// It reads the record first. A create retried after a lost response finds
// its own record there, and that record's live replica fills the prefix,
// so without the read the retry would be told the address holds a deleted
// instance's data (or, in an S3 outage, to try again). It gets the door's
// ordinary duplicate answer instead. A failed read is not "absent": it gets
// the door's own failure. The create stays create-only, so a record that
// appears after this read still loses the CAS in the write.
//
// The rail gets its own instanceStoreTimeout and runs off the request's
// context rather than the door's, whose budget earlier reads have spent
// some of. The door must likewise take a fresh budget for its write after
// this returns.
func (s *Server) leftoverReplicaRefusal(parent context.Context, c leftoverCreate) *doorRefusal {
	if !s.leftoverCheckApplies(c.region, c.pool) {
		return nil
	}
	ctx, cancel := context.WithTimeout(parent, instanceStoreTimeout)
	defer cancel()
	exists, err := c.exists(ctx)
	if err != nil {
		log.Printf("console: %s: read existing record for %s: %v", c.logTag, c.host, err)
		return &c.readFailed
	}
	if exists {
		return &c.duplicate
	}
	held, err := s.hostHoldsReplicaData(ctx, c.host)
	if err != nil {
		log.Printf("console: %s: leftover replica check for %s: %v", c.logTag, c.host, err)
		return &doorRefusal{http.StatusServiceUnavailable, c.unsure}
	}
	if held {
		log.Printf("console: %s: refused: %s still holds replica data from a deleted record (#199); purge it to free the host", c.logTag, c.host)
		return &doorRefusal{http.StatusConflict, c.taken}
	}
	return nil
}