// The AWS provisioner (§3): S3 + IAM behind the Provisioner seam, path-scoped
// to exactly the bucket a store's user owns. Every idempotence mapping here is
// load-bearing — the grant, rotate and teardown walks in this package are
// re-runnable after a half-failure only because each verb below treats
// "already exists" and "already gone" as success rather than error.
package objstore

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

	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/iam"
	iamtypes "github.com/aws/aws-sdk-go-v2/service/iam/types"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/smithy-go"
	"github.com/carlosframework/platform/internal/rootaws"
)

// policyName is the inline policy every store user carries. One name, one
// document per user — PutUserPolicy overwrites it wholesale, so there is
// nothing to merge or version.
const policyName = "carlos-store"

// s3API is the subset of *s3.Client awsProvisioner calls, narrowed so a test
// can stand a stub in for a live client without a network or credentials.
type s3API interface {
	CreateBucket(ctx context.Context, in *awss3.CreateBucketInput, optFns ...func(*awss3.Options)) (*awss3.CreateBucketOutput, error)
	DeleteBucket(ctx context.Context, in *awss3.DeleteBucketInput, optFns ...func(*awss3.Options)) (*awss3.DeleteBucketOutput, error)
	ListObjectsV2(ctx context.Context, in *awss3.ListObjectsV2Input, optFns ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error)
	DeleteObjects(ctx context.Context, in *awss3.DeleteObjectsInput, optFns ...func(*awss3.Options)) (*awss3.DeleteObjectsOutput, error)
}

// iamAPI is the subset of *iam.Client awsProvisioner calls.
type iamAPI interface {
	CreateUser(ctx context.Context, in *iam.CreateUserInput, optFns ...func(*iam.Options)) (*iam.CreateUserOutput, error)
	GetUser(ctx context.Context, in *iam.GetUserInput, optFns ...func(*iam.Options)) (*iam.GetUserOutput, error)
	PutUserPolicy(ctx context.Context, in *iam.PutUserPolicyInput, optFns ...func(*iam.Options)) (*iam.PutUserPolicyOutput, error)
	DeleteUserPolicy(ctx context.Context, in *iam.DeleteUserPolicyInput, optFns ...func(*iam.Options)) (*iam.DeleteUserPolicyOutput, error)
	DeleteUser(ctx context.Context, in *iam.DeleteUserInput, optFns ...func(*iam.Options)) (*iam.DeleteUserOutput, error)
	CreateAccessKey(ctx context.Context, in *iam.CreateAccessKeyInput, optFns ...func(*iam.Options)) (*iam.CreateAccessKeyOutput, error)
	DeleteAccessKey(ctx context.Context, in *iam.DeleteAccessKeyInput, optFns ...func(*iam.Options)) (*iam.DeleteAccessKeyOutput, error)
	ListAccessKeys(ctx context.Context, in *iam.ListAccessKeysInput, optFns ...func(*iam.Options)) (*iam.ListAccessKeysOutput, error)
}

// awsProvisioner implements Provisioner against real AWS S3 + IAM.
type awsProvisioner struct {
	s3     s3API
	iam    iamAPI
	region string
	// boundary is the permissions-boundary policy ARN stamped on every store
	// user this provisioner creates. See EnsureUser for why it is not
	// optional in any deployment whose console role can reach IAM.
	boundary string
}

var _ Provisioner = (*awsProvisioner)(nil)

