package spindle

import (
	"context"
	"encoding/json"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

	"tangled.org/core/api/tangled"
	"tangled.org/core/idresolver"
	"tangled.org/core/jetstream"
	"tangled.org/core/notifier"
	"tangled.org/core/rbac"
	"tangled.org/core/spindle/artifactstore"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/secrets"
	"tangled.org/core/spindle/webhook"
	"tangled.org/core/tapc"
	"tangled.org/core/workflow"
)

func TestDeclareTapInterestBootstrapsConfiguredOwner(t *testing.T) {
	d, _ := newTestSpindleDB(t)
	owner := syntax.DID("did:plc:spindleowner")
	requests := make(chan []syntax.DID, 1)
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost || r.URL.Path != "/repos/add" {
			http.NotFound(w, r)
			return
		}
		var body struct {
			DIDs []syntax.DID `json:"dids"`
		}
		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		requests <- body.DIDs
		w.WriteHeader(http.StatusOK)
	}))
	defer server.Close()

	cfg := &config.Config{}
	cfg.Server.Owner = owner.String()
	s := &Spindle{
		db:  d,
		l:   slog.Default(),
		cfg: cfg,
		wh:  webhook.New(d, false),
	}
	tapClient := tapc.NewClient(server.URL, "")
	s.tap = &Tap{spindle: s, tap: tapClient}

	s.declareTapInterest(context.Background())

	select {
	case dids := <-requests:
		if len(dids) != 1 || dids[0] != owner {
			t.Fatalf("tap interest = %v, want [%s]", dids, owner)
		}
	default:
		t.Fatal("configured owner was not declared to tap")
	}
}

func TestProcessRepo_MembershipCheck(t *testing.T) {
	d, e := newTestSpindleDB(t)

	cfg := &config.Config{}
	cfg.Server.Hostname = "spindle.test"

	jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false)
	if jcerr != nil {
		t.Fatalf("NewJetstreamClient: %v", jcerr)
	}
	tapServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
	}))
	t.Cleanup(tapServer.Close)

	s := &Spindle{
		db:      d,
		e:       e,
		vault:   newTestVault(t),
		l:       slog.Default(),
		cfg:     cfg,
		jc:      jc,
		rootCtx: context.Background(),
		wh:      webhook.New(d, false),
	}

	tap := &Tap{
		spindle: s,
		logger:  slog.Default(),
		tap:     tapc.NewClient(tapServer.URL, "admin"),
	}

	ownerDid := syntax.DID("did:plc:memberowner")
	nonMemberDid := syntax.DID("did:plc:nonmemberowner")
	repoDid := "did:plc:testrepo123"

	err := e.AddSpindle(rbac.ThisServer)
	if err != nil {
		t.Fatalf("AddSpindle: %v", err)
	}
	err = e.AddSpindleMember(rbac.ThisServer, ownerDid.String())
	if err != nil {
		t.Fatalf("AddSpindleMember: %v", err)
	}

	recNonMember := tangled.Repo{
		Knot:      "knot.test",
		RepoDid:   &repoDid,
		Spindle:   &cfg.Server.Hostname,
		CreatedAt: time.Now().Format(time.RFC3339),
	}
	recNonMemberJson, _ := json.Marshal(recNonMember)

	err = tap.processRepo(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        nonMemberDid,
		Rkey:       "test-repo-rkey",
		Collection: syntax.NSID(tangled.RepoNSID),
		Action:     tapc.RecordCreateAction,
		Record:     recNonMemberJson,
	})
	if err != nil {
		t.Fatalf("processRepo returned error for non-member: %v", err)
	}

	_, err = d.GetRepoByOwnerRkey(nonMemberDid, "test-repo-rkey")
	if err == nil {
		t.Fatal("repo for non-member was registered in DB, expected rejection")
	}

	recMember := tangled.Repo{
		Knot:      "knot.test",
		RepoDid:   &repoDid,
		Spindle:   &cfg.Server.Hostname,
		CreatedAt: time.Now().Format(time.RFC3339),
	}
	recMemberJson, _ := json.Marshal(recMember)

	err = tap.processRepo(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        ownerDid,
		Rkey:       "test-repo-rkey",
		Collection: syntax.NSID(tangled.RepoNSID),
		Action:     tapc.RecordCreateAction,
		Record:     recMemberJson,
	})
	if err != nil {
		t.Fatalf("processRepo returned unexpected error for member: %v", err)
	}

	_, err = d.GetRepoByOwnerRkey(ownerDid, "test-repo-rkey")
	if err != nil {
		t.Fatalf("repo for member was not registered in DB: %v", err)
	}
}

