package sitefeed

import (
	"bytes"
	"context"
	"errors"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"tangled.org/core/appview/cloudflare"
	"tangled.org/core/appview/config"
	"tangled.org/core/appview/db"
	"tangled.org/core/appview/models"
	"tangled.org/core/knotfeed"
	"tangled.org/core/log"
	"tangled.org/core/tapc"

	cbg "github.com/whyrusleeping/cbor-gen"
)

const (
	feedTestHost      = "knot.invalid"
	feedTestOtherHost = "other.knot.invalid"
	feedTestRepoDid   = "did:plc:limpet"
	feedTestOwner     = "did:plc:akshay"
	feedTestSha       = "bcfa6359b037bc411e9626260cc3dd9a28408d2c"
	feedTestOtherSha  = "9eab6c4f29bc9e609ec52888cd6a4ef716b567c9"
)

var errBoom = errors.New("boom")

func feedTestDB(t *testing.T) *db.DB {
	t.Helper()
	d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "appview.db"))
	if err != nil {
		t.Fatalf("db.Make: %v", err)
	}
	t.Cleanup(func() { d.Close() })
	return d
}

func seedFeedRepo(t *testing.T, d *db.DB) {
	t.Helper()
	tx, err := d.Begin()
	if err != nil {
		t.Fatalf("begin: %v", err)
	}
	if err := db.AddRepo(tx, &models.Repo{
		Did:     feedTestOwner,
		Knot:    feedTestHost,
		RepoDid: feedTestRepoDid,
		Name:    "anemone",
	}); err != nil {
		t.Fatalf("AddRepo: %v", err)
	}
	if err := tx.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}
}

func setSiteConfig(t *testing.T, d *db.DB) {
	t.Helper()
	if err := db.SetRepoSiteConfig(d, feedTestRepoDid, "main", "/", false); err != nil {
		t.Fatalf("SetRepoSiteConfig: %v", err)
	}
}

func enabledCf(t *testing.T) *cloudflare.Client {
	t.Helper()
	cf, err := cloudflare.New(&config.Config{Cloudflare: config.Cloudflare{
		AccountId: "acct",
		KV:        config.KVConfig{NamespaceId: "ns"},
		R2:        config.R2Config{Bucket: "bucket"},
	}})
	if err != nil {
		t.Fatalf("cloudflare.New: %v", err)
	}
	return cf
}

type deployCall struct {
	repoDid string
	branch  string
	dir     string
}

type deployRecorder struct {
	mu    sync.Mutex
	calls []deployCall
}

func (r *deployRecorder) hook(err error) func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
	return func(_ context.Context, _ *cloudflare.Client, _ *config.Config, repo *models.Repo, branch, dir string) error {
		r.mu.Lock()
		r.calls = append(r.calls, deployCall{repoDid: repo.RepoDid, branch: branch, dir: dir})
		r.mu.Unlock()
		return err
	}
}

func (r *deployRecorder) got() []deployCall {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]deployCall(nil), r.calls...)
}

func writeText(w *bytes.Buffer, s string) {
	if err := cbg.CborWriteHeader(w, cbg.MajTextString, uint64(len(s))); err != nil {
		w.Reset()
	}
	w.WriteString(s)
}

func encodeRefRecord(t *testing.T, sha string) []byte {
	t.Helper()
	var out bytes.Buffer
	if err := cbg.CborWriteHeader(&out, cbg.MajMap, 1); err != nil {
		t.Fatalf("map header: %v", err)
	}
	writeText(&out, "sha")
	writeText(&out, sha)
	return out.Bytes()
}

func refOpFor(t *testing.T, refname string, action tapc.RecordAction, sha string) knotfeed.RecordOp {
	t.Helper()
	rkey, _ := knotfeed.EscapeRefname(refname)
	op := knotfeed.RecordOp{Action: action, Collection: knotfeed.GitRefCollection, Rkey: rkey}
	if sha != "" {
		op.Bytes = encodeRefRecord(t, sha)
	}
	return op
}

func object(t *testing.T, hex string) knotfeed.ObjectID {
	t.Helper()
	sha, ok := knotfeed.ParseObjectID(hex)
	if !ok {
		t.Fatalf("ParseObjectID(%q)", hex)
	}
	return sha
}

func pollFor(t *testing.T, what string, ready func() bool) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if ready() {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("%s never appeared", what)
}

