package hostactors

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/carlosframework/platform/internal/release"
)

// Group membership, confirmed by the console (region HA spec 2 §5 step 5,
// H16, H28). Before it invites a box its region's group launched, and
// again before it enrolls it, the console asks the cloud, read-only and
// with its own credentials, whether the instance is a current member of
// the region's group. No box writes or contributes membership evidence:
// a confirmation exists only inside this package, built by
// confirmMembership from one MembershipReader answer, and nothing
// exported takes one in.
//
// What the cloud says is the reader's business; the AWS adapter is
// internal/hostactors/awsmembership. This file names no provider (§25).

// MembershipReader answers, read-only, from the cloud the roster's region
// entries name. It is the console's own client.
type MembershipReader interface {
	// Member describes instance in awsRegion: its group, if any, and its
	// state. An instance in no group is a Membership with no GroupName. An
	// error is a read that failed, never an answer; the Membership beside
	// it then carries only the RequestIDs of the calls made, so the log
	// keeps whatever ids the cloud gave.
	Member(ctx context.Context, awsRegion, instance string) (Membership, error)
}

// Membership is one instance's group membership, as the cloud reported it.
type Membership struct {
	GroupName              string
	AvailabilityZoneRegion string
	LifecycleState         string
	InstanceState          string
	Owner                  string
	// RequestIDs are the cloud's request ids for the reads, one per call
	// made, in order: the id the cloud gave, in its answer or in its error
	// response, or NoRequestID for a call that failed before the cloud
	// answered.
	RequestIDs []string
}

// NoRequestID stands in a Membership's RequestIDs for a call that failed
// before the cloud answered, so it gave no id. It is never an id: a read
// that carries it is not an answer.
const NoRequestID = "none"

// MembershipEvidence is what a confirmation read, for the console's log:
// the cloud's request ids, and when the read began.
type MembershipEvidence struct {
	RequestIDs []string
	ObservedAt time.Time
}

// membershipFreshness is the write deadline: a read's write is issued
// less than membershipFreshness after the read's anchor, never at or after
// it.
const membershipFreshness = time.Minute

var (
	// ErrMembershipUnknown is a membership read that failed. It says
	// nothing about the instance: the console tries again with the same
	// attempt.
	ErrMembershipUnknown = errors.New("hostactors: group membership could not be read")
	// ErrNotMember is the cloud's positive answer that the instance is not
	// a current, running member of the region's group in the deployment's
	// account and AWS region.
	ErrNotMember = errors.New("hostactors: the instance is not a current member of the region's group")
	// ErrMembershipStale is a read that was already membershipFreshness old
	// or more when the describe returned: the cloud was slow, and the
	// answer counts against the attempt.
	ErrMembershipStale = errors.New("hostactors: the membership read is more than a minute old")
	// ErrWriteDeadline is a write the deadline stopped (writeBound): no
	// store call was issued at or after it, but one already in flight was
	// cancelled, so whether it committed is unknown. The next pass reads
	// back and describes again.
	ErrWriteDeadline = errors.New("hostactors: the write would be issued more than a minute after its membership read")
)

// membershipConfirmation is one read of an attempt's instance. It is built
// only by confirmMembership, from the reader's answer, and never leaves
// this package. A read the cloud answered carries its request ids and
// time whether or not it found a member, so every read can be logged.
type membershipConfirmation struct {
	requestIDs []string
	observedAt time.Time
}