func TestProcessPull_PushAllowedCheck(t *testing.T) {
	d, e := newTestSpindleDB(t)

	cfg := &config.Config{}
	cfg.Server.Hostname = "spindle.test"

	jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false)
	if jcerr != nil {
		t.Fatalf("NewJetstreamClient: %v", jcerr)
	}
	s := &Spindle{
		db:      d,
		e:       e,
		l:       slog.Default(),
		cfg:     cfg,
		res:     idresolver.DefaultResolver("https://plc.test"),
		jc:      jc,
		rootCtx: context.Background(),
		wh:      webhook.New(d, false),
	}

	repoOwnerDid := syntax.DID("did:plc:repoowner")
	nonPusherDid := syntax.DID("did:plc:nonpusher")
	pusherDid := syntax.DID("did:plc:pusher")
	repoDid := syntax.DID("did:plc:testrepo123")

	err := d.AddRepo(db.Repo{
		Knot:      "knot.test",
		Owner:     repoOwnerDid,
		Rkey:      "test-repo-rkey",
		RepoDid:   repoDid,
		CreatedAt: time.Now().Format(time.RFC3339),
	})
	if err != nil {
		t.Fatalf("AddRepo: %v", err)
	}

	err = e.AddRepo(repoOwnerDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil {
		t.Fatalf("AddRepo permissions: %v", err)
	}
	err = e.AddCollaborator(pusherDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil {
		t.Fatalf("AddCollaborator: %v", err)
	}

	pullRecord := tangled.RepoPull{
		Target: &tangled.RepoPull_Target{
			Branch: "main",
			Repo:   repoDid.String(),
		},
		Source: &tangled.RepoPull_Source{
			Branch: "feature",
			Repo:   nil, // branch-based PR (source repo is nil)
		},
	}
	pullRecordJson, _ := json.Marshal(pullRecord)

	err = s.processPull(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        nonPusherDid,
		Rkey:       "pull-rkey-1",
		Collection: syntax.NSID(tangled.RepoPullNSID),
		Action:     tapc.RecordCreateAction,
		Record:     pullRecordJson,
	})
	if err != nil {
		t.Fatalf("processPull returned error for non-pusher: %v", err)
	}

	err = s.processPull(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        pusherDid,
		Rkey:       "pull-rkey-2",
		Collection: syntax.NSID(tangled.RepoPullNSID),
		Action:     tapc.RecordCreateAction,
		Record:     pullRecordJson,
	})
	if err != nil {
		t.Fatalf("processPull returned error for pusher: %v", err)
	}

	// legacy round-based records fall back to fetching the patch blob; the
	// fetch fails here (plc/pds are not real) which proves we did not skip
	legacyRecord := pullRecord
	legacyRecord.Rounds = []*tangled.RepoPull_Round{{CreatedAt: time.Now().Format(time.RFC3339)}}
	legacyRecordJson, _ := json.Marshal(legacyRecord)

	err = s.processPull(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        pusherDid,
		Rkey:       "pull-rkey-3",
		Collection: syntax.NSID(tangled.RepoPullNSID),
		Action:     tapc.RecordCreateAction,
		Record:     legacyRecordJson,
	})
	if err == nil || !strings.Contains(err.Error(), "resolve PR owner") {
		t.Fatalf("expected legacy round-based PR to fetch the patch blob, got: %v", err)
	}

	// versions take precedence over rounds
	bothRecord := legacyRecord
	bothRecord.Versions = []*tangled.RepoPull_Version{{Head: "deadbeef", Base: "cafe", CreatedAt: time.Now().Format(time.RFC3339)}}
	bothRecordJson, _ := json.Marshal(bothRecord)

	err = s.processPull(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        pusherDid,
		Rkey:       "pull-rkey-4",
		Collection: syntax.NSID(tangled.RepoPullNSID),
		Action:     tapc.RecordCreateAction,
		Record:     bothRecordJson,
	})
	if err != nil {
		t.Fatalf("processPull returned error for versioned PR: %v", err)
	}

	// the pipeline itself swallows downstream failures, so assert the push
	// access check directly
	allowed, err := s.isPullTriggerAuthorized(nonPusherDid.String(), nonPusherDid.String(), repoDid.String())
	if err != nil {
		t.Fatalf("isPullTriggerAuthorized(non-pusher): %v", err)
	}
	if allowed {
		t.Fatal("non-pusher was authorized to trigger a pull pipeline")
	}

	allowed, err = s.isPullTriggerAuthorized(pusherDid.String(), pusherDid.String(), repoDid.String())
	if err != nil {
		t.Fatalf("isPullTriggerAuthorized(pusher): %v", err)
	}
	if !allowed {
		t.Fatal("pusher was not authorized to trigger a pull pipeline")
	}
}

