//go:build linux && rootrunner

package host

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"syscall"
	"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/instancectl"
)

func TestRootServingParkStoppedWriterHandover(t *testing.T) {
	runServingParkAuthenticatedPeer(t, true, true)
}

// Only the fixture unit starts this helper. Keep SQLite open until SIGTERM;
// the root stop verifier must independently observe descriptor loss.
func TestRootServingParkWriterPayload(t *testing.T) {
	path, readyPath := os.Getenv("SERVING_WRITER_DB"), os.Getenv("SERVING_WRITER_READY")
	if path == "" || readyPath == "" {
		t.Fatal("invoke through TestRootServingParkStoppedWriterHandover")
	}
	stop := make(chan os.Signal, 1)
	signal.Notify(stop, syscall.SIGTERM)
	defer signal.Stop(stop)
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	conn, err := db.Conn(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	for _, query := range []string{"PRAGMA journal_mode=WAL", "CREATE TABLE serving_witness(value TEXT)", "INSERT INTO serving_witness VALUES ('tenant committed bytes')"} {
		if _, err := conn.ExecContext(t.Context(), query); err != nil {
			t.Fatal(err)
		}
	}
	start, err := claimConstructionProcessStart()
	if err != nil {
		t.Fatal(err)
	}
	body, err := json.Marshal(StopPeer{PID: os.Getpid(), ProcessStart: start})
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(readyPath, body, 0600); err != nil {
		t.Fatal(err)
	}
	select {
	case <-stop:
	case <-t.Context().Done():
		t.Fatal(t.Context().Err())
	}
}

func runServingPeerStoppedWriter(t *testing.T, f *servingRootFixture, binary string, launch hostbox.LaunchInventory, peer StopPeer, command func(servingPeerCommand, bool) servingPeerReply, kernelResults <-chan error) {
	t.Helper()
	ctx, cancel := context.WithTimeout(t.Context(), 45*time.Second)
	defer cancel()
	unitName := instancectl.UnitName(instancectl.RoleRun, f.current.Host)
	slices := []string{instancectl.SliceName(f.current.Host), "carlos-instances.slice", "carlos.slice"}
	probe, probeErr := exec.CommandContext(ctx, "systemctl", "show", unitName, "-p", "LoadState", "--value").Output()
	if strings.TrimSpace(string(probe)) != "not-found" {
		t.Fatalf("refusing existing/unreadable writer unit: %q %v", probe, probeErr)
	}
	for _, name := range slices {
		out, err := exec.CommandContext(ctx, "systemctl", "show", name, "-p", "ActiveState", "--value").Output()
		if strings.TrimSpace(string(out)) != "inactive" {
			t.Fatalf("refusing existing/unreadable fixture unit %s: %q %v", name, out, err)
		}
	}
	executable, err := stopIdentity(binary)
	if err != nil {
		t.Fatal(err)
	}
	reservation := hostbox.LaunchReservation{ID: "writer:main", Role: "main", InvocationID: "writer", Inputs: "sqlite-witness", Target: launch.Target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup(f.current.Host)}, Executable: hostbox.LaunchFile(executable)}
	// Fixture unit creation and launch journal writes are explicit seams. The
	// identity observations, stop verifier and stop transport are production.
	if err := f.f.journal.IntendUnit(ctx, reservation); err != nil {
		t.Fatal(err)
	}
	readyPath := filepath.Join(filepath.Dir(binary), "writer-ready")
	content := fmt.Sprintf("[Unit]\nDescription=Serving park SQLite witness\n[Service]\nType=simple\nUser=root\nSlice=%s\nExecStart=%s -test.run=^TestRootServingParkWriterPayload$ -test.v -test.timeout=60s\nEnvironment=SERVING_WRITER_DB=%s SERVING_WRITER_READY=%s\nIPAddressDeny=any\nTimeoutStopSec=10s\n", slices[0], binary, f.db, readyPath)
	unitPath := filepath.Join("/run/systemd/system", unitName)
	unit, err := os.OpenFile(unitPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0644)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		for _, name := range append([]string{unitName}, slices...) {
			if out, err := exec.CommandContext(ctx, "systemctl", "stop", name).CombinedOutput(); err != nil {
				t.Errorf("stop fixture %s: %v %s", name, err, out)
			}
		}
		if err := os.Remove(unitPath); err != nil {
			t.Error(err)
		}
		if out, err := exec.CommandContext(ctx, "systemctl", "daemon-reload").CombinedOutput(); err != nil {
			t.Errorf("reload fixture: %v %s", err, out)
		}
	})
	_, writeErr := unit.WriteString(content)
	if err := errors.Join(writeErr, unit.Close()); err != nil {
		t.Fatal(err)
	}
	servingRunnerSystemctl(t, "daemon-reload")
	servingRunnerSystemctl(t, "start", unitName)
	var process StopPeer
	for {
		body, err := os.ReadFile(readyPath)
		if err == nil && json.Unmarshal(body, &process) == nil {
			break
		}
		select {
		case <-ctx.Done():
			t.Fatal("writer readiness", ctx.Err())
		case <-time.After(20 * time.Millisecond):
		}
	}
	properties := func(name string) string {
		t.Helper()
		out, err := exec.CommandContext(ctx, "systemctl", "show", unitName, "-p", name, "--value").Output()
		if err != nil {
			t.Fatal(err)
		}
		return strings.TrimSpace(string(out))
	}
	pid, err := strconv.Atoi(properties("MainPID"))
	if err != nil || pid != process.PID || pid <= 0 {
		t.Fatal("writer MainPID", pid, process.PID, err)
	}
	start, err := claimLaunchProcessStart("/proc", pid)
	if err != nil || process.ProcessStart != start {
		t.Fatalf("writer identity: %+v %v", process, err)
	}
	group := reservation.Group
	group.InvocationID = properties("InvocationID")
	if properties("ControlGroup") != "/"+group.RelativePath || group.InvocationID == "" {
		t.Fatal("writer unit identity")
	}
	groupID, err := stopIdentity(filepath.Join("/sys/fs/cgroup", group.RelativePath))
	if err != nil {
		t.Fatal(err)
	}
	group.Device, group.Inode = groupID.Device, groupID.Inode
	if err := f.f.journal.StartedUnit(ctx, reservation, group, pid, start); err != nil {
		t.Fatal(err)
	}
	id, err := stopIdentity(f.db)
	if err != nil {
		t.Fatal(err)
	}
	if err := f.f.journal.RecordFile(ctx, launch.Target, hostbox.LaunchFile(id)); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(f.db+".gen", []byte(f.current.Generation), 0600); err != nil {
		t.Fatal(err)
	}
	current := f.f.read()
	req := StopRequest{Region: f.actor.Region, Host: current.Host, Incarnation: f.current.Incarnation, InstallationStamp: f.current.Stamp, WriterSession: f.current.Session, Box: f.actor.Box, Enrollment: f.actor.Enrollment, BootID: peer.BootID, ControllerID: peer.ControllerID, ExpectedClaimETag: current.ETag, ExpectedDirectoryRevision: "7", RosterRevision: 7}
	stop := servingPeerCommand{Action: "stop", Stop: req}
	checkNoReceipt := func() {
		t.Helper()
		receipts, err := f.f.guard.CompletedServingStops(ctx, current.Host, hostguard.Captured{Incarnation: f.current.Incarnation, Stamp: f.current.Stamp}, f.current.Session, f.actor.Box)
		if !errors.Is(err, hostguard.ErrNoFenceReceipt) || len(receipts) != 0 {
			t.Fatal("root trusted a drain acknowledgement while a writer remained")
		}
	}
	command(stop, true)
	select {
	case err := <-kernelResults:
		if err == nil {
			t.Fatal("live writer escaped kernel verification")
		}
	default:
		t.Fatal("stop refused before reaching the production kernel verifier")
	}
	checkNoReceipt()
	servingRunnerSystemctl(t, "stop", unitName)
	if properties("MainPID") != "0" || properties("ActiveState") != "inactive" || properties("Result") != "success" {
		t.Fatal("writer unit did not stop cleanly")
	}
	if _, err := claimLaunchProcessStart("/proc", pid); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("writer process still observable: %v", err)
	}
	if err := f.f.journal.SettleUnit(ctx, launch.Target, reservation.ID); err != nil {
		t.Fatal(err)
	}
	// Even an empty group and settled journal cannot hide an independent FD.
	openDB, err := os.Open(f.db)
	if err != nil {
		t.Fatal(err)
	}
	defer openDB.Close()
	command(stop, true)
	select {
	case err := <-kernelResults:
		if err == nil || !strings.Contains(err.Error(), "descriptor remains") {
			t.Fatalf("independent FD refusal: %v", err)
		}
	default:
		t.Fatal("FD refusal did not reach the production kernel verifier")
	}
	checkNoReceipt()
	if err := openDB.Close(); err != nil {
		t.Fatal(err)
	}
	stopped := command(stop, false).Stopped
	select {
	case err := <-kernelResults:
		if err != nil {
			t.Fatal("stopped writer kernel verification", err)
		}
	default:
		t.Fatal("stop succeeded without production kernel verification")
	}
	if err := verifyRetirementReceipt(ctx, hostguard.FenceReceipt{Target: stopped.Target, Bytes: stopped.Bytes, EvidenceAt: stopped.EvidenceAt}); err != nil {
		t.Fatal("independent stop receipt", err)
	}
	if replay := command(stop, false).Stopped; !reflect.DeepEqual(replay, stopped) {
		t.Fatal("stop replay changed receipt")
	}
	before, err := os.ReadFile(f.db)
	if err != nil {
		t.Fatal(err)
	}
	previous := f.f.read()
	h := previous.Claim.Holder
	admitted, err := hostclaim.Update(ctx, f.f.claims, previous, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: f.actor.Box, Region: f.actor.Region, PredecessorRegion: f.actor.Region, Mode: "cold", NewGeneration: "G3"}))
	if err != nil {
		t.Fatal(err)
	}
	prior := f.current
	f.current, err = servingWriterIdentity(admitted, f.actor.Box, f.actor.Region)
	if err != nil {
		t.Fatal(err)
	}
	command(servingPeerCommand{Action: "handover", Previous: previous, Admitted: admitted, Actor: f.actor, Target: stopped.Target}, false)
	command(servingPeerCommand{Action: "prelaunch", Target: recoveryTuple(f.f.read(), claimStopLocal{Actor: f.actor})}, false)
	q := servingPeerCommand{Action: "park", Current: f.current, Prior: prior}
	parked := command(q, false).Authorization
	f.tags(f.current.Stamp)
	command(servingPeerCommand{Action: "complete", Current: f.current, ID: parked.ID}, false)
	record, err := f.p.ledger.read(parked.ID)
	if err != nil || !record.Complete || record.Disposition != servingParkRetain || len(record.Receipts) != 1 || !bytes.Equal(record.Receipts[0].Bytes, stopped.Bytes) {
		t.Fatalf("park did not use independent stop: %+v %v", record, err)
	}
	parkPath := filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", parked.ID, filepath.Base(f.db))
	after, err := os.ReadFile(parkPath)
	if err != nil || !bytes.Equal(before, after) {
		t.Fatal("park changed committed SQLite bytes", err)
	}
	if _, err := os.Lstat(f.db); !errors.Is(err, os.ErrNotExist) {
		t.Fatal("park left live database", err)
	}
	if replay := command(q, false).Authorization; replay != parked {
		t.Fatal("stopped-writer park replay changed authority")
	}
	for name, want := range map[string]string{"files/upload": "tenant upload", "env": "tenant configuration"} {
		body, err := os.ReadFile(filepath.Join(filepath.Dir(f.db), name))
		if err != nil || string(body) != want {
			t.Fatal("writer retirement changed preserved state", name, err)
		}
	}
	if err := f.p.ledger.fs.CheckParked(&record); err != nil {
		t.Fatal(err)
	}
	servingRestoreAfterStoppedWriter(t, f, servingPeerRetirement{command: command})
	check, err := f.p.ledger.read(parked.ID)
	if err != nil || !reflect.DeepEqual(record, check) {
		t.Fatal("stopped-writer park replay changed record", err)
	}
}

// Adapt the fixture's control connection to the production client in the edge
// process. Root never receives a retirement request from the test controller.
type servingPeerRetirement struct {
	command func(servingPeerCommand, bool) servingPeerReply
}

func (p servingPeerRetirement) Park(ctx context.Context, current, prior servingArtifactIdentity) (servingParkAuthorization, error) {
	if err := ctx.Err(); err != nil {
		return servingParkAuthorization{}, err
	}
	return p.command(servingPeerCommand{Action: "park", Current: current, Prior: prior}, false).Authorization, nil
}

func (p servingPeerRetirement) Complete(ctx context.Context, id string, current servingArtifactIdentity) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	p.command(servingPeerCommand{Action: "complete", Current: current, ID: id}, false)
	return nil
}

func (servingPeerRetirement) ParkLegacy(context.Context, servingArtifactIdentity, servingLegacyPredecessor) (servingParkAuthorization, error) {
	return servingParkAuthorization{}, hostbox.ErrUnsupported
}