/*
 * Copyright (c) 2025 Huawei Technologies Co., Ltd.
 * openFuyao is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */
package controller

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"

	serverlessdbv1 "serverlessdb/api/v1"
	"serverlessdb/internal/cache"
	dbtypes "serverlessdb/internal/types"
)

const requeueInterval = 30 * time.Second

type DBResourcePoolReconciler struct {
	client.Client
	Scheme     *runtime.Scheme
	PoolCache  *cache.PoolCache
	ServiceTTL time.Duration
}

func (r *DBResourcePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	var pool serverlessdbv1.DBResourcePool
	if err := r.Get(ctx, req.NamespacedName, &pool); err != nil {
		if errors.IsNotFound(err) {
			logger.Info("DBResourcePool not found, skipping")
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, err
	}

	if pool.DeletionTimestamp != nil {
		return r.reconcileDeletion(ctx, &pool)
	}

	if !controllerutil.ContainsFinalizer(&pool, serverlessdbv1.PoolFinalizer) {
		patch := client.MergeFrom(pool.DeepCopy())
		controllerutil.AddFinalizer(&pool, serverlessdbv1.PoolFinalizer)
		if err := r.Patch(ctx, &pool, patch); err != nil {
			return ctrl.Result{}, fmt.Errorf("add finalizer: %w", err)
		}
		return ctrl.Result{RequeueAfter: time.Second}, nil
	}

	logger.Info("Reconciling DBResourcePool", "pool", pool.Name, "replicas", pool.Spec.Replicas)

	var instanceList serverlessdbv1.DBInstanceList
	if err := r.List(ctx, &instanceList,
		client.InNamespace(pool.Namespace),
		client.MatchingLabels{serverlessdbv1.LabelKeyDBResourcePool: pool.Name},
	); err != nil {
		return ctrl.Result{}, fmt.Errorf("list DBInstances: %w", err)
	}

	pgPort := extractPgPort(pool.Spec.PodTemplate)

	r.syncInstanceStatuses(ctx, &pool, &instanceList, pgPort)

	currentCount := int32(len(instanceList.Items))
	desired := pool.Spec.Replicas

	if currentCount < desired {
		r.scaleUp(ctx, &pool, desired-currentCount, pgPort)
	}

	if currentCount > desired {
		r.scaleDown(ctx, &instanceList, currentCount-desired)
	}

	if err := r.updatePoolStatus(ctx, &pool); err != nil {
		return ctrl.Result{}, err
	}

	if err := r.PoolCache.Sync(ctx, r.Client, pool.Namespace); err != nil {
		logger.Error(err, "failed to sync pool cache")
	}

	for i := range instanceList.Items {
		if instanceList.Items[i].Status.Phase == serverlessdbv1.DBInstancePhaseScaling {
			return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
		}
	}

	hasPendingServices, err := r.reconcileStaleServices(ctx, pool.Namespace)
	if err != nil {
		logger.Error(err, "failed to reconcile stale services")
	}
	if hasPendingServices {
		return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
	}

	return ctrl.Result{}, nil
}

func (r *DBResourcePoolReconciler) reconcileDeletion(ctx context.Context, pool *serverlessdbv1.DBResourcePool) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	if !controllerutil.ContainsFinalizer(pool, serverlessdbv1.PoolFinalizer) {
		logger.Info("DBResourcePool finalizer already removed, deletion in progress")
		return ctrl.Result{}, nil
	}

	var delInstanceList serverlessdbv1.DBInstanceList
	if err := r.List(ctx, &delInstanceList,
		client.InNamespace(pool.Namespace),
		client.MatchingLabels{serverlessdbv1.LabelKeyDBResourcePool: pool.Name},
	); err != nil {
		return ctrl.Result{}, fmt.Errorf("list DBInstances during deletion: %w", err)
	}

	allocatedCount := 0
	for _, inst := range delInstanceList.Items {
		if inst.Labels[serverlessdbv1.LabelKeyAppRef] != "" {
			allocatedCount++
		}
	}

	if allocatedCount > 0 {
		logger.Info("waiting for allocated instances to be released before deletion",
			"allocatedCount", allocatedCount, "pool", pool.Name)
		return ctrl.Result{RequeueAfter: requeueInterval}, nil
	}

	for i := range delInstanceList.Items {
		inst := &delInstanceList.Items[i]
		if err := r.Delete(ctx, inst); err != nil && !errors.IsNotFound(err) {
			logger.Error(err, "failed to delete DBInstance during cleanup", "name", inst.Name)
		}
	}

	patch := client.MergeFrom(pool.DeepCopy())
	controllerutil.RemoveFinalizer(pool, serverlessdbv1.PoolFinalizer)
	if err := r.Patch(ctx, pool, patch); err != nil {
		return ctrl.Result{}, fmt.Errorf("remove finalizer: %w", err)
	}
	logger.Info("removed finalizer, Pool deletion proceeding", "pool", pool.Name)
	return ctrl.Result{}, nil
}

