package spindle

import (
	"context"
	"encoding/json"
	"testing"

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

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/jetstream"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/tapc"
)

func recordEvent(did syntax.DID, collection, rkey string, action tapc.RecordAction, record any) tapc.Event {
	raw, _ := json.Marshal(record)
	return tapc.Event{
		Type: tapc.EvtRecord,
		Record: &tapc.RecordEventData{
			Live:       true,
			Did:        did,
			Collection: syntax.NSID(collection),
			Rkey:       syntax.RecordKey(rkey),
			Action:     action,
			Record:     raw,
		},
	}
}

func declarationEvent(owner, repo syntax.DID, action tapc.RecordAction) tapc.Event {
	return recordEvent(owner, org_tangled.RepoDeclarationNSID, "repo", action, org_tangled.RepoDeclaration{Repo: repo.String()})
}

func processOK(t *testing.T, env *ensureEnv, evt tapc.Event) {
	t.Helper()
	if err := env.s.processEvent(context.Background(), evt); err != nil {
		t.Fatal(err)
	}
}

func TestProcessEventIgnoresOtherCollections(t *testing.T) {
	env := newEnsureEnv(t)
	for _, collection := range []string{org_tangled.FeedStarNSID, org_tangled.FeedCommentNSID, org_tangled.FeedReactionNSID, org_tangled.FeedSubscriptionNSID} {
		evt := recordEvent("did:plc:someone", collection, "3mrhpypucbsg4", tapc.RecordCreateAction, map[string]string{"title": "x"})
		if err := env.s.processEvent(context.Background(), evt); err != nil {
			t.Fatalf("processEvent(%s): %v", collection, err)
		}
	}
	if env.knot.count() != 0 {
		t.Fatalf("knot was asked %d times for records spindle no longer reads", env.knot.count())
	}
}

func TestManifestEventForAnUnknownRepoNeedsToNameThisSpindle(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")

	other := org_tangled.RepoManifest{Spindle: ptr("did:web:other.example.com")}
	evt := recordEvent(testRepoDid, org_tangled.RepoManifestNSID, "self", tapc.RecordUpdateAction, other)
	processOK(t, env, evt)
	if env.knot.count() != 0 || len(env.rows(t)) != 0 {
		t.Fatalf("a manifest for another spindle cost %d knot requests, rows %+v", env.knot.count(), env.rows(t))
	}

	processOK(t, env, markEvent(testRepoDid, env.spindleDid(), testOwner))
	if rows := env.rows(t); len(rows) != 1 {
		t.Fatalf("rows = %+v", rows)
	}
}

func TestManifestEventForAKnownRepoReadsTheKnot(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)

	// the event body is empty on purpose, the knot's manifest is what counts
	env.manifest("repo", "did:web:other.example.com")
	evt := recordEvent(testRepoDid, org_tangled.RepoManifestNSID, "self", tapc.RecordUpdateAction, org_tangled.RepoManifest{})
	processOK(t, env, evt)
	if rows := env.rows(t); len(rows) != 0 {
		t.Fatalf("moved repo kept, rows = %+v", rows)
	}
}

func TestManifestDeleteForAKnownRepoForgetsIt(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)

	env.knot.erase(testRepoDid, org_tangled.RepoManifestNSID, "self")
	evt := recordEvent(testRepoDid, org_tangled.RepoManifestNSID, "self", tapc.RecordDeleteAction, nil)
	processOK(t, env, evt)
	if rows := env.rows(t); len(rows) != 0 {
		t.Fatalf("rows = %+v", rows)
	}
}

func TestCollaboratorEventsOnlyMatterForKnownRepos(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo", testBob)
	evt := recordEvent(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String(), tapc.RecordCreateAction, map[string]string{})

	processOK(t, env, evt)
	if env.knot.count() != 0 {
		t.Fatalf("collaborator of an unknown repo cost %d knot requests", env.knot.count())
	}

	env.ensure(t)
	env.knot.erase(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String())
	evt = recordEvent(testRepoDid, org_tangled.GraphCollaboratorNSID, testBob.String(), tapc.RecordDeleteAction, nil)
	processOK(t, env, evt)
	if env.collaborator(t, testBob) {
		t.Error("bob kept access after his collaborator record went away")
	}
}

