package hostactors

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/boxid"
	"github.com/carlosframework/platform/internal/boxid/boxidtest"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// fakeMembership is the cloud, as the console's reader sees it: each
// instance's membership. err fails every call, with failIDs the request
// ids of the calls made; during runs inside each call (a slow describe,
// say).
type fakeMembership struct {
	members map[string]Membership
	err     error
	failIDs []string
	during  func()
	calls   []string
}

func (f *fakeMembership) Member(_ context.Context, awsRegion, instance string) (Membership, error) {
	f.calls = append(f.calls, awsRegion+" "+instance)
	if f.during != nil {
		f.during()
	}
	if f.err != nil {
		return Membership{RequestIDs: f.failIDs}, f.err
	}
	return f.members[instance], nil
}

func member(group string) Membership {
	return Membership{GroupName: group, AvailabilityZoneRegion: "eu-west-1", LifecycleState: "InService", InstanceState: "running", Owner: "123", RequestIDs: []string{"asg-req", "ec2-req"}}
}

// memberWorld is fixture()'s roster with ie's group "routers-ie", and a
// box (instance i-new, reference newRef) the group launched, invited, with an open
// attempt.
type memberWorld struct {
	d     Directory
	sg    release.Signer
	o     Observed
	v     boxid.Verifier
	raw   []byte
	in    EnrollmentRequest
	cloud *fakeMembership
	clock time.Time
}

func newMemberWorld(t *testing.T, m Membership) *memberWorld {
	t.Helper()
	ctx := context.Background()
	d, sg, o := fixture(t)
	aws := boxidtest.NewSigner(t)
	_, raw := aws.Record(t, newRef, "123", "eu-west-1", "i-new")
	next := Clone(o.Roster)
	next.Regions = map[string]Region{"ie": {Account: "123", AWSRegion: "eu-west-1", GroupName: "routers-ie", GroupARN: "arn:aws:autoscaling:eu-west-1:123:autoScalingGroup:x:autoScalingGroupName/routers-ie", Namespaces: []string{"platform"}, Pools: []string{"default"}}}
	o, err := d.Replace(ctx, sg, o, next)
	if err != nil {
		t.Fatal(err)
	}
	w := &memberWorld{d: d, sg: sg, v: boxid.Verifier{Account: "123", Certs: aws.Certs()}, raw: raw, clock: time.Now().UTC(),
		cloud: &fakeMembership{members: map[string]Membership{"i-new": m}},
		in:    EnrollmentRequest{Reference: newRef, Namespace: "platform", Label: newRef, Region: "ie", Pool: "default", AWSRegion: "eu-west-1"}}
	if _, err := d.Invite(ctx, sg, newRef, raw); err != nil {
		t.Fatal(err)
	}
	o, a, err := d.BeginEnrollmentAttempt(ctx, sg, o, w.in, raw, "i-new")
	if err != nil {
		t.Fatal(err)
	}
	w.o, w.in.AttemptID, w.in.ConfirmationID = o, a.AttemptID, a.ConfirmationID
	return w
}

func (w *memberWorld) now() time.Time { return w.clock }

func (w *memberWorld) enroll() (Observed, MembershipEvidence, error) {
	return w.d.EnrollAutomatic(context.Background(), w.sg, w.o, w.v, w.in, w.raw, "i-new", w.cloud, w.now)
}

// The console reads membership itself, once, in the attempt's AWS region,
// and enrolls on it; the read's request ids come back for its log.
func TestEnrollAutomaticReadsMembershipItself(t *testing.T) {
	w := newMemberWorld(t, member("routers-ie"))
	next, ev, err := w.enroll()
	if err != nil {
		t.Fatal(err)
	}
	if b := next.Roster.Boxes[newRef]; b.State != "active" || b.Group != "routers-ie" || b.Cloud.InstanceID != "i-new" {
		t.Fatalf("roster entry %+v", b)
	}
	if !reflect.DeepEqual(w.cloud.calls, []string{"eu-west-1 i-new"}) || !reflect.DeepEqual(ev.RequestIDs, []string{"asg-req", "ec2-req"}) || !ev.ObservedAt.Equal(w.clock) {
		t.Fatalf("calls %v, evidence %+v", w.cloud.calls, ev)
	}
	// A replay of the consumed attempt is refused before the cloud is
	// asked again.
	w.o = next
	if _, _, err := w.enroll(); !errors.Is(err, ErrRoster) || len(w.cloud.calls) != 1 {
		t.Fatalf("replay: %v, calls %v", err, w.cloud.calls)
	}
	if _, _, err := w.d.checkConfirmed(context.Background(), next, w.v, w.in, w.raw, "i-new", w.cloud, w.now); !errors.Is(err, ErrRoster) {
		t.Fatalf("replayed check: %v", err)
	}
}

