package console

import (
	"context"
	"errors"
	"net/http"
	"os"
	"sync/atomic"
	"testing"
	"time"

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

// afterRecordRead runs then after each of the first n reads of key: what
// another request does between the delete door's import check and its
// delete.
type afterRecordRead struct {
	release.Store
	key   string
	n     int32
	reads atomic.Int32
	then  func()
}

func (s *afterRecordRead) Get(ctx context.Context, key string) ([]byte, string, error) {
	raw, tag, err := s.Store.Get(ctx, key)
	if key == s.key && s.reads.Add(1) <= s.n {
		s.then()
	}
	return raw, tag, err
}

func (s *afterRecordRead) DeleteIfMatch(ctx context.Context, key, etag string) error {
	return release.DeleteIfMatch(ctx, s.Store, key, etag)
}

// plainRecord writes a plain record of shop on host.
func (f fromFixture) plainRecord(t *testing.T, host string) {
	t.Helper()
	if err := provision.CreateRecord(context.Background(), f.fx.rel, provision.Record{Schema: 1, Host: host, Slug: "old",
		App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}); err != nil {
		t.Fatal(err)
	}
}

// Final review F7: the delete door's check finds no record, and an import
// is begun before the delete would have run. Absence is the delete's own
// idempotent answer: the ordinary delete is not run, and the import's
// record stands.
func TestImportDeleteOfAnAbsentHostLeavesAnImportBegunSince(t *testing.T) {
	f := newImportFixture(t)
	id := newSeedID(t)
	key := provision.RecordKey("bab", fromApp, importHost)
	f.fx.srv.cfg.Releases = &afterRecordRead{Store: f.fx.rel, key: key, n: 1, then: func() {
		f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StateAwaitingUpload)
	}}
	if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
		t.Fatalf("delete of an absent host = %d %v, want the idempotent 200", code, out)
	}
	rec, err := f.record(t, importHost)
	if err != nil {
		t.Fatalf("the import begun since was deleted: %v", err)
	}
	requireImportState(t, rec, id, seeds.StateAwaitingUpload, "")
}

// The check reads a plain record; before the delete, that instance is
// deleted and an import of the host begins and swaps. The ordinary delete
// is bound to the plain record's etag, so it does not land: the door reads
// the host again and answers as it answers for the import, SD-N57, with
// the import's record standing.
func TestImportDeleteReclassifiesARecordThatBecameAnImport(t *testing.T) {
	f := newImportFixture(t)
	f.plainRecord(t, importHost)
	id := newSeedID(t)
	key := provision.RecordKey("bab", fromApp, importHost)
	f.fx.srv.cfg.Releases = &afterRecordRead{Store: f.fx.rel, key: key, n: 1, then: func() {
		if err := f.fx.rel.Delete(context.Background(), key); err != nil {
			t.Error(err)
		}
		f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StateSwapped)
	}}
	if code, out := deleteInstance(t, f, importHost); code != http.StatusConflict || str(t, out, "error") != importAbortFirst(importHost) {
		t.Fatalf("delete = %d %v, want 409 SD-N57", code, out)
	}
	rec, err := f.record(t, importHost)
	if err != nil {
		t.Fatalf("the import was deleted the ordinary way: %v", err)
	}
	requireImportState(t, rec, id, seeds.StateSwapped, "")
	if rows := importAudit(t, f, "instance-delete"); len(rows) != 0 {
		t.Fatalf("a refused delete audited: %q", rows)
	}
}

// A record that changes under the delete twice is not chased: the second
// conflict answers main's retry sentence, and the record stands.
func TestImportDeleteGivesUpAfterASecondChange(t *testing.T) {
	f := newImportFixture(t)
	f.plainRecord(t, importHost)
	key := provision.RecordKey("bab", fromApp, importHost)
	f.fx.srv.cfg.Releases = &afterRecordRead{Store: f.fx.rel, key: key, n: 2, then: func() {
		rec, etag, err := provision.GetRecordETag(context.Background(), f.fx.rel, "bab", fromApp, importHost)
		if err != nil {
			t.Error(err)
			return
		}
		rec.Steering = map[string]string{"": "latency", "latency": ""}[rec.Steering]
		if err := provision.UpdateRecord(context.Background(), f.fx.rel, *rec, etag); err != nil {
			t.Error(err)
		}
	}}
	if code, out := deleteInstance(t, f, importHost); code != http.StatusBadGateway || str(t, out, "error") != importDeleteRetry {
		t.Fatalf("delete = %d %v, want 502 %q", code, out, importDeleteRetry)
	}
	if _, err := f.record(t, importHost); errors.Is(err, os.ErrNotExist) {
		t.Fatal("a record that kept changing was deleted")
	}
}

// The ordinary case is unchanged: a plain record the check read is
// deleted, and a second delete of the gone host is the idempotent 200.
func TestImportDeleteOfAPlainRecord(t *testing.T) {
	f := newImportFixture(t)
	f.plainRecord(t, importHost)
	for range 2 {
		if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
			t.Fatalf("delete = %d %v", code, out)
		}
		f.requireRecordGone(t, importHost)
	}
	if rows := importAudit(t, f, "instance-delete"); len(rows) != 2 {
		t.Fatalf("instance-delete rows = %q, want one per delete", rows)
	}
}