package console

import (
	"bytes"
	"context"
	"crypto/ecdsa"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/carlosframework/platform/internal/park"
	"io"
	"net/http"
	"os"
	"reflect"
	"strconv"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/superfly/ltx"
)

const dbHandleTTL = 15 * time.Minute

var (
	errDBNotAdopted    = errors.New("host not adopted")
	errDBClaimChanged  = errors.New("host claim changed; retry the database pull")
	errDBClaimClosed   = errors.New("host is not accepting database reads")
	errDBHandle        = errors.New("invalid database file handle")
	errDBHandleExpired = errors.New("database file handle expired; retry the database pull")
)

// dbFileHandle is encoded as this ordered JSON struct, never as a map.
// Account and App bind the capability to the authenticated route as well.
type dbFileHandle struct {
	V             int     `json:"v"`
	Host          string  `json:"host"`
	Location      string  `json:"location"`
	Key           string  `json:"key"`
	VersionID     string  `json:"version_id"`
	Incarnation   string  `json:"incarnation"`
	Expiry        int64   `json:"expiry"`
	Account       string  `json:"account"`
	App           string  `json:"app"`
	Size          int64   `json:"size"`
	SnapshotCRC64 *uint64 `json:"snapshot_crc64,omitempty"`
}

func dbHandleSecret(key *ecdsa.PrivateKey) []byte {
	mac := hmac.New(sha256.New, key.D.FillBytes(make([]byte, 32)))
	mac.Write([]byte("carlos-db-handle-key-v1\x00"))
	return mac.Sum(nil)
}

func dbHandleMAC(secret, body []byte) []byte {
	mac := hmac.New(sha256.New, secret)
	mac.Write([]byte("carlos-db-file-v1\x00"))
	mac.Write(body)
	return mac.Sum(nil)
}

func signDBHandle(secret []byte, h dbFileHandle) (string, error) {
	if len(secret) != sha256.Size {
		return "", errDBHandle
	}
	body, err := json.Marshal(h)
	if err != nil {
		return "", err
	}
	return base64.RawURLEncoding.EncodeToString(body) + "." + base64.RawURLEncoding.EncodeToString(dbHandleMAC(secret, body)), nil
}

func verifyDBHandle(secret []byte, token string, now time.Time) (dbFileHandle, error) {
	var h dbFileHandle
	if len(secret) != sha256.Size || len(token) > 8192 {
		return h, errDBHandle
	}
	a, b, ok := strings.Cut(token, ".")
	if !ok {
		return h, errDBHandle
	}
	body, err := base64.RawURLEncoding.DecodeString(a)
	if err != nil {
		return h, errDBHandle
	}
	sig, err := base64.RawURLEncoding.DecodeString(b)
	if err != nil || !hmac.Equal(sig, dbHandleMAC(secret, body)) {
		return h, errDBHandle
	}
	dec := json.NewDecoder(bytes.NewReader(body))
	dec.DisallowUnknownFields()
	if err := dec.Decode(&h); err != nil {
		return dbFileHandle{}, errDBHandle
	}
	canonical, err := json.Marshal(h)
	if err != nil || !bytes.Equal(body, canonical) || h.V != 1 || h.VersionID == "" || h.VersionID == "null" || h.Size < 0 {
		return dbFileHandle{}, errDBHandle
	}
	if now.Unix() >= h.Expiry {
		return dbFileHandle{}, errDBHandleExpired
	}
	return h, nil
}

func dbLevelPrefix(location string, level int) string {
	// litestream/s3 uses hex level directories, unlike its file replica.
	return fmt.Sprintf("%s/%04x/", location, level)
}

func dbVersionKey(location string, level int, lo, hi uint64) string {
	return dbLevelPrefix(location, level) + ltx.FormatFilename(ltx.TXID(lo), ltx.TXID(hi))
}

