// Copyright 2024 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package tasks

import (
	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"github.com/pingcap/tidb-operator/apis/core/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/client"
	maputil "github.com/pingcap/tidb-operator/pkg/utils/map"
	"github.com/pingcap/tidb-operator/pkg/utils/task/v2"
	"github.com/pingcap/tidb-operator/pkg/volumes"
)

type TaskPVC struct {
	Client         client.Client
	Logger         logr.Logger
	VolumeModifier volumes.Modifier
}

func NewTaskPVC(logger logr.Logger, c client.Client, vm volumes.Modifier) task.Task[ReconcileContext] {
	return &TaskPVC{
		Client:         c,
		Logger:         logger,
		VolumeModifier: vm,
	}
}

func (*TaskPVC) Name() string {
	return "PVC"
}

func (t *TaskPVC) Sync(ctx task.Context[ReconcileContext]) task.Result {
	rtx := ctx.Self()

	pvcs := newPVCs(rtx.TiKV)
	if wait, err := volumes.SyncPVCs(rtx, t.Client, pvcs, t.VolumeModifier, t.Logger); err != nil {
		return task.Fail().With("failed to sync pvcs: %w", err)
	} else if wait {
		return task.Complete().With("waiting for pvcs to be synced")
	}

	return task.Complete().With("pvcs are synced")
}

func newPVCs(tikv *v1alpha1.TiKV) []*corev1.PersistentVolumeClaim {
	pvcs := make([]*corev1.PersistentVolumeClaim, 0, len(tikv.Spec.Volumes))
	for i := range tikv.Spec.Volumes {
		vol := tikv.Spec.Volumes[i]
		pvcs = append(pvcs, &corev1.PersistentVolumeClaim{
			ObjectMeta: metav1.ObjectMeta{
				Name:      PersistentVolumeClaimName(tikv.PodName(), vol.Name),
				Namespace: tikv.Namespace,
				Labels: maputil.Merge(tikv.Labels, map[string]string{
					v1alpha1.LabelKeyInstance: tikv.Name,
				}),
				OwnerReferences: []metav1.OwnerReference{
					*metav1.NewControllerRef(tikv, v1alpha1.SchemeGroupVersion.WithKind("TiKV")),
				},
			},
			Spec: corev1.PersistentVolumeClaimSpec{
				AccessModes: []corev1.PersistentVolumeAccessMode{
					corev1.ReadWriteOnce,
				},
				Resources: corev1.VolumeResourceRequirements{
					Requests: corev1.ResourceList{
						corev1.ResourceStorage: vol.Storage,
					},
				},
				StorageClassName:          vol.StorageClassName,
				VolumeAttributesClassName: vol.VolumeAttributesClassName,
			},
		})
	}

	return pvcs
}