func (r *DBResourcePoolReconciler) syncInstanceStatuses(ctx context.Context, pool *serverlessdbv1.DBResourcePool, instanceList *serverlessdbv1.DBInstanceList, pgPort int32) {
	logger := log.FromContext(ctx)

	for i := range instanceList.Items {
		instance := &instanceList.Items[i]

		podName := instance.Status.PodName
		if podName == "" {
			podName = instance.Name
		}

		var pod corev1.Pod
		podErr := r.Get(ctx, types.NamespacedName{Namespace: pool.Namespace, Name: podName}, &pod)

		if podErr != nil && errors.IsNotFound(podErr) {
			r.recreateMissingPod(ctx, pool, instance, pgPort)
			continue
		} else if podErr != nil {
			logger.Error(podErr, "failed to get pod", "podName", podName)
			continue
		}

		needPatch := false
		patch := client.MergeFrom(instance.DeepCopy())

		if pod.Status.PodIP != instance.Status.PodIP || string(pod.UID) != instance.Status.PodUID {
			instance.Status.PodIP = pod.Status.PodIP
			instance.Status.PodUID = string(pod.UID)
			needPatch = true
		}

		if instance.Status.Phase == serverlessdbv1.DBInstancePhaseScaling {
			r.applyResize(ctx, instance, &pod)
			needPatch = true
		} else {
			if r.syncPhaseFromPod(instance, &pod) {
				needPatch = true
			}
			needPatch = r.syncRunningInstanceFields(instance, pool.Namespace, pgPort, needPatch)
		}

		// Always reflect the pod's actual compute resources in status, so
		// currentResources stays in sync with what the pod really has (during
		// and after resize) rather than the requested values.
		if want := currentResourcesFromPod(&pod); currentResourcesChanged(instance.Status.CurrentResources, want) {
			instance.Status.CurrentResources = want
			needPatch = true
		}

		if needPatch {
			if statusErr := r.Status().Patch(ctx, instance, patch); statusErr != nil {
				logger.Error(statusErr, "failed to patch DBInstance status")
				continue
			}
			logger.Info("synced DBInstance status from pod", "instance", instance.Name, "phase", instance.Status.Phase, "podIP", pod.Status.PodIP)
		}
	}
}

func (r *DBResourcePoolReconciler) syncRunningInstanceFields(instance *serverlessdbv1.DBInstance, namespace string, pgPort int32, needPatch bool) bool {
	if instance.Status.Phase != serverlessdbv1.DBInstancePhaseRunning {
		return needPatch
	}
	appRef := instance.Labels[serverlessdbv1.LabelKeyAppRef]
	if appRef == "" {
		return needPatch
	}
	if instance.Status.AllocatedAt == nil {
		now := metav1.Now()
		instance.Status.AllocatedAt = &now
		needPatch = true
	}
	connStr := fmt.Sprintf("%s.%s.svc.cluster.local:%d", appRef, namespace, pgPort)
	if instance.Status.ConnectionString != connStr {
		instance.Status.ConnectionString = connStr
		needPatch = true
	}
	return needPatch
}

func (r *DBResourcePoolReconciler) recreateMissingPod(ctx context.Context, pool *serverlessdbv1.DBResourcePool, instance *serverlessdbv1.DBInstance, pgPort int32) {
	logger := log.FromContext(ctx)

	newPod, err := buildPod(pool, instance, r.Scheme)
	if err != nil {
		logger.Error(err, "failed to build pod", "instance", instance.Name)
		return
	}
	if createErr := r.Create(ctx, newPod); createErr != nil {
		logger.Error(createErr, "failed to create pod", "instance", instance.Name)
		return
	}
	logger.Info("created pod for DBInstance", "pod", newPod.Name, "instance", instance.Name)

	patch := client.MergeFrom(instance.DeepCopy())
	instance.Status.PodName = newPod.Name
	instance.Status.PgPort = pgPort
	if instance.Status.Phase == "" {
		instance.Status.Phase = serverlessdbv1.DBInstancePhaseWakingUp
	}
	if statusErr := r.Status().Patch(ctx, instance, patch); statusErr != nil {
		logger.Error(statusErr, "failed to patch DBInstance status after pod creation")
	}
}

