package controller

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"testing"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	bindv1alpha1 "git.unkin.net/unkin/bind-operator/api/v1alpha1"
	"git.unkin.net/unkin/bind-operator/internal/bind"
)

func testScheme(t *testing.T) *runtime.Scheme {
	t.Helper()
	s := runtime.NewScheme()
	if err := clientgoscheme.AddToScheme(s); err != nil {
		t.Fatal(err)
	}
	if err := bindv1alpha1.AddToScheme(s); err != nil {
		t.Fatal(err)
	}
	return s
}

// fakeExec mirrors bind.Executor: stdout is returned, stderr only inside the
// error. It serves `cat` from mounted (per pod) and counts reconfigs.
type fakeExec struct {
	mounted         map[string]string
	catErr          map[string]error
	reconfigErr     error
	checkconfStderr string
	reconfigs       map[string]int
}

func (f *fakeExec) Exec(_ context.Context, _, pod string, cmd []string, _ string) (string, error) {
	if len(cmd) == 2 && cmd[0] == bind.CheckconfBin {
		return "", fmt.Errorf("exec %v: command terminated with exit code 1 (stderr: %s)", cmd, f.checkconfStderr)
	}
	if len(cmd) != 3 || cmd[0] != "cat" || cmd[1] != bind.NamedConfRun || cmd[2] != bind.KeysConfPath {
		return "", errors.New("unexpected command")
	}
	if err := f.catErr[pod]; err != nil {
		return "", err
	}
	return f.mounted[pod], nil
}

func (f *fakeExec) Reconfig(_ context.Context, _, pod string) error {
	f.reconfigs[pod]++
	return f.reconfigErr
}

func readyPod(cluster, name string) *corev1.Pod {
	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", Labels: commonLabels(cluster)},
		Status:     corev1.PodStatus{Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}},
	}
}

func TestPodTemplateIgnoresConfigButTracksPodSpec(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{
		ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"},
		Spec:       bindv1alpha1.BindClusterSpec{Replicas: 2},
	}
	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster).Build()
	r := &BindClusterReconciler{Client: cl, Scheme: scheme}
	ctx := context.Background()

	template := func() corev1.PodTemplateSpec {
		t.Helper()
		sts, err := r.reconcileStatefulSet(ctx, cluster)
		if err != nil {
			t.Fatal(err)
		}
		return sts.Spec.Template
	}
	setConfig := func(conf, keys string) {
		t.Helper()
		if err := r.upsertConfigMap(ctx, cluster, configMapName("c"), map[string]string{"named.conf.secondary": conf}); err != nil {
			t.Fatal(err)
		}
		if err := r.upsertSecret(ctx, cluster, keysSecretName("c"), map[string][]byte{"keys.conf": []byte(keys)}); err != nil {
			t.Fatal(err)
		}
	}

	setConfig("options { recursion yes; };", "")
	base := template()
	setConfig(`options { recursion no; }; zone "x" { type forward; forwarders { 1.1.1.1; }; };`, "key k {};")
	if got := template(); !equality.Semantic.DeepEqual(base, got) {
		t.Fatal("pod template changed on a rendered-config change; that restarts every pod")
	}

	cluster.Spec.Image = "internetsystemsconsortium/bind9:9.21"
	withImage := template()
	if equality.Semantic.DeepEqual(base, withImage) {
		t.Fatal("pod template must change when the image changes")
	}

	cluster.Spec.Resources = corev1.ResourceRequirements{Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("1Gi")}}
	if equality.Semantic.DeepEqual(withImage, template()) {
		t.Fatal("pod template must change when resources change")
	}
}