func TestProcessRepo_HijackRepoDidCheck(t *testing.T) {
	d, e := newTestSpindleDB(t)

	cfg := &config.Config{}
	cfg.Server.Hostname = "spindle.test"

	jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false)
	if jcerr != nil {
		t.Fatalf("NewJetstreamClient: %v", jcerr)
	}
	s := &Spindle{
		db:      d,
		e:       e,
		l:       slog.Default(),
		cfg:     cfg,
		jc:      jc,
		rootCtx: context.Background(),
		wh:      webhook.New(d, false),
	}

	tap := &Tap{
		spindle: s,
		logger:  slog.Default(),
	}

	aliceDid := syntax.DID("did:plc:alice")
	bobDid := syntax.DID("did:plc:bob")
	repoDid := "did:plc:sharedrepo"

	err := e.AddSpindle(rbac.ThisServer)
	if err != nil {
		t.Fatalf("AddSpindle: %v", err)
	}
	err = e.AddSpindleMember(rbac.ThisServer, aliceDid.String())
	if err != nil {
		t.Fatalf("AddSpindleMember alice: %v", err)
	}
	err = e.AddSpindleMember(rbac.ThisServer, bobDid.String())
	if err != nil {
		t.Fatalf("AddSpindleMember bob: %v", err)
	}

	err = d.AddRepo(db.Repo{
		Knot:      "knot.test",
		Owner:     aliceDid,
		Rkey:      "alice-repo",
		RepoDid:   syntax.DID(repoDid),
		CreatedAt: time.Now().Format(time.RFC3339),
	})
	if err != nil {
		t.Fatalf("d.AddRepo: %v", err)
	}

	// bob tries to register alice's repo did, must reject the hijack
	recBob := tangled.Repo{
		Knot:      "knot.test",
		RepoDid:   &repoDid,
		Spindle:   &cfg.Server.Hostname,
		CreatedAt: time.Now().Format(time.RFC3339),
	}
	recBobJson, _ := json.Marshal(recBob)

	err = tap.processRepo(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        bobDid,
		Rkey:       "bob-repo",
		Collection: syntax.NSID(tangled.RepoNSID),
		Action:     tapc.RecordCreateAction,
		Record:     recBobJson,
	})
	if err != nil {
		t.Fatalf("processRepo returned error on duplicate repoDid hijack attempt: %v", err)
	}

	_, err = d.GetRepoByOwnerRkey(bobDid, "bob-repo")
	if err == nil {
		t.Fatal("bob successfully hijacked alice's repoDid in DB, expected rejection")
	}
}

