package host

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/http"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

func TestClaimCompletedProducersBoundStopAndRetry(t *testing.T) {
	t.Parallel()
	c, req, store := claimStopFixture(t)
	nextID := 0
	c.NewID = func() string { nextID++; return fmt.Sprintf("stop-%d", nextID) }
	target := hostguard.FenceTarget{Host: req.Host, Incarnation: req.Incarnation, Stamp: req.InstallationStamp, Session: req.WriterSession, Box: req.Box, Enrollment: req.Enrollment, BootID: req.BootID, ControllerID: req.ControllerID}
	path := filepath.Join(t.TempDir(), "launch.db")
	j, err := hostbox.OpenLaunchJournal(path)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { j.Close() })
	if err := j.Prelaunch(t.Context(), hostbox.LaunchInventory{Target: target, ControllerPID: 42, ProcessStart: "start-42", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "parent", Device: 1, Inode: 2}}}); err != nil {
		t.Fatal(err)
	}
	p, err := NewClaimProducers(target, "start-42", j.RecordProducer, j.CompleteProducer)
	if err != nil {
		t.Fatal(err)
	}
	const deliveries = 1024
	producerID := func(n int) string { return fmt.Sprintf("%01024d", n) }
	set := &ClaimProducerSet{}
	if err := set.Add(p); err != nil {
		t.Fatal(err)
	}
	source := hostbox.WorkSource{Guard: c.Guard, Box: req.Box, Region: "ie", Now: time.Now,
		Policy: func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error { return nil },
		Local:  func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }}
	work, err := source.Select(t.Context(), req.Host, "timer")
	if err != nil {
		t.Fatal(err)
	}
	route := registry.Route{Host: req.Host, ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: req.Incarnation, ClaimStamp: req.InstallationStamp}}
	timer := registry.Timer{Host: req.Host, ClaimOwnership: route.ClaimOwnership}
	seq := 0
	deliver := ClaimTimerDelivery(set, func() string { id := producerID(seq); seq++; return id })
	oldIDs := make([]string, 0, deliveries)
	for n := 0; n < deliveries; n++ {
		id := "timer-" + producerID(n)
		oldIDs = append(oldIDs, id)
		_, finish, err := deliver(t.Context(), work, route, timer)
		if err != nil {
			t.Fatal(err)
		}
		finish()
		finish()
		if len(p.workers) != 0 {
			t.Fatalf("completed delivery %d remains in memory", n)
		}
	}
	legacy, _ := json.Marshal(DrainClaimAcknowledgement("fence", req.ControllerID, "start-42", oldIDs))
	if len(legacy) <= 1<<20 {
		t.Fatal("fixture does not exceed the transport limit")
	}
	if err := j.Close(); err != nil {
		t.Fatal(err)
	}
	j, err = hostbox.OpenLaunchJournal(path)
	if err != nil {
		t.Fatal(err)
	}
	in, err := j.Read(t.Context(), target)
	if err != nil {
		t.Fatal(err)
	}
	raw, _ := json.Marshal(in)
	if len(in.ProducerIDs) != 0 || in.Completed.Count != deliveries || len(in.Completed.Digest) != 64 || len(raw) > 4096 {
		t.Fatalf("unbounded or missing durable completion: %d bytes, %+v", len(raw), in.Completed)
	}
	checks := 0
	c.Kernel = stopKernelFunc(func(ctx context.Context, target hostguard.FenceTarget, _ StopInventory, ack DrainAcknowledgement) (StopKernelEvidence, error) {
		checks++
		in, err := j.ReadStopKernelJournal(ctx, target)
		if err != nil {
			return StopKernelEvidence{}, err
		}
		if err := validateClaimDrainProducers(in.ProducerIDs, ack, in.Completed); err != nil {
			return StopKernelEvidence{}, err
		}
		bad := ack
		bad.Completed.Count++
		if validateClaimDrainProducers(in.ProducerIDs, bad, in.Completed) == nil {
			t.Fatal("changed completion accepted")
		}
		return StopKernelEvidence{InventorySHA256: strings.Repeat("b", 64), LocalJournalSHA256: digestStop(in), CgroupIdentities: []string{"cg-1"}}, nil
	})
	peer, _ := c.Peers.AuthenticatedPeer(t.Context())
	call := func() (StopResult, error) {
		client, serverConn := net.Pipe()
		ln := &stopPipeListener{conn: serverConn, closed: make(chan struct{})}
		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
		h := ClaimStopHTTP{Coordinator: c, Authenticate: func(context.Context, net.Conn) (StopPeer, error) { return peer, nil }}
		s := instancectl.Server{PeerOK: func(net.Conn) error { return nil }, ClaimStop: h}
		done := make(chan error, 1)
		go func() { done <- s.Serve(ctx, ln) }()
		result, err := collectClaimStop(ctx, "unused", req, p, func(context.Context, string, string) (net.Conn, error) { return client, nil })
		cancel()
		if serveErr := <-done; serveErr != nil {
			t.Fatal(serveErr)
		}
		return result, err
	}
	store.err = errors.New("publication failure")
	if _, err := call(); err == nil {
		t.Fatal("publication failure hidden")
	}
	first, err := call()
	if err != nil {
		t.Fatal(err)
	}
	again, err := call()
	if err != nil || first.Digest != again.Digest || !first.EvidenceAt.Equal(again.EvidenceAt) || checks != 1 {
		t.Fatalf("receipt retry changed evidence: %v checks=%d", err, checks)
	}
}