func TestDeclarationEventsLeadToTheRepo(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")

	evt := declarationEvent(testOwner, testRepoDid, tapc.RecordCreateAction)
	processOK(t, env, evt)
	if rows := env.rows(t); len(rows) != 1 {
		t.Fatalf("a declaration backfill didn't register the repo, rows = %+v", rows)
	}

	env.knot.undeclare(testOwner, "repo")
	evt = recordEvent(testOwner, org_tangled.RepoDeclarationNSID, "repo", tapc.RecordDeleteAction, nil)
	processOK(t, env, evt)
	if rows := env.rows(t); len(rows) != 1 {
		t.Fatalf("deleting the declaration forgot the repo, rows = %+v", rows)
	}
}

func TestDeclarationDeleteForARowWithoutARepoDidIsSkipped(t *testing.T) {
	env := newEnsureEnv(t)
	if err := env.s.db.AddRepo(db.Repo{Knot: env.knot.host(), Owner: testOwner, Rkey: "repo"}); err != nil {
		t.Fatal(err)
	}

	evt := recordEvent(testOwner, org_tangled.RepoDeclarationNSID, "repo", tapc.RecordDeleteAction, nil)
	processOK(t, env, evt)
	if env.knot.count() != 0 {
		t.Fatalf("a legacy row cost %d knot requests", env.knot.count())
	}
}

func TestDeclarationFromAnOutsiderCostsNothing(t *testing.T) {
	env := newEnsureEnv(t)
	env.s.cfg.Server.Admission = config.AdmissionClosed
	env.serve("repo")
	if err := db.AddToAllowlist(env.s.db, "did:plc:someoneelse"); err != nil {
		t.Fatal(err)
	}

	evt := declarationEvent(testOwner, testRepoDid, tapc.RecordCreateAction)
	processOK(t, env, evt)
	if env.knot.count() != 0 {
		t.Fatalf("an owner outside the allow-list cost %d knot requests", env.knot.count())
	}
}

func TestDeclarationFromAnAdminIsServedWhileClosed(t *testing.T) {
	env := newEnsureEnv(t)
	env.s.cfg.Server.Admission = config.AdmissionClosed
	env.s.cfg.Server.Admins = []string{testOwner.String()}
	env.serve("repo")

	evt := declarationEvent(testOwner, testRepoDid, tapc.RecordCreateAction)
	processOK(t, env, evt)
	if rows := env.rows(t); len(rows) != 1 {
		t.Fatalf("an admin's declaration wasn't served, rows = %+v", rows)
	}
}

func TestJetstreamToTapEventMarksRecordsLive(t *testing.T) {
	tests := []struct {
		name      string
		operation string
		action    tapc.RecordAction
	}{
		{name: "create", operation: jetstream.CommitOperationCreate, action: tapc.RecordCreateAction},
		{name: "update", operation: jetstream.CommitOperationUpdate, action: tapc.RecordUpdateAction},
		{name: "delete", operation: jetstream.CommitOperationDelete, action: tapc.RecordDeleteAction},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			event, ok := jetstreamToTapEvent(&jetstream.Event{
				Did:  "did:plc:jge3zxi7lgrfnvhzcgrimeo7",
				Kind: jetstream.EventKindCommit,
				Commit: &jetstream.Commit{
					Operation:  tt.operation,
					Collection: org_tangled.RepoManifestNSID,
					RKey:       "self",
					Record:     json.RawMessage(`{"declaration":{}}`),
				},
			})
			if !ok {
				t.Fatal("jetstreamToTapEvent() rejected a valid event")
			}
			if event.Record == nil || !event.Record.Live {
				t.Fatalf("converted event = %+v, want a live record", event.Record)
			}
			if event.Record.Collection.String() != org_tangled.RepoManifestNSID {
				t.Errorf("collection = %q", event.Record.Collection)
			}
			if event.Record.Action != tt.action {
				t.Errorf("action = %q, want %q", event.Record.Action, tt.action)
			}
		})
	}
}