// NewAWSProvisioner builds a Provisioner against AWS S3 and IAM in region.
// The S3 client applies s3compat-style options when endpoint is non-empty
// (hand-rolled here rather than calling internal/s3compat.Options directly,
// so this package stays independent of CARLOS_S3_ENDPOINT — the
// deployment's Endpoint is threaded in explicitly instead). These four lines
// mirror s3compat.Options' universal (non-AWS) block line for line and MUST
// move in lockstep with it: BaseEndpoint + UsePathStyle are the addressing
// fix, and the checksum pair exists because default-on flexible checksums
// wrap bodies in aws-chunked framing some S3-compatibles reject — DeleteObjects
// (EmptyBucket's batch delete) is exactly a checksummed request, so dropping
// either checksum line silently breaks teardown against those stores. Not
// applied: s3compat's per-provider Quirks table (e.g. the Hetzner If-Match
// unquoting) — none of it is conditional writes, which this provisioner
// never issues.
//
// The IAM client never takes an endpoint override: IAM has no S3-compatible
// twin, so a non-AWS backend (endpoint set, "aws" backend) still means
// IAM-against-real-AWS in v1 — there is nothing else for it to point at.
func NewAWSProvisioner(ctx context.Context, region, endpoint, boundary string) (Provisioner, error) {
	s3Client, err := S3ClientFor(ctx, region, endpoint)
	if err != nil {
		return nil, err
	}
	cfg, err := rootaws.LoadDefaultConfig(ctx, awsconfig.WithRegion(region))
	if err != nil {
		return nil, fmt.Errorf("objstore: load aws config: %w", err)
	}
	iamClient := iam.NewFromConfig(cfg)
	return &awsProvisioner{s3: s3Client, iam: iamClient, region: region, boundary: boundary}, nil
}

// S3ClientFor builds a bare *s3.Client the same way NewAWSProvisioner builds
// its own — region + the s3compat-style option layering (BaseEndpoint,
// UsePathStyle, the checksum pair) applied only when endpoint is non-empty —
// for callers that need the raw client rather than the Provisioner seam.
// The economics storage sweep (console.sizeProvisionedStore) is the first:
// it sizes a tenant's provisioned bucket directly, with no IAM involved.
func S3ClientFor(ctx context.Context, region, endpoint string) (*awss3.Client, error) {
	cfg, err := rootaws.LoadDefaultConfig(ctx, awsconfig.WithRegion(region))
	if err != nil {
		return nil, fmt.Errorf("objstore: load aws config: %w", err)
	}
	return awss3.NewFromConfig(cfg, func(o *awss3.Options) {
		if endpoint != "" {
			o.BaseEndpoint = aws.String(endpoint)
			o.UsePathStyle = true
			o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
			o.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
		}
	}), nil
}

// ---- policy document --------------------------------------------------

type policyStatement struct {
	Effect   string   `json:"Effect"`
	Action   []string `json:"Action"`
	Resource string   `json:"Resource"`
	// Condition is omitted entirely when nil. The value side is `any` because
	// IAM condition values are sometimes a string (StringEquals on one tag)
	// and sometimes a list (ForAnyValue:StringEquals over tag KEYS); a
	// map[string]string could only express the first, and the second is what
	// closes the upload-tagging route below.
	Condition map[string]map[string]any `json:"Condition,omitempty"`
}

type policyDocument struct {
	Version   string            `json:"Version"`
	Statement []policyStatement `json:"Statement"`
}