func TestProcessCollaborator_RBAC(t *testing.T) {
	d, e := newTestSpindleDB(t)

	cfg := &config.Config{}
	cfg.Server.Hostname = "spindle.test"

	ownerDid := syntax.DID("did:plc:repoowner")
	otherDid := syntax.DID("did:plc:otheractor")
	subjectDid := syntax.DID("did:plc:collabsubject")
	repoDid := syntax.DID("did:plc:testrepo123")

	h, err := syntax.ParseHandle("collabsubject.test")
	if err != nil {
		t.Fatalf("syntax.ParseHandle: %v", err)
	}
	mockIdent := &identity.Identity{
		DID:    subjectDid,
		Handle: h,
	}
	resolver := idresolver.NewMockResolver(idresolver.MockDirectory{Ident: mockIdent})

	jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false)
	if jcerr != nil {
		t.Fatalf("NewJetstreamClient: %v", jcerr)
	}
	s := &Spindle{
		db:      d,
		e:       e,
		l:       slog.Default(),
		cfg:     cfg,
		res:     resolver,
		jc:      jc,
		rootCtx: context.Background(),
		wh:      webhook.New(d, false),
	}

	tap := &Tap{
		spindle: s,
		logger:  slog.Default(),
	}

	err = d.AddRepo(db.Repo{
		Knot:      "knot.test",
		Owner:     ownerDid,
		Rkey:      "test-repo-rkey",
		RepoDid:   repoDid,
		CreatedAt: time.Now().Format(time.RFC3339),
	})
	if err != nil {
		t.Fatalf("AddRepo: %v", err)
	}

	collabRecord := tangled.RepoCollaborator{
		Subject: subjectDid.String(),
		Repo:    repoDid.String(),
	}
	collabRecordJson, _ := json.Marshal(collabRecord)

	err = tap.processCollaborator(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        otherDid,
		Rkey:       "collab-rkey-1",
		Collection: syntax.NSID(tangled.RepoCollaboratorNSID),
		Action:     tapc.RecordCreateAction,
		Record:     collabRecordJson,
	})
	if err != nil {
		t.Fatalf("processCollaborator returned error: %v", err)
	}

	_, err = d.GetRepoCollaborator(otherDid, "collab-rkey-1")
	if err == nil {
		t.Fatal("collaborator from non-owner was registered in DB")
	}

	err = tap.processCollaborator(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        ownerDid,
		Rkey:       "collab-rkey-2",
		Collection: syntax.NSID(tangled.RepoCollaboratorNSID),
		Action:     tapc.RecordCreateAction,
		Record:     collabRecordJson,
	})
	if err != nil {
		t.Fatalf("processCollaborator returned error: %v", err)
	}
	_, err = d.GetRepoCollaborator(ownerDid, "collab-rkey-2")
	if err == nil {
		t.Fatal("collaborator registered despite missing Casbin invite permission")
	}

	err = e.AddRepo(ownerDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil {
		t.Fatalf("AddRepo permissions: %v", err)
	}

	err = tap.processCollaborator(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        ownerDid,
		Rkey:       "collab-rkey-3",
		Collection: syntax.NSID(tangled.RepoCollaboratorNSID),
		Action:     tapc.RecordCreateAction,
		Record:     collabRecordJson,
	})
	if err != nil {
		t.Fatalf("processCollaborator failed for authorized owner: %v", err)
	}

	c, err := d.GetRepoCollaborator(ownerDid, "collab-rkey-3")
	if err != nil {
		t.Fatalf("GetRepoCollaborator error: %v", err)
	}
	if c.Subject != subjectDid || c.RepoDid != repoDid {
		t.Fatalf("unexpected collaborator: %+v", c)
	}

	ok, err := e.IsRepoCollaborator(subjectDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil || !ok {
		t.Fatalf("Casbin policy for collaborator missing or err: %v", err)
	}

	err = tap.processCollaborator(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        ownerDid,
		Rkey:       "collab-rkey-3",
		Collection: syntax.NSID(tangled.RepoCollaboratorNSID),
		Action:     tapc.RecordDeleteAction,
	})
	if err != nil {
		t.Fatalf("delete collaborator process returned error: %v", err)
	}

	_, err = d.GetRepoCollaborator(ownerDid, "collab-rkey-3")
	if err == nil {
		t.Fatal("collaborator DB row remained after deletion")
	}

	ok, err = e.IsRepoCollaborator(subjectDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil || ok {
		t.Fatal("Casbin policy for collaborator remained after deletion")
	}
}

