//go:build linux

package host

import (
	"context"
	"errors"
	"net"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"golang.org/x/sys/unix"
)

func recoveryPage(t *testing.T, p *servingParkRoot) servingParkRecoveryPage {
	t.Helper()
	page, err := p.recoverServingParks(t.Context(), "", servingParkRecoveryLimit)
	if err != nil || page.More || len(page.Results) != 1 {
		t.Fatalf("recovery page: %+v %v", page, err)
	}
	return page
}

func TestServingParkStartupResumesRecordedIntent(t *testing.T) {
	for _, boundary := range []string{"first-rename", "last-rename", "parked", "retagged", "complete-slot-pending"} {
		t.Run(boundary, func(t *testing.T) {
			f := newServingRootFixture(t, true)
			fs := f.p.ledger.fs.(*servingParkFS)
			count := 0
			fs.rename = func(a int, b string, c int, d string, flags uint) error {
				err := unix.Renameat2(a, b, c, d, flags)
				count++
				if err == nil && (boundary == "first-rename" && count == 1 || boundary == "last-rename" && count == 4) {
					return unix.EIO
				}
				return err
			}
			a, err := f.park()
			if strings.HasSuffix(boundary, "rename") {
				if !errors.Is(err, unix.EIO) {
					t.Fatalf("lost rename reply: %v", err)
				}
				a.ID = servingRetirementID(f.prior, f.current, false)
			} else if err != nil {
				t.Fatal(err)
			}
			before, err := f.p.ledger.read(a.ID)
			if err != nil {
				t.Fatal(err)
			}
			if boundary == "retagged" || boundary == "complete-slot-pending" {
				f.tags(f.current.Stamp)
			}
			if boundary == "complete-slot-pending" {
				before.Complete = true
				if err = f.p.ledger.save(before); err != nil {
					t.Fatal(err)
				}
			}
			claimBefore := f.f.read()
			// New authority instance, same durable stores. Startup has no agent
			// connection and must not invoke a transport peer callback.
			restarted := *f.p
			ledger := *f.p.ledger
			restarted.ledger = &ledger
			newfs := newServingParkFS(fs.data, fs.state)
			newfs.owner, newfs.mount = fs.owner, fs.mount
			ledger.fs = newfs
			restarted.root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { panic("startup authenticated an agent") }
			page := recoveryPage(t, &restarted)
			wantComplete := boundary == "retagged" || boundary == "complete-slot-pending"
			if wantComplete && page.Results[0].Err != nil || !wantComplete && !errors.Is(page.Results[0].Err, errServingParkRetagPending) {
				t.Fatalf("unexpected recovery: %+v", page.Results)
			}
			after, err := ledger.read(a.ID)
			if err != nil || !after.Parked || after.Complete != wantComplete || after.Authorization != before.Authorization || !sameServingRootStops(after.Receipts, before.Receipts) {
				t.Fatalf("changed intent or incomplete rename: %+v %v", after, err)
			}
			if !before.ParkedAt.IsZero() && (!after.ParkedAt.Equal(before.ParkedAt) || !after.ReapAfter.Equal(before.ReapAfter)) {
				t.Fatal("startup refreshed retention")
			}
			if !reflect.DeepEqual(f.f.read(), claimBefore) {
				t.Fatal("startup changed claim or retry state")
			}
			for suffix, want := range map[string]string{"": "served bytes", ".gen": "G", "-wal": "WAL", "-shm": "SHM"} {
				body, err := os.ReadFile(filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", a.ID, f.current.Host+".db"+suffix))
				if err != nil || string(body) != want {
					t.Fatalf("retained bytes %s: %q %v", suffix, body, err)
				}
			}
			if !wantComplete {
				f.tags(f.current.Stamp)
				if page = recoveryPage(t, &restarted); page.Results[0].Err != nil {
					t.Fatal(page.Results[0].Err)
				}
			}
			slot, err := ledger.slot(after.Authorization)
			if err != nil || slot.Pending != "" || slot.Last != a.ID || slot.ID != a.ID {
				t.Fatalf("completion replay references: %+v %v", slot, err)
			}
		})
	}
}

func TestServingParkStartupRefusesChangedAuthority(t *testing.T) {
	for _, change := range []string{"enrollment", "binding", "claim", "tags", "directory", "receipt", "launch-intent", "census", "cancel"} {
		t.Run(change, func(t *testing.T) {
			f := newServingRootFixture(t, true)
			a, err := f.park()
			if err != nil {
				t.Fatal(err)
			}
			f.tags(f.current.Stamp)
			before, _ := f.disk.Read("retirement-" + a.ID + ".json")
			ctx := t.Context()
			switch change {
			case "enrollment":
				f.actor.Enrollment = "new-enrollment"
			case "binding":
				f.actor.Binding = "8"
			case "claim":
				_, err = hostclaim.Update(ctx, f.f.claims, f.f.read(), func(c *hostclaim.Claim) error { c.Holder.WriterBox = "other-box"; return nil })
			case "tags":
				f.tags(f.current.Stamp + 1)
			case "directory":
				err = os.Rename(filepath.Dir(f.db), filepath.Dir(f.db)+"-saved")
				if err == nil {
					err = os.Mkdir(filepath.Dir(f.db), 0700)
				}
			case "receipt":
				f.mutateRoot(func(m map[string]any) { m["State"].(map[string]any)["fence_journal"] = []any{} })
			case "launch-intent":
				target := f.receipt.Target
				target.Stamp, target.Session = f.current.Stamp, f.current.Session
				err = f.f.journal.Prelaunch(ctx, hostbox.LaunchInventory{Target: target, ControllerPID: 77, ProcessStart: "edge-start", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/parent", Device: 1, Inode: 2}}})
				if err == nil {
					err = f.f.journal.IntendUnit(ctx, hostbox.LaunchReservation{ID: "new:main", Role: "main", InvocationID: "new", Inputs: "digest", Target: target, Group: hostbox.LaunchGroup{RelativePath: "system.slice/carlos-app@shop.service"}, Executable: hostbox.LaunchFile{Device: 1, Inode: 3}})
				}
			case "census":
				f.p.root.RequireNewWork = func(context.Context) error { return hostbox.ErrChanged }
			case "cancel":
				var cancel context.CancelFunc
				ctx, cancel = context.WithCancel(ctx)
				cancel()
			}
			if err != nil {
				t.Fatal(err)
			}
			page, err := f.p.recoverServingParks(ctx, "", 128)
			if err == nil && (len(page.Results) != 1 || page.Results[0].Err == nil) {
				t.Fatalf("accepted changed %s: %+v", change, page)
			}
			after, _ := f.disk.Read("retirement-" + a.ID + ".json")
			if string(after) != string(before) {
				t.Fatal("refusal changed private retirement")
			}
		})
	}
}

func TestServingParkStartupDeletionAndEmptyEvent(t *testing.T) {
	for _, boundary := range []string{"unlink", "rmdir", "closed-old-event", "replacement", "unknown-child", "missing-no-intent"} {
		t.Run(boundary, func(t *testing.T) {
			f := newServingParkDiskFixture(t)
			f.put("", "retain")
			a, receipts := f.evidence(10, true, "missing")
			_, err := f.park(a, receipts)
			if err != nil {
				t.Fatal(err)
			}
			f.complete(a)
			f.now = f.now.Add(7 * 24 * time.Hour)
			event := filepath.Dir(f.path(a, ""))
			held, err := os.Open(event)
			if err != nil {
				t.Fatal(err)
			}
			defer held.Close()
			f.fs.unlink = func(fd int, name string, flags int) error {
				if boundary == "closed-old-event" && flags == unix.AT_REMOVEDIR {
					return unix.EIO
				}
				err := unix.Unlinkat(fd, name, flags)
				if err == nil && (boundary == "unlink" && flags == 0 || boundary != "unlink" && flags == unix.AT_REMOVEDIR) {
					return unix.EIO
				}
				return err
			}
			err = f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) })
			if !errors.Is(err, unix.EIO) {
				t.Fatalf("missing fault: %v", err)
			}
			r, err := f.ledger.read(a.ID)
			if err != nil {
				t.Fatal(err)
			}
			if boundary == "closed-old-event" {
				r.Closed, r.EventRemoveIntent = true, false
				if err = f.ledger.save(r); err != nil {
					t.Fatal(err)
				}
			}
			if boundary == "replacement" || boundary == "unknown-child" {
				if err = os.Mkdir(event, 0700); err != nil {
					t.Fatal(err)
				}
				if boundary == "unknown-child" {
					err = os.WriteFile(filepath.Join(event, "victim"), []byte("keep"), 0600)
				}
				if err != nil {
					t.Fatal(err)
				}
			}
			if boundary == "missing-no-intent" {
				r.EventRemoveIntent = false
				if err = f.ledger.save(r); err != nil {
					t.Fatal(err)
				}
			}
			f.fs.unlink = unix.Unlinkat
			// No live claim/root host installation is needed for a deletion
			// already authorized by a complete private retirement.
			err = f.ledger.recoverCompleted(t.Context(), a.ID)
			refuse := boundary == "replacement" || boundary == "unknown-child" || boundary == "missing-no-intent"
			if (err != nil) != refuse {
				t.Fatalf("recovery: %v, refuse=%v", err, refuse)
			}
			if refuse {
				if boundary == "replacement" || boundary == "unknown-child" {
					if _, err = os.Lstat(event); err != nil {
						t.Fatal("refusal removed replacement", err)
					}
				}
				if boundary == "unknown-child" {
					body, err := os.ReadFile(filepath.Join(event, "victim"))
					if err != nil || string(body) != "keep" {
						t.Fatal("refusal removed unknown child", err)
					}
				}
				return
			}
			r, err = f.ledger.read(a.ID)
			if err != nil || !r.Closed || !r.EventRemoved {
				t.Fatalf("unfinished cleanup: %+v %v", r, err)
			}
			if _, err = os.Lstat(event); !errors.Is(err, os.ErrNotExist) {
				t.Fatal("empty event survived", err)
			}
			slot, err := f.ledger.slot(a)
			if err != nil || slot.ID != a.ID || slot.Last != a.ID {
				t.Fatal("lost replay reference", err)
			}
			if err = f.ledger.recoverCompleted(t.Context(), a.ID); err != nil {
				t.Fatal("cleanup replay", err)
			}
		})
	}
}