func TestReconfigPodsWaitsForMountThenReconfigsOnce(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"}}
	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(readyPod("c", "c-0"), readyPod("c", "c-1")).Build()
	ex := &fakeExec{
		mounted:   map[string]string{"c-0": "old-primary", "c-1": "old-secondary"},
		reconfigs: map[string]int{},
	}
	r := &BindClusterReconciler{Client: cl, Scheme: scheme, Exec: ex}
	ctx := context.Background()
	const primary, secondary, keys = "new-primary;", "new-secondary;", "key k {};"

	if d := r.reconfigPods(ctx, cluster, primary, secondary, keys); d != requeueShort {
		t.Fatalf("stale mounts should requeue short, got %v", d)
	}
	if len(ex.reconfigs) != 0 {
		t.Fatalf("reconfig ran before the mount caught up: %v", ex.reconfigs)
	}
	if c := meta.FindStatusCondition(cluster.Status.Conditions, "ConfigApplied"); c == nil || c.Reason != "Propagating" {
		t.Fatalf("want Propagating condition, got %+v", c)
	}

	// Only the primary's mount has propagated.
	ex.mounted["c-0"] = primary + keys
	if d := r.reconfigPods(ctx, cluster, primary, secondary, keys); d != requeueShort {
		t.Fatalf("secondary still stale should requeue, got %v", d)
	}
	if ex.reconfigs["c-0"] != 1 || ex.reconfigs["c-1"] != 0 {
		t.Fatalf("want only c-0 reconfigured, got %v", ex.reconfigs)
	}

	ex.mounted["c-1"] = secondary + keys
	for range 3 {
		if d := r.reconfigPods(ctx, cluster, primary, secondary, keys); d != 0 {
			t.Fatalf("converged cluster should not requeue, got %v", d)
		}
	}
	if ex.reconfigs["c-0"] != 1 || ex.reconfigs["c-1"] != 1 {
		t.Fatalf("each pod must reconfig exactly once, got %v", ex.reconfigs)
	}
	if c := meta.FindStatusCondition(cluster.Status.Conditions, "ConfigApplied"); c == nil || c.Status != metav1.ConditionTrue {
		t.Fatalf("want ConfigApplied=True, got %+v", c)
	}

	// A pod that no longer exists drops out of the applied map.
	if err := cl.Delete(ctx, readyPod("c", "c-1")); err != nil {
		t.Fatal(err)
	}
	r.reconfigPods(ctx, cluster, primary, secondary, keys)
	if _, ok := cluster.Status.AppliedConfig["c-1"]; ok {
		t.Fatal("applied map kept a deleted pod")
	}
}

func TestReconfigPodsSurfacesFailure(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"}}
	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(readyPod("c", "c-0")).Build()
	ex := &fakeExec{
		mounted:         map[string]string{"c-0": "bad;"},
		reconfigErr:     errors.New("rndc: 'reconfig' failed: unexpected token"),
		reconfigs:       map[string]int{},
		checkconfStderr: "/run/named/named.conf:6: expected IP address near 'not-an-ip'",
	}
	r := &BindClusterReconciler{Client: cl, Scheme: scheme, Exec: ex}

	if d := r.reconfigPods(context.Background(), cluster, "bad;", "", ""); d != requeueLong {
		t.Fatalf("failure should requeue long, got %v", d)
	}
	c := meta.FindStatusCondition(cluster.Status.Conditions, "ConfigApplied")
	if c == nil || c.Status != metav1.ConditionFalse || c.Reason != "ReconfigFailed" {
		t.Fatalf("want ReconfigFailed condition, got %+v", c)
	}
	if !strings.Contains(c.Message, "named.conf:6: expected IP address") {
		t.Fatalf("condition should carry the checkconf detail, got %q", c.Message)
	}
	if _, ok := cluster.Status.AppliedConfig["c-0"]; ok {
		t.Fatal("failed reconfig must not be recorded as applied")
	}
}

func notReadyPod(cluster, name string) *corev1.Pod {
	p := readyPod(cluster, name)
	p.Status.Conditions[0].Status = corev1.ConditionFalse
	return p
}

