package controller

import (
	"context"
	"strings"
	"testing"

	cephv1 "git.unkin.net/unkin/cephrgw-operator/api/v1alpha1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/types"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/tools/record"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
)

func TestClusterReconcile(t *testing.T) {
	requireEnvtest(t)
	ctx := context.Background()
	ns := newNamespace(t, ctx)
	createDestinations(t, ctx, ns)

	cluster := newCluster()
	cluster.SetName("pg")
	cluster.SetNamespace(ns)
	cluster.SetAnnotations(map[string]string{
		annSchedule:    "@daily",
		annDestination: "cephs3_ec4_1",
	})
	_ = unstructured.SetNestedField(cluster.Object, int64(1), "spec", "instances")
	if err := k8sClient.Create(ctx, cluster); err != nil {
		t.Fatalf("create cluster: %v", err)
	}

	r := &ClusterReconciler{baseReconciler{
		Client:        k8sClient,
		Scheme:        clientgoscheme.Scheme,
		Recorder:      record.NewFakeRecorder(16),
		DestNamespace: ns,
		DestConfigMap: "autobackup-destinations",
	}}
	req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: ns, Name: "pg"}}

	// First reconcile provisions the bucket stack but blocks on access readiness.
	if _, err := r.Reconcile(ctx, req); err != nil {
		t.Fatalf("first reconcile: %v", err)
	}
	var access cephv1.BucketAccess
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: accessName("pg")}, &access); err != nil {
		t.Fatalf("BucketAccess not created: %v", err)
	}
	access.Status.Phase = "Ready"
	access.Status.SecretName = credSecretName("pg")
	if err := k8sClient.Status().Update(ctx, &access); err != nil {
		t.Fatalf("update access status: %v", err)
	}

	if _, err := r.Reconcile(ctx, req); err != nil {
		t.Fatalf("second reconcile: %v", err)
	}

	// Cluster barmanObjectStore patched.
	got := newCluster()
	if err := k8sClient.Get(ctx, req.NamespacedName, got); err != nil {
		t.Fatalf("get cluster: %v", err)
	}
	path, _, _ := unstructured.NestedString(got.Object, "spec", "backup", "barmanObjectStore", "destinationPath")
	if want := "s3://" + bucketName(ns, "pg"); path != want {
		t.Errorf("destinationPath = %q, want %q", path, want)
	}
	endpoint, _, _ := unstructured.NestedString(got.Object, "spec", "backup", "barmanObjectStore", "endpointURL")
	if endpoint != "https://s3.ceph.unkin.net" {
		t.Errorf("endpointURL = %q", endpoint)
	}
	server, _, _ := unstructured.NestedString(got.Object, "spec", "backup", "barmanObjectStore", "serverName")
	if server != "pg" {
		t.Errorf("serverName = %q", server)
	}
	akName, _, _ := unstructured.NestedString(got.Object, "spec", "backup", "barmanObjectStore", "s3Credentials", "accessKeyId", "name")
	if akName != credSecretName("pg") {
		t.Errorf("accessKeyId secret name = %q", akName)
	}
	caName, _, _ := unstructured.NestedString(got.Object, "spec", "backup", "barmanObjectStore", "endpointCA", "name")
	if caName != "vault-ca-cert" {
		t.Errorf("endpointCA name = %q", caName)
	}
	rp, _, _ := unstructured.NestedString(got.Object, "spec", "backup", "retentionPolicy")
	if rp != "30d" {
		t.Errorf("retentionPolicy = %q", rp)
	}

	// ScheduledBackup created with 6-field schedule.
	sb := newScheduledBackup()
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: scheduleName("pg")}, sb); err != nil {
		t.Fatalf("ScheduledBackup not created: %v", err)
	}
	sched, _, _ := unstructured.NestedString(sb.Object, "spec", "schedule")
	if sched != "0 0 0 * * *" {
		t.Errorf("scheduledbackup schedule = %q, want 0 0 0 * * *", sched)
	}
	method, _, _ := unstructured.NestedString(sb.Object, "spec", "method")
	if method != "barmanObjectStore" {
		t.Errorf("method = %q", method)
	}
	clusterName, _, _ := unstructured.NestedString(sb.Object, "spec", "cluster", "name")
	if clusterName != "pg" {
		t.Errorf("cluster.name = %q", clusterName)
	}
	owner, _, _ := unstructured.NestedString(sb.Object, "spec", "backupOwnerReference")
	if owner != "self" {
		t.Errorf("backupOwnerReference = %q", owner)
	}
}

func TestClusterReconcileDestinationConflict(t *testing.T) {
	requireEnvtest(t)
	ctx := context.Background()
	ns := newNamespace(t, ctx)
	createDestinations(t, ctx, ns)

	cluster := newCluster()
	cluster.SetName("pg2")
	cluster.SetNamespace(ns)
	cluster.SetAnnotations(map[string]string{annSchedule: "@daily", annDestination: "cephs3_ec4_1"})
	_ = unstructured.SetNestedField(cluster.Object, int64(1), "spec", "instances")
	// User already configured a different destinationPath.
	_ = unstructured.SetNestedField(cluster.Object, "s3://user-owned-bucket", "spec", "backup", "barmanObjectStore", "destinationPath")
	if err := k8sClient.Create(ctx, cluster); err != nil {
		t.Fatalf("create cluster: %v", err)
	}

	rec := record.NewFakeRecorder(16)
	r := &ClusterReconciler{baseReconciler{
		Client: k8sClient, Scheme: clientgoscheme.Scheme, Recorder: rec,
		DestNamespace: ns, DestConfigMap: "autobackup-destinations",
	}}
	req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: ns, Name: "pg2"}}

	if _, err := r.Reconcile(ctx, req); err != nil {
		t.Fatalf("first reconcile: %v", err)
	}
	var access cephv1.BucketAccess
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: accessName("pg2")}, &access); err != nil {
		t.Fatalf("BucketAccess not created: %v", err)
	}
	access.Status.Phase = "Ready"
	access.Status.SecretName = credSecretName("pg2")
	if err := k8sClient.Status().Update(ctx, &access); err != nil {
		t.Fatalf("update access status: %v", err)
	}
	if _, err := r.Reconcile(ctx, req); err != nil {
		t.Fatalf("second reconcile: %v", err)
	}

	got := newCluster()
	if err := k8sClient.Get(ctx, req.NamespacedName, got); err != nil {
		t.Fatalf("get cluster: %v", err)
	}
	path, _, _ := unstructured.NestedString(got.Object, "spec", "backup", "barmanObjectStore", "destinationPath")
	if path != "s3://user-owned-bucket" {
		t.Errorf("operator clobbered user destinationPath: got %q", path)
	}

	// ScheduledBackup is still created; only the barman patch is skipped.
	sb := newScheduledBackup()
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: scheduleName("pg2")}, sb); err != nil {
		t.Fatalf("ScheduledBackup should still be created: %v", err)
	}

	var sawConflict bool
	for drained := false; !drained; {
		select {
		case e := <-rec.Events:
			if strings.Contains(e, "DestinationConflict") {
				sawConflict = true
			}
		default:
			drained = true
		}
	}
	if !sawConflict {
		t.Errorf("expected a DestinationConflict warning event")
	}
}