func (s *Server) dbClaim(ctx context.Context, g dbRequest, rec provision.Record) (hostclaim.Observed, error) {
	obs, err := hostclaim.Read(ctx, s.cfg.Releases, g.host)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	if obs.Absent {
		return obs, errDBNotAdopted
	}
	h := obs.Claim.Holder
	if h == nil {
		return obs, errDBClaimClosed
	}
	inc, err := hostclaim.Incarnation(g.acct.Sqid, g.app, g.host, rec.CreatedAt)
	if err != nil {
		return obs, err
	}
	if h.Family != "instance" || h.Account != g.acct.Sqid || h.App != g.app || h.Key != provision.RecordKey(g.acct.Sqid, g.app, g.host) || h.Incarnation != inc || !h.CreatedAt.Equal(rec.CreatedAt) {
		return obs, os.ErrNotExist
	}
	switch obs.Claim.Op.Kind {
	case hostclaim.KindDeleting, hostclaim.KindParking, hostclaim.KindPurging:
		return obs, errDBClaimClosed
	case hostclaim.KindMoving, hostclaim.KindLanding, hostclaim.KindReverting, hostclaim.KindRestoring, hostclaim.KindRelocating:
		// Every move kind, in every phase, landing/committed and the
		// opened phases included (moves spec §19.1): the door reopens at
		// the close, when the claim is serving again.
		return obs, errDBClaimClosed
	}
	if h.Location == "" {
		return obs, errDBClaimClosed
	}
	return obs, nil
}

func sameDBAdmission(a, b hostclaim.Observed) bool {
	if a.Absent || b.Absent || a.Claim == nil || b.Claim == nil {
		return false
	}
	// §6.5 permits unrelated chore revisions; compare all holder/op content,
	// including stamp, session, admissions, journal and execution ownership.
	return a.Host == b.Host && reflect.DeepEqual(a.Claim.Holder, b.Claim.Holder) && reflect.DeepEqual(a.Claim.Op, b.Claim.Op)
}

func (s *Server) dbClaimFailure(w http.ResponseWriter, r *http.Request, g dbRequest, err error) {
	switch {
	case errors.Is(err, os.ErrNotExist):
		http.NotFound(w, r)
	case errors.Is(err, errDBNotAdopted), errors.Is(err, errDBClaimChanged), errors.Is(err, errDBClaimClosed):
		w.Header().Set("Retry-After", "1")
		jsonError(w, http.StatusConflict, err.Error())
	case errors.Is(err, errDBHandleExpired):
		jsonError(w, http.StatusGone, err.Error())
	case errors.Is(err, errDBHandle):
		jsonError(w, http.StatusBadRequest, err.Error())
	default:
		s.dbStoreFailure(w, g, "claim/version read", err)
	}
}

func (cfg Config) claimPlacementRegion(code string) string {
	region := park.PlacementRegion(code)
	if region == park.HomeRegion && park.PlacementRegion(cfg.HomeRegion) == cfg.HomeRegion && cfg.HomeRegion != "" {
		return cfg.HomeRegion
	}
	return region
}

// dbClaimHomeRegion binds the old fallback only for this configured home
// bucket. Admission contents never choose the compatibility name.
func (cfg Config) dbClaimHomeRegion() string {
	if cfg.HomeRegion != "" {
		return cfg.HomeRegion
	}
	legacy := strings.TrimSpace(cfg.HibernateRegion)
	if cfg.HostClaim && cfg.Region == "" && cfg.HibernateBucket != "" && legacy != "" && park.PlacementRegion(legacy) == legacy {
		return legacy
	}
	return ""
}

func (cfg Config) dbClaimRegion() string {
	if cfg.HostClaim {
		if cfg.Region == "" && cfg.HomeRegion == "" && strings.TrimSpace(cfg.HibernateRegion) == "" {
			return ""
		}
		return cfg.claimPlacementRegion(cfg.Region)
	}
	for _, region := range []string{cfg.Region, cfg.HomeRegion, cfg.HibernateRegion} {
		if region = strings.TrimSpace(region); region != "" {
			return region
		}
	}
	return ""
}

func (s *Server) listDBVersions(ctx context.Context, g dbRequest, level int) (dbdoor.ListResponse, error) {
	// Select the complete baseline once; the client retains it across levels.
	out := dbdoor.ListResponse{Protocol: 2, Plan: true, Files: []dbdoor.LTXFile{}}
	h := g.claim.Claim.Holder
	plan, err := genselect.Select(ctx, s.cfg.DBVersions, genselect.Request{
		Bucket: s.cfg.HibernateBucket, Region: s.cfg.dbClaimRegion(), Host: g.host, Holder: park.HolderRegionView(*h, s.cfg.dbClaimHomeRegion(), s.cfg.dbClaimRegion()),
	})
	if err != nil {
		return dbdoor.ListResponse{}, errDBClaimChanged
	}
	for _, f := range plan.Files {
		var crc *uint64
		if plan.Pin != nil && f.Key == plan.Pin.SnapshotKey && f.VersionID == plan.Pin.SnapshotVersion {
			crc = &plan.Pin.SnapshotCRC64
		}
		token, err := signDBHandle(s.dbHandleKey, dbFileHandle{V: 1, Host: g.host, Location: h.Location, Key: f.Key, VersionID: f.VersionID, Incarnation: h.Incarnation, Expiry: time.Now().Add(dbHandleTTL).Unix(), Account: g.acct.Sqid, App: g.app, Size: f.Size, SnapshotCRC64: crc})
		if err != nil {
			return dbdoor.ListResponse{}, err
		}
		out.Files = append(out.Files, dbdoor.LTXFile{Level: f.Level, MinTXID: f.MinTXID, MaxTXID: f.MaxTXID, Size: f.Size, Handle: token, SnapshotCRC64: crc})
	}
	o2, err := hostclaim.Read(ctx, s.cfg.Releases, g.host)
	if err != nil {
		return dbdoor.ListResponse{}, err
	}
	if !sameDBAdmission(g.claim, o2) {
		return dbdoor.ListResponse{}, errDBClaimChanged
	}
	return out, nil
}