// Every positive answer that the instance is not a current member of the
// region's group refuses, and writes nothing.
func TestEnrollAutomaticRefusesANonMember(t *testing.T) {
	for name, m := range map[string]Membership{
		"another group":     member("routers-uk"),
		"no group":          {Owner: "123", InstanceState: "running", AvailabilityZoneRegion: "eu-west-1", RequestIDs: []string{"a", "b"}},
		"leaving the group": func() Membership { m := member("routers-ie"); m.LifecycleState = "Terminating"; return m }(),
		"stopped":           func() Membership { m := member("routers-ie"); m.InstanceState = "stopped"; return m }(),
		"terminated":        func() Membership { m := member("routers-ie"); m.InstanceState = "terminated"; return m }(),
		"another account":   func() Membership { m := member("routers-ie"); m.Owner = "999"; return m }(),
		"another zone":      func() Membership { m := member("routers-ie"); m.AvailabilityZoneRegion = "eu-west-2"; return m }(),
	} {
		t.Run(name, func(t *testing.T) {
			w := newMemberWorld(t, m)
			if _, _, err := w.d.checkConfirmed(context.Background(), w.o, w.v, w.in, w.raw, "i-new", w.cloud, w.now); !errors.Is(err, ErrNotMember) {
				t.Fatalf("check: %v", err)
			}
			if _, _, err := w.enroll(); !errors.Is(err, ErrNotMember) {
				t.Fatalf("enroll: %v", err)
			}
			o, err := w.d.Read(context.Background())
			if err != nil || o.Roster.Revision != w.o.Roster.Revision {
				t.Fatalf("a refused enrollment wrote the roster: %v", err)
			}
		})
	}
}

// A read that fails, or is not an answer, is unknown: never a member and
// never a non-member.
func TestAMembershipReadThatIsNotAnAnswerIsUnknown(t *testing.T) {
	for name, edit := range map[string]func(*fakeMembership){
		"the call fails": func(f *fakeMembership) { f.err = errors.New("Throttling") },
		"EC2 does not list it": func(f *fakeMembership) {
			m := f.members["i-new"]
			m.InstanceState, m.Owner = "", ""
			f.members["i-new"] = m
		},
		"one request id": func(f *fakeMembership) { m := f.members["i-new"]; m.RequestIDs = []string{"a"}; f.members["i-new"] = m },
		"a blank request id": func(f *fakeMembership) {
			m := f.members["i-new"]
			m.RequestIDs = []string{"a", ""}
			f.members["i-new"] = m
		},
		// A call that failed before the cloud answered has no id, so the
		// reply is not an answer however it got this far.
		"no id from a call": func(f *fakeMembership) {
			m := f.members["i-new"]
			m.RequestIDs = []string{"a", NoRequestID}
			f.members["i-new"] = m
		},
		// The ids are checked before anything the reply claims: another
		// group with one id is still no answer, not a refusal.
		"another group, one request id": func(f *fakeMembership) {
			m := f.members["i-new"]
			m.GroupName, m.RequestIDs = "routers-uk", []string{"a"}
			f.members["i-new"] = m
		},
		"no group and no ids": func(f *fakeMembership) {
			f.members["i-new"] = Membership{}
		},
	} {
		t.Run(name, func(t *testing.T) {
			w := newMemberWorld(t, member("routers-ie"))
			edit(w.cloud)
			if _, _, err := w.d.checkConfirmed(context.Background(), w.o, w.v, w.in, w.raw, "i-new", w.cloud, w.now); !errors.Is(err, ErrMembershipUnknown) || errors.Is(err, ErrNotMember) {
				t.Fatalf("check: %v", err)
			}
			_, _, err := w.enroll()
			if !errors.Is(err, ErrMembershipUnknown) || errors.Is(err, ErrNotMember) {
				t.Fatalf("%v", err)
			}
			// The attempt is kept as it was, and nothing is written.
			o, err := w.d.Read(context.Background())
			a := o.Roster.EnrollmentAttempts[newRef]
			if err != nil || o.Roster.Revision != w.o.Roster.Revision || a.Consumed || a.AttemptID != w.in.AttemptID || a.ConfirmationID != w.in.ConfirmationID {
				t.Fatalf("an unknown read moved the attempt: %v, %+v", err, a)
			}
		})
	}
}