func TestTeardownRepo_RBAC(t *testing.T) {
	d, e := newTestSpindleDB(t)

	cfg := &config.Config{}
	cfg.Server.Hostname = "spindle.test"
	cfg.Server.RepoDir = t.TempDir()
	cfg.Server.LogDir = t.TempDir()

	jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false)
	if jcerr != nil {
		t.Fatalf("NewJetstreamClient: %v", jcerr)
	}
	vault, verr := secrets.NewSQLiteManager(filepath.Join(t.TempDir(), "secrets.db"))
	if verr != nil {
		t.Fatalf("NewSQLiteManager: %v", verr)
	}
	stores, serr := artifactstore.NewStores(config.ArtifactStores{
		Disk: config.ArtifactStoreDisk{Dir: t.TempDir()},
	}, "", "")
	if serr != nil {
		t.Fatalf("NewStores: %v", serr)
	}
	s := &Spindle{
		db:      d,
		e:       e,
		l:       slog.Default(),
		cfg:     cfg,
		jc:      jc,
		vault:   vault,
		stores:  stores,
		n:       ptr(notifier.New()),
		engs:    map[string]models.Engine{},
		rootCtx: context.Background(),
		wh:      webhook.New(d, false),
	}

	tap := &Tap{
		spindle: s,
		logger:  slog.Default(),
	}

	ownerDid := syntax.DID("did:plc:repoowner")
	repoDid := syntax.DID("did:plc:testrepo123")
	collabDid := syntax.DID("did:plc:collab")

	err := d.AddRepo(db.Repo{
		Knot:      "knot.test",
		Owner:     ownerDid,
		Rkey:      "test-repo-rkey",
		RepoDid:   repoDid,
		CreatedAt: time.Now().Format(time.RFC3339),
	})
	if err != nil {
		t.Fatalf("AddRepo DB: %v", err)
	}

	err = e.AddRepo(ownerDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil {
		t.Fatalf("AddRepo policy: %v", err)
	}

	err = d.AddRepoCollaborator(db.RepoCollaborator{
		OwnerDid: ownerDid,
		Rkey:     "collab-rkey",
		Subject:  collabDid,
		RepoDid:  repoDid,
	})
	if err != nil {
		t.Fatalf("AddCollaborator DB: %v", err)
	}

	err = e.AddCollaborator(collabDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil {
		t.Fatalf("AddCollaborator policy: %v", err)
	}

	// entities the delete path only reaches through DeleteAllData
	if err := vault.AddSecret(context.Background(), secrets.UnlockedSecret{Key: "api_key", Value: "v", Repo: secrets.RepoIdentifier(repoDid.String()), CreatedBy: ownerDid}); err != nil {
		t.Fatalf("AddSecret: %v", err)
	}
	if err := d.EnqueueJob(context.Background(), repoDid.String(), models.PipelineId("p1"), nil, tangled.Pipeline{}, "", ""); err != nil {
		t.Fatalf("EnqueueJob: %v", err)
	}

	err = tap.processRepo(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        ownerDid,
		Rkey:       "test-repo-rkey",
		Collection: syntax.NSID(tangled.RepoNSID),
		Action:     tapc.RecordDeleteAction,
	})
	if err != nil {
		t.Fatalf("processRepo delete returned error: %v", err)
	}

	_, err = d.GetRepoByOwnerRkey(ownerDid, "test-repo-rkey")
	if err == nil {
		t.Fatal("repo remained in DB after delete")
	}

	collabs, err := d.ListCollaboratorsByRepoDid(repoDid)
	if err != nil {
		t.Fatalf("ListCollaboratorsByRepoDid: %v", err)
	}
	if len(collabs) > 0 {
		t.Fatal("collaborators remained in DB after delete")
	}

	ok, err := e.IsRepoOwner(ownerDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil || ok {
		t.Fatal("repo owner policy remained in Casbin after delete")
	}

	ok, err = e.IsRepoCollaborator(collabDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil || ok {
		t.Fatal("collaborator policy remained in Casbin after delete")
	}

	if got, kerr := vault.GetSecretsUnlocked(context.Background(), secrets.RepoIdentifier(repoDid.String())); kerr != nil || len(got) > 0 {
		t.Fatalf("secrets remained in vault after delete: %v %v", got, kerr)
	}

	var njobs int
	if err := d.QueryRow(`select count(*) from jobs where repo_did = ?`, repoDid.String()).Scan(&njobs); err != nil || njobs != 0 {
		t.Fatalf("jobs remained in DB after delete: %d %v", njobs, err)
	}
}

func TestProcessRepo_ForgeDeleteRejection(t *testing.T) {
	d, e := newTestSpindleDB(t)

	cfg := &config.Config{}
	cfg.Server.Hostname = "spindle.test"

	jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false)
	if jcerr != nil {
		t.Fatalf("NewJetstreamClient: %v", jcerr)
	}

	s := &Spindle{
		db:      d,
		e:       e,
		l:       slog.Default(),
		cfg:     cfg,
		jc:      jc,
		rootCtx: context.Background(),
		wh:      webhook.New(d, false),
	}

	tap := &Tap{
		spindle: s,
		logger:  slog.Default(),
	}

	aliceDid := syntax.DID("did:plc:alice")
	bobDid := syntax.DID("did:plc:bob")
	repoDid := syntax.DID("did:plc:sharedrepo")

	err := d.AddRepo(db.Repo{
		Knot:      "knot.test",
		Owner:     aliceDid,
		Rkey:      "test-repo-rkey",
		RepoDid:   repoDid,
		CreatedAt: time.Now().Format(time.RFC3339),
	})
	if err != nil {
		t.Fatalf("AddRepo DB: %v", err)
	}

	err = e.AddRepo(aliceDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil {
		t.Fatalf("AddRepo policy: %v", err)
	}

	// bob tries to delete alice's repo, must reject forged delete
	err = tap.processRepo(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        bobDid,
		Rkey:       "test-repo-rkey",
		Collection: syntax.NSID(tangled.RepoNSID),
		Action:     tapc.RecordDeleteAction,
	})
	if err != nil {
		t.Fatalf("processRepo returned error on delete: %v", err)
	}

	_, err = d.GetRepoByOwnerRkey(aliceDid, "test-repo-rkey")
	if err != nil {
		t.Fatalf("Alice's repo was deleted or error: %v", err)
	}

	ok, err := e.IsRepoOwner(aliceDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil || !ok {
		t.Fatal("Alice's owner policy was removed from Casbin by forged delete")
	}
}

func TestProcessCollaborator_ForgeDeleteRejection(t *testing.T) {
	d, e := newTestSpindleDB(t)

	cfg := &config.Config{}
	cfg.Server.Hostname = "spindle.test"

	jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false)
	if jcerr != nil {
		t.Fatalf("NewJetstreamClient: %v", jcerr)
	}

	s := &Spindle{
		db:      d,
		e:       e,
		l:       slog.Default(),
		cfg:     cfg,
		res:     idresolver.DefaultResolver("https://plc.test"),
		jc:      jc,
		rootCtx: context.Background(),
		wh:      webhook.New(d, false),
	}

	tap := &Tap{
		spindle: s,
		logger:  slog.Default(),
	}

	ownerDid := syntax.DID("did:plc:repoowner")
	bobDid := syntax.DID("did:plc:bob")
	collabDid := syntax.DID("did:plc:collab")
	repoDid := syntax.DID("did:plc:testrepo123")

	err := d.AddRepo(db.Repo{
		Knot:      "knot.test",
		Owner:     ownerDid,
		Rkey:      "test-repo-rkey",
		RepoDid:   repoDid,
		CreatedAt: time.Now().Format(time.RFC3339),
	})
	if err != nil {
		t.Fatalf("AddRepo: %v", err)
	}

	err = e.AddRepo(ownerDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil {
		t.Fatalf("AddRepo permissions: %v", err)
	}

	err = d.AddRepoCollaborator(db.RepoCollaborator{
		OwnerDid: ownerDid,
		Rkey:     "collab-rkey",
		Subject:  collabDid,
		RepoDid:  repoDid,
	})
	if err != nil {
		t.Fatalf("AddRepoCollaborator: %v", err)
	}

	err = e.AddCollaborator(collabDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil {
		t.Fatalf("AddCollaborator policy: %v", err)
	}

	// bob tries to delete alice's collaborator, must reject forged delete
	err = tap.processCollaborator(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        bobDid,
		Rkey:       "collab-rkey",
		Collection: syntax.NSID(tangled.RepoCollaboratorNSID),
		Action:     tapc.RecordDeleteAction,
	})
	if err != nil {
		t.Fatalf("processCollaborator delete returned error: %v", err)
	}

	_, err = d.GetRepoCollaborator(ownerDid, "collab-rkey")
	if err != nil {
		t.Fatalf("collaborator was deleted from DB: %v", err)
	}

	ok, err := e.IsRepoCollaborator(collabDid.String(), rbac.ThisServer, repoDid.String())
	if err != nil || !ok {
		t.Fatal("collaborator policy was removed from Casbin by forged delete")
	}
}

func TestPullStatusAction(t *testing.T) {
	tests := []struct {
		name       string
		status     string
		wantAction string
		wantOK     bool
	}{
		{"open maps to reopened", tangled.RepoPullStatusOpen, workflow.PullRequestActionReopened, true},
		{"closed maps to closed", tangled.RepoPullStatusClosed, workflow.PullRequestActionClosed, true},
		{"merged maps to merged", tangled.RepoPullStatusMerged, workflow.PullRequestActionMerged, true},
		{"unknown status is rejected", "sh.tangled.repo.pull.status.bogus", "", false},
		{"empty status is rejected", "", "", false},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			action, ok := pullStatusAction(tt.status)
			if ok != tt.wantOK {
				t.Fatalf("pullStatusAction(%q) ok = %v, want %v", tt.status, ok, tt.wantOK)
			}
			if action != tt.wantAction {
				t.Fatalf("pullStatusAction(%q) action = %q, want %q", tt.status, action, tt.wantAction)
			}
		})
	}
}