func dbHandleCoordinate(h dbFileHandle, level int, lo, hi uint64) bool {
	if h.Key == dbVersionKey(h.Location, level, lo, hi) {
		return true
	}
	rel, ok := strings.CutPrefix(h.Key, h.Location+"/g-")
	if !ok {
		return false
	}
	gen, tail, ok := strings.Cut(rel, "/")
	return ok && gen != "" && gen != "." && gen != ".." && gen != hostclaim.RootGeneration && !strings.ContainsAny(gen, "\\\x00") && tail == dbVersionKey("", level, lo, hi)[1:]
}

func (s *Server) pinnedDBReader(r *http.Request, g dbRequest, level int, lo, hi uint64) (*dbVersionReader, error) {
	h, err := verifyDBHandle(s.dbHandleKey, r.Header.Get(dbdoor.HandleHeader), time.Now())
	if err != nil {
		return nil, err
	}
	holder := g.claim.Claim.Holder
	if h.Host != g.host || h.Account != g.acct.Sqid || h.App != g.app || h.Location != holder.Location || h.Incarnation != holder.Incarnation || !dbHandleCoordinate(h, level, lo, hi) {
		return nil, errDBHandle
	}
	return &dbVersionReader{store: s.cfg.DBVersions, bucket: s.cfg.HibernateBucket, handle: h}, nil
}

// This adapter keeps the existing stream budget, watchdog and audit path.
// Its only object read names the signed version, including on every resume.
type dbVersionReader struct {
	store  lease.VersionedStore
	bucket string
	handle dbFileHandle
}

func (v *dbVersionReader) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
	return nil, errDBHandle
}
func (v *dbVersionReader) OpenLTXFile(ctx context.Context, level int, lo, hi ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	if !dbHandleCoordinate(v.handle, level, uint64(lo), uint64(hi)) {
		return nil, errDBHandle
	}
	n, err := dbVersionRange(v.handle.Size, offset, size)
	if err != nil {
		return nil, err
	}
	var rc io.ReadCloser
	ranged, ok := v.store.(lease.VersionRangeOpener)
	if ok && n > 0 {
		rc, err = ranged.OpenVersionRange(ctx, v.bucket, v.handle.Key, v.handle.VersionID, offset, n)
	} else {
		rc, err = v.store.OpenVersion(ctx, v.bucket, v.handle.Key, v.handle.VersionID)
	}
	if errors.Is(err, lease.ErrMirrorNotFound) {
		return nil, os.ErrNotExist
	}
	if err != nil {
		return nil, err
	}
	if !ok {
		if _, err = io.CopyN(io.Discard, rc, offset); err != nil {
			rc.Close()
			return nil, err
		}
	}
	return &dbRangeReader{Reader: io.LimitReader(rc, n), Closer: rc}, nil
}

type dbRangeReader struct {
	io.Reader
	io.Closer
}

func dbVersionRange(total, offset, size int64) (int64, error) {
	if total < 0 || offset < 0 || size < 0 || offset > total || (offset == total && total != 0) {
		return 0, errDBHandle
	}
	n := total - offset
	if size != 0 && size < n {
		n = size
	}
	return n, nil
}
func dbVersionHeaders(w http.ResponseWriter, r *http.Request, total, offset, size int64) error {
	n, err := dbVersionRange(total, offset, size)
	if err != nil {
		return err
	}
	w.Header().Set("Content-Length", strconv.FormatInt(n, 10))
	w.Header().Set(dbdoor.HandleHeader, r.Header.Get(dbdoor.HandleHeader))
	if r.Header.Get("Range") != "" && n > 0 {
		w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+n-1, total))
	}
	return nil
}