// A read that failed still logs the ids the cloud gave: the call that
// answered, and the failed call's own id from its error response.
func TestAFailedReadKeepsTheIdsTheCloudGave(t *testing.T) {
	for name, ids := range map[string][]string{
		"the second call fails":                 {"asg-req", "ec2-err"},
		"the first call fails":                  {"asg-err"},
		"a call fails before the cloud answers": {"asg-req", NoRequestID},
	} {
		t.Run(name, func(t *testing.T) {
			w := newMemberWorld(t, member("routers-ie"))
			w.cloud.err, w.cloud.failIDs = errors.New("UnauthorizedOperation"), ids
			_, c, err := w.d.checkConfirmed(context.Background(), w.o, w.v, w.in, w.raw, "i-new", w.cloud, w.now)
			ev := c.evidence()
			if !errors.Is(err, ErrMembershipUnknown) || !reflect.DeepEqual(ev.RequestIDs, ids) || !ev.ObservedAt.Equal(w.clock) {
				t.Fatalf("check: %v, evidence %+v", err, ev)
			}
			_, ev, err = w.enroll()
			if !errors.Is(err, ErrMembershipUnknown) || !reflect.DeepEqual(ev.RequestIDs, ids) {
				t.Fatalf("enroll: %v, evidence %+v", err, ev)
			}
			c, err = confirmMembership(context.Background(), w.cloud, w.o.Roster.Regions["ie"], "i-new", w.now)
			if !errors.Is(err, ErrMembershipUnknown) || !reflect.DeepEqual(c.requestIDs, ids) {
				t.Fatalf("confirmMembership: %v, %v", err, c.requestIDs)
			}
		})
	}
}

// C2: the write deadline is 60 seconds from the read's anchor, taken
// before the describe and so never later than its completion, and a read
// sixty seconds old or more is refused. A describe that took exactly a
// minute leaves the roster as it was.
func TestEnrollAutomaticRefusesAReadOlderThanAMinuteAtTheWrite(t *testing.T) {
	w := newMemberWorld(t, member("routers-ie"))
	w.cloud.during = func() { w.clock = w.clock.Add(60 * time.Second) }
	if _, _, err := w.enroll(); !errors.Is(err, ErrMembershipStale) {
		t.Fatalf("slow read: %v", err)
	}
	o, err := w.d.Read(context.Background())
	if err != nil || o.Roster.Revision != w.o.Roster.Revision || o.Roster.EnrollmentAttempts[newRef].Consumed {
		t.Fatalf("a stale read wrote the roster: %v", err)
	}
	// A read that was fresh when it was checked but is over a minute old
	// by the write: here every look at the clock takes 31 seconds, so the
	// read is 31 seconds old when checked and 62 when the roster write is
	// about to be made. The write bound refuses it there, as unknown.
	u := newMemberWorld(t, member("routers-ie"))
	ticking := func() time.Time { u.clock = u.clock.Add(31 * time.Second); return u.clock }
	if _, _, err := u.d.EnrollAutomatic(context.Background(), u.sg, u.o, u.v, u.in, u.raw, "i-new", u.cloud, ticking); !errors.Is(err, ErrWriteDeadline) || errors.Is(err, ErrMembershipStale) {
		t.Fatalf("a read gone stale before the write: %v", err)
	}
	if o, err := u.d.Read(context.Background()); err != nil || o.Roster.Revision != u.o.Roster.Revision {
		t.Fatalf("a read gone stale before the write wrote the roster: %v", err)
	}
	// Fifty-nine seconds is inside the minute.
	v := newMemberWorld(t, member("routers-ie"))
	v.cloud.during = func() { v.clock = v.clock.Add(59 * time.Second) }
	if _, _, err := v.enroll(); err != nil {
		t.Fatalf("a read inside the minute: %v", err)
	}
}

// No exported identifier of this package carries a membership
// confirmation: the only one is built inside it, from a reader's answer.
func TestNoExportedPathTakesAConfirmation(t *testing.T) {
	// Every exported method of Directory, EnrollAutomatic among them: the
	// admission check itself (checkConfirmed) is not exported.
	if _, ok := reflect.TypeOf(Directory{}).MethodByName("EnrollAutomatic"); !ok {
		t.Fatal("no EnrollAutomatic")
	}
	if _, ok := reflect.TypeOf(Directory{}).MethodByName("CheckConfirmed"); ok {
		t.Fatal("the admission check is exported again")
	}
	for j := 0; j < reflect.TypeOf(Directory{}).NumMethod(); j++ {
		m := reflect.TypeOf(Directory{}).Method(j)
		name := m.Name
		for i := 0; i < m.Type.NumIn(); i++ {
			in := m.Type.In(i)
			if strings.Contains(strings.ToLower(in.Name()), "confirmation") || in == reflect.TypeOf(MembershipEvidence{}) || in == reflect.TypeOf(Membership{}) {
				t.Fatalf("%s takes a %s", name, in)
			}
		}
	}
}