func TestReconfigPodsSurfacesExecErrorAndBacksOff(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"}}
	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(readyPod("c", "c-0")).Build()
	ex := &fakeExec{
		catErr:    map[string]error{"c-0": errors.New(`pods "c-0" is forbidden: cannot create resource "pods/exec"`)},
		reconfigs: map[string]int{},
	}
	r := &BindClusterReconciler{Client: cl, Scheme: scheme, Exec: ex}

	if d := r.reconfigPods(context.Background(), cluster, "p;", "s;", ""); d != requeueLong {
		t.Fatalf("exec error should back off with requeueLong, got %v", d)
	}
	c := meta.FindStatusCondition(cluster.Status.Conditions, "ConfigApplied")
	if c == nil || c.Status != metav1.ConditionFalse || c.Reason != "ExecFailed" || !strings.Contains(c.Message, "forbidden") {
		t.Fatalf("want ExecFailed condition carrying the cause, got %+v", c)
	}
	if ex.reconfigs["c-0"] != 0 {
		t.Fatal("reconfig ran after a failed mount check")
	}
}

func TestReconfigPodsNotReadyStalePodBlocksApplied(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"}}
	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(readyPod("c", "c-0"), notReadyPod("c", "c-1")).Build()
	const primary, secondary = "p;", "s;"
	ex := &fakeExec{
		mounted:   map[string]string{"c-0": primary},
		catErr:    map[string]error{"c-1": errors.New("must not exec into a not-ready pod")},
		reconfigs: map[string]int{},
	}
	r := &BindClusterReconciler{Client: cl, Scheme: scheme, Exec: ex}

	if d := r.reconfigPods(context.Background(), cluster, primary, secondary, ""); d != requeueShort {
		t.Fatalf("not-ready stale pod should requeue short, got %v", d)
	}
	c := meta.FindStatusCondition(cluster.Status.Conditions, "ConfigApplied")
	if c == nil || c.Status != metav1.ConditionFalse || c.Reason != "Propagating" || !strings.Contains(c.Message, "c-1") {
		t.Fatalf("want Propagating naming c-1, got %+v", c)
	}
	if ex.reconfigs["c-0"] != 1 || ex.reconfigs["c-1"] != 0 {
		t.Fatalf("want only the ready pod reconfigured, got %v", ex.reconfigs)
	}
}

func TestReconfigPodsNewPodReconfigsOnceThenRecorded(t *testing.T) {
	scheme := testScheme(t)
	const primary, secondary = "p;", "s;"
	cluster := &bindv1alpha1.BindCluster{
		ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"},
		Status:     bindv1alpha1.BindClusterStatus{AppliedConfig: map[string]string{"c-0": sha256Hex(primary)}},
	}
	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(readyPod("c", "c-0"), readyPod("c", "c-1")).Build()
	ex := &fakeExec{mounted: map[string]string{"c-0": primary, "c-1": secondary}, reconfigs: map[string]int{}}
	r := &BindClusterReconciler{Client: cl, Scheme: scheme, Exec: ex}

	for range 3 {
		if d := r.reconfigPods(context.Background(), cluster, primary, secondary, ""); d != 0 {
			t.Fatalf("converged cluster should not requeue, got %v", d)
		}
	}
	if ex.reconfigs["c-0"] != 0 || ex.reconfigs["c-1"] != 1 {
		t.Fatalf("want exactly one reconfig for the new pod only, got %v", ex.reconfigs)
	}
	if cluster.Status.AppliedConfig["c-1"] != sha256Hex(secondary) {
		t.Fatalf("new pod not recorded: %v", cluster.Status.AppliedConfig)
	}
}

