// Copyright IBM Corp. 2021, 2026
// SPDX-License-Identifier: MPL-2.0

package storage

import (
	"bufio"
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/base64"
	stderr "errors"
	"fmt"
	"io"
	"os"
	"path"
	"strings"
	"sync"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/google/uuid"
	cred "github.com/hashicorp/boundary-plugin-aws/internal/credential"
	"github.com/hashicorp/boundary-plugin-aws/internal/errors"
	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
	"github.com/hashicorp/go-multierror"
	"github.com/hashicorp/go-secure-stdlib/parseutil"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/structpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// Ensure that we are implementing StoragePluginServiceServer
var _ pb.StoragePluginServiceServer = (*StoragePlugin)(nil)

// clientCache caches previously initialized S3API clients created with a RoleARN attribute.
type clientCache struct {
	// cache is a map of cached s3 clients. The key of the map
	// is the public id of the storage bucket
	cache map[string]S3API
	sync.RWMutex
}

// StoragePlugin implements the StoragePluginServiceServer interface for the
// AWS storage service plugin.
type StoragePlugin struct {
	pb.UnimplementedStoragePluginServiceServer

	// testCredStateOpts are passed in to the stored state to control test behavior
	testCredStateOpts []cred.AwsCredentialPersistedStateOption

	// testStorageStateOpts are passed in to the stored state to control test behavior
	testStorageStateOpts []awsStoragePersistedStateOption

	clients *clientCache
}

// New creates a new StoragePlugin
func New() *StoragePlugin {
	c := &clientCache{
		cache: make(map[string]S3API),
	}
	return &StoragePlugin{
		clients: c,
	}
}

// getClient returns an S3API client for the given storage bucket id.
func (p *StoragePlugin) getClient(ctx context.Context,
	storageBucketId string,
	storageState *awsStoragePersistedState,
	opt ...s3Option) (S3API, error) {
	if storageBucketId == "" {
		// No storage bucket ID to key cache on, create new client and return
		client, err := storageState.s3Client(ctx, opt...)
		if err != nil {
			return nil, errors.BadRequestStatusf("error creating S3 client: %s", err)
		}
		return client, nil
	}

	opts, err := getOpts(opt...)
	if err != nil {
		return nil, fmt.Errorf("error parsing options in cache: %w", err)
	}

	if opts.withCacheRefresh {
		// Forced cache refresh
		p.clients.Lock()
		defer p.clients.Unlock()
		client, err := storageState.s3Client(ctx, opt...)
		if err != nil {
			return nil, errors.BadRequestStatusf("error creating S3 client: %s", err)
		}
		p.clients.cache[storageBucketId] = client
		return client, nil
	}

	p.clients.RLock()
	client, ok := p.clients.cache[storageBucketId]
	p.clients.RUnlock()

	if !ok || client.Credentials().Expired() {
		// We got a cache miss or the credentials has expired, time to refresh
		p.clients.Lock()
		defer p.clients.Unlock()

		// Check cache again in case another caller updated it since we got the lock
		client, ok = p.clients.cache[storageBucketId]
		if ok && !client.Credentials().Expired() {
			// Got a cache hit with valid credentials, return cached client
			return client, nil
		}

		// Create new client and cache it
		var err error
		client, err = storageState.s3Client(ctx, opt...)
		if err != nil {
			return nil, errors.BadRequestStatusf("error creating S3 client: %s", err)
		}
		p.clients.cache[storageBucketId] = client
	}
	return client, nil
}

type s3Caller func(client S3API) (any, error)

// call gets the s3client and passes it to the s3caller function, if a permission error is returned
// it forces a cache refresh and tries again.
func (p *StoragePlugin) call(
	ctx context.Context,
	fn s3Caller,
	storageBucketId string,
	storageState *awsStoragePersistedState,
	opts ...s3Option) (any, error) {

	s3Client, err := p.getClient(ctx, storageBucketId, storageState, opts...)
	if err != nil {
		return nil, errors.BadRequestStatusf("error getting S3 client: %s", err)
	}
	firstAttempt := true
	for {
		resp, err := fn(s3Client)
		if err != nil {
			st, _ := errors.ParseAWSError("", err)
			if st.Code() == codes.PermissionDenied && firstAttempt {
				// We got a permission error, this is the first time so refresh cache and try again
				firstAttempt = false
				opts = append(opts, WithCacheRefresh(true))
				s3Client, err = p.getClient(ctx, storageBucketId, storageState, opts...)
				if err != nil {
					return nil, errors.BadRequestStatusf("error refreshing S3 client: %s", err)
				}
				continue
			}
			return nil, err
		}
		return resp, nil
	}
}

// NormalizeStorageBucketData is called by boundary before calling OnCreate or OnUpdate
//
// aws storage supports a total of 8 attributes, 6 as part of credential.CredentialAttributes
// and 2 more from storage.StorageAttributes. of these, only endpoint_url is normalized.
func (sp *StoragePlugin) NormalizeStorageBucketData(ctx context.Context, req *pb.NormalizeStorageBucketDataRequest) (*pb.NormalizeStorageBucketDataResponse, error) {
	sa, err := getStorageAttributes(req.GetAttributes())
	if err != nil {
		return nil, err
	}

	fields := req.GetAttributes().GetFields()
	if _, ok := fields[ConstAwsEndpointUrl]; ok {
		if sa.EndpointUrl, err = parseutil.NormalizeAddr(sa.EndpointUrl); err != nil {
			return nil, status.New(codes.InvalidArgument, fmt.Sprintf("failed to normalize provided %s: %s", ConstAwsEndpointUrl, err.Error())).Err()
		}
		fields[ConstAwsEndpointUrl] = structpb.NewStringValue(sa.EndpointUrl)
	}

	return &pb.NormalizeStorageBucketDataResponse{
		Attributes: &structpb.Struct{
			Fields: fields,
		},
	}, nil
}

// OnCreateStorageBucket is called when a storage bucket is created.
func (p *StoragePlugin) OnCreateStorageBucket(ctx context.Context, req *pb.OnCreateStorageBucketRequest) (*pb.OnCreateStorageBucketResponse, error) {
	bucket := req.GetBucket()
	if bucket == nil {
		return nil, errors.BadRequestStatus("bucket is required")
	}
	if bucket.BucketName == "" {
		return nil, errors.BadRequestStatus("bucketName is required")
	}

	attrs := bucket.GetAttributes()
	if attrs == nil {
		return nil, errors.BadRequestStatus("attributes is required")
	}

	storageAttributes, err := getStorageAttributes(attrs)
	if err != nil {
		return nil, err
	}

	credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
	if err != nil {
		return nil, err
	}

	// Set up the creds in our persisted state.
	credState, err := cred.NewAwsCredentialPersistedState(
		append([]cred.AwsCredentialPersistedStateOption{
			cred.WithCredentialsConfig(credConfig),
		}, p.testCredStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	// Try to rotate AWS static credentials
	if cred.GetCredentialType(credConfig) == cred.StaticAWS && !storageAttributes.DisableCredentialRotation {
		if err := credState.RotateCreds(ctx); err != nil {
			return nil, err
		}
	}

	storageState, err := newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	// perform dry run to ensure we can interact with the bucket as expected.
	if st := dryRunValidation(ctx, storageState, storageAttributes, bucket); st != nil {
		return nil, st.Err()
	}

	persistedProto, err := storageState.toProto()
	if err != nil {
		return nil, errors.BadRequestStatus(err.Error())
	}

	return &pb.OnCreateStorageBucketResponse{
		Persisted: persistedProto,
	}, nil
}

// OnUpdateStorageBucket is called when a storage bucket is updated.
func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpdateStorageBucketRequest) (*pb.OnUpdateStorageBucketResponse, error) {
	// For storage buckets we only need to really validate what is
	// currently in the new copy of the storage bucket data, so skip
	// fetching the new copy to keep things a little less messy.
	newBucket := req.GetNewBucket()
	if newBucket == nil {
		return nil, errors.BadRequestStatus("new bucket is required")
	}
	if newBucket.BucketName == "" {
		return nil, errors.BadRequestStatus("new bucketName is required")
	}
	currentBucket := req.GetCurrentBucket()
	if currentBucket == nil {
		return nil, errors.BadRequestStatus("current bucket is required")
	}

	oldAttrs := currentBucket.GetAttributes()
	if oldAttrs == nil {
		return nil, errors.BadRequestStatus("current bucket attributes is required")
	}
	oldStorageAttributes, err := getStorageAttributes(oldAttrs)
	if err != nil {
		return nil, err
	}
	newAttrs := newBucket.GetAttributes()
	if newAttrs == nil {
		return nil, errors.BadRequestStatus("new bucket attributes is required")
	}
	newStorageAttributes, err := getStorageAttributes(newAttrs)
	if err != nil {
		return nil, err
	}

	credState, err := cred.AwsCredentialPersistedStateFromProto(
		req.GetPersisted().GetData(),
		oldStorageAttributes.CredentialAttributes,
		oldStorageAttributes.DualStack,
		p.testCredStateOpts...)
	if err != nil {
		return nil, errors.BadRequestStatusf("error loading persisted state: %s", err)
	}

	// Verify the incoming credentials are valid and return any errors to the
	// user if they're not. Note this doesn't validate the credentials against
	// AWS - it only does logical validation on the fields.
	updatedCredentials, err := cred.GetCredentialsConfig(newBucket.GetSecrets(), newStorageAttributes.CredentialAttributes, newStorageAttributes.DualStack)
	if err != nil {
		return nil, err
	}

	// We will be updating credentials when one of two changes occur:
	// 1. the new bucket has secrets for static credentials, in which case we need to delete the old static credentials
	// 2. the new bucket has a different roleARN value for dynamic credentials
	if newBucket.GetSecrets() != nil || newStorageAttributes.RoleArn != oldStorageAttributes.RoleArn {
		// Ensure the incoming credentials are valid for interaction with the S3
		// Bucket before replacing them.
		newCredState, err := cred.NewAwsCredentialPersistedState(
			append([]cred.AwsCredentialPersistedStateOption{
				cred.WithCredentialsConfig(updatedCredentials),
			}, p.testCredStateOpts...)...,
		)
		if err != nil {
			return nil, errors.BadRequestStatusf("error setting up new credential persisted state: %s", err)
		}
		newStorageState, err := newAwsStoragePersistedState(
			append([]awsStoragePersistedStateOption{
				withCredentials(newCredState),
			}, p.testStorageStateOpts...)...,
		)
		if err != nil {
			return nil, errors.BadRequestStatusf("error loading persisted state: %s", err)
		}
		if st := dryRunValidation(ctx, newStorageState, newStorageAttributes, newBucket); st != nil {
			return nil, st.Err()
		}

		// Replace the existing credential state.
		// This checks the timestamp on the last rotation time as well
		// and deletes the credentials if we are managing them
		// (ie: if we've rotated them before).
		if err := credState.ReplaceCreds(ctx, updatedCredentials); err != nil {
			return nil, err
		}
	}

	if cred.GetCredentialType(credState.CredentialsConfig) == cred.StaticAWS {
		// This is a validate check to make sure that we aren't disabling
		// rotation for credentials currently being managed by rotation.
		// This is not allowed.
		if newStorageAttributes.DisableCredentialRotation && newBucket.GetSecrets() == nil {
			if !credState.CredsLastRotatedTime.IsZero() {
				return nil, errors.BadRequestStatus("cannot disable rotation for already-rotated credentials")
			}
		}

		// If we're enabling rotation now but didn't before, or have
		// freshly replaced credentials, we can rotate here.
		if !newStorageAttributes.DisableCredentialRotation && credState.CredsLastRotatedTime.IsZero() {
			if err := credState.RotateCreds(ctx); err != nil {
				return nil, err
			}
		}
	}

	storageState, err := newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error loading persisted state: %s", err)
	}

	// perform dry run to ensure we can interact with the bucket as expected.
	if st := dryRunValidation(ctx, storageState, newStorageAttributes, newBucket); st != nil {
		return nil, st.Err()
	}

	persistedProto, err := storageState.toProto()
	if err != nil {
		return nil, errors.BadRequestStatus(err.Error())
	}

	return &pb.OnUpdateStorageBucketResponse{
		Persisted: persistedProto,
	}, nil
}

// OnDeleteStorageBucket is called when a storage bucket is deleted.
func (p *StoragePlugin) OnDeleteStorageBucket(ctx context.Context, req *pb.OnDeleteStorageBucketRequest) (*pb.OnDeleteStorageBucketResponse, error) {
	bucket := req.GetBucket()
	if bucket == nil {
		return nil, errors.BadRequestStatus("bucket is required")
	}

	attrs := bucket.GetAttributes()
	if attrs == nil {
		return nil, errors.BadRequestStatus("attributes is required")
	}

	storageAttributes, err := getStorageAttributes(attrs)
	if err != nil {
		return nil, err
	}

	// Get the persisted data.
	// NOTE: We return on error here, blocking the delete. This may or
	// may not be an overzealous approach to maintaining database/state
	// integrity. May need to be changed at later time if there are
	// scenarios where we might be deleting things and any secret state
	// may be corrupt/and or legitimately missing.
	credState, err := cred.AwsCredentialPersistedStateFromProto(
		req.GetPersisted().GetData(),
		storageAttributes.CredentialAttributes,
		storageAttributes.DualStack,
		p.testCredStateOpts...)
	if err != nil {
		return nil, errors.BadRequestStatusf("error loading persisted state: %s", err)
	}

	_, err = newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error loading persisted state: %s", err)
	}

	// try to delete static credentials
	if cred.GetCredentialType(credState.CredentialsConfig) == cred.StaticAWS {
		if !credState.CredsLastRotatedTime.IsZero() {
			// Delete old/existing credentials. This is done with the same
			// credentials to ensure that it has the proper permissions to do
			// it.
			if err := credState.DeleteCreds(ctx); err != nil {
				return nil, err
			}
		}
	}

	return &pb.OnDeleteStorageBucketResponse{}, nil
}