func manifestCommit(env *ensureEnv) *jetstream.Event {
	raw, _ := json.Marshal(org_tangled.RepoManifest{Spindle: ptr(env.spindleDid()), Declaration: &org_tangled.RepoManifest_Declaration{Owner: testOwner.String(), Name: "repo"}})
	return &jetstream.Event{
		Did:  testRepoDid.String(),
		Kind: jetstream.EventKindCommit,
		Commit: &jetstream.Commit{
			Operation:  jetstream.CommitOperationCreate,
			Collection: org_tangled.RepoManifestNSID,
			RKey:       "self",
			Record:     raw,
		},
	}
}

func TestIngestRoutesJetstreamCommitsIntoEnsureRepo(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo", testBob)

	if err := env.s.ingest()(context.Background(), manifestCommit(env)); err != nil {
		t.Fatal(err)
	}
	if rows := env.rows(t); len(rows) != 1 || !env.collaborator(t, testBob) {
		t.Fatalf("jetstream commit didn't register the repo, rows = %+v", rows)
	}
}

func TestDeclarationArrivingAfterTheManifestEnrollsTheRepo(t *testing.T) {
	env := newEnsureEnv(t)
	env.manifest("repo", env.spindleDid())

	processOK(t, env, markEvent(testRepoDid, env.spindleDid(), testOwner))
	if rows := env.rows(t); len(rows) != 0 {
		t.Fatalf("repo served before its owner declared it, rows = %+v", rows)
	}

	env.knot.declare(testOwner, "repo", testRepoDid)
	processOK(t, env, declarationEvent(testOwner, testRepoDid, tapc.RecordCreateAction))
	if rows := env.rows(t); len(rows) != 1 {
		t.Fatalf("the declaration didn't enrol the repo, rows = %+v", rows)
	}
}

func TestIngestReturnsAnEventCutByShutdownSoTheCursorStaysBehindIt(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo", testBob)
	event := manifestCommit(env)

	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if err := env.s.ingest()(ctx, event); err == nil {
		t.Fatal("an event cut by shutdown was reported handled")
	}
	if rows := env.rows(t); len(rows) != 0 {
		t.Fatalf("a cancelled event still registered the repo: %+v", rows)
	}
	if err := env.s.ingest()(context.Background(), event); err != nil {
		t.Fatal(err)
	}
}

func TestDeclarationFromANonOwnerDoesNotRefreshAServedRepo(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)
	before := env.knot.count()

	evt := recordEvent("did:plc:newowner", org_tangled.RepoDeclarationNSID, "whatever", tapc.RecordCreateAction, org_tangled.RepoDeclaration{Repo: testRepoDid.String()})
	processOK(t, env, evt)
	if got := env.knot.count(); got != before {
		t.Fatalf("a stranger's declaration cost %d knot requests", got-before)
	}

	evt = declarationEvent(testOwner, testRepoDid, tapc.RecordUpdateAction)
	processOK(t, env, evt)
	if env.knot.count() == before {
		t.Error("the owner's own declaration no longer refreshes the repo")
	}
}

func TestAMarkThatFailsOnTheDbIsReportedInsteadOfSwallowed(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)
	if _, err := env.s.db.Exec(`drop table record_retries`); err != nil {
		t.Fatal(err)
	}
	event := &jetstream.Event{Did: testRepoDid.String(), Kind: jetstream.EventKindCommit, Commit: &jetstream.Commit{
		Operation: jetstream.CommitOperationCreate, Collection: org_tangled.TrackTicketNSID, RKey: "3kticket00001", Record: []byte("{}"),
	}}

	if err := env.s.ingestEvent(withMarkOnly(context.Background()), event); err == nil {
		t.Fatal("an event whose mark couldn't be written was reported as handled")
	}
}