func feedFor(t *testing.T, d *db.DB, hook func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error) *Feed {
	t.Helper()
	f := New(d, nil, &config.Config{}, enabledCf(t), log.New("test"))
	f.deploy = hook
	t.Cleanup(func() {
		f.Unsubscribe(feedTestHost)
		f.Unsubscribe(feedTestOtherHost)
	})
	return f
}

func TestFeedRefOp_BranchMatchTriggersDeploy(t *testing.T) {
	d := feedTestDB(t)
	seedFeedRepo(t, d)
	setSiteConfig(t, d)
	r := &deployRecorder{}
	f := feedFor(t, d, r.hook(nil))

	f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha))
	pollFor(t, "the deploy call", func() bool { return len(r.got()) == 1 })
	calls := r.got()
	if calls[0] != (deployCall{repoDid: feedTestRepoDid, branch: "main", dir: "/"}) {
		t.Fatalf("deploys = %+v, want one push on main", calls)
	}

	var deploy *models.SiteDeploy
	pollFor(t, "the deploy row", func() bool {
		rows, err := db.GetSiteDeploys(d, feedTestRepoDid, 10)
		if err == nil && len(rows) > 0 {
			deploy = &rows[0]
			return true
		}
		return false
	})
	if deploy.Status != models.SiteDeployStatusSuccess {
		t.Errorf("status = %q, want success", deploy.Status)
	}
	if deploy.Trigger != models.SiteDeployTriggerPush {
		t.Errorf("trigger = %q, want push", deploy.Trigger)
	}
	if deploy.CommitSHA != feedTestSha || deploy.Branch != "main" || deploy.Dir != "/" {
		t.Errorf("deploy row = %+v, want the pushed sha, branch and dir", deploy)
	}
}

func TestFeedSkipsNonDeployingOps(t *testing.T) {
	seedSiteRepo := func(t *testing.T, d *db.DB) {
		seedFeedRepo(t, d)
		setSiteConfig(t, d)
	}
	for _, tc := range []struct {
		name string
		seed func(*testing.T, *db.DB)
		op   func(*testing.T, *Feed)
	}{
		{
			name: "push to a branch other than the site branch",
			seed: seedSiteRepo,
			op: func(_ *testing.T, f *Feed) {
				f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid}, "other", object(t, feedTestSha))
			},
		},
		{
			name: "push without a site config",
			seed: seedFeedRepo,
			op: func(_ *testing.T, f *Feed) {
				f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid}, "main", object(t, feedTestSha))
			},
		},
		{
			name: "tag push",
			seed: seedSiteRepo,
			op: func(t *testing.T, f *Feed) {
				f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/tags/v1", tapc.RecordCreateAction, feedTestSha))
			},
		},
		{
			name: "deleted ref",
			seed: seedSiteRepo,
			op: func(t *testing.T, f *Feed) {
				f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordDeleteAction, ""))
			},
		},
		{
			name: "record from a knot that doesn't host the repo",
			seed: seedSiteRepo,
			op: func(t *testing.T, f *Feed) {
				f.refOp(context.Background(), feedTestOtherHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha))
			},
		},
		{
			name: "repo missing from the index",
			op: func(t *testing.T, f *Feed) {
				f.refOp(context.Background(), feedTestHost, "did:plc:unknown", refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha))
			},
		},
		{
			name: "garbage rkey",
			seed: seedSiteRepo,
			op: func(_ *testing.T, f *Feed) {
				f.refOp(context.Background(), feedTestHost, feedTestRepoDid, knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: "~zz", Bytes: []byte{0xff}})
			},
		},
		{
			name: "garbage record bytes",
			seed: seedSiteRepo,
			op: func(_ *testing.T, f *Feed) {
				f.refOp(context.Background(), feedTestHost, feedTestRepoDid, knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: "refs~2fheads~2fmain", Bytes: []byte{0xff}})
			},
		},
		{
			name: "commit with only foreign collections",
			seed: seedFeedRepo,
			op: func(t *testing.T, f *Feed) {
				msg := knotfeed.Message{Type: knotfeed.TypeCommit, Commit: &knotfeed.Commit{
					Repo: feedTestRepoDid,
					Seq:  7,
					Records: []knotfeed.RecordOp{
						{Action: tapc.RecordCreateAction, Collection: "sh.tangled.repo", Rkey: "abc", Bytes: []byte("x")},
					},
				}}
				if err := f.handle(feedTestHost)(context.Background(), msg); err != nil {
					t.Errorf("handle: %v", err)
				}
			},
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			d := feedTestDB(t)
			if tc.seed != nil {
				tc.seed(t, d)
			}
			r := &deployRecorder{}
			f := feedFor(t, d, r.hook(nil))
			tc.op(t, f)
			if calls := r.got(); len(calls) != 0 {
				t.Errorf("deploys = %v, want none", calls)
			}
		})
	}
}