// HeadObject is called to get the metadata of an object.
func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectRequest) (*pb.HeadObjectResponse, error) {
	if req.GetKey() == "" {
		return nil, errors.BadRequestStatus("key is required")
	}

	bucket := req.GetBucket()
	if bucket == nil {
		return nil, errors.BadRequestStatus("bucket is required")
	}

	if bucket.BucketName == "" {
		return nil, errors.BadRequestStatus("bucketName is required")
	}

	attrs := bucket.GetAttributes()
	if attrs == nil {
		return nil, errors.BadRequestStatus("attributes is required")
	}

	storageAttributes, err := getStorageAttributes(attrs)
	if err != nil {
		return nil, err
	}

	credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
	if err != nil {
		return nil, err
	}

	// Set up the creds in our persisted state.
	credState, err := cred.NewAwsCredentialPersistedState(
		append([]cred.AwsCredentialPersistedStateOption{
			cred.WithCredentialsConfig(credConfig),
		}, p.testCredStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	storageState, err := newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	opts := []s3Option{}
	if storageAttributes.EndpointUrl != "" {
		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
	}
	if storageAttributes.DualStack {
		opts = append(opts, WithDualStack(storageAttributes.DualStack))
	}

	objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
	headCall := func(s3Client S3API) (any, error) {
		return s3Client.HeadObject(ctx, &s3.HeadObjectInput{
			Bucket: aws.String(bucket.GetBucketName()),
			Key:    aws.String(objectKey),
		})
	}
	headResp, err := p.call(ctx, headCall, bucket.GetId(), storageState, opts...)
	if err != nil {
		return nil, parseS3Error("head object", err, req).Err()
	}
	resp := headResp.(*s3.HeadObjectOutput)

	return &pb.HeadObjectResponse{
		ContentLength: aws.ToInt64(resp.ContentLength),
		LastModified:  timestamppb.New(*resp.LastModified),
	}, nil
}

// ValidatePermissions is called to validate the secrets associated with the storage bucket.
func (p *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.ValidatePermissionsRequest) (*pb.ValidatePermissionsResponse, error) {
	bucket := req.GetBucket()
	if bucket == nil {
		return nil, errors.BadRequestStatus("bucket is required")
	}

	if bucket.BucketName == "" {
		return nil, errors.BadRequestStatus("bucketName is required")
	}

	attrs := bucket.GetAttributes()
	if attrs == nil {
		return nil, errors.BadRequestStatus("attributes is required")
	}

	storageAttributes, err := getStorageAttributes(attrs)
	if err != nil {
		return nil, err
	}

	credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
	if err != nil {
		return nil, err
	}

	// Set up the creds in our persisted state.
	credState, err := cred.NewAwsCredentialPersistedState(
		append([]cred.AwsCredentialPersistedStateOption{
			cred.WithCredentialsConfig(credConfig),
		}, p.testCredStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	storageState, err := newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	// perform dry run to ensure we can interact with the bucket as expected.
	if st := dryRunValidation(ctx, storageState, storageAttributes, bucket); st != nil {
		return nil, st.Err()
	}

	return &pb.ValidatePermissionsResponse{}, nil
}

// GetObject is called when retrieving objects from an s3 bucket.
// GetObject is a blocking call until the stream has been recieved in full.
func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePluginService_GetObjectServer) error {
	if req.GetKey() == "" {
		return errors.BadRequestStatus("key is required")
	}

	bucket := req.GetBucket()
	if bucket == nil {
		return errors.BadRequestStatus("bucket is required")
	}

	if bucket.GetBucketName() == "" {
		return errors.BadRequestStatus("bucketName is required")
	}

	attrs := bucket.GetAttributes()
	if attrs == nil {
		return errors.BadRequestStatus("attributes is required")
	}

	storageAttributes, err := getStorageAttributes(attrs)
	if err != nil {
		return err
	}

	credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
	if err != nil {
		return err
	}

	// Set up the creds in our persisted state.
	credState, err := cred.NewAwsCredentialPersistedState(
		append([]cred.AwsCredentialPersistedStateOption{
			cred.WithCredentialsConfig(credConfig),
		}, p.testCredStateOpts...)...,
	)
	if err != nil {
		return errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	storageState, err := newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	opts := []s3Option{}
	if storageAttributes.EndpointUrl != "" {
		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
	}
	if storageAttributes.DualStack {
		opts = append(opts, WithDualStack(storageAttributes.DualStack))
	}

	objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
	getCall := func(s3Client S3API) (any, error) {
		return s3Client.GetObject(stream.Context(), &s3.GetObjectInput{
			Bucket: aws.String(bucket.GetBucketName()),
			Key:    aws.String(objectKey),
		})
	}
	getResp, err := p.call(stream.Context(), getCall, bucket.GetId(), storageState, opts...)
	if err != nil {
		return parseS3Error("get object", err, req).Err()
	}
	resp := getResp.(*s3.GetObjectOutput)

	defer resp.Body.Close()
	reader := bufio.NewReader(resp.Body)
	chunkSize := req.GetChunkSize()
	if chunkSize == 0 {
		chunkSize = defaultStreamChunkSize
	}
	for {
		buffer := make([]byte, chunkSize)
		n, err := reader.Read(buffer)
		if err != nil && err != io.EOF {
			return errors.UnknownStatusf("error reading chunk from s3: %s", err)
		}
		if n > 0 {
			if err := stream.Send(&pb.GetObjectResponse{
				FileChunk: buffer[:n],
			}); err != nil {
				return errors.UnknownStatusf("error sending chunk to client: %s", err)
			}
		}
		if err == io.EOF {
			break
		}
	}

	return nil
}

const (
	// multipartThreshold is the file size, in bytes, above which PutObject
	// uploads an object using a multipart upload instead of a single PutObject
	// request. 8 MiB matches the AWS CLI and boto3 defaults.
	multipartThreshold = 8 * 1024 * 1024 // 8 MiB

	// defaultMultipartPartSize is the default size, in bytes, of each part
	// uploaded during a multipart upload. It is kept equal to
	// multipartThreshold so a multipart upload always has at least two parts
	// (S3 requires every part except the last to be at least 5 MiB).
	defaultMultipartPartSize = 8 * 1024 * 1024 // 8 MiB

	// maxMultipartParts is the maximum number of parts S3 permits for a single
	// multipart upload.
	maxMultipartParts = 10000
)

// PutObject is called when putting objects into an s3 bucket.
func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest) (*pb.PutObjectResponse, error) {
	if req == nil {
		return nil, errors.BadRequestStatus("request is required")
	}
	if req.GetKey() == "" {
		return nil, errors.BadRequestStatus("key is required")
	}
	bucket := req.GetBucket()
	if bucket == nil {
		return nil, errors.BadRequestStatus("bucket is required")
	}
	if bucket.GetBucketName() == "" {
		return nil, errors.BadRequestStatus("bucketName is required")
	}
	attrs := bucket.GetAttributes()
	if attrs == nil {
		return nil, errors.BadRequestStatus("attributes is required")
	}
	if req.GetPath() == "" {
		return nil, errors.BadRequestStatus("path is required")
	}
	file, err := os.Open(req.GetPath())
	if err != nil {
		return nil, errors.BadRequestStatus("failed to open file")
	}
	info, err := file.Stat()
	if err != nil {
		return nil, errors.BadRequestStatus("failed to read file info")
	}
	if info.IsDir() {
		return nil, errors.BadRequestStatus("path is not a file")
	}
	if info.Size() == 0 {
		return nil, errors.BadRequestStatus("file is empty")
	}

	storageAttributes, err := getStorageAttributes(attrs)
	if err != nil {
		return nil, err
	}

	credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
	if err != nil {
		return nil, err
	}

	// Set up the creds in our persisted state.
	credState, err := cred.NewAwsCredentialPersistedState(
		append([]cred.AwsCredentialPersistedStateOption{
			cred.WithCredentialsConfig(credConfig),
		}, p.testCredStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	storageState, err := newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	opts := []s3Option{}
	if storageAttributes.EndpointUrl != "" {
		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
	}
	if storageAttributes.DualStack {
		opts = append(opts, WithDualStack(storageAttributes.DualStack))
	}

	hash := sha256.New()
	if _, err := io.Copy(hash, file); err != nil {
		return nil, errors.UnknownStatus("failed to calcualte hash")
	}
	checksum := base64.StdEncoding.EncodeToString(hash.Sum(nil))

	_, err = file.Seek(0, io.SeekStart)
	if err != nil {
		return nil, errors.UnknownStatus("failed to rewind file pointer")
	}
	objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())

	// decodedChecksum is the raw SHA-256 digest of the entire object. Boundary
	// compares this byte-for-byte against the hash it computed locally before
	// the upload, so it must always be the whole-object checksum regardless of
	// whether a single PutObject or a multipart upload is used.
	decodedChecksum, err := base64.StdEncoding.DecodeString(checksum)
	if err != nil {
		return nil, errors.UnknownStatus("failed to decode checksum value")
	}

	// for larger files, use a multipart upload instead of a single PutObject
	// request.
	if info.Size() > multipartThreshold {
		partSize := int64(defaultMultipartPartSize)
		// S3 allows at most maxMultipartParts parts per upload, so grow the
		// part size for very large objects to stay within that limit.
		if sz := info.Size(); sz/partSize >= maxMultipartParts {
			partSize = (sz + maxMultipartParts - 1) / maxMultipartParts
		}
		if err := p.putObjectMultipart(ctx, req, storageState, bucket.GetBucketName(), objectKey, file, partSize, opts...); err != nil {
			return nil, err
		}
		return &pb.PutObjectResponse{
			ChecksumSha_256: decodedChecksum,
		}, nil
	}

	putCall := func(s3Client S3API) (any, error) {
		return s3Client.PutObject(ctx, &s3.PutObjectInput{
			Bucket:            aws.String(bucket.GetBucketName()),
			Key:               aws.String(objectKey),
			Body:              file,
			ChecksumAlgorithm: types.ChecksumAlgorithmSha256,
			ChecksumSHA256:    aws.String(checksum),
		})
	}
	putResp, err := p.call(ctx, putCall, bucket.GetId(), storageState, opts...)
	if err != nil {
		return nil, parseS3Error("put object", err, req).Err()
	}
	resp := putResp.(*s3.PutObjectOutput)

	if resp.ChecksumSHA256 == nil {
		return nil, errors.UnknownStatus("missing checksum response from aws")
	}
	if checksum != *resp.ChecksumSHA256 {
		return nil, checksumMistmatchStatus()
	}
	return &pb.PutObjectResponse{
		ChecksumSha_256: decodedChecksum,
	}, nil
}

// putObjectMultipart uploads file to the given bucket and key using an S3
// multipart upload. The file is split into parts of partSize bytes (the final
// part may be smaller) and each part is uploaded with its own SHA-256 checksum
// so S3 can verify part integrity on the wire.
//
// If any step fails the upload is aborted on a best-effort basis to avoid
// leaving orphaned parts behind, which would otherwise continue to accrue
// storage charges.
func (p *StoragePlugin) putObjectMultipart(
	ctx context.Context,
	req *pb.PutObjectRequest,
	storageState *awsStoragePersistedState,
	bucketName, objectKey string,
	file *os.File,
	partSize int64,
	opts ...s3Option,
) error {
	bucketId := req.GetBucket().GetId()

	createCall := func(s3Client S3API) (any, error) {
		return s3Client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{
			Bucket:            aws.String(bucketName),
			Key:               aws.String(objectKey),
			ChecksumAlgorithm: types.ChecksumAlgorithmSha256,
		})
	}
	createResp, err := p.call(ctx, createCall, bucketId, storageState, opts...)
	if err != nil {
		return parseS3Error("create multipart upload", err, req).Err()
	}
	uploadId := createResp.(*s3.CreateMultipartUploadOutput).UploadId

	// once the upload has been created, any subsequent failure must abort it so
	// uploaded parts are not orphaned.
	abort := func() {
		abortCall := func(s3Client S3API) (any, error) {
			return s3Client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
				Bucket:   aws.String(bucketName),
				Key:      aws.String(objectKey),
				UploadId: uploadId,
			})
		}
		// best effort: the failure that triggered the abort is more useful to
		// the caller than an error from the abort itself.
		_, _ = p.call(ctx, abortCall, bucketId, storageState, opts...)
	}

	var completedParts []types.CompletedPart
	buf := make([]byte, partSize)
	var partNumber int32 = 1
	for {
		n, readErr := io.ReadFull(file, buf)
		if readErr == io.EOF {
			// previous iteration consumed the final bytes of the file
			break
		}
		if readErr != nil && readErr != io.ErrUnexpectedEOF {
			abort()
			return errors.UnknownStatusf("failed to read file: %s", readErr)
		}

		partData := buf[:n]
		sum := sha256.Sum256(partData)
		partChecksum := base64.StdEncoding.EncodeToString(sum[:])

		pn := partNumber
		uploadCall := func(s3Client S3API) (any, error) {
			return s3Client.UploadPart(ctx, &s3.UploadPartInput{
				Bucket:            aws.String(bucketName),
				Key:               aws.String(objectKey),
				UploadId:          uploadId,
				PartNumber:        aws.Int32(pn),
				Body:              bytes.NewReader(partData),
				ContentLength:     aws.Int64(int64(n)),
				ChecksumAlgorithm: types.ChecksumAlgorithmSha256,
				ChecksumSHA256:    aws.String(partChecksum),
			})
		}
		uploadResp, err := p.call(ctx, uploadCall, bucketId, storageState, opts...)
		if err != nil {
			abort()
			return parseS3Error("upload part", err, req).Err()
		}
		part := uploadResp.(*s3.UploadPartOutput)
		completedParts = append(completedParts, types.CompletedPart{
			ETag:           part.ETag,
			ChecksumSHA256: part.ChecksumSHA256,
			PartNumber:     aws.Int32(pn),
		})

		partNumber++
		if readErr == io.ErrUnexpectedEOF {
			// the final, short part
			break
		}
	}

	completeCall := func(s3Client S3API) (any, error) {
		return s3Client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
			Bucket:   aws.String(bucketName),
			Key:      aws.String(objectKey),
			UploadId: uploadId,
			MultipartUpload: &types.CompletedMultipartUpload{
				Parts: completedParts,
			},
		})
	}
	if _, err := p.call(ctx, completeCall, bucketId, storageState, opts...); err != nil {
		abort()
		return parseS3Error("complete multipart upload", err, req).Err()
	}

	return nil
}

