/*
Copyright 2024 The Ceph-CSI Authors.

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 rbd

import (
	"context"
	"errors"

	"github.com/container-storage-interface/spec/lib/go/csi"
	"github.com/kubernetes-csi/csi-lib-utils/protosanitizer"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	rbderrors "github.com/ceph/ceph-csi/internal/rbd/errors"
	"github.com/ceph/ceph-csi/internal/rbd/types"
	"github.com/ceph/ceph-csi/internal/util"
	"github.com/ceph/ceph-csi/internal/util/log"
)

// CreateVolumeGroupSnapshot receives a list of volume handles and is requested
// to create a list of snapshots that are created at the same time. This is
// similar (although not exactly the same) to consistency groups.
//
// RBD has a limitation that an image can only belong to a single group. It is
// therefore required to create a temporary group, add all images, create the
// group snapshot and remove all images from the group again. This leaves the
// group and its snapshot around, the group snapshot can be inspected to list
// the snapshots of the images.
//
//nolint:gocyclo,cyclop,gocognit // TODO: reduce complexity.
func (cs *ControllerServer) CreateVolumeGroupSnapshot(
	ctx context.Context,
	req *csi.CreateVolumeGroupSnapshotRequest,
) (*csi.CreateVolumeGroupSnapshotResponse, error) {
	var (
		err           error
		vg            types.VolumeGroup
		groupSnapshot types.VolumeGroupSnapshot

		// the VG and VGS should not have the same name
		vgName  = req.GetName() + "-vg" // stable temporary name
		vgsName = req.GetName()

		vgNeedsDeletion = true
	)

	err = cs.Driver.ValidateGroupControllerServiceRequest(
		csi.GroupControllerServiceCapability_RPC_CREATE_DELETE_GET_VOLUME_GROUP_SNAPSHOT,
	)
	if err != nil {
		log.ErrorLog(ctx, "invalid create groupsnapshot req: %v", protosanitizer.StripSecrets(req))

		return nil, err
	}

	// Existence and conflict checks
	if acquired := cs.VolumeGroupLocks.TryAcquire(vgsName); !acquired {
		log.ErrorLog(ctx, util.SnapshotOperationAlreadyExistsFmt, vgsName)

		return nil, status.Errorf(codes.Aborted, util.SnapshotOperationAlreadyExistsFmt, vgsName)
	}
	defer cs.VolumeGroupLocks.Release(vgsName)

	mgr := NewManager(cs.Driver.GetInstanceID(), req.GetParameters(), req.GetSecrets())
	defer mgr.Destroy(ctx)

	// resolve all volumes, free them later on
	volumes := make([]types.Volume, len(req.GetSourceVolumeIds()))
	defer func() {
		for _, volume := range volumes {
			if vg != nil && vgNeedsDeletion {
				// 'normal' cleanup, remove all images from the group
				vgErr := vg.RemoveVolume(ctx, volume)
				if vgErr != nil {
					log.ErrorLog(
						ctx,
						"failed to remove volume %q from volume group %q: %v",
						volume, vg, vgErr)
				}
			}

			// free all allocated volumes
			if volume != nil {
				volume.Destroy(ctx)
			}
		}

		if vg != nil && vgNeedsDeletion {
			// the VG should always be deleted, volumes can only belong to a single VG
			log.DebugLog(ctx, "removing temporary volume group %q", vg)

			vgErr := vg.Delete(ctx)
			if vgErr != nil {
				log.ErrorLog(ctx, "failed to remove temporary volume group %q: %v", vg, vgErr)
			}

			// free the resources of the VolumeGroup
			vg.Destroy(ctx)
		}
	}()
	for i, id := range req.GetSourceVolumeIds() {
		var vol types.Volume
		vol, err = mgr.GetVolumeByID(ctx, id)
		if err != nil {
			return nil, status.Errorf(
				codes.InvalidArgument,
				"failed to find required volume %q for volume group snapshot %q: %s",
				id,
				vgsName,
				err.Error())
		}
		volumes[i] = vol
	}

	log.DebugLog(ctx, "all %d Volumes for VolumeGroup %q have been found", len(volumes), vgsName)

	// all volumes should belong to the same group, or to no group at all
	groupMatches, err := mgr.VolumesInSameGroup(ctx, volumes)
	if err != nil {
		return nil, status.Errorf(
			codes.Internal,
			"failed to compare %d volumes: %s",
			len(volumes),
			err.Error())
	} else if !groupMatches {
		return nil, status.Error(
			codes.Internal,
			"not all volumes belong to the same group")
	}

	// check for pre-existing volume group name
	vgHandle, err := volumes[0].GetVolumeGroupID(ctx, mgr)
	if err != nil && !errors.Is(err, rbderrors.ErrGroupNotFound) {
		return nil, status.Errorf(
			codes.Internal,
			"could not get volume group for volume %q: %v",
			volumes[0],
			err)
	} else if vgHandle != "" {
		vg, err = mgr.GetVolumeGroupByID(ctx, vgHandle)
		// vg.Destroy(ctx) is called in a defer further up ^
		if err != nil {
			return nil, status.Errorf(
				codes.InvalidArgument,
				"failed to resolve volume group with id %q: %s",
				vgHandle,
				err.Error())
		}

		vgName, err = vg.GetName(ctx)
		if err != nil {
			return nil, status.Errorf(
				codes.InvalidArgument,
				"failed to get name of volume group %q: %s",
				vg,
				err.Error())
		}

		vgNeedsDeletion = false
	}

	groupSnapshot, err = mgr.GetVolumeGroupSnapshotByName(ctx, vgsName)
	if groupSnapshot != nil {
		defer groupSnapshot.Destroy(ctx)

		csiVGS, csiErr := groupSnapshot.ToCSI(ctx)
		if csiErr != nil {
			return nil, status.Error(codes.Aborted, csiErr.Error())
		}

		return &csi.CreateVolumeGroupSnapshotResponse{
			GroupSnapshot: csiVGS,
		}, nil
	}
	if err != nil {
		log.DebugLog(ctx, "need to create new volume group snapshot, "+
			"failed to get existing one with name %q: %v", vgsName, err)
	}

	creds, err := util.NewUserCredentials(req.GetSecrets())
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	defer creds.DeleteCredentials()

	errList := make([]error, 0)
	for _, volume := range volumes {
		err = volume.PrepareVolumeForSnapshot(ctx, creds)
		if err != nil {
			errList = append(errList, err)
		}
	}
	if len(errList) > 0 {
		// FIXME: we should probably choose a error code that has more priority.
		return nil, status.Errorf(
			status.Code(errList[0]),
			"failed to prepare volumes for snapshot: %v",
			errList)
	}

	if vg == nil {
		// create a temporary VolumeGroup with a different name
		vg, err = mgr.CreateVolumeGroup(ctx, vgName)
		if err != nil {
			return nil, status.Errorf(
				codes.Internal,
				"failed to create volume group %q: %s",
				vgName,
				err.Error())
		}
		// vg.Destroy(ctx) is called in a defer further up ^

		log.DebugLog(ctx, "VolumeGroup %q has been created: %+v", vgName, vg)
	}

	vols, err := vg.ListVolumes(ctx)
	if err != nil {
		return nil, status.Errorf(
			codes.Internal,
			"failed to list volumes of volume group %q: %s",
			vg,
			err.Error())
	} else if len(vols) == 0 {
		// add images to the group
		for _, volume := range volumes {
			err = vg.AddVolume(ctx, volume)
			if err != nil {
				return nil, status.Error(codes.Aborted, err.Error())
			}
		}
	}

	groupSnapshot, err = mgr.CreateVolumeGroupSnapshot(ctx, vg, vgsName)
	if err != nil {
		return nil, status.Error(codes.Aborted, err.Error())
	}
	defer groupSnapshot.Destroy(ctx)

	csiVGS, err := groupSnapshot.ToCSI(ctx)
	if err != nil {
		return nil, status.Error(codes.Aborted, err.Error())
	}

	return &csi.CreateVolumeGroupSnapshotResponse{
		GroupSnapshot: csiVGS,
	}, nil
}

func (cs *ControllerServer) DeleteVolumeGroupSnapshot(
	ctx context.Context,
	req *csi.DeleteVolumeGroupSnapshotRequest,
) (*csi.DeleteVolumeGroupSnapshotResponse, error) {
	err := cs.Driver.ValidateGroupControllerServiceRequest(
		csi.GroupControllerServiceCapability_RPC_CREATE_DELETE_GET_VOLUME_GROUP_SNAPSHOT,
	)
	if err != nil {
		log.ErrorLog(ctx, "invalid create groupsnapshot req: %v", protosanitizer.StripSecrets(req))

		return nil, err
	}

	// FIXME: more checking of the request in needed
	// 1. verify that all snapshots in the request are all snapshots in the group
	// 2. delete the group

	groupSnapshotID := req.GetGroupSnapshotId()

	// Existence and conflict checks
	if acquired := cs.VolumeGroupLocks.TryAcquire(groupSnapshotID); !acquired {
		log.ErrorLog(ctx, util.SnapshotOperationAlreadyExistsFmt, groupSnapshotID)

		return nil, status.Errorf(codes.Aborted, util.SnapshotOperationAlreadyExistsFmt, groupSnapshotID)
	}
	defer cs.VolumeGroupLocks.Release(groupSnapshotID)

	mgr := NewManager(cs.Driver.GetInstanceID(), nil, req.GetSecrets())
	defer mgr.Destroy(ctx)

	groupSnapshot, err := mgr.GetVolumeGroupSnapshotByID(ctx, groupSnapshotID)
	if err != nil {
		if errors.Is(err, rbderrors.ErrGroupNotFound) {
			log.ErrorLog(ctx, "VolumeGroupSnapshot %q doesn't exists", groupSnapshotID)

			return &csi.DeleteVolumeGroupSnapshotResponse{}, nil
		}

		return nil, status.Errorf(
			codes.Internal,
			"could not fetch volume group snapshot with id %q: %s",
			groupSnapshotID,
			err.Error())
	}
	defer groupSnapshot.Destroy(ctx)

	err = groupSnapshot.Delete(ctx)
	if err != nil {
		return nil, status.Errorf(
			codes.Internal,
			"failed to delete volume group snapshot %q: %v",
			groupSnapshot, err)
	}

	return &csi.DeleteVolumeGroupSnapshotResponse{}, nil
}

// GetVolumeGroupSnapshot is sortof optional, only used for
// static/pre-provisioned VolumeGroupSnapshots.
func (cs *ControllerServer) GetVolumeGroupSnapshot(
	ctx context.Context,
	req *csi.GetVolumeGroupSnapshotRequest,
) (*csi.GetVolumeGroupSnapshotResponse, error) {
	err := cs.Driver.ValidateGroupControllerServiceRequest(
		csi.GroupControllerServiceCapability_RPC_CREATE_DELETE_GET_VOLUME_GROUP_SNAPSHOT,
	)
	if err != nil {
		log.ErrorLog(ctx, "invalid create groupsnapshot req: %v", protosanitizer.StripSecrets(req))

		return nil, err
	}

	groupSnapshotID := req.GetGroupSnapshotId()

	// Existence and conflict checks
	if acquired := cs.VolumeGroupLocks.TryAcquire(groupSnapshotID); !acquired {
		log.ErrorLog(ctx, util.SnapshotOperationAlreadyExistsFmt, groupSnapshotID)

		return nil, status.Errorf(codes.Aborted, util.SnapshotOperationAlreadyExistsFmt, groupSnapshotID)
	}
	defer cs.VolumeGroupLocks.Release(groupSnapshotID)

	mgr := NewManager(cs.Driver.GetInstanceID(), nil, req.GetSecrets())
	defer mgr.Destroy(ctx)

	groupSnapshot, err := mgr.GetVolumeGroupSnapshotByID(ctx, groupSnapshotID)
	if err != nil {
		if errors.Is(err, rbderrors.ErrGroupNotFound) {
			log.ErrorLog(ctx, "VolumeGroupSnapshot %q doesn't exists", groupSnapshotID)

			return nil, status.Errorf(
				codes.NotFound,
				"failed to get volume group snapshot with id %q: %v",
				groupSnapshotID, err)
		}

		return nil, status.Errorf(
			codes.Internal,
			"could not fetch volume group snapshot with id %q: %s",
			groupSnapshotID,
			err.Error())
	}
	defer groupSnapshot.Destroy(ctx)

	csiVGS, err := groupSnapshot.ToCSI(ctx)
	if err != nil {
		return nil, status.Error(codes.Aborted, err.Error())
	}

	return &csi.GetVolumeGroupSnapshotResponse{
		GroupSnapshot: csiVGS,
	}, nil
}