// confirmMembership asks r about instance in region's AWS region, and
// returns the confirmation only when the answer is a current member of
// region's group: Pending or InService, pending or running, owned by the
// region's account, in its AWS region, with a request id for each call.
// The read's anchor, its time, is taken before the call: never later than
// the describe's completion, so its age is never understated.
//
// The call gets at most storeCallTimeout, as each bound store call does:
// neither describe sets a deadline of its own, and a connection that
// stalls would otherwise hold the serial pass, and every box behind this
// one, for as long as it stalls. A call cut off, by the timeout or by the
// pass's own context, is unknown whatever it returned: the attempt stands
// and its wait does not grow.
func confirmMembership(ctx context.Context, r MembershipReader, region Region, instance string, now func() time.Time) (membershipConfirmation, error) {
	if r == nil || now == nil {
		return membershipConfirmation{}, fmt.Errorf("%w: no reader", ErrMembershipUnknown)
	}
	at := now().UTC()
	rctx, cancel := context.WithTimeout(ctx, storeCallTimeout)
	m, err := r.Member(rctx, region.AWSRegion, instance)
	cut := rctx.Err()
	cancel()
	// The ids the cloud gave are kept whatever came of the read, failed
	// calls included, for the console's log.
	c := membershipConfirmation{requestIDs: append([]string(nil), m.RequestIDs...), observedAt: at}
	if cut != nil {
		return c, fmt.Errorf("%w: the read did not finish in time: %v", ErrMembershipUnknown, cut)
	}
	if err != nil {
		return c, fmt.Errorf("%w: %v", ErrMembershipUnknown, err)
	}
	// A reply without an id from each call is not an answer, whatever it
	// claims, so it is never counted against the attempt.
	if len(m.RequestIDs) < 2 {
		return c, fmt.Errorf("%w: the cloud gave no request ids", ErrMembershipUnknown)
	}
	for _, id := range m.RequestIDs {
		if id == "" || id == NoRequestID {
			return c, fmt.Errorf("%w: the cloud gave a blank request id", ErrMembershipUnknown)
		}
	}
	switch {
	case m.GroupName != region.GroupName:
		return c, fmt.Errorf("%w: it is in group %q", ErrNotMember, m.GroupName)
	case m.LifecycleState != "Pending" && m.LifecycleState != "InService":
		return c, fmt.Errorf("%w: its lifecycle state is %q", ErrNotMember, m.LifecycleState)
	case m.InstanceState == "" || m.Owner == "":
		// Auto Scaling knows it but EC2 does not list it (yet, or any
		// more): no answer.
		return c, fmt.Errorf("%w: EC2 does not list it", ErrMembershipUnknown)
	case m.InstanceState != "pending" && m.InstanceState != "running":
		return c, fmt.Errorf("%w: it is %q", ErrNotMember, m.InstanceState)
	case m.Owner != region.Account || m.AvailabilityZoneRegion != region.AWSRegion:
		return c, fmt.Errorf("%w: it is in account %q, region %q", ErrNotMember, m.Owner, m.AvailabilityZoneRegion)
	}
	return c, nil
}

// fresh refuses a confirmation membershipFreshness old or more at now, or
// one from the future: the describe itself was too slow.
func (c membershipConfirmation) fresh(now time.Time) error {
	if c.observedAt.IsZero() || now.Sub(c.observedAt) >= membershipFreshness || c.observedAt.After(now.Add(5*time.Second)) {
		return fmt.Errorf("%w (read at %s, checked at %s)", ErrMembershipStale, c.observedAt.Format(time.RFC3339), now.UTC().Format(time.RFC3339))
	}
	return nil
}

// deadline is the last moment a write this read allows may be issued.
func (c membershipConfirmation) deadline() time.Time { return c.observedAt.Add(membershipFreshness) }

// atWrite is the bound store's check before each call: at or after the
// deadline it is ErrWriteDeadline, unknown, never a refusal.
func (c membershipConfirmation) atWrite(now time.Time) error {
	if c.observedAt.IsZero() || !now.Before(c.deadline()) {
		return fmt.Errorf("%w (read at %s, write at %s)", ErrWriteDeadline, c.observedAt.Format(time.RFC3339), now.UTC().Format(time.RFC3339))
	}
	return nil
}

func (c membershipConfirmation) evidence() MembershipEvidence {
	return MembershipEvidence{RequestIDs: append([]string(nil), c.requestIDs...), ObservedAt: c.observedAt}
}

