package lease

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"sort"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/smithy-go"
	"github.com/aws/smithy-go/encoding/httpbinding"
	smithyhttp "github.com/aws/smithy-go/transport/http"
)

// ObjectVersion is one entry of a versioned listing: an object version, or
// a delete marker (DeleteMarker true, no ETag or Size).
type ObjectVersion struct {
	Key          string
	VersionID    string
	ETag         string // unquoted
	Size         int64
	IsLatest     bool
	DeleteMarker bool
	LastModified time.Time
}

var (
	ErrExists            = errors.New("lease: object already exists")
	ErrDestinationExists = errors.New("lease: copy destination already exists")
	ErrIsDeleteMarker    = errors.New("lease: version is a delete marker")
	ErrNoVersionID       = errors.New("lease: a version id is required")
)

// VersionedStore is the data bucket addressed by exact object version (host
// lifecycle claim spec C3, §6.2, §6.5). No method deletes or overwrites by
// bare key.
type VersionedStore interface {
	// ListVersions returns every version and delete marker under prefix:
	// key order; within a key the IsLatest entry first, the rest newest
	// first with ties unordered. Any page error fails the whole call.
	ListVersions(ctx context.Context, bucket, prefix string) ([]ObjectVersion, error)
	// OpenVersion streams one exact version. Missing: ErrMirrorNotFound.
	// A delete marker: ErrIsDeleteMarker.
	OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error)
	// GetVersion reads one exact version whole (OpenVersion's errors).
	GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error)
	// CopyVersion copies one exact source version to dstKey. createOnly
	// adds If-None-Match: *; an existing current destination object gives
	// ErrDestinationExists. It returns the destination's new version id
	// and ETag.
	CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (dstVersionID, dstETag string, err error)
	// DeleteVersion removes one exact version; gone already is success.
	// An empty versionID is ErrNoVersionID, without any request.
	DeleteVersion(ctx context.Context, bucket, key, versionID string) error
	// PutIfAbsent writes body only if key holds no current object version.
	PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (versionID string, err error)
}

// VersionRangeOpener optionally streams a range of one exact object version.
// A negative length reads to the end; a positive length bounds the read.
// Zero length and negative offsets are invalid. Errors match OpenVersion.
type VersionRangeOpener interface {
	OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error)
}

// versionedS3 binds every bucket to its own regional client explicitly.
// An unbound bucket is an error, never a guess.
type versionedS3 struct {
	clients map[string]*awss3.Client // bucket -> client for its region
}

// NewVersionedS3 builds one client per region from the default credential
// chain (with s3compat options, as newLeaserClient does) and binds each
// bucket to its region's client.
func NewVersionedS3(ctx context.Context, regionByBucket map[string]string) (VersionedStore, error) {
	byRegion := map[string]*awss3.Client{}
	out := &versionedS3{clients: map[string]*awss3.Client{}}
	for bucket, region := range regionByBucket {
		c, ok := byRegion[region]
		if !ok {
			var err error
			if c, _, err = newLeaserClient(ctx, region); err != nil {
				return nil, fmt.Errorf("lease: versioned client for %s: %w", region, err)
			}
			byRegion[region] = c
		}
		out.clients[bucket] = c
	}
	return out, nil
}

func (s *versionedS3) client(bucket string) (*awss3.Client, error) {
	c, ok := s.clients[bucket]
	if !ok {
		return nil, fmt.Errorf("lease: bucket %q is not bound to a region", bucket)
	}
	return c, nil
}