// storePolicyJSON is the exact document every store user's inline policy
// carries: ListBucket on the bucket itself, object verbs scoped under it, and
// — when deny is non-nil — a Deny on reading an object the scanner has tagged
// infected (spec §8).
//
// s3:GetObjectTagging is in the Allow so an app can read the verdict on its
// own object. s3:PutObjectTagging is NOT, and MUST NEVER BE: every verdict
// this platform reports is only worth anything while an app cannot mark its
// own object clean.
//
// ABSENCE IS NOT ENOUGH, which cost this design a real hole. There are TWO
// ways to put a tag on an object, and leaving the permission out of the Allow
// only closes one of them: PutObjectTagging, the explicit call. The other is
// the `x-amz-tagging` header on PutObject itself, which sets tags during the
// upload without ever naming the tagging action. AWS documents that header as
// requiring s3:PutObjectTagging; a gateway that does not enforce that reading
// hands any app with s3:PutObject the ability to upload a payload already
// marked clean — and the sweep, finding a version-matching clean tag, will
// never scan it. `carlos store scan status` then reports it clean forever.
//
// This is not hypothetical. Against MinIO RELEASE.2025-09-07T16-13-09Z, run
// from CI's own contract fixture (.amadan/ci.d/30-minio-contract), the upload
// succeeded with the tag attached. Adding an explicit Deny on
// s3:PutObjectTagging — the obvious fix, and the first one tried — did NOT
// close it, because the gateway never attributed the header to that action.
// Only denying s3:PutObject itself when the request carries one of our tag
// keys did. Both Denys are therefore below, unconditionally, for every store
// whether or not enforcement is on: an explicit Deny costs nothing where the
// permission was already absent, and it is the only form that travels to a
// backend whose IAM does not read the header the way AWS's does.
//
// TestDeliveredContractMinio holds the live half of this against a real
// server on every push. Do not weaken either statement without watching that
// test fail first.
//
// Adding GetObjectTagging changes this document for every existing store,
// which is picked up on the next grant re-walk. That is a deliberate one-time
// rewrite, not a regression: PutUserPolicy overwrites wholesale and is
// idempotent, so the cost is one extra IAM call per re-walk.
//
// The deny is weaker than AWS's own tag-based access control example, which
// denies reads UNLESS an object is tagged clean and so blocks unscanned
// objects too. This one blocks only known-infected objects; unscanned ones
// stay readable. That is the right trade when the goal is evidence rather
// than enforcement, but anyone quoting it in a compliance answer needs to
// know which of the two they have.
func storePolicyJSON(bucketARN string, deny *DenyTag, denyWrites bool) string {
	doc := policyDocument{
		Version: "2012-10-17",
		Statement: []policyStatement{
			{Effect: "Allow", Action: []string{"s3:ListBucket"}, Resource: bucketARN},
			{Effect: "Allow", Action: []string{
				"s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:GetObjectTagging",
			}, Resource: bucketARN + "/*"},
		},
	}
	// The two anti-forgery Denys, always. See above for why the second one
	// exists and why the first is not sufficient on its own.
	doc.Statement = append(doc.Statement,
		policyStatement{
			Effect: "Deny",
			Action: []string{
				"s3:PutObjectTagging", "s3:DeleteObjectTagging",
				// Nothing here ever calls PutBucketVersioning, so these two
				// are unreachable today. They cost nothing and close the shape
				// before a bucket is ever versioned — which is the cheap half
				// of the lesson this policy has already learned once.
				"s3:PutObjectVersionTagging", "s3:DeleteObjectVersionTagging",
			},
			Resource: bucketARN + "/*",
		},
		policyStatement{
			Effect:   "Deny",
			Action:   []string{"s3:PutObject"},
			Resource: bucketARN + "/*",
			Condition: map[string]map[string]any{
				"ForAnyValue:StringEquals": {
					"s3:RequestObjectTagKeys": []string{TagVerdict, TagVersion, TagAt},
				},
			},
		},
	)
	if deny != nil {
		doc.Statement = append(doc.Statement, policyStatement{
			Effect:   "Deny",
			Action:   []string{"s3:GetObject"},
			Resource: bucketARN + "/*",
			Condition: map[string]map[string]any{
				"StringEquals": {"s3:ExistingObjectTag/" + deny.Key: deny.Value},
			},
		})
	}
	if denyWrites {
		// Plans 4b: a store past its plan's storage grace refuses new
		// uploads. Unconditional, and on PutObject only: reads, lists and
		// deletes stay allowed so the customer can clean up.
		doc.Statement = append(doc.Statement, policyStatement{
			Effect:   "Deny",
			Action:   []string{"s3:PutObject"},
			Resource: bucketARN + "/*",
		})
	}
	b, err := json.Marshal(doc)
	if err != nil {
		// doc is a static struct of strings/string-slices; Marshal cannot fail.
		panic(fmt.Sprintf("objstore: marshal store policy: %v", err))
	}
	return string(b)
}