func TestProcessPullStatus(t *testing.T) {
	d, e := newTestSpindleDB(t)

	cfg := &config.Config{}
	cfg.Server.Hostname = "spindle.test"

	jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false)
	if jcerr != nil {
		t.Fatalf("NewJetstreamClient: %v", jcerr)
	}
	s := &Spindle{
		db:      d,
		e:       e,
		l:       slog.Default(),
		cfg:     cfg,
		res:     idresolver.DefaultResolver("https://plc.test"),
		jc:      jc,
		rootCtx: context.Background(),
		wh:      webhook.New(d, false),
	}

	statusRecord := func(status, pullUri string) []byte {
		rec := tangled.RepoPullStatus{
			Status:    status,
			Pull:      pullUri,
			CreatedAt: time.Now().Format(time.RFC3339),
		}
		b, _ := json.Marshal(rec)
		return b
	}

	validPullUri := "at://did:plc:pullowner/sh.tangled.repo.pull/pull-rkey-1"

	// non-create actions are ignored
	if err := s.processPullStatus(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        syntax.DID("did:plc:actor"),
		Rkey:       "status-rkey",
		Collection: syntax.NSID(tangled.RepoPullStatusNSID),
		Action:     tapc.RecordUpdateAction,
		Record:     statusRecord(tangled.RepoPullStatusClosed, validPullUri),
	}); err != nil {
		t.Fatalf("update action should be a no-op, got: %v", err)
	}

	// unknown status variant is skipped without error
	if err := s.processPullStatus(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        syntax.DID("did:plc:actor"),
		Rkey:       "status-rkey",
		Collection: syntax.NSID(tangled.RepoPullStatusNSID),
		Action:     tapc.RecordCreateAction,
		Record:     statusRecord("sh.tangled.repo.pull.status.bogus", validPullUri),
	}); err != nil {
		t.Fatalf("unknown status should be skipped, got: %v", err)
	}

	// a malformed pull at-uri is skipped without error
	if err := s.processPullStatus(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        syntax.DID("did:plc:actor"),
		Rkey:       "status-rkey",
		Collection: syntax.NSID(tangled.RepoPullStatusNSID),
		Action:     tapc.RecordCreateAction,
		Record:     statusRecord(tangled.RepoPullStatusClosed, "not-an-at-uri"),
	}); err != nil {
		t.Fatalf("invalid pull at-uri should be skipped, got: %v", err)
	}

	// a status pointing at a non-pull subject is skipped without error
	if err := s.processPullStatus(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        syntax.DID("did:plc:actor"),
		Rkey:       "status-rkey",
		Collection: syntax.NSID(tangled.RepoPullStatusNSID),
		Action:     tapc.RecordCreateAction,
		Record:     statusRecord(tangled.RepoPullStatusClosed, "at://did:plc:x/sh.tangled.repo.issue/y"),
	}); err != nil {
		t.Fatalf("non-pull subject should be skipped, got: %v", err)
	}

	// a valid close event resolves the pull record; fetch fails because plc/pds
	// are not real, confirming we reached the fetch stage with a mapped action.
	err := s.processPullStatus(context.Background(), &tapc.RecordEventData{
		Live:       true,
		Did:        syntax.DID("did:plc:actor"),
		Rkey:       "status-rkey",
		Collection: syntax.NSID(tangled.RepoPullStatusNSID),
		Action:     tapc.RecordCreateAction,
		Record:     statusRecord(tangled.RepoPullStatusClosed, validPullUri),
	})
	if err == nil {
		t.Fatal("expected error fetching pull record against fake pds, got nil")
	}
	if !strings.Contains(err.Error(), "fetch pull record") {
		t.Fatalf("expected fetch pull record error, got: %v", err)
	}
}

