package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"net"
	"net/http"
	"slices"
	"sort"
	"time"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/instancectl"
)

type ClaimStopHTTP struct {
	Coordinator  ClaimStopCoordinator
	Authenticate func(context.Context, net.Conn) (StopPeer, error)
}
type stopPeerValue struct{ peer StopPeer }

func (p stopPeerValue) AuthenticatedPeer(context.Context) (StopPeer, error) { return p.peer, nil }

type stopFrame struct {
	FenceID string
	Result  *StopResult
	Failed  bool
}
type stopStreamDrain struct {
	w       http.ResponseWriter
	decode  *json.Decoder
	peer    StopPeer
	recheck func(context.Context) (StopPeer, error)
}

func (s stopStreamDrain) DrainClaim(ctx context.Context, req StopRequest, fence string) (DrainAcknowledgement, error) {
	if err := json.NewEncoder(s.w).Encode(stopFrame{FenceID: fence}); err != nil {
		return DrainAcknowledgement{}, err
	}
	if err := http.NewResponseController(s.w).Flush(); err != nil {
		return DrainAcknowledgement{}, err
	}
	var ack DrainAcknowledgement
	if err := s.decode.Decode(&ack); err != nil {
		return ack, err
	}
	if ack.ProcessStart != s.peer.ProcessStart || ack.ControllerID != s.peer.ControllerID || ack.FenceID != fence {
		return ack, ErrClaimStopIdentity
	}
	current, err := s.recheck(ctx)
	if err != nil || current != s.peer {
		return ack, ErrClaimStopIdentity
	}
	return ack, ctx.Err()
}
func (h ClaimStopHTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost || h.Authenticate == nil {
		w.WriteHeader(http.StatusServiceUnavailable)
		return
	}
	conn := instancectl.RequestConn(r.Context())
	if conn == nil {
		w.WriteHeader(http.StatusUnauthorized)
		return
	}
	peer, err := h.Authenticate(r.Context(), conn)
	if err != nil {
		w.WriteHeader(http.StatusUnauthorized)
		return
	}
	if r.URL.Path == "/claim/writers-stopped" {
		h.serveWritersStopped(w, r, peer)
		return
	}
	rc := http.NewResponseController(w)
	if err := rc.EnableFullDuplex(); err != nil {
		w.WriteHeader(http.StatusServiceUnavailable)
		return
	}
	deadline := time.Now().Add(2 * time.Minute)
	_ = rc.SetReadDeadline(deadline)
	_ = rc.SetWriteDeadline(deadline)
	decoder := json.NewDecoder(io.LimitReader(r.Body, 1<<20))
	decoder.DisallowUnknownFields()
	var req StopRequest
	if err := decoder.Decode(&req); err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	w.Header().Set("Content-Type", "application/x-ndjson")
	c := h.Coordinator
	c.Peers = stopPeerValue{peer}
	c.Drainer = stopStreamDrain{w: w, decode: decoder, peer: peer, recheck: func(ctx context.Context) (StopPeer, error) { return h.Authenticate(ctx, conn) }}
	result, err := c.FenceAndStop(r.Context(), req)
	if err != nil {
		_ = json.NewEncoder(w).Encode(stopFrame{Failed: true})
		return
	}
	_ = json.NewEncoder(w).Encode(stopFrame{Result: &result})
}
func digestStop(v any) string {
	b, _ := json.Marshal(v)
	s := sha256.Sum256(b)
	return hex.EncodeToString(s[:])
}
func DrainClaimAcknowledgement(fence, controller, start string, ids []string, completed ...hostbox.ProducerCompletion) DrainAcknowledgement {
	ids = append([]string{}, ids...)
	sort.Strings(ids)
	ack := DrainAcknowledgement{FenceID: fence, ControllerID: controller, ProcessStart: start, ProducerIDs: ids}
	if len(completed) == 1 {
		ack.Completed = completed[0]
	}
	ack.Digest = claimDrainDigest(ack)
	return ack
}
func CollectClaimStop(ctx context.Context, path string, req StopRequest, drain StopDrainer) (StopResult, error) {
	return collectClaimStop(ctx, path, req, drain, func(ctx context.Context, _, _ string) (net.Conn, error) {
		return (&net.Dialer{}).DialContext(ctx, "unix", path)
	})
}
func collectClaimStop(ctx context.Context, path string, req StopRequest, drain StopDrainer, dial func(context.Context, string, string) (net.Conn, error)) (StopResult, error) {
	if drain == nil {
		return StopResult{}, ErrClaimStopUnavailable
	}
	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
	defer cancel()
	reader, writer := io.Pipe()
	defer reader.Close()
	defer writer.Close()
	stopCancel := context.AfterFunc(ctx, func() {
		_ = reader.CloseWithError(ctx.Err())
		_ = writer.CloseWithError(ctx.Err())
	})
	defer stopCancel()
	transport := &http.Transport{DisableKeepAlives: true, DialContext: dial}
	defer transport.CloseIdleConnections()
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://control/claim/stop", reader)
	if err != nil {
		return StopResult{}, err
	}
	written := make(chan error, 1)
	go func() { written <- json.NewEncoder(writer).Encode(req) }()
	response, err := (&http.Client{Transport: transport}).Do(request)
	if err != nil {
		return StopResult{}, err
	}
	defer func() {
		_ = writer.Close()
		_ = response.Body.Close()
	}()
	if response.StatusCode != http.StatusOK {
		return StopResult{}, ErrClaimStopUnavailable
	}
	if err := <-written; err != nil {
		return StopResult{}, err
	}
	decoder := json.NewDecoder(io.LimitReader(response.Body, 1<<20))
	decoder.DisallowUnknownFields()
	var frame stopFrame
	if err := decoder.Decode(&frame); err != nil {
		return StopResult{}, err
	}
	if frame.Failed {
		return StopResult{}, ErrClaimStopUnavailable
	}
	if frame.Result != nil {
		return validateTransportStop(req, *frame.Result)
	}
	if frame.FenceID == "" {
		return StopResult{}, ErrClaimStopIdentity
	}
	ack, err := drain.DrainClaim(ctx, req, frame.FenceID)
	if err != nil {
		return StopResult{}, err
	}
	if err := json.NewEncoder(writer).Encode(ack); err != nil {
		return StopResult{}, err
	}
	_ = writer.Close()
	if err := decoder.Decode(&frame); err != nil {
		return StopResult{}, err
	}
	if frame.Failed || frame.Result == nil {
		return StopResult{}, ErrClaimStopUnavailable
	}
	result := *frame.Result
	if result.Digest != digestRawStop(result.Bytes) {
		return StopResult{}, errors.New("claim stop receipt digest mismatch")
	}
	return validateTransportStop(req, result)
}
func digestRawStop(b []byte) string { s := sha256.Sum256(b); return hex.EncodeToString(s[:]) }