func (r *DBResourcePoolReconciler) applyResize(ctx context.Context, instance *serverlessdbv1.DBInstance, pod *corev1.Pod) {
	logger := log.FromContext(ctx)

	specJSON := instance.Annotations[serverlessdbv1.AnnotationKeyResizeSpec]

	// No new resize request: check if an in-flight resize has completed.
	if specJSON == "" {
		if instance.Status.Phase != serverlessdbv1.DBInstancePhaseScaling {
			return
		}
		if isPodResizeInProgress(pod) {
			logger.Info("pod resize in progress, waiting",
				"instance", instance.Name, "pod", pod.Name)
			return
		}
		logger.Info("pod resize completed", "instance", instance.Name, "pod", pod.Name)
		instance.Status.Phase = serverlessdbv1.DBInstancePhaseRunning
		return
	}

	// New resize request: parse, apply to pod spec, and submit via /resize subresource.
	var containers []dbtypes.ContainerResource
	if err := json.Unmarshal([]byte(specJSON), &containers); err != nil {
		logger.Error(err, "failed to unmarshal resize spec", "instance", instance.Name)
		instance.Status.Phase = serverlessdbv1.DBInstancePhaseRunning
		return
	}

	// Build the desired pod spec on a deep copy so the in-memory pod passed in
	// by the caller keeps reflecting the cluster's actual state (needed by
	// currentResourcesFromPod, which reads the real pod resources afterwards).
	podCopy := pod.DeepCopy()
	// MergeFrom holds a live reference to its base object and computes the diff
	// lazily at Patch time, so the base must be a separate snapshot taken BEFORE
	// mutating podCopy. Passing podCopy itself would make original == modified
	// (both post-mutation) and yield an empty patch, leaving resources unchanged.
	podPatch := client.MergeFrom(pod.DeepCopy())
	for _, cr := range containers {
		idx := findContainerIndex(podCopy.Spec.Containers, cr.Name)
		if idx < 0 {
			logger.Info("container not found in pod, skipping", "instance", instance.Name, "pod", pod.Name, "container", cr.Name)
			continue
		}
		applyContainerResize(podCopy, idx, cr, logger)
	}

	// Submit the resize via the /resize subresource only. Do not fall back to a
	// direct pod patch on failure: if the subresource is unavailable the request
	// is left to be retried on the next reconcile (the resize-spec trigger
	// annotation is preserved on error).
	if err := r.SubResource("resize").Patch(ctx, podCopy, podPatch); err != nil {
		logger.Error(err, "failed to patch Pod resources via resize subresource",
			"instance", instance.Name, "pod", pod.Name)
		return
	}
	logger.Info("submitted pod resize request", "instance", instance.Name, "pod", pod.Name, "containers", len(containers))

	// Resize submitted: clear the trigger annotation and persist the latest
	// resize spec for the record. Subsequent resize requests overwrite this.
	metaPatch := client.MergeFrom(instance.DeepCopy())
	delete(instance.Annotations, serverlessdbv1.AnnotationKeyResizeSpec)
	if instance.Annotations == nil {
		instance.Annotations = map[string]string{}
	}
	instance.Annotations[serverlessdbv1.AnnotationKeyLastResizeSpec] = specJSON
	if err := r.Patch(ctx, instance, metaPatch); err != nil {
		logger.Error(err, "failed to update resize annotations", "instance", instance.Name)
	}
	// Keep phase = Scaling; next reconcile will check pod.Status.Resize for completion.
}

func findContainerIndex(containers []corev1.Container, name string) int {
	for i, c := range containers {
		if c.Name == name {
			return i
		}
	}
	return -1
}