func (s *versionedS3) ListVersions(ctx context.Context, bucket, prefix string) ([]ObjectVersion, error) {
	c, err := s.client(bucket)
	if err != nil {
		return nil, err
	}
	p := awss3.NewListObjectVersionsPaginator(c, &awss3.ListObjectVersionsInput{
		Bucket: aws.String(bucket), Prefix: aws.String(prefix),
	}) // StopOnDuplicateToken stays off: a key legitimately repeats across version pages
	var out []ObjectVersion
	for p.HasMorePages() {
		page, err := p.NextPage(ctx)
		if err != nil {
			return nil, fmt.Errorf("lease: list versions %s/%s: %w", bucket, prefix, err)
		}
		for _, v := range page.Versions {
			out = append(out, ObjectVersion{Key: aws.ToString(v.Key), VersionID: aws.ToString(v.VersionId),
				ETag: strings.Trim(aws.ToString(v.ETag), `"`), Size: aws.ToInt64(v.Size),
				IsLatest: aws.ToBool(v.IsLatest), LastModified: aws.ToTime(v.LastModified)})
		}
		for _, d := range page.DeleteMarkers {
			out = append(out, ObjectVersion{Key: aws.ToString(d.Key), VersionID: aws.ToString(d.VersionId),
				IsLatest: aws.ToBool(d.IsLatest), DeleteMarker: true, LastModified: aws.ToTime(d.LastModified)})
		}
	}
	sort.SliceStable(out, func(i, j int) bool {
		if out[i].Key != out[j].Key {
			return out[i].Key < out[j].Key
		}
		if out[i].IsLatest != out[j].IsLatest {
			return out[i].IsLatest
		}
		return out[i].LastModified.After(out[j].LastModified) // ties: unspecified (Global Constraints)
	})
	return out, nil
}

// apiCode returns the smithy error code and HTTP status of err.
func apiCode(err error) (string, int) {
	var ae smithy.APIError
	code := ""
	if errors.As(err, &ae) {
		code = ae.ErrorCode()
	}
	var re *smithyhttp.ResponseError
	status := 0
	if errors.As(err, &re) {
		status = re.HTTPStatusCode()
	}
	return code, status
}

// mapReadErr normalizes only object or version absence. A missing bucket
// (NoSuchBucket, also a 404) stays an error: it is never "already gone".
func mapReadErr(err error) error {
	code, status := apiCode(err)
	switch {
	case code == "NoSuchBucket":
		return err
	case code == "NoSuchKey" || code == "NoSuchVersion" || code == "NotFound" || status == http.StatusNotFound:
		return ErrMirrorNotFound
	case status == http.StatusMethodNotAllowed:
		return ErrIsDeleteMarker // S3 answers 405 for a delete marker's version id
	}
	return err
}

func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
	return s.openVersion(ctx, bucket, key, versionID, nil)
}

func (s *versionedS3) OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) {
	if versionID == "" {
		return nil, ErrNoVersionID
	}
	if offset < 0 || length == 0 || (length > 0 && offset > (1<<63-1)-(length-1)) {
		return nil, errors.New("lease: invalid version range")
	}
	rangeHeader := fmt.Sprintf("bytes=%d-", offset)
	if length > 0 {
		rangeHeader += fmt.Sprint(offset + length - 1)
	}
	return s.openVersion(ctx, bucket, key, versionID, &rangeHeader)
}

func (s *versionedS3) openVersion(ctx context.Context, bucket, key, versionID string, byteRange *string) (io.ReadCloser, error) {
	if versionID == "" {
		return nil, ErrNoVersionID
	}
	c, err := s.client(bucket)
	if err != nil {
		return nil, err
	}
	out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID), Range: byteRange})
	if err != nil {
		if m := mapReadErr(err); m != err {
			return nil, m
		}
		return nil, fmt.Errorf("lease: get %s/%s@%s: %w", bucket, key, versionID, err)
	}
	return out.Body, nil
}

func (s *versionedS3) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
	rc, err := s.OpenVersion(ctx, bucket, key, versionID)
	if err != nil {
		return nil, err
	}
	defer rc.Close()
	return io.ReadAll(rc)
}