// DeleteObjects is used to delete one or many objects from an s3 bucket.
func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjectsRequest) (*pb.DeleteObjectsResponse, error) {
	if req.GetKeyPrefix() == "" {
		return nil, errors.BadRequestStatus("key prefix is required")
	}

	bucket := req.GetBucket()
	if bucket == nil {
		return nil, errors.BadRequestStatus("bucket is required")
	}

	if bucket.GetBucketName() == "" {
		return nil, errors.BadRequestStatus("bucketName is required")
	}

	attrs := bucket.GetAttributes()
	if attrs == nil {
		return nil, errors.BadRequestStatus("attributes is required")
	}

	storageAttributes, err := getStorageAttributes(attrs)
	if err != nil {
		return nil, err
	}

	credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
	if err != nil {
		return nil, err
	}

	// Set up the creds in our persisted state.
	credState, err := cred.NewAwsCredentialPersistedState(
		append([]cred.AwsCredentialPersistedStateOption{
			cred.WithCredentialsConfig(credConfig),
		}, p.testCredStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	storageState, err := newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	opts := []s3Option{}
	if storageAttributes.EndpointUrl != "" {
		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
	}

	prefix := path.Join(bucket.GetBucketPrefix(), req.GetKeyPrefix())
	if strings.HasSuffix(req.GetKeyPrefix(), "/") {
		// path.Join ends by "cleaning" the path, including removing a trailing slash, if
		// it exists. given that a slash is used to denote a folder, it is required here.
		prefix += "/"
	}

	if !req.Recursive {
		deleteCall := func(s3Client S3API) (any, error) {
			return s3Client.DeleteObject(ctx, &s3.DeleteObjectInput{
				Bucket: aws.String(bucket.GetBucketName()),
				Key:    aws.String(prefix),
			})
		}
		_, err := p.call(ctx, deleteCall, bucket.GetId(), storageState, opts...)
		if err != nil {
			return nil, parseS3Error("delete object", err, req).Err()
		}
		return &pb.DeleteObjectsResponse{
			ObjectsDeleted: uint32(1),
		}, nil
	}

	const maxkeys = 1000
	objects := []types.ObjectIdentifier{}
	client, err := p.getClient(ctx, bucket.GetId(), storageState, opts...)
	if err != nil {
		return nil, errors.BadRequestStatusf("error getting S3 client: %s", err)
	}
	var conToken *string
	truncated, firstAttempt := true, true
	for truncated {
		res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
			Bucket:            aws.String(bucket.GetBucketName()),
			Prefix:            aws.String(prefix),
			MaxKeys:           aws.Int32(maxkeys),
			ContinuationToken: conToken,
		})
		if err != nil {
			st, _ := errors.ParseAWSError("", err)
			if st.Code() == codes.PermissionDenied && firstAttempt {
				// This is first attempt and we get a permission error, refresh cache and try again
				firstAttempt = false
				opts = append(opts, WithCacheRefresh(true))
				client, err = p.getClient(ctx, bucket.GetId(), storageState, opts...)
				if err != nil {
					return nil, errors.BadRequestStatusf("error refreshing S3 client: %s", err)
				}
				continue
			}

			return nil, parseS3Error("list objects", err, req).Err()
		}
		truncated = aws.ToBool(res.IsTruncated)
		conToken = res.NextContinuationToken
		for _, o := range res.Contents {
			objects = append(objects, types.ObjectIdentifier{
				Key: o.Key,
			})
		}
	}

	deleted := 0

	for i := 0; i < len(objects); i += maxkeys {
		toDelete := objects[i:min(len(objects), i+maxkeys)] // min is required to avoid an out of bounds panic
		res, err := client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
			Bucket: aws.String(bucket.GetBucketName()),
			Delete: &types.Delete{
				Objects: toDelete,
			},
		})
		if err != nil {
			return nil, parseS3Error("delete objects", err, req).Err()
		}
		deleted += len(res.Deleted)
	}

	return &pb.DeleteObjectsResponse{
		ObjectsDeleted: uint32(deleted),
	}, nil
}