// writeBound is d with every store call bound to c's deadline (C2, S1),
// the single enforcement point of the write's freshness. Each store call
// (a read, the write, the read-back after a lost write) is refused before
// it is issued once the deadline is reached, by either clock, the
// context's or now; the returned context ends at the deadline; and each
// call it does issue gets at most storeCallTimeout (U2). The sealed
// writers (Invite, Replace and the checkMembership it makes) run on it
// unchanged, so no read, signature's Put or read-back of theirs is issued
// 60 seconds or more after the membership read's anchor, and any outcome
// not confirmed in time is ErrWriteDeadline, unknown.
func writeBound(ctx context.Context, d Directory, c membershipConfirmation, now func() time.Time) (Directory, context.Context, context.CancelFunc) {
	bctx, cancel := context.WithTimeout(ctx, c.deadline().Sub(now()))
	bd := d
	bd.Store = deadlineStore{inner: d.Store, check: func(callCtx context.Context) error {
		if bctx.Err() != nil || callCtx.Err() != nil {
			return fmt.Errorf("%w: the deadline passed", ErrWriteDeadline)
		}
		return c.atWrite(now())
	}}
	return bd, bctx, cancel
}

// storeCallTimeout is the hard limit on one store call of a bound write
// (U2): a call the store has not answered within it is unknown, and its
// context is cancelled, so an S3 request is abandoned at the HTTP layer.
// It bounds how long the console waits, not when a write commits: a
// call cut off may still commit later, and is settled from the roster on
// a later pass. confirmMembership gives each membership read the same
// limit. A var so tests can shorten it.
var storeCallTimeout = 15 * time.Second

// deadlineStore refuses every call once check says the deadline has been
// reached, before the call reaches the store underneath, and gives each
// call it issues at most storeCallTimeout. The bound cannot rest on the
// store alone: release.DirStore ignores its context, so the wrapper
// returns on its own when the call outlives the timeout or the deadline,
// whatever the store does. Every outcome it cannot confirm in time is
// ErrWriteDeadline, unknown: a call cut off in flight, one that never
// answered, and one that answered, even successfully, only after the
// deadline.
//
// The store underneath is a named field, not embedded, and every method of
// release.Store is written out here: a method added to the interface later
// fails to compile instead of reaching the store past the bound.
type deadlineStore struct {
	inner release.Store
	check func(context.Context) error
}

var _ release.Store = deadlineStore{}

// call runs one store call under the bound.
func (s deadlineStore) call(ctx context.Context, f func(context.Context) error) error {
	if err := s.check(ctx); err != nil {
		return err
	}
	cctx, cancel := context.WithTimeout(ctx, storeCallTimeout)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- f(cctx) }()
	var err error
	select {
	case err = <-done:
	case <-cctx.Done():
	}
	// Cut off, whether the store gave up on the context or not: by the
	// deadline, or by the hard timeout. Whatever it answered, if anything,
	// is not known to be the outcome.
	if cctx.Err() != nil {
		if ctx.Err() != nil {
			return fmt.Errorf("%w: in flight at the deadline", ErrWriteDeadline)
		}
		return fmt.Errorf("%w: the store did not answer in time", ErrWriteDeadline)
	}
	if err == nil {
		if late := s.check(ctx); late != nil {
			return fmt.Errorf("%w: it answered only after the deadline", ErrWriteDeadline)
		}
	}
	return err
}

func (s deadlineStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	var body []byte
	var etag string
	err := s.call(ctx, func(cctx context.Context) error {
		var err error
		body, etag, err = s.inner.Get(cctx, key)
		return err
	})
	if err != nil {
		return nil, "", err
	}
	return body, etag, nil
}

func (s deadlineStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	var etag string
	err := s.call(ctx, func(cctx context.Context) error {
		var err error
		etag, err = s.inner.Put(cctx, key, body, ifMatch)
		return err
	})
	if err != nil {
		return "", err
	}
	return etag, nil
}

func (s deadlineStore) List(ctx context.Context, prefix string) ([]string, error) {
	var keys []string
	err := s.call(ctx, func(cctx context.Context) error {
		var err error
		keys, err = s.inner.List(cctx, prefix)
		return err
	})
	if err != nil {
		return nil, err
	}
	return keys, nil
}

func (s deadlineStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	var dirs []string
	err := s.call(ctx, func(cctx context.Context) error {
		var err error
		dirs, err = s.inner.ListDirs(cctx, prefix)
		return err
	})
	if err != nil {
		return nil, err
	}
	return dirs, nil
}

func (s deadlineStore) Delete(ctx context.Context, key string) error {
	return s.call(ctx, func(cctx context.Context) error { return s.inner.Delete(cctx, key) })
}