// newRef is the joining box's reference: the group reference of
// instance i-new in account 123.
var newRef = GroupBoxReference("123", "i-new")

// The ruling's first test: two group boxes never share a reference. The
// reference is a pseudonym of account and instance, stable, and outside
// the label space.
func TestGroupBoxReferencesDoNotCollide(t *testing.T) {
	if GroupBoxReference("123", "i-0a61f580df1acfa65") != GroupBoxReference("123", "i-0a61f580df1acfa65") {
		t.Fatal("not stable")
	}
	if GroupBoxReference("123", "i-a") == GroupBoxReference("124", "i-a") {
		t.Fatal("the account is not in the reference")
	}
	seen := map[string]string{}
	for i := 0; i < 200000; i++ {
		inst := fmt.Sprintf("i-%017x", i)
		ref := GroupBoxReference("123456789012", inst)
		if len(ref) != 22 || !strings.HasPrefix(ref, "g-") || !segment(ref) {
			t.Fatalf("%s: %q", inst, ref)
		}
		if other, ok := seen[ref]; ok {
			t.Fatalf("%s and %s share %s", inst, other, ref)
		}
		seen[ref] = inst
	}
	if strings.Contains(GroupBoxReference("123", "i-0a61f580df1acfa65"), "0a61f580") {
		t.Fatal("the reference shows the instance id")
	}
}

// The console reads a group box's report under the reference it derives
// from the roster's pinned instance, never the label in the entry; every
// other box under its reference, exactly as before.
func TestPrivateBoxReportDerivesAGroupBoxsKey(t *testing.T) {
	d, _, _ := fixture(t)
	ctx := context.Background()
	put := func(key, instance string) {
		raw, _ := json.Marshal(status.Box{GeneratedAt: time.Now(), Metrics: &status.BoxMetrics{InstanceID: instance}})
		if _, err := d.Store.Put(ctx, key, raw, ""); err != nil {
			t.Fatal(err)
		}
	}
	derived := GroupBoxReference("123", "i-g")
	put("status/boxes/"+derived+".json", "i-g")
	put("status/boxes/chosen.json", "i-chosen")
	put("status/boxes/annie-welding.json", "i-old")
	r, err := d.PrivateBoxReport(ctx, "chosen", Box{Label: "chosen", Group: "routers-ie", Cloud: Cloud{Account: "123", InstanceID: "i-g"}})
	if err != nil || r.InstanceID != "i-g" {
		t.Fatalf("group box: %+v %v", r, err)
	}
	r, err = d.PrivateBoxReport(ctx, "annie-welding", Box{Label: "annie-welding", Cloud: Cloud{Account: "123", InstanceID: "i-old"}})
	if err != nil || r.InstanceID != "i-old" {
		t.Fatalf("a box an operator enrolled: %+v %v", r, err)
	}
	if _, err := d.PrivateBoxReport(ctx, derived, Box{Group: "routers-ie"}); !errors.Is(err, ErrRoster) {
		t.Fatalf("a group box with no pinned instance: %v", err)
	}
}

// A group box is admitted only under the reference its instance derives:
// an attempt under a reference the box chose is refused before the cloud
// is asked, even for a member.
func TestEnrollAutomaticRefusesAReferenceTheBoxChose(t *testing.T) {
	w := newMemberWorld(t, member("routers-ie"))
	in := w.in
	in.Reference, in.Label = "chosen", "chosen"
	o, a, err := w.d.BeginEnrollmentAttempt(context.Background(), w.sg, w.o, in, w.raw, "i-new")
	if err != nil {
		t.Fatal(err)
	}
	in.AttemptID, in.ConfirmationID = a.AttemptID, a.ConfirmationID
	if _, _, err := w.d.EnrollAutomatic(context.Background(), w.sg, o, w.v, in, w.raw, "i-new", w.cloud, w.now); !errors.Is(err, ErrRoster) || len(w.cloud.calls) != 0 {
		t.Fatalf("a chosen reference: %v, calls %v", err, w.cloud.calls)
	}
}