func applyContainerResize(pod *corev1.Pod, idx int, cr dbtypes.ContainerResource, logger logr.Logger) {
	if pod.Spec.Containers[idx].Resources.Requests == nil {
		pod.Spec.Containers[idx].Resources.Requests = corev1.ResourceList{}
	}
	if pod.Spec.Containers[idx].Resources.Limits == nil {
		pod.Spec.Containers[idx].Resources.Limits = corev1.ResourceList{}
	}
	ctx := resizeLogCtx{logger: logger, containerName: cr.Name}
	setResourceQuantity(pod.Spec.Containers[idx].Resources.Requests, corev1.ResourceCPU, cr.Resources.Cpu, "cpu", ctx)
	setResourceQuantity(pod.Spec.Containers[idx].Resources.Requests, corev1.ResourceMemory, cr.Resources.Memory, "memory", ctx)
	setResourceQuantity(pod.Spec.Containers[idx].Resources.Limits, corev1.ResourceCPU, cr.Resources.CpuLimit, "cpu_limit", ctx)
	setResourceQuantity(pod.Spec.Containers[idx].Resources.Limits, corev1.ResourceMemory, cr.Resources.MemoryLimit, "memory_limit", ctx)
}

type resizeLogCtx struct {
	logger        logr.Logger
	containerName string
}

func setResourceQuantity(list corev1.ResourceList, name corev1.ResourceName, value, label string, ctx resizeLogCtx) {
	if value == "" {
		return
	}
	q, err := resource.ParseQuantity(value)
	if err != nil {
		ctx.logger.Error(err, "invalid "+label+" quantity", "container", ctx.containerName, "value", value)
		return
	}
	list[name] = q
}

func (r *DBResourcePoolReconciler) syncPhaseFromPod(instance *serverlessdbv1.DBInstance, pod *corev1.Pod) bool {
	isReady := isPodReady(pod)

	if isReady {
		appRef := instance.Labels[serverlessdbv1.LabelKeyAppRef]
		if appRef != "" {
			if instance.Status.Phase != serverlessdbv1.DBInstancePhaseRunning {
				instance.Status.Phase = serverlessdbv1.DBInstancePhaseRunning
				return true
			}
		} else {
			if instance.Status.Phase != serverlessdbv1.DBInstancePhaseWarmReady {
				instance.Status.Phase = serverlessdbv1.DBInstancePhaseWarmReady
				return true
			}
		}
	} else {
		if instance.Status.Phase != serverlessdbv1.DBInstancePhaseWakingUp && instance.Status.Phase != serverlessdbv1.DBInstancePhaseFailed {
			instance.Status.Phase = serverlessdbv1.DBInstancePhaseFailed
			return true
		}
	}
	return false
}

// currentResourcesFromPod reads the actual resources of all containers in the
// pod spec and returns them as a slice of ContainerResources snapshots. Fields
// not present on a container are left empty. Returns nil if the pod has no
// containers.
func currentResourcesFromPod(pod *corev1.Pod) []serverlessdbv1.ContainerResources {
	if len(pod.Spec.Containers) == 0 {
		return nil
	}
	out := make([]serverlessdbv1.ContainerResources, 0, len(pod.Spec.Containers))
	for _, c := range pod.Spec.Containers {
		res := serverlessdbv1.InstanceResources{}
		if q, ok := c.Resources.Requests[corev1.ResourceCPU]; ok {
			res.Cpu = q.String()
		}
		if q, ok := c.Resources.Requests[corev1.ResourceMemory]; ok {
			res.Memory = q.String()
		}
		if q, ok := c.Resources.Limits[corev1.ResourceCPU]; ok {
			res.CpuLimit = q.String()
		}
		if q, ok := c.Resources.Limits[corev1.ResourceMemory]; ok {
			res.MemoryLimit = q.String()
		}
		out = append(out, serverlessdbv1.ContainerResources{Name: c.Name, Resources: res})
	}
	return out
}

// currentResourcesChanged returns true when the desired snapshot slice differs
// from the one already on the instance status. Comparison is by container name
// (order-insensitive). A nil slice and an empty slice are treated as equal.
func currentResourcesChanged(cur, want []serverlessdbv1.ContainerResources) bool {
	if len(cur) == 0 && len(want) == 0 {
		return false
	}
	if len(cur) != len(want) {
		return true
	}
	curMap := make(map[string]serverlessdbv1.InstanceResources, len(cur))
	for _, c := range cur {
		curMap[c.Name] = c.Resources
	}
	for _, w := range want {
		c, ok := curMap[w.Name]
		if !ok || c != w.Resources {
			return true
		}
	}
	return false
}