func TestFeedTrigger_RecordsDeployFailure(t *testing.T) {
	d := feedTestDB(t)
	seedFeedRepo(t, d)
	setSiteConfig(t, d)
	r := &deployRecorder{}
	f := feedFor(t, d, r.hook(errBoom))

	f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid, Knot: feedTestHost}, "main", object(t, feedTestSha))

	pollFor(t, "the deploy call", func() bool { return len(r.got()) == 1 })
	var deploy *models.SiteDeploy
	pollFor(t, "the deploy row", func() bool {
		rows, err := db.GetSiteDeploys(d, feedTestRepoDid, 10)
		if err == nil && len(rows) > 0 {
			deploy = &rows[0]
			return true
		}
		return false
	})
	if deploy.Status != models.SiteDeployStatusFailure {
		t.Errorf("status = %q, want failure", deploy.Status)
	}
	if deploy.Error != "boom" {
		t.Errorf("error = %q, want the sync failure text", deploy.Error)
	}
}

func TestFeedHandleFailsTheCommitWhenTheDatabaseIsDown(t *testing.T) {
	d := feedTestDB(t)
	seedFeedRepo(t, d)
	setSiteConfig(t, d)
	r := &deployRecorder{}
	f := feedFor(t, d, r.hook(nil))

	if err := d.Close(); err != nil {
		t.Fatalf("Close: %v", err)
	}

	msg := knotfeed.Message{Type: knotfeed.TypeCommit, Commit: &knotfeed.Commit{
		Repo: feedTestRepoDid,
		Seq:  9,
		Records: []knotfeed.RecordOp{
			refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha),
		},
	}}
	if err := f.handle(feedTestHost)(context.Background(), msg); err == nil {
		t.Fatal("handle accepted a commit while the database was down; the cursor would advance past a dropped deploy")
	}
	if calls := r.got(); len(calls) != 0 {
		t.Errorf("deploys = %v, want none while the database is down", calls)
	}
}

func TestFeedSubscribeDeduplicatesAndUnsubscribes(t *testing.T) {
	d := feedTestDB(t)
	f := feedFor(t, d, nil)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	f.Subscribe(ctx, feedTestHost)
	f.Subscribe(ctx, feedTestHost)

	f.mu.Lock()
	count := len(f.sources)
	f.mu.Unlock()
	if count != 1 {
		t.Fatalf("sources = %d, want 1 after a duplicate subscribe", count)
	}

	f.Unsubscribe(feedTestHost)
	f.mu.Lock()
	count = len(f.sources)
	f.mu.Unlock()
	if count != 0 {
		t.Fatalf("sources = %d, want 0 after unsubscribe", count)
	}

	f.Unsubscribe(feedTestOtherHost)
	f.mu.Lock()
	count = len(f.sources)
	f.mu.Unlock()
	if count != 0 {
		t.Fatalf("sources = %d, want 0 after unsubscribing a host with no source", count)
	}
}

func TestFeedTrigger_LastPushWinsOverARacingDeploy(t *testing.T) {
	d := feedTestDB(t)
	seedFeedRepo(t, d)
	setSiteConfig(t, d)
	r := &deployRecorder{}
	f := feedFor(t, d, r.hook(nil))
	repo := &models.Repo{RepoDid: feedTestRepoDid, Knot: feedTestHost}

	f.maybeDeploy(context.Background(), repo, "main", object(t, feedTestSha))
	f.maybeDeploy(context.Background(), repo, "main", object(t, feedTestOtherSha))

	var winner models.SiteDeploy
	pollFor(t, "a successful deploy row for the later push", func() bool {
		rows, err := db.GetSiteDeploys(d, feedTestRepoDid, 10)
		if err != nil {
			t.Fatalf("GetSiteDeploys: %v", err)
		}
		for _, row := range rows {
			if row.Id > winner.Id {
				winner = row
			}
		}
		return winner.CommitSHA == feedTestOtherSha && winner.Status == models.SiteDeployStatusSuccess
	})
}