// ---- error classification ----------------------------------------------

// isNoSuchEntity reports whether err is IAM's NoSuchEntityException — the
// "already gone" answer ListAccessKeys, DeleteAccessKey, DeleteUserPolicy and
// DeleteUser all use for their respective absent target.
func isNoSuchEntity(err error) bool {
	var nse *iamtypes.NoSuchEntityException
	return errors.As(err, &nse)
}

// isNoSuchBucket reports whether err is S3's "bucket does not exist" answer.
// DeleteBucket and DeleteObjects surface the typed s3types.NoSuchBucket, but
// ListObjectsV2 has been observed to surface it only as a generic smithy
// APIError with code "NoSuchBucket" — so both forms are checked.
func isNoSuchBucket(err error) bool {
	var nsb *s3types.NoSuchBucket
	if errors.As(err, &nsb) {
		return true
	}
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) && apiErr.ErrorCode() == "NoSuchBucket" {
		return true
	}
	return false
}

// ---- S3 -------------------------------------------------------------------

// EnsureBucket creates name if absent and answers the region it was
// provisioned in. AWS returns BucketAlreadyOwnedByYou (not us-east-1) or a
// bare success (us-east-1's historical quirk) when the caller's own prior
// CreateBucket already succeeded, so that shape is success too — this is what
// makes the grant walk's bucket step re-runnable. BucketAlreadyExists (a
// different account owns the global name) is a real conflict, not treated as
// idempotent: this provisioner has no way to tell "you already made this"
// from "someone else got the name first", and only the former is safe to
// swallow.
func (p *awsProvisioner) EnsureBucket(ctx context.Context, name string) (string, error) {
	in := &awss3.CreateBucketInput{Bucket: aws.String(name)}
	if p.region != "" && p.region != "us-east-1" {
		in.CreateBucketConfiguration = &s3types.CreateBucketConfiguration{
			LocationConstraint: s3types.BucketLocationConstraint(p.region),
		}
	}
	if _, err := p.s3.CreateBucket(ctx, in); err != nil {
		var owned *s3types.BucketAlreadyOwnedByYou
		if !errors.As(err, &owned) {
			return "", fmt.Errorf("objstore: create bucket %q: %w", name, err)
		}
	}
	return p.region, nil
}

// EmptyBucket deletes every object in name, paginating ListObjectsV2 and
// batching DeleteObjects at <=1000 keys (the API's own ceiling) so a crash
// mid-empty resumes cleanly on the next page. An absent bucket — including
// discovering that only partway through, on a later ListObjectsV2 page — is
// success: teardown resumed after a crash between DeleteBucket and DeleteUser
// re-runs this against a bucket that is already gone.
func (p *awsProvisioner) EmptyBucket(ctx context.Context, name string) error {
	var token *string
	for {
		out, err := p.s3.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
			Bucket:            aws.String(name),
			ContinuationToken: token,
		})
		if err != nil {
			if isNoSuchBucket(err) {
				return nil
			}
			return fmt.Errorf("objstore: list objects in %q: %w", name, err)
		}
		if err := p.deleteObjectBatches(ctx, name, out.Contents); err != nil {
			return err
		}
		if !aws.ToBool(out.IsTruncated) {
			return nil
		}
		token = out.NextContinuationToken
	}
}