func (r *DBResourcePoolReconciler) scaleUp(ctx context.Context, pool *serverlessdbv1.DBResourcePool, count int32, pgPort int32) {
	logger := log.FromContext(ctx)

	for i := int32(0); i < count; i++ {
		instanceName := generateInstanceName(pool.Name)
		instance := &serverlessdbv1.DBInstance{
			ObjectMeta: metav1.ObjectMeta{
				Name:      instanceName,
				Namespace: pool.Namespace,
				Labels: map[string]string{
					serverlessdbv1.LabelKeyDBResourcePool: pool.Name,
				},
			},
			Spec:   serverlessdbv1.DBInstanceSpec{},
			Status: serverlessdbv1.DBInstanceStatus{Phase: serverlessdbv1.DBInstancePhaseWakingUp},
		}

		if err := ctrlSetControllerReference(pool, instance, r.Scheme); err != nil {
			logger.Error(err, "failed to set DBInstance ownerRef")
			continue
		}

		if createErr := r.Create(ctx, instance); createErr != nil {
			logger.Error(createErr, "failed to create DBInstance", "name", instanceName)
			continue
		}
		logger.Info("created DBInstance", "name", instanceName)

		pod, err := buildPod(pool, instance, r.Scheme)
		if err != nil {
			logger.Error(err, "failed to build pod for new instance", "instance", instanceName)
			continue
		}

		if createErr := r.Create(ctx, pod); createErr != nil {
			logger.Error(createErr, "failed to create pod", "name", pod.Name)
			continue
		}
		logger.Info("created pod", "name", pod.Name)

		patch := client.MergeFrom(instance.DeepCopy())
		instance.Status.PodName = pod.Name
		instance.Status.PgPort = pgPort
		instance.Status.Phase = serverlessdbv1.DBInstancePhaseWakingUp
		if statusErr := r.Status().Patch(ctx, instance, patch); statusErr != nil {
			logger.Error(statusErr, "failed to patch DBInstance initial status")
			continue
		}
	}
}

func (r *DBResourcePoolReconciler) scaleDown(ctx context.Context, instanceList *serverlessdbv1.DBInstanceList, count int32) {
	logger := log.FromContext(ctx)

	var deletableInstances []serverlessdbv1.DBInstance
	for i := range instanceList.Items {
		inst := instanceList.Items[i]
		if inst.Labels[serverlessdbv1.LabelKeyAppRef] == "" {
			deletableInstances = append(deletableInstances, inst)
		}
	}

	if len(deletableInstances) == 0 && count > 0 {
		logger.Info("scaling down desired but all instances are allocated; waiting for release",
			"count", count)
	}

	for i := int32(0); i < count && i < int32(len(deletableInstances)); i++ {
		if err := r.Delete(ctx, &deletableInstances[i]); err != nil {
			logger.Error(err, "failed to delete excess unallocated DBInstance", "name", deletableInstances[i].Name)
		} else {
			logger.Info("deleted excess unallocated DBInstance", "name", deletableInstances[i].Name)
		}
	}
}

func (r *DBResourcePoolReconciler) updatePoolStatus(ctx context.Context, pool *serverlessdbv1.DBResourcePool) error {
	logger := log.FromContext(ctx)

	var updatedList serverlessdbv1.DBInstanceList
	if err := r.List(ctx, &updatedList,
		client.InNamespace(pool.Namespace),
		client.MatchingLabels{serverlessdbv1.LabelKeyDBResourcePool: pool.Name},
	); err != nil {
		return fmt.Errorf("list DBInstances for status update: %w", err)
	}

	warm := 0
	available := 0
	allocated := 0
	for _, inst := range updatedList.Items {
		if inst.Status.Phase == serverlessdbv1.DBInstancePhaseWarmReady {
			warm++
			if inst.Labels[serverlessdbv1.LabelKeyAppRef] == "" {
				available++
			}
		}
		if inst.Labels[serverlessdbv1.LabelKeyAppRef] != "" {
			allocated++
		}
	}

	poolPatch := client.MergeFrom(pool.DeepCopy())
	pool.Status.TotalInstances = int32(len(updatedList.Items))
	pool.Status.WarmInstances = int32(warm)
	pool.Status.AvailableInstances = int32(available)
	pool.Status.AllocatedInstances = int32(allocated)
	pool.Status.LabelSelector = fmt.Sprintf("%s=%s", serverlessdbv1.LabelKeyDBResourcePool, pool.Name)

	if err := r.Status().Patch(ctx, pool, poolPatch); err != nil {
		logger.Error(err, "failed to patch DBResourcePool status")
		return err
	}
	return nil
}