func TestIsPullTriggerAuthorized(t *testing.T) {
	d, e := newTestSpindleDB(t)
	s := &Spindle{
		db: d,
		e:  e,
		l:  slog.Default(),
		wh: webhook.New(d, false),
	}

	repoDid := syntax.DID("did:plc:targetrepo")
	ownerDid := syntax.DID("did:plc:owner")   // has push (repo owner)
	collaboratorDid := "did:plc:collaborator" // has push
	noPushAuthorDid := "did:plc:nopushauthor" // pull author without push
	strangerDid := "did:plc:stranger"         // no push, not the author

	if err := e.AddRepo(ownerDid.String(), rbac.ThisServer, repoDid.String()); err != nil {
		t.Fatalf("AddRepo permissions: %v", err)
	}
	if err := e.AddCollaborator(collaboratorDid, rbac.ThisServer, repoDid.String()); err != nil {
		t.Fatalf("AddCollaborator: %v", err)
	}

	cases := []struct {
		name     string
		eventDid string
		pullDid  string
		allowed  bool
	}{
		// direct pr path: event author == pull author
		{"direct pr by pushing author", ownerDid.String(), ownerDid.String(), true},
		{"direct pr by non-pushing author", noPushAuthorDid, noPushAuthorDid, false},

		// status path: actor differs from pull author, pull author has push
		{"actor with push acts on authorized pr", collaboratorDid, ownerDid.String(), true},
		{"pull author acts on own authorized pr", ownerDid.String(), ownerDid.String(), true},
		{"stranger without push acts on authorized pr", strangerDid, ownerDid.String(), false},

		// pull author must always have push, even if the actor does
		{"pushing actor on unauthorized pull author", ownerDid.String(), noPushAuthorDid, false},
		{"non-pushing actor on unauthorized pull author", strangerDid, noPushAuthorDid, false},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			allowed, err := s.isPullTriggerAuthorized(tc.eventDid, tc.pullDid, repoDid.String())
			if err != nil {
				t.Fatalf("isPullTriggerAuthorized: %v", err)
			}
			if allowed != tc.allowed {
				t.Fatalf("event=%q pull=%q allowed=%v, want %v", tc.eventDid, tc.pullDid, allowed, tc.allowed)
			}
		})
	}
}