func validateTransportStop(req StopRequest, result StopResult) (StopResult, error) {
	if result.Digest != digestRawStop(result.Bytes) {
		return StopResult{}, ErrClaimStopReceipt
	}
	var receipt StopReceipt
	if err := json.Unmarshal(result.Bytes, &receipt); err != nil {
		return StopResult{}, err
	}
	if receipt.Region != req.Region || receipt.Host != req.Host || receipt.Incarnation != req.Incarnation || receipt.InstallationStamp != req.InstallationStamp || receipt.WriterSession != req.WriterSession || receipt.OperationID != req.OperationID || receipt.Phase != req.Phase || receipt.ObligationID != req.ObligationID || receipt.Box != req.Box || receipt.Enrollment != req.Enrollment || receipt.BootID != req.BootID || receipt.ControllerID != req.ControllerID || receipt.Binding != req.ExpectedDirectoryRevision || !result.EvidenceAt.Equal(receipt.EvidenceAt) {
		return StopResult{}, ErrClaimStopIdentity
	}
	// The numeral is the fence's, not this request's: a retry after roster
	// writes returns the original receipt with its original numeral.
	numeral, ok := receiptRosterRevision(receipt.DirectoryRevision)
	if !ok {
		return StopResult{}, ErrClaimStopReceipt
	}

	expected := hostguard.FenceTarget{Region: receipt.Region, FenceID: receipt.FenceID, ClaimGen: receipt.ClaimGen, ObligationID: receipt.ObligationID, Host: receipt.Host, Incarnation: receipt.Incarnation, Stamp: receipt.InstallationStamp, Session: receipt.WriterSession, Op: receipt.OperationID, Phase: receipt.Phase, Box: receipt.Box, Enrollment: receipt.Enrollment, BootID: receipt.BootID, ControllerID: receipt.ControllerID, ClaimETag: receipt.ClaimETag, DirectoryRevision: receipt.Binding, RosterRevision: numeral}
	if result.Target != expected {
		return StopResult{}, ErrClaimStopIdentity
	}
	return result, nil
}

// validateClaimDrainProducers is shared by the kernel verifier and fixtures.
func validateClaimDrainProducers(producers []string, ack DrainAcknowledgement, completed ...hostbox.ProducerCompletion) error {
	expected := hostbox.ProducerCompletion{}
	if len(completed) > 1 {
		return ErrClaimStopIdentity
	}
	if len(completed) == 1 {
		expected = completed[0]
	}
	ids := append([]string(nil), producers...)
	sort.Strings(ids)
	if !expected.Valid() || expected != ack.Completed || !slices.Equal(ids, ack.ProducerIDs) || ack.Digest != claimDrainDigest(ack) {
		return ErrClaimStopIdentity
	}
	return nil
}

func claimDrainDigest(ack DrainAcknowledgement) string {
	if ack.Completed == (hostbox.ProducerCompletion{}) {
		return digestStop(struct {
			FenceID, ControllerID, ProcessStart string
			ProducerIDs                         []string
		}{ack.FenceID, ack.ControllerID, ack.ProcessStart, ack.ProducerIDs})
	}
	return digestStop(struct {
		FenceID, ControllerID, ProcessStart string
		ProducerIDs                         []string
		Completed                           hostbox.ProducerCompletion
	}{ack.FenceID, ack.ControllerID, ack.ProcessStart, ack.ProducerIDs, ack.Completed})
}