// deleteObjectBatches issues DeleteObjects in chunks of <=1000 keys, the
// API's own limit per call.
func (p *awsProvisioner) deleteObjectBatches(ctx context.Context, bucket string, objs []s3types.Object) error {
	const batchSize = 1000
	for start := 0; start < len(objs); start += batchSize {
		end := start + batchSize
		if end > len(objs) {
			end = len(objs)
		}
		ids := make([]s3types.ObjectIdentifier, end-start)
		for i, obj := range objs[start:end] {
			ids[i] = s3types.ObjectIdentifier{Key: obj.Key}
		}
		out, err := p.s3.DeleteObjects(ctx, &awss3.DeleteObjectsInput{
			Bucket: aws.String(bucket),
			Delete: &s3types.Delete{Objects: ids},
		})
		if err != nil {
			if isNoSuchBucket(err) {
				return nil
			}
			return fmt.Errorf("objstore: delete objects in %q: %w", bucket, err)
		}
		// DeleteObjects is a 200-with-body API: a per-key failure (AccessDenied,
		// object-lock, ...) never surfaces as err — it lands in out.Errors while
		// the call itself reports success. Left unchecked, EmptyBucket returns
		// nil with objects still in the bucket, and Teardown then dies later at
		// DeleteBucket's BucketNotEmpty with no key and no reason, retrying
		// forever without ever naming the object that's actually stuck.
		if len(out.Errors) > 0 {
			first := out.Errors[0]
			return fmt.Errorf("objstore: delete objects in %q: %d failed, first %s: %s (%s)",
				bucket, len(out.Errors), aws.ToString(first.Key), aws.ToString(first.Code), aws.ToString(first.Message))
		}
	}
	return nil
}

// DeleteBucket removes name. Absent is success: it is the re-run shape after
// a crash right after a prior DeleteBucket succeeded but before that was
// recorded.
func (p *awsProvisioner) DeleteBucket(ctx context.Context, name string) error {
	if _, err := p.s3.DeleteBucket(ctx, &awss3.DeleteBucketInput{Bucket: aws.String(name)}); err != nil {
		if isNoSuchBucket(err) {
			return nil
		}
		return fmt.Errorf("objstore: delete bucket %q: %w", name, err)
	}
	return nil
}

// ---- IAM --------------------------------------------------------------

// EnsureUser creates the principal at path if absent, scoped to bucketARN by
// an inline policy that is written unconditionally either way. A user that
// already exists (EntityAlreadyExistsException — the grant walk re-run after
// a crash between CreateUser and PutUserPolicy) is resolved to its ARN via
// GetUser rather than treated as an error; PutUserPolicy is an idempotent
// overwrite, so it always runs regardless of which branch produced the ARN.
// The user is created UNDER A PERMISSIONS BOUNDARY, and that is the load
// bearing half of this function rather than a hardening extra.
//
// A role that may call CreateUser and PutUserPolicy can, without a boundary,
// mint a principal carrying any policy it likes — including one far wider
// than the role itself. Scoping those two actions to a path is not a fix:
// the path constrains WHICH user is made, never what that user may do. So
// the deployment's own IAM conditions both calls on this exact boundary ARN
// (infra/modules/region-host), and the boundary permits nothing but S3 on
// the deployment's own bucket prefix. The inline policy below then narrows
// that ceiling to one bucket; effective permissions are the intersection,
// and the ceiling holds even if this code one day writes the wrong document.
//
// An empty boundary is legal here only because MemProvisioner and the
// contract tests construct provisioners with no IAM at all. Every real
// console refuses to boot without one (cmd/console/main.go, storeBackend).
func (p *awsProvisioner) EnsureUser(ctx context.Context, name, path, bucketARN string, deny *DenyTag, denyWrites bool) (string, error) {
	var arn string
	in := &iam.CreateUserInput{
		UserName: aws.String(name),
		Path:     aws.String(path),
	}
	if p.boundary != "" {
		in.PermissionsBoundary = aws.String(p.boundary)
	}
	out, err := p.iam.CreateUser(ctx, in)
	if err != nil {
		var exists *iamtypes.EntityAlreadyExistsException
		if !errors.As(err, &exists) {
			return "", fmt.Errorf("objstore: create user %q: %w", name, err)
		}
		got, gerr := p.iam.GetUser(ctx, &iam.GetUserInput{UserName: aws.String(name)})
		if gerr != nil {
			return "", fmt.Errorf("objstore: get existing user %q: %w", name, gerr)
		}
		arn = aws.ToString(got.User.Arn)
	} else {
		arn = aws.ToString(out.User.Arn)
	}

	_, err = p.iam.PutUserPolicy(ctx, &iam.PutUserPolicyInput{
		UserName:       aws.String(name),
		PolicyName:     aws.String(policyName),
		PolicyDocument: aws.String(storePolicyJSON(bucketARN, deny, denyWrites)),
	})
	if err != nil {
		return "", fmt.Errorf("objstore: put policy for user %q: %w", name, err)
	}
	return arn, nil
}

