package controller

import (
	"context"
	"testing"

	cephv1 "git.unkin.net/unkin/cephrgw-operator/api/v1alpha1"
	k8upv1 "github.com/k8up-io/k8up/v2/api/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/tools/record"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
)

func pvcWith(ann map[string]string) *corev1.PersistentVolumeClaim {
	return &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "data", Namespace: "ns", Annotations: ann}}
}

func TestSchedulePredicate(t *testing.T) {
	annotated := pvcWith(map[string]string{annSchedule: "@daily"})
	plain := pvcWith(nil)

	if schedulePredicate.Create(event.CreateEvent{Object: plain}) {
		t.Error("create of unannotated PVC should be filtered out")
	}
	if !schedulePredicate.Create(event.CreateEvent{Object: annotated}) {
		t.Error("create of annotated PVC should enqueue")
	}
	if schedulePredicate.Generic(event.GenericEvent{Object: plain}) {
		t.Error("generic event for unannotated PVC should be filtered out")
	}
	// Annotation removed: old had it, new does not -> must still enqueue so
	// teardown runs.
	if !schedulePredicate.Update(event.UpdateEvent{ObjectOld: annotated, ObjectNew: plain}) {
		t.Error("annotation removal should enqueue for teardown")
	}
	// Annotation added.
	if !schedulePredicate.Update(event.UpdateEvent{ObjectOld: plain, ObjectNew: annotated}) {
		t.Error("annotation addition should enqueue")
	}
	// Neither old nor new annotated -> no enqueue.
	if schedulePredicate.Update(event.UpdateEvent{ObjectOld: plain, ObjectNew: plain}) {
		t.Error("update of unannotated PVC should be filtered out")
	}
}

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

	pvc := &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "data",
			Namespace: ns,
			Annotations: map[string]string{
				annSchedule:    "@daily",
				annDestination: "cephs3_ec4_1",
			},
		},
		Spec: corev1.PersistentVolumeClaimSpec{
			AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
			Resources: corev1.VolumeResourceRequirements{
				Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1Gi")},
			},
		},
	}
	if err := k8sClient.Create(ctx, pvc); err != nil {
		t.Fatalf("create pvc: %v", err)
	}

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

	res, err := r.Reconcile(ctx, req)
	if err != nil {
		t.Fatalf("first reconcile: %v", err)
	}
	if res.RequeueAfter != requeueShort {
		t.Errorf("expected requeueShort while access not ready, got %v", res.RequeueAfter)
	}

	// Bucket stack was created and wired.
	var user cephv1.ObjectStoreUser
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: userName("data")}, &user); err != nil {
		t.Fatalf("ObjectStoreUser not created: %v", err)
	}
	var bucket cephv1.Bucket
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: bucketName(ns, "data")}, &bucket); err != nil {
		t.Fatalf("Bucket not created: %v", err)
	}
	if bucket.Spec.OwnerRef != userName("data") || !bucket.Spec.Versioning || bucket.Spec.PlacementTarget != "cephs3_ec4_1" {
		t.Errorf("unexpected bucket spec: %+v", bucket.Spec)
	}
	if !bucket.Spec.RetainOnDelete {
		t.Errorf("expected RetainOnDelete true for data safety")
	}
	var access cephv1.BucketAccess
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: accessName("data")}, &access); err != nil {
		t.Fatalf("BucketAccess not created: %v", err)
	}
	if access.Spec.Level != cephv1.AccessReadWrite || access.Spec.SecretName != credSecretName("data") {
		t.Errorf("unexpected access spec: %+v", access.Spec)
	}

	// No Schedule until the BucketAccess is Ready.
	var sched k8upv1.Schedule
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: scheduleName("data")}, &sched); !apierrors.IsNotFound(err) {
		t.Fatalf("Schedule should not exist before access Ready, got err=%v", err)
	}

	// Simulate cephrgw marking the access Ready.
	access.Status.Phase = "Ready"
	access.Status.SecretName = credSecretName("data")
	if err := k8sClient.Status().Update(ctx, &access); err != nil {
		t.Fatalf("update access status: %v", err)
	}

	res, err = r.Reconcile(ctx, req)
	if err != nil {
		t.Fatalf("second reconcile: %v", err)
	}
	if res.RequeueAfter != requeueSteady {
		t.Errorf("expected requeueSteady after ready, got %v", res.RequeueAfter)
	}

	// Restic password secret exists with a password.
	var restic corev1.Secret
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: resticSecretName("data")}, &restic); err != nil {
		t.Fatalf("restic secret not created: %v", err)
	}
	if len(restic.Data["password"]) == 0 {
		t.Errorf("restic secret has empty password")
	}

	// Schedule created and wired to the cephrgw + restic secrets.
	if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: ns, Name: scheduleName("data")}, &sched); err != nil {
		t.Fatalf("Schedule not created: %v", err)
	}
	if sched.Spec.Backend == nil || sched.Spec.Backend.S3 == nil {
		t.Fatalf("Schedule backend not set: %+v", sched.Spec)
	}
	s3 := sched.Spec.Backend.S3
	if s3.Bucket != bucketName(ns, "data") || s3.Endpoint != "https://s3.ceph.unkin.net" {
		t.Errorf("unexpected S3 backend: %+v", s3)
	}
	if s3.AccessKeyIDSecretRef.Name != credSecretName("data") || s3.AccessKeyIDSecretRef.Key != "AWS_ACCESS_KEY_ID" {
		t.Errorf("unexpected access key ref: %+v", s3.AccessKeyIDSecretRef)
	}
	if sched.Spec.Backend.RepoPasswordSecretRef.Name != resticSecretName("data") {
		t.Errorf("unexpected repo password ref: %+v", sched.Spec.Backend.RepoPasswordSecretRef)
	}
	if sched.Spec.Backup == nil || string(sched.Spec.Backup.Schedule) != "@daily" {
		t.Errorf("unexpected backup schedule: %+v", sched.Spec.Backup)
	}
	vols := sched.Spec.Backup.Volumes
	if vols == nil || len(*vols) != 1 || (*vols)[0].PersistentVolumeClaim == nil || (*vols)[0].PersistentVolumeClaim.ClaimName != "data" {
		t.Errorf("expected schedule scoped to PVC data, got %+v", vols)
	}
}