// ListObjects retrieves files and directories from an S3 bucket under a given prefix.
// If the "recursive" flag in the request is false (default), it returns only the top-level objects directly under the prefix.
// If the "recursive" is true, it returns all objects under the prefix recursively.
func (p *StoragePlugin) ListObjects(ctx context.Context, req *pb.ListObjectsRequest) (*pb.ListObjectsResponse, error) {
	bucket := req.GetBucket()
	if bucket == nil {
		return nil, errors.BadRequestStatus("bucket is required")
	}

	if bucket.BucketName == "" {
		return nil, errors.BadRequestStatus("bucketName is required")
	}

	attrs := bucket.GetAttributes()
	if attrs == nil {
		return nil, errors.BadRequestStatus("attributes is required")
	}

	storageAttributes, err := getStorageAttributes(attrs)
	if err != nil {
		return nil, err
	}

	credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, false)
	if err != nil {
		return nil, err
	}

	// Set up the creds in our persisted state.
	credState, err := cred.NewAwsCredentialPersistedState(
		append([]cred.AwsCredentialPersistedStateOption{
			cred.WithCredentialsConfig(credConfig),
		}, p.testCredStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	storageState, err := newAwsStoragePersistedState(
		append([]awsStoragePersistedStateOption{
			withCredentials(credState),
		}, p.testStorageStateOpts...)...,
	)
	if err != nil {
		return nil, errors.BadRequestStatusf("error setting up persisted state: %s", err)
	}

	var opts []s3Option
	if storageAttributes.EndpointUrl != "" {
		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
	}
	s3Client, err := storageState.s3Client(ctx, opts...)
	if err != nil {
		return nil, errors.BadRequestStatusf("error getting S3 client: %s", err)
	}

	prefix := path.Join(bucket.GetBucketPrefix(), req.GetKeyPrefix())
	if strings.HasSuffix(req.GetKeyPrefix(), "/") {
		// path.Join ends by "cleaning" the path, including removing a trailing slash, if
		// it exists. given that a slash is used to denote a folder, it is required here.
		prefix += "/"
	}
	var objects []*pb.Object

	if req.GetRecursive() {
		objects, err = listObjectsRecursive(ctx, s3Client, bucket.GetBucketName(), prefix, req)
	} else {
		objects, err = listObjectsTopLevel(ctx, s3Client, bucket.GetBucketName(), prefix, req)
	}
	if err != nil {
		return nil, err
	}

	return &pb.ListObjectsResponse{
		Objects: objects,
	}, nil
}

// listObjectsTopLevel lists only the objects that exist directly under the specified prefix (This does not traverse subdirectories)
// It uses the S3 "Delimiter" option to group results so that subdirectories appear as CommonPrefixes.
func listObjectsTopLevel(ctx context.Context, s3Client S3API, bucketName, prefix string, req *pb.ListObjectsRequest) ([]*pb.Object, error) {
	var objects []*pb.Object
	paginator := s3.NewListObjectsV2Paginator(s3Client, &s3.ListObjectsV2Input{
		Bucket:    aws.String(bucketName),
		Prefix:    aws.String(prefix),
		Delimiter: aws.String("/"),
	})

	for paginator.HasMorePages() {
		if err := ctx.Err(); err != nil {
			switch {
			case stderr.Is(err, context.Canceled):
				return nil, status.Error(codes.Canceled, "request canceled")
			case stderr.Is(err, context.DeadlineExceeded):
				return nil, status.Error(codes.DeadlineExceeded, "request deadline exceeded")
			default:
				return nil, status.Error(codes.Internal, "unknown context error")
			}
		}
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return nil, parseS3Error("failed to list objects", err, req).Err()
		}

		for _, c := range page.Contents {
			if c.Key != nil {
				objects = append(objects, &pb.Object{Key: *c.Key, IsDir: false})
			}
		}
		for _, cp := range page.CommonPrefixes {
			if cp.Prefix != nil {
				objects = append(objects, &pb.Object{Key: *cp.Prefix, IsDir: true})
			}
		}
	}
	return objects, nil
}

