package host

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"sort"
	"strings"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/credentials"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"

	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/objstore"
)

// The storage probes of the target's report: T4's probe of every store
// bound in the composed environment, and T10's return lane in the source's
// bucket. armPhase1 wires them as moveCfg.ProbeStore and LaneProbe.

// probeStores is T4 in production (§7.3): for each store bound in the
// composed environment (objstore.EnvNames), with that store's own
// credentials, list one key, then put, get and delete
// .carlos-move-probe/<move id>. It returns the first failing binding's
// prefix with its error. The client is built the way a member app builds
// it from the delivered variables (objstore's delivered contract): static
// credentials, the delivered region, and the endpoint path-style when one
// is delivered.
func probeStores(ctx context.Context, env map[string]string, id string) (string, error) {
	for _, p := range storePrefixes(env) {
		if err := probeStore(ctx, env, p, id); err != nil {
			return p, err
		}
	}
	return "", nil
}

func probeStore(ctx context.Context, env map[string]string, prefix, id string) error {
	bn, rn, en, ak, sk := objstore.EnvNames(prefix)
	region := env[rn]
	if region == "" {
		region = "us-east-1"
	}
	opts := awss3.Options{
		Region:                     region,
		Credentials:                credentials.NewStaticCredentialsProvider(env[ak], env[sk], ""),
		RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
		ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
	}
	if ep := env[en]; ep != "" {
		opts.BaseEndpoint = aws.String(ep)
		opts.UsePathStyle = true
	}
	c := awss3.New(opts)
	bucket := aws.String(env[bn])
	key := aws.String(".carlos-move-probe/" + id)
	body := []byte("carlos move probe " + id)
	if _, err := c.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{Bucket: bucket, MaxKeys: aws.Int32(1)}); err != nil {
		return fmt.Errorf("list: %w", err)
	}
	if _, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: bucket, Key: key, Body: bytes.NewReader(body)}); err != nil {
		return fmt.Errorf("put: %w", err)
	}
	out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: bucket, Key: key})
	if err != nil {
		return fmt.Errorf("get: %w", err)
	}
	got, err := io.ReadAll(io.LimitReader(out.Body, 1<<16))
	out.Body.Close()
	if err != nil {
		return fmt.Errorf("get: %w", err)
	}
	if !bytes.Equal(got, body) {
		return errors.New("get: the probe object read back different bytes")
	}
	if _, err := c.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: bucket, Key: key}); err != nil {
		return fmt.Errorf("delete: %w", err)
	}
	return nil
}

// laneProbe is T10 in production: put-if-absent then delete-by-version of
// moves/_lane/<id> in the source's bucket. A probe object left by an earlier
// pass whose delete failed is removed by version and the check fails this
// pass; the next one starts clean.
func laneProbe(versioned func(ctx context.Context, bucket, region string) (lease.VersionedStore, error)) func(context.Context, string, string, string) error {
	return func(ctx context.Context, bucket, region, id string) error {
		vs, err := versioned(ctx, bucket, region)
		if err != nil {
			return err
		}
		key := moves.LaneKey(id)
		ver, err := vs.PutIfAbsent(ctx, bucket, key, []byte(`{"probe":"`+id+`"}`))
		if errors.Is(err, lease.ErrExists) {
			vers, lerr := vs.ListVersions(ctx, bucket, key)
			if lerr != nil {
				return fmt.Errorf("a probe object is left and cannot be listed: %w", lerr)
			}
			for _, v := range vers {
				if v.Key == key {
					if derr := vs.DeleteVersion(ctx, bucket, key, v.VersionID); derr != nil {
						return derr
					}
				}
			}
			return errors.New("a probe object was left by an earlier pass; removed")
		}
		if err != nil {
			return err
		}
		return vs.DeleteVersion(ctx, bucket, key, ver)
	}
}

// storePrefixes are the store bindings in an environment: every prefix P
// with P_BUCKET, P_ACCESS_KEY and P_SECRET_KEY set (objstore.EnvNames).
func storePrefixes(env map[string]string) []string {
	var out []string
	for name := range env {
		p, ok := strings.CutSuffix(name, "_BUCKET")
		if !ok || p == "" {
			continue
		}
		_, _, _, ak, sk := objstore.EnvNames(p)
		if env[ak] != "" && env[sk] != "" {
			out = append(out, p)
		}
	}
	sort.Strings(out)
	return out
}