// reconcileStaleServices cleans up orphaned headless Services that have no backing pod
// for longer than ServiceTTL. Returns hasPending=true if any service is still within TTL.
func (r *DBResourcePoolReconciler) reconcileStaleServices(ctx context.Context, namespace string) (bool, error) {
	if r.ServiceTTL == 0 {
		return false, nil
	}

	svcList := &corev1.ServiceList{}
	if err := r.List(ctx, svcList,
		client.InNamespace(namespace),
		client.MatchingLabels{serverlessdbv1.LabelKeyManaged: "true"},
	); err != nil {
		return false, fmt.Errorf("list managed services: %w", err)
	}

	hasPending := false
	now := time.Now()
	for i := range svcList.Items {
		if r.processStaleService(ctx, &svcList.Items[i], namespace, now) {
			hasPending = true
		}
	}
	return hasPending, nil
}

func (r *DBResourcePoolReconciler) processStaleService(ctx context.Context, svc *corev1.Service, namespace string, now time.Time) bool {
	logger := log.FromContext(ctx)
	appRef := svc.Name

	podList := &corev1.PodList{}
	if err := r.List(ctx, podList,
		client.InNamespace(namespace),
		client.MatchingLabels{serverlessdbv1.LabelKeyAppRef: appRef},
	); err != nil {
		logger.Error(err, "failed to list pods for service", "service", appRef)
		return false
	}

	if len(podList.Items) > 0 {
		r.clearPodAbsentAnnotation(ctx, svc, appRef)
		return false
	}

	absentSinceStr, hasAnnotation := svc.Annotations[serverlessdbv1.AnnotationKeyPodAbsentSince]
	if !hasAnnotation {
		r.markPodAbsent(ctx, svc, appRef, now)
		return true
	}

	absentSince, err := time.Parse(time.RFC3339, absentSinceStr)
	if err != nil {
		logger.Error(err, "failed to parse pod-absent-since, re-marking", "service", appRef, "value", absentSinceStr)
		r.markPodAbsent(ctx, svc, appRef, now)
		return true
	}

	elapsed := now.Sub(absentSince)
	if elapsed >= r.ServiceTTL {
		if err := r.Delete(ctx, svc); err != nil && !errors.IsNotFound(err) {
			logger.Error(err, "failed to delete stale service", "service", appRef)
		} else {
			logger.Info("deleted stale service after TTL", "service", appRef, "elapsed", elapsed, "ttl", r.ServiceTTL)
		}
		return false
	}
	return true
}

func (r *DBResourcePoolReconciler) markPodAbsent(ctx context.Context, svc *corev1.Service, appRef string, now time.Time) {
	logger := log.FromContext(ctx)
	patch := client.MergeFrom(svc.DeepCopy())
	if svc.Annotations == nil {
		svc.Annotations = map[string]string{}
	}
	svc.Annotations[serverlessdbv1.AnnotationKeyPodAbsentSince] = metav1.Time{Time: now}.Format(time.RFC3339)
	if err := r.Patch(ctx, svc, patch); err != nil {
		logger.Error(err, "failed to set pod-absent-since annotation", "service", appRef)
	} else {
		logger.Info("service has no backing pod, marked absent", "service", appRef, "ttl", r.ServiceTTL)
	}
}

func (r *DBResourcePoolReconciler) clearPodAbsentAnnotation(ctx context.Context, svc *corev1.Service, appRef string) {
	logger := log.FromContext(ctx)
	if _, ok := svc.Annotations[serverlessdbv1.AnnotationKeyPodAbsentSince]; !ok {
		return
	}
	patch := client.MergeFrom(svc.DeepCopy())
	delete(svc.Annotations, serverlessdbv1.AnnotationKeyPodAbsentSince)
	if err := r.Patch(ctx, svc, patch); err != nil {
		logger.Error(err, "failed to clear pod-absent-since annotation", "service", appRef)
	} else {
		logger.Info("cleared pod-absent-since annotation, backing pod found", "service", appRef)
	}
}

func (r *DBResourcePoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&serverlessdbv1.DBResourcePool{}).
		Owns(&serverlessdbv1.DBInstance{}).
		Watches(
			&corev1.Pod{},
			handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []ctrl.Request {
				poolName := obj.GetLabels()[serverlessdbv1.LabelKeyDBResourcePool]
				if poolName == "" {
					return nil
				}
				return []ctrl.Request{
					{NamespacedName: types.NamespacedName{Namespace: obj.GetNamespace(), Name: poolName}},
				}
			}),
			builder.WithPredicates(podPredicates()...),
		).
		Complete(r)
}