// listObjectsRecursive lists all objects under the specified prefix, traversing all subdirectories.
// It does not use the S3 "Delimiter" option, so every object key that matches the prefix is returned.
// This mode is useful for fully scanning all objects in a folder tree.
func listObjectsRecursive(ctx context.Context, s3Client S3API, bucketName, prefix string, req *pb.ListObjectsRequest) ([]*pb.Object, error) {
	var objects []*pb.Object
	folderSet := make(map[string]struct{})
	paginator := s3.NewListObjectsV2Paginator(s3Client, &s3.ListObjectsV2Input{
		Bucket: aws.String(bucketName),
		Prefix: aws.String(prefix),
	})

	for paginator.HasMorePages() {
		if err := ctx.Err(); err != nil {
			switch {
			case stderr.Is(err, context.Canceled):
				return nil, status.Error(codes.Canceled, "request canceled")
			case stderr.Is(err, context.DeadlineExceeded):
				return nil, status.Error(codes.DeadlineExceeded, "request deadline exceeded")
			default:
				return nil, status.Error(codes.Internal, "unknown context error")
			}
		}
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return nil, parseS3Error("failed to list objects", err, req).Err()
		}

		for _, c := range page.Contents {
			// We intentionally ignore "empty objects" that simulate folders.
			// S3 is not a traditional file system and does not have real directories.
			// Folders are an abstraction; they are represented by zero-byte objects
			// whose keys end with a slash (e.g., "myfolder/"). These objects exist
			// solely as placeholders to give the appearance of directories in S3 tools.
			// Here, we only include actual objects with content (non-zero length).
			if c.Key != nil && !strings.HasSuffix(*c.Key, "/") {
				objects = append(objects, &pb.Object{Key: *c.Key, IsDir: false})
			}

			// Extract all parent folder paths from the object key.
			// We walk up the directory hierarchy using path.Dir,
			// adding each parent folder to folderSet until we reach
			// the root ("/" or ".") or leave the current prefix subtree.
			// This ensures all intermediate "folders" appear in the result even if
			// no explicit zero-byte object exists for them.
			dir := path.Dir(*c.Key)
			for dir != "." && dir != "/" {
				folder := dir + "/"
				if !strings.HasPrefix(folder, prefix) {
					break // left the subtree, stop climbing up
				}
				if folder != prefix {
					folderSet[folder] = struct{}{}
				}
				dir = path.Dir(dir)
			}
		}
	}
	// Add folders
	for folder := range folderSet {
		objects = append(objects, &pb.Object{
			Key:   folder,
			IsDir: true,
		})
	}
	return objects, nil
}