func TestServingParkStartupBoundedScanContinuesAfterCorruptRecord(t *testing.T) {
	f := newServingRootFixture(t, true)
	a, err := f.park()
	if err != nil {
		t.Fatal(err)
	}
	f.tags(f.current.Stamp)
	if err = f.disk.Write("retirement-0000.json", []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	page, err := f.p.recoverServingParks(t.Context(), "", 1)
	if err != nil || !page.More || len(page.Results) != 1 || page.Results[0].Host != "unknown" || page.Results[0].Err == nil {
		t.Fatalf("first page: %+v %v", page, err)
	}
	page, err = f.p.recoverServingParks(t.Context(), page.Next, 1)
	if err != nil || page.More || len(page.Results) != 1 || page.Results[0].ID != a.ID || page.Results[0].Err != nil {
		t.Fatalf("second page: %+v %v", page, err)
	}
}

func TestServingParkStartupLegacyKeepsConsumption(t *testing.T) {
	for _, empty := range []bool{false, true} {
		t.Run(map[bool]string{false: "legacy", true: "empty"}[empty], func(t *testing.T) {
			f, _ := newServingLegacyRootFixture(t, empty)
			a, err := f.p.parkLegacyServingDatabase(t.Context(), f.conn, f.current, servingLegacyPredecessor{Empty: empty})
			if err != nil {
				t.Fatal(err)
			}
			key := servingLegacyKey("legacy", f.current)
			consumedBefore, err := f.disk.Read(key)
			if err != nil {
				t.Fatal(err)
			}
			stopBefore, err := f.disk.Read(servingLegacyKey("legacy-stop", f.current))
			if err != nil {
				t.Fatal(err)
			}
			before, err := f.p.ledger.read(a.ID)
			if err != nil {
				t.Fatal(err)
			}
			f.tags(f.current.Stamp)
			page := recoveryPage(t, f.p)
			if page.Results[0].Err != nil {
				t.Fatal(page.Results[0].Err)
			}
			after, err := f.p.ledger.read(a.ID)
			if err != nil || !after.Complete || after.Authorization != before.Authorization || !after.ReapAfter.Equal(before.ReapAfter) {
				t.Fatalf("legacy recovery: %+v %v", after, err)
			}
			consumedAfter, _ := f.disk.Read(key)
			stopAfter, _ := f.disk.Read(servingLegacyKey("legacy-stop", f.current))
			if string(consumedBefore) != string(consumedAfter) || string(stopBefore) != string(stopAfter) {
				t.Fatal("startup changed legacy evidence")
			}
		})
	}
}

func TestServingParkStartupRuntimeJoinsBeforeClose(t *testing.T) {
	f := newServingRootFixture(t, true)
	if _, err := f.park(); err != nil {
		t.Fatal(err)
	}
	f.tags(f.current.Stamp)
	entered, release := make(chan struct{}), make(chan struct{})
	f.p.root.RequireNewWork = func(ctx context.Context) error {
		close(entered)
		<-ctx.Done()
		<-release
		return ctx.Err()
	}
	closed := &servingCloseStore{Store: f.disk}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	p := &servingParkRuntime{authority: f.p, state: closed, ctx: ctx, cancel: cancel, idle: make(chan struct{})}
	done := make(chan error, 1)
	go func() { done <- p.recoverPending(t.Context()) }()
	select {
	case <-entered:
	case <-time.After(5 * time.Second):
		t.Fatal("recovery did not enter authority")
	}
	short, stop := context.WithTimeout(t.Context(), 20*time.Millisecond)
	defer stop()
	if err := p.Close(short); !errors.Is(err, context.DeadlineExceeded) || closed.closed.Load() {
		t.Fatal("closed underneath recovery", err)
	}
	close(release)
	if err := <-done; err == nil {
		t.Fatal("cancelled recovery succeeded")
	}
	if err := p.Close(t.Context()); err != nil || !closed.closed.Load() {
		t.Fatal("recovery did not join", err)
	}
	if err := p.recoverPending(t.Context()); err == nil {
		t.Fatal("closed runtime recovered")
	}
}

func TestServingParkStartupRunsAfterOtherRecoveryFailure(t *testing.T) {
	f := newServingRootFixture(t, true)
	a, err := f.park()
	if err != nil {
		t.Fatal(err)
	}
	f.tags(f.current.Stamp)
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	p := &servingParkRuntime{authority: f.p, state: f.disk, ctx: ctx, cancel: cancel, idle: make(chan struct{})}
	defer p.Close(t.Context())
	h := f.p.root
	h.servingParks = p
	bad := filepath.Join(t.TempDir(), "not-a-directory")
	if err = os.WriteFile(bad, nil, 0600); err != nil {
		t.Fatal(err)
	}
	h.Reconcile.Retirement.JournalDir = bad
	if err = h.RecoverPending(t.Context()); err == nil {
		t.Fatal("other recovery error was swallowed")
	}
	r, err := f.p.ledger.read(a.ID)
	if err != nil || !r.Complete {
		t.Fatalf("serving recovery skipped: %+v %v", r, err)
	}
}