package lease

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"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/benbjohnson/litestream"
	lss3 "github.com/benbjohnson/litestream/s3"
)

// errLockPresent is createLease's answer when a lock object exists, live or
// expired: litestream's AcquireLease decides it, as it always has.
var errLockPresent = errors.New("lease: a lock is present")

// createLease is litestream's AcquireLease for the common case, the lock
// absent (a release deletes it), in one round trip instead of two. It
// writes what AcquireLease writes for an absent lock: generation 1, the
// leaser's owner, now plus its TTL, the same JSON and content type, and the
// same create-only condition (If-None-Match: *). So the lock it leaves, and
// what a reader of it decides, are AcquireLease's. A 412 means a lock is
// there, and errLockPresent hands the decision back to AcquireLease (the
// takeover of an expired lock, the refusal of a live one). Any other error
// is the acquire's, as AcquireLease's own write failing would be.
func createLease(ctx context.Context, client *awss3.Client, l *lss3.Leaser) (*litestream.Lease, error) {
	lease := &litestream.Lease{Generation: 1, ExpiresAt: time.Now().Add(l.TTL), Owner: l.Owner}
	data, err := json.Marshal(lease)
	if err != nil {
		return nil, fmt.Errorf("marshal lock file: %w", err)
	}
	key := lss3.DefaultLeasePath
	if l.Path != "" {
		key = l.Path + "/" + lss3.DefaultLeasePath
	}
	out, err := client.PutObject(ctx, &awss3.PutObjectInput{
		Bucket:      aws.String(l.Bucket),
		Key:         aws.String(key),
		Body:        bytes.NewReader(data),
		ContentType: aws.String("application/json"),
		IfNoneMatch: aws.String("*"),
	})
	if err != nil {
		if preconditionFailed(err) {
			return nil, errLockPresent
		}
		return nil, fmt.Errorf("put lock file: %w", err)
	}
	if out.ETag != nil {
		lease.ETag = *out.ETag
	}
	return lease, nil
}

// preconditionFailed is litestream's isPreconditionFailed (unexported
// there), plus the 409 S3 answers when two conditional writes of one key
// race: either way a lock is, or is about to be, present.
func preconditionFailed(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		switch apiErr.ErrorCode() {
		case "PreconditionFailed", "412", "ConditionalRequestConflict":
			return true
		}
	}
	var opErr *smithy.OperationError
	if errors.As(err, &opErr) {
		if h, ok := opErr.Err.(interface{ HTTPStatusCode() int }); ok {
			return h.HTTPStatusCode() == 412 || h.HTTPStatusCode() == 409
		}
	}
	return false
}