// dryRunValidation validates that the IAM role policy attached to the given secrets of the
// storage bucket has the minimum required permissions needed for this plugin to function
// as expected. This function will create an object in the s3 bucket to validate it has
// PutObject permissions. This function will read the same file it had uploaded to validate
// it has GetObject permissions. This function will read the metadata of the same file it
// had uploaded to validate it has GetObjectAttributes permissions.
func dryRunValidation(ctx context.Context, state *awsStoragePersistedState, attrs *StorageAttributes, bucket *storagebuckets.StorageBucket) *status.Status {
	if state == nil {
		return status.New(codes.InvalidArgument, "persisted state is required")
	}
	if attrs == nil {
		return status.New(codes.InvalidArgument, "attributes is required")
	}
	if bucket == nil {
		return status.New(codes.InvalidArgument, "bucket is required")
	}

	opts := []s3Option{}
	if attrs.EndpointUrl != "" {
		opts = append(opts, WithEndpoint(attrs.EndpointUrl))
	}
	if attrs.DualStack {
		opts = append(opts, WithDualStack(attrs.DualStack))
	}

	client, err := state.s3Client(ctx, opts...)
	if err != nil {
		return status.New(codes.InvalidArgument, fmt.Sprintf("error getting S3 client: %s", err))
	}

	var errs *multierror.Error
	permissions := &pb.Permissions{}
	// we track the codes returned by ParseAWSError and pick the most severe one by using the max func

	objectKey := path.Join(bucket.GetBucketPrefix(), uuid.New().String())
	if _, err := client.PutObject(ctx, &s3.PutObjectInput{
		Bucket: aws.String(bucket.GetBucketName()),
		Key:    aws.String(objectKey),
		Body:   bytes.NewReader([]byte("hashicorp boundary aws plugin access test")),
	}); err != nil {
		var st *status.Status
		st, permissions.Write = errors.ParseAWSError("put object", err)
		errs = multierror.Append(errs, st.Err())
	} else {
		permissions.Write = &pb.Permission{State: pb.StateType_STATE_TYPE_OK, CheckedAt: timestamppb.Now()}
	}

	if _, err := client.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(bucket.GetBucketName()),
		Key:    aws.String(objectKey),
	}); err != nil {
		var st *status.Status
		st, permissions.Read = errors.ParseAWSError("get object", err)
		errs = multierror.Append(errs, st.Err())
	} else {
		permissions.Read = &pb.Permission{State: pb.StateType_STATE_TYPE_OK, CheckedAt: timestamppb.Now()}
	}

	if _, err := client.HeadObject(ctx, &s3.HeadObjectInput{
		Bucket: aws.String(bucket.GetBucketName()),
		Key:    aws.String(objectKey),
	}); err != nil {
		var st *status.Status
		st, permissions.Read = errors.ParseAWSError("head object", err)
		errs = multierror.Append(errs, st.Err())
	}

	if res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
		Bucket: aws.String(bucket.GetBucketName()),
		Prefix: aws.String(objectKey),
	}); err != nil {
		var st *status.Status
		st, permissions.Read = errors.ParseAWSError("list object", err)
		errs = multierror.Append(errs, st.Err())
	} else if res == nil || len(res.Contents) != 1 || *res.Contents[0].Key != objectKey {
		permissions.Read = &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, CheckedAt: timestamppb.Now()}
		errs = multierror.Append(errs, status.New(codes.Aborted, fmt.Sprintf("list response did not contain the expected key: %+v", res)).Err())
	}

	if _, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{
		Bucket: aws.String(bucket.GetBucketName()),
		Key:    aws.String(objectKey),
	}); err != nil {
		var st *status.Status
		st, permissions.Delete = errors.ParseAWSError("delete object", err)
		errs = multierror.Append(errs, st.Err())
	} else {
		permissions.Delete = &pb.Permission{State: pb.StateType_STATE_TYPE_OK, CheckedAt: timestamppb.Now()}
	}

	if errs != nil {
		st := status.New(codes.FailedPrecondition, fmt.Sprintf("failed to validate provided aws credentials: %v", errs.Unwrap()))
		state := &pb.StorageBucketCredentialState{State: permissions}
		if st, err = st.WithDetails(state); err != nil {
			st = status.New(codes.Internal, err.Error())
		}
		return st
	}

	return nil
}