func TestReconcilePersistsAppliedConfig(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{
		ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"},
		Spec:       bindv1alpha1.BindClusterSpec{Replicas: 1},
	}
	cl := fake.NewClientBuilder().WithScheme(scheme).
		WithObjects(cluster, readyPod("c", "c-0")).
		WithStatusSubresource(&bindv1alpha1.BindCluster{}).
		Build()
	ex := &fakeExec{mounted: map[string]string{}, reconfigs: map[string]int{}}
	r := &BindClusterReconciler{Client: cl, Scheme: scheme, Exec: ex}
	ctx := context.Background()
	req := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "ns", Name: "c"}}

	if _, err := r.Reconcile(ctx, req); err != nil {
		t.Fatal(err)
	}
	var cm corev1.ConfigMap
	if err := cl.Get(ctx, types.NamespacedName{Namespace: "ns", Name: configMapName("c")}, &cm); err != nil {
		t.Fatal(err)
	}
	var keys corev1.Secret
	if err := cl.Get(ctx, types.NamespacedName{Namespace: "ns", Name: keysSecretName("c")}, &keys); err != nil {
		t.Fatal(err)
	}
	mounted := cm.Data["named.conf.primary"] + string(keys.Data["keys.conf"])
	ex.mounted["c-0"] = mounted

	if _, err := r.Reconcile(ctx, req); err != nil {
		t.Fatal(err)
	}
	var got bindv1alpha1.BindCluster
	if err := cl.Get(ctx, req.NamespacedName, &got); err != nil {
		t.Fatal(err)
	}
	if got.Status.AppliedConfig["c-0"] != sha256Hex(mounted) {
		t.Fatalf("appliedConfig not persisted: %v", got.Status.AppliedConfig)
	}
	if _, err := r.Reconcile(ctx, req); err != nil {
		t.Fatal(err)
	}
	if ex.reconfigs["c-0"] != 1 {
		t.Fatalf("persisted appliedConfig should prevent a second reconfig, got %d", ex.reconfigs["c-0"])
	}
}

func TestReconfigPodsFailureBacksOffWhileOthersWait(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"}}
	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(readyPod("c", "c-0"), notReadyPod("c", "c-1")).Build()
	ex := &fakeExec{
		mounted:     map[string]string{"c-0": "p;"},
		reconfigErr: errors.New("rndc: 'reconfig' failed"),
		reconfigs:   map[string]int{},
	}
	r := &BindClusterReconciler{Client: cl, Scheme: scheme, Exec: ex}

	if d := r.reconfigPods(context.Background(), cluster, "p;", "s;", ""); d != requeueLong {
		t.Fatalf("a failed pod must back off with requeueLong even while another waits, got %v", d)
	}
	if c := meta.FindStatusCondition(cluster.Status.Conditions, "ConfigApplied"); c == nil || c.Reason != "ReconfigFailed" {
		t.Fatalf("want ReconfigFailed, got %+v", c)
	}
}

func TestReconcileFailureBacksOffWhileStatefulSetNotReady(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{
		ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"},
		Spec:       bindv1alpha1.BindClusterSpec{Replicas: 2},
	}
	cl := fake.NewClientBuilder().WithScheme(scheme).
		WithObjects(cluster, readyPod("c", "c-0")).
		WithStatusSubresource(&bindv1alpha1.BindCluster{}).
		Build()
	ex := &fakeExec{
		catErr:    map[string]error{"c-0": errors.New("forbidden")},
		reconfigs: map[string]int{},
	}
	r := &BindClusterReconciler{Client: cl, Scheme: scheme, Exec: ex}

	res, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "ns", Name: "c"}})
	if err != nil {
		t.Fatal(err)
	}
	if res.RequeueAfter != requeueLong {
		t.Fatalf("a failed pod must back off with requeueLong while the StatefulSet is not Ready, got %v", res.RequeueAfter)
	}
}

func TestPodTemplateEntrypointHashTracksScript(t *testing.T) {
	scheme := testScheme(t)
	cluster := &bindv1alpha1.BindCluster{
		ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "ns"},
		Spec:       bindv1alpha1.BindClusterSpec{Replicas: 1},
	}
	r := &BindClusterReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster).Build(), Scheme: scheme}
	sts, err := r.reconcileStatefulSet(context.Background(), cluster)
	if err != nil {
		t.Fatal(err)
	}
	got := sts.Spec.Template.Annotations[entrypointHashAnnotation]
	if got != sha256Hex(entrypointScript()) {
		t.Fatalf("entrypoint hash annotation %q does not match the script", got)
	}
	if got == sha256Hex(entrypointScript()+"\n# changed") {
		t.Fatal("entrypoint hash annotation must change when the script changes")
	}
}