func (s *versionedS3) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
	if srcVersionID == "" {
		return "", "", ErrNoVersionID
	}
	if _, err := s.client(srcBucket); err != nil {
		return "", "", err
	}
	c, err := s.client(dstBucket) // a copy executes in the destination's region
	if err != nil {
		return "", "", err
	}
	in := &awss3.CopyObjectInput{
		Bucket:     aws.String(dstBucket),
		Key:        aws.String(dstKey),
		CopySource: aws.String(httpbinding.EscapePath(srcBucket+"/"+srcKey, false) + "?versionId=" + url.QueryEscape(srcVersionID)),
	}
	if createOnly {
		in.IfNoneMatch = aws.String("*")
	}
	out, err := c.CopyObject(ctx, in)
	if err != nil {
		code, status := apiCode(err)
		var ae smithy.APIError
		switch {
		case createOnly && (code == "PreconditionFailed" || status == http.StatusPreconditionFailed):
			return "", "", ErrDestinationExists
		case code == "InvalidRequest" && errors.As(err, &ae) && strings.Contains(strings.ToLower(ae.ErrorMessage()), "delete marker"):
			return "", "", ErrIsDeleteMarker
		default:
			if m := mapReadErr(err); m != err {
				return "", "", m
			}
		}
		return "", "", fmt.Errorf("lease: copy %s/%s@%s -> %s/%s: %w", srcBucket, srcKey, srcVersionID, dstBucket, dstKey, err)
	}
	etag := ""
	if out.CopyObjectResult != nil {
		etag = strings.Trim(aws.ToString(out.CopyObjectResult.ETag), `"`)
	}
	return aws.ToString(out.VersionId), etag, nil
}

func (s *versionedS3) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
	if versionID == "" {
		return ErrNoVersionID // never a bare-key delete (spec C3), and no request is sent
	}
	c, err := s.client(bucket)
	if err != nil {
		return err
	}
	if _, err := c.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID)}); err != nil {
		if mapReadErr(err) == ErrMirrorNotFound {
			return nil // already gone
		}
		return fmt.Errorf("lease: delete %s/%s@%s: %w", bucket, key, versionID, err)
	}
	return nil
}

func (s *versionedS3) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
	c, err := s.client(bucket)
	if err != nil {
		return "", err
	}
	out, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key),
		Body: bytes.NewReader(body), IfNoneMatch: aws.String("*")})
	if err != nil {
		if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed {
			return "", ErrExists
		}
		return "", fmt.Errorf("lease: put-if-absent %s/%s: %w", bucket, key, err)
	}
	return aws.ToString(out.VersionId), nil
}

// FilePutIfAbsent is PutIfAbsent streaming a file from disk, for objects
// too large to hold in memory: a move's sealed archive, written create-only
// into the source's own move-snapshots/ (moves spec §8.3). The S3 store
// implements it; an existing current object is ErrExists.
type FilePutIfAbsent interface {
	PutFileIfAbsent(ctx context.Context, bucket, key, path string) (versionID string, err error)
}

func (s *versionedS3) PutFileIfAbsent(ctx context.Context, bucket, key, path string) (string, error) {
	c, err := s.client(bucket)
	if err != nil {
		return "", err
	}
	f, err := os.Open(path)
	if err != nil {
		return "", err
	}
	defer f.Close()
	st, err := f.Stat()
	if err != nil {
		return "", err
	}
	// An *os.File is seekable, so the SDK hashes the payload by reading it
	// and seeking back, never by buffering it.
	out, err := c.PutObject(ctx, &awss3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key),
		Body: f, ContentLength: aws.Int64(st.Size()), IfNoneMatch: aws.String("*")})
	if err != nil {
		if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed {
			return "", ErrExists
		}
		return "", fmt.Errorf("lease: put-if-absent %s/%s: %w", bucket, key, err)
	}
	return aws.ToString(out.VersionId), nil
}

var _ VersionedStore = (*versionedS3)(nil)

var _ FilePutIfAbsent = (*versionedS3)(nil)

var _ VersionRangeOpener = (*versionedS3)(nil)