func TestClaimStopDisconnectedPeerCancelsPipe(t *testing.T) {
	t.Parallel()
	c, req, _ := claimStopFixture(t)
	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()
	serverDone := make(chan struct{})
	go func() {
		defer close(serverDone)
		defer server.Close()
		r, err := http.ReadRequest(bufio.NewReader(server))
		if err == nil {
			_ = json.NewDecoder(r.Body).Decode(new(StopRequest))
		}
	}()
	ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
	defer cancel()
	done := make(chan error, 1)
	go func() {
		_, err := collectClaimStop(ctx, "unused", req, c.Drainer, func(context.Context, string, string) (net.Conn, error) { return client, nil })
		done <- err
	}()
	select {
	case err := <-done:
		if err == nil {
			t.Fatal("disconnected peer accepted")
		}
	case <-time.After(2 * time.Second):
		t.Fatal("request pipe ignored cancellation")
	}
	<-serverDone
}

func TestClaimProducerCompactionFailureRetriesBeforeAdmission(t *testing.T) {
	_, req, _ := claimStopFixture(t)
	target := hostguard.FenceTarget{Host: req.Host, Incarnation: req.Incarnation, Stamp: req.InstallationStamp, Session: req.WriterSession, Box: req.Box, Enrollment: req.Enrollment, BootID: req.BootID, ControllerID: req.ControllerID}
	j, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "launch.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer j.Close()
	if err := j.Prelaunch(t.Context(), hostbox.LaunchInventory{Target: target, ControllerPID: 42, ProcessStart: "start-42", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "parent", Device: 1, Inode: 2}}}); err != nil {
		t.Fatal(err)
	}
	fail := true
	p, err := NewClaimProducers(target, "start-42", j.RecordProducer, func(ctx context.Context, target hostguard.FenceTarget, id string, previous hostbox.ProducerCompletion) (hostbox.ProducerCompletion, error) {
		if fail {
			return previous, errors.New("disk failure")
		}
		return j.CompleteProducer(ctx, target, id, previous)
	})
	if err != nil {
		t.Fatal(err)
	}
	_, finish, err := p.Register(t.Context(), target, "timer-first")
	if err != nil {
		t.Fatal(err)
	}
	finish(nil)
	if _, _, err := p.Register(t.Context(), target, "timer-next"); err == nil {
		t.Fatal("admission ignored failed compaction")
	}
	if _, err := p.DrainClaim(t.Context(), req, "fence"); err == nil {
		t.Fatal("drain ignored failed compaction")
	}
	fail = false
	ack, err := p.DrainClaim(t.Context(), req, "fence")
	if err != nil || ack.Completed.Count != 1 || len(p.workers) != 0 {
		t.Fatalf("compaction retry failed: %+v %v", ack, err)
	}
	// A lost successful commit response must also be safe to replay.
	completed, err := j.CompleteProducer(t.Context(), target, "timer-first", hostbox.ProducerCompletion{})
	if err != nil || completed != ack.Completed {
		t.Fatalf("commit replay: %+v %v", completed, err)
	}
}