// MintKey creates a fresh access key for user.
func (p *awsProvisioner) MintKey(ctx context.Context, user string) (string, string, error) {
	out, err := p.iam.CreateAccessKey(ctx, &iam.CreateAccessKeyInput{UserName: aws.String(user)})
	if err != nil {
		return "", "", fmt.Errorf("objstore: create access key for %q: %w", user, err)
	}
	return aws.ToString(out.AccessKey.AccessKeyId), aws.ToString(out.AccessKey.SecretAccessKey), nil
}

// DeleteKey removes access key id from user. Absent is success: it is the
// shape Rotate step-1 and RotateFinish's PrevKeyID escape hatch both produce
// on a re-run after the key was already deleted.
func (p *awsProvisioner) DeleteKey(ctx context.Context, user, id string) error {
	_, err := p.iam.DeleteAccessKey(ctx, &iam.DeleteAccessKeyInput{
		UserName:    aws.String(user),
		AccessKeyId: aws.String(id),
	})
	if err != nil && !isNoSuchEntity(err) {
		return fmt.Errorf("objstore: delete access key %q for %q: %w", id, user, err)
	}
	return nil
}

// ReapKeys deletes every access key of user not named in keep. A user that no
// longer exists (NoSuchEntity from ListAccessKeys) is success with nothing to
// reap — the grant walk's crash-safety depends on this list-and-delete being
// safe to re-run against a user that got torn down out from under it.
func (p *awsProvisioner) ReapKeys(ctx context.Context, user string, keep []string) error {
	keepSet := make(map[string]bool, len(keep))
	for _, id := range keep {
		keepSet[id] = true
	}
	var marker *string
	for {
		out, err := p.iam.ListAccessKeys(ctx, &iam.ListAccessKeysInput{
			UserName: aws.String(user),
			Marker:   marker,
		})
		if err != nil {
			if isNoSuchEntity(err) {
				return nil
			}
			return fmt.Errorf("objstore: list access keys for %q: %w", user, err)
		}
		for _, k := range out.AccessKeyMetadata {
			id := aws.ToString(k.AccessKeyId)
			if keepSet[id] {
				continue
			}
			if err := p.DeleteKey(ctx, user, id); err != nil {
				return err
			}
		}
		if !out.IsTruncated {
			return nil
		}
		marker = out.Marker
	}
}

// DeleteUser removes user entirely: every access key, then the inline
// policy, then the user itself — IAM refuses to delete a user that still
// owns either. A pending-finish PrevKeyID (a key Rotate minted but no
// RotateFinish ever reaped) is caught here by ReapKeys(keep=nil) during
// teardown even if nothing else ever reaped it. Absent at any step
// (NoSuchEntity) is success throughout, including a bare re-run against a
// user already fully deleted.
func (p *awsProvisioner) DeleteUser(ctx context.Context, user string) error {
	if err := p.ReapKeys(ctx, user, nil); err != nil {
		return err
	}
	if _, err := p.iam.DeleteUserPolicy(ctx, &iam.DeleteUserPolicyInput{
		UserName:   aws.String(user),
		PolicyName: aws.String(policyName),
	}); err != nil && !isNoSuchEntity(err) {
		return fmt.Errorf("objstore: delete policy for user %q: %w", user, err)
	}
	if _, err := p.iam.DeleteUser(ctx, &iam.DeleteUserInput{UserName: aws.String(user)}); err != nil && !isNoSuchEntity(err) {
		return fmt.Errorf("objstore: delete user %q: %w", user, err)
	}
	return nil
}