package spindle

import (
	"context"
	"testing"
	"time"

	comatproto "github.com/bluesky-social/indigo/api/atproto"
	"github.com/bluesky-social/indigo/atproto/syntax"

	"tangled.org/core/jetstream"
	"tangled.org/core/spindle/secrets"
)

func takedownEvent(did syntax.DID) *jetstream.Event {
	status := "takendown"
	return &jetstream.Event{
		Did:     did.String(),
		Kind:    jetstream.EventKindAccount,
		Account: &comatproto.SyncSubscribeRepos_Account{Did: did.String(), Active: false, Status: &status},
	}
}

type hangingVault struct{ secrets.Manager }

func (hangingVault) RemoveAllSecrets(ctx context.Context, _ secrets.RepoIdentifier) error {
	<-ctx.Done()
	return ctx.Err()
}

func TestAnAccountJobStopsWaitingOnAHungWipeAndMovesTheCursorOn(t *testing.T) {
	s := newWipeTestSpindle(t)
	repoDid, owner := syntax.DID("did:plc:repo1"), syntax.DID("did:plc:owner1")
	seedWipeRepo(t, s, repoDid, owner)
	s.vault = hangingVault{s.vault}
	defer func(old time.Duration) { accountWipeBudget = old }(accountWipeBudget)
	accountWipeBudget = 100 * time.Millisecond

	var log ticketLog
	finished := goErr(func() error {
		s.ingestAccountJob(context.Background(), takedownEvent(owner), namedTicket{"account", &log})
		return nil
	})
	waitErr(t, finished, 5*time.Second, "an account job waited on a hung wipe far past its budget")

	if done, dropped := log.snapshot(); len(done) != 1 || len(dropped) != 0 {
		t.Fatalf("done = %v, dropped = %v, want the ticket finished so the cursor moves on", done, dropped)
	}
	if exists, err := s.db.WipeExists(repoDid); err != nil || !exists {
		t.Fatalf("wipe row exists %v, %v, want it left for the repo jobs to finish", exists, err)
	}
}

func TestAnAccountEventWaitsForASlotOnTheReader(t *testing.T) {
	env := newEnsureEnv(t)
	slots := make(chan struct{}, 1)
	slots <- struct{}{}
	ctx, cancel := context.WithCancel(context.Background())
	var log ticketLog

	returned := goErr(func() error {
		return env.s.startAccountJob(ctx, takedownEvent("did:plc:nobody"), namedTicket{"account", &log}, slots)
	})
	select {
	case err := <-returned:
		t.Fatalf("an account event took a slot that wasn't free, returned %v", err)
	case <-time.After(200 * time.Millisecond):
	}

	cancel()
	if err := <-returned; err == nil {
		t.Fatal("a cancelled wait for a slot reported success")
	}
	if done, dropped := log.snapshot(); len(done) != 0 || len(dropped) != 1 {
		t.Fatalf("done = %v, dropped = %v, want the ticket dropped so the cursor stays before it", done, dropped)
	}
}

func TestAnAccountEventWithAFreeSlotReturnsAtOnceAndFinishesItsTicket(t *testing.T) {
	env := newEnsureEnv(t)
	slots := make(chan struct{}, 1)
	var log ticketLog
	if err := env.s.startAccountJob(context.Background(), takedownEvent("did:plc:nobody"), namedTicket{"account", &log}, slots); err != nil {
		t.Fatal(err)
	}
	eventually(t, 5*time.Second, "the account job to finish its ticket", func() bool {
		done, _ := log.snapshot()
		return len(done) == 1
	})
	eventually(t, 5*time.Second, "the job to give its slot back", func() bool { return len(slots) == 0 })
}