package spindle

import (
	"context"
	"encoding/json"
	"os"
	"path/filepath"
	"tangled.org/core/api/org_tangled"
	"testing"
	"time"

	comatproto "github.com/bluesky-social/indigo/api/atproto"
	"github.com/bluesky-social/indigo/atproto/syntax"
	"tangled.org/core/jetstream"
	"tangled.org/core/log"
	"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"
)

func newWipeTestSpindle(t *testing.T) *Spindle {
	t.Helper()
	tmp := t.TempDir()
	ctx := context.Background()

	d, err := db.Make(ctx, filepath.Join(tmp, "spindle.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { d.Close() })

	e, err := rbac.NewEnforcer(filepath.Join(tmp, "rbac.db"))
	if err != nil {
		t.Fatal(err)
	}

	vault, err := secrets.NewSQLiteManager(filepath.Join(tmp, "secrets.db"))
	if err != nil {
		t.Fatal(err)
	}

	stores, err := artifactstore.NewStores(config.ArtifactStores{
		Disk: config.ArtifactStoreDisk{Dir: filepath.Join(tmp, "artifacts")},
	}, "", "")
	if err != nil {
		t.Fatal(err)
	}

	jc, err := jetstream.NewJetstreamClient("", nil, nil, log.New("test"), d, false, false)
	if err != nil {
		t.Fatal(err)
	}

	return &Spindle{
		l:      log.New("test"),
		db:     d,
		e:      e,
		vault:  vault,
		stores: stores,
		jc:     jc,
		n:      ptr(notifier.New()),
		engs:   map[string]models.Engine{},
		cfg: &config.Config{
			Server: config.Server{
				Hostname: "spindle.example.com",
				RepoDir:  filepath.Join(tmp, "repos"),
				LogDir:   filepath.Join(tmp, "logs"),
			},
		},
	}
}

func seedWipeRepo(t *testing.T, s *Spindle, repoDid, owner syntax.DID) {
	t.Helper()
	ctx := context.Background()

	sfx := repoDid.String()[len(repoDid.String())-1:]

	if err := s.db.AddRepo(db.Repo{Knot: "knot.example.com", Owner: owner, Rkey: syntax.RecordKey("r" + sfx), RepoDid: repoDid}); err != nil {
		t.Fatal(err)
	}

	if err := s.db.AddRepoCollaborator(db.RepoCollaborator{OwnerDid: owner, Rkey: syntax.RecordKey("c" + sfx), Subject: "did:plc:collab", RepoDid: repoDid}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec(`insert into jobs (repo_did, pipeline_id, tpl, created_at_ns) values (?, ?, '{}', 0)`, repoDid.String(), "p"+sfx); err != nil {
		t.Fatal(err)
	}
	if err := s.db.SaveMillLease(db.MillLease{LeaseID: "l" + sfx, NodeID: "n1", Epoch: "e1", Engine: "microvm", PipelineID: "p" + sfx, Workflow: "w" + sfx, State: "active", RepoDID: repoDid.String()}); err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec(`insert into mill_artifacts (lease_id, repo_did, pipeline_id, workflow, ref, hash) values (?, ?, ?, ?, ?, 'h')`, "l"+sfx, repoDid.String(), "p"+sfx, "w"+sfx, "out/l"+sfx+".bin"); err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec(`insert into quota_allocations (repo_did, resource, kind, key, amount) values (?, 'compute', 'generic', 'k', 1)`, repoDid.String()); err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec(`insert into webhooks (repo_did, url, events) values (?, 'https://example.com/hook', 'push')`, repoDid.String()); err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec(`insert into webhook_deliveries (webhook_id, event, delivery_id, url, success) values ((select max(id) from webhooks), 'push', ?, 'https://example.com/hook', 1)`, "d"+sfx); err != nil {
		t.Fatal(err)
	}
	if err := s.db.AddPullTicket(repoDid, "pull1", db.PullTicket{Author: owner, State: "open"}); err != nil {
		t.Fatal(err)
	}
	if err := s.db.AddPullPatch(repoDid, db.PullPatch{Rkey: "patch1", Ticket: "pull1", Head: "abc", CreatedAt: "2026-01-01T00:00:00Z"}, db.PullEvent{RepoDid: repoDid, Ticket: "pull1", Pull: []byte("{}")}); err != nil {
		t.Fatal(err)
	}
	if err := s.db.PutRecordRetry(db.RecordRetry{RepoDid: repoDid, Collection: "org.tangled.track.ticket", Rkey: "pull1"}, 1, time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := s.vault.AddSecret(ctx, secrets.UnlockedSecret{Key: "api_key", Value: "v", Repo: secrets.RepoIdentifier(repoDid.String()), CreatedBy: owner}); err != nil {
		t.Fatal(err)
	}
	pipeline := &org_tangled.CiDefs_Pipeline{
		Id:        "p" + sfx,
		Repo:      repoDid.String(),
		Workflows: []*org_tangled.CiDefs_WorkflowRun{{Id: "w" + sfx, Name: ptr("w" + sfx), Status: "pending"}},
	}
	payload, err := json.Marshal(pipeline)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec(`insert into pipelines (pipeline_id, repo_did, commit_sha, kind, payload) values (?, ?, '', '', ?)`, pipeline.Id, repoDid.String(), string(payload)); err != nil {
		t.Fatal(err)
	}
	if _, err := s.db.Exec(`insert into workflow_statuses (pipeline_id, workflow, status, created_at) values (?, ?, 'pending', 'now')`, pipeline.Id, "w"+sfx); err != nil {
		t.Fatal(err)
	}

	if err := os.MkdirAll(s.newRepoPath(repoDid), 0755); err != nil {
		t.Fatal(err)
	}
	for _, ref := range []string{"out/l" + sfx + ".bin", "logs/l" + sfx + ".log"} {
		for _, put := range s.stores.PutFile(ctx, ref, writeTempFile(t, "data")) {
			if put.Err != nil {
				t.Fatal(put.Err)
			}
		}
	}
}

func writeTempFile(t *testing.T, content string) string {
	t.Helper()
	p := filepath.Join(t.TempDir(), "src")
	if err := os.WriteFile(p, []byte(content), 0644); err != nil {
		t.Fatal(err)
	}
	return p
}

func TestWipeRepoRemovesAllState(t *testing.T) {
	s := newWipeTestSpindle(t)
	ctx := context.Background()
	repoDid := syntax.DID("did:plc:repo1")
	owner := syntax.DID("did:plc:owner1")

	seedWipeRepo(t, s, repoDid, owner)

	if err := s.WipeRepo(ctx, repoDid, "test"); err != nil {
		t.Fatalf("WipeRepo: %v", err)
	}

	var n int
	for _, table := range []string{"repos", "jobs", "mill_leases", "mill_artifacts", "quota_allocations", "repo_collaborators", "pipelines", "workflow_statuses", "webhooks", "webhook_deliveries", "pull_tickets", "pull_patches", "pull_events", "record_retries"} {
		if err := s.db.QueryRow(`select count(*) from ` + table).Scan(&n); err != nil {
			t.Fatal(err)
		}
		if n != 0 {
			t.Fatalf("%s: expected empty, got %d rows", table, n)
		}
	}

	got, err := s.vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid.String()))
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 0 {
		t.Fatalf("expected secrets gone, got %d", len(got))
	}

	if _, err := os.Stat(s.newRepoPath(repoDid)); !os.IsNotExist(err) {
		t.Fatalf("expected clone dir gone, stat err: %v", err)
	}
	if _, err := s.stores.Open(ctx, "out/l1.bin"); err == nil {
		t.Fatal("expected artifact gone")
	}
	if _, err := s.stores.Open(ctx, "logs/l1.log"); err == nil {
		t.Fatal("expected log artifact gone")
	}

	if err := s.WipeRepo(ctx, repoDid, "test again"); err != nil {
		t.Fatalf("second WipeRepo: %v", err)
	}
}

func TestWipeShouldKeepAllowListedOwners(t *testing.T) {
	s := newWipeTestSpindle(t)
	ctx := context.Background()
	repoDid := syntax.DID("did:plc:repo1")
	owner := syntax.DID("did:plc:owner1")

	seedWipeRepo(t, s, repoDid, owner)
	if err := db.AddToAllowlist(s.db, owner); err != nil {
		t.Fatal(err)
	}

	if err := s.WipeRepo(ctx, repoDid, "test"); err != nil {
		t.Fatalf("WipeRepo: %v", err)
	}

	onList, err := db.AllowlistContains(s.db, owner)
	if err != nil {
		t.Fatal(err)
	}
	if !onList {
		t.Fatal("A wipe mustn't take the owner off the allow-list.")
	}
}

func TestWipeRepoRemovesLocalEngineArtifacts(t *testing.T) {
	s := newWipeTestSpindle(t)
	ctx := context.Background()
	repoDid := syntax.DID("did:plc:repo1")
	owner := syntax.DID("did:plc:owner1")

	seedWipeRepo(t, s, repoDid, owner)

	wid := models.WorkflowId{PipelineId: models.PipelineId("p1"), Name: "w1"}
	localRef := "logs/" + wid.String() + ".log"
	if err := s.db.SaveArtifactRef(wid.String(), repoDid.String(), wid, localRef, "h"); err != nil {
		t.Fatal(err)
	}
	for _, put := range s.stores.PutFile(ctx, localRef, writeTempFile(t, "log")) {
		if put.Err != nil {
			t.Fatal(put.Err)
		}
	}

	if err := s.WipeRepo(ctx, repoDid, "test"); err != nil {
		t.Fatalf("WipeRepo: %v", err)
	}

	if _, err := s.stores.Open(ctx, localRef); err == nil {
		t.Fatal("expected local engine artifact gone")
	}
	var n int
	if err := s.db.QueryRow(`select count(*) from mill_artifacts`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	if n != 0 {
		t.Fatalf("expected mill_artifacts empty, got %d rows", n)
	}
}

func TestWipeOwnerWipesAllOwnedRepos(t *testing.T) {
	s := newWipeTestSpindle(t)
	ctx := context.Background()
	owner := syntax.DID("did:plc:owner1")
	other := syntax.DID("did:plc:owner2")

	seedWipeRepo(t, s, "did:plc:repo1", owner)
	seedWipeRepo(t, s, "did:plc:repo2", owner)
	seedWipeRepo(t, s, "did:plc:repo3", other)

	if err := s.WipeOwner(ctx, owner, "test"); err != nil {
		t.Fatalf("WipeOwner: %v", err)
	}

	var n int
	if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	if n != 1 {
		t.Fatalf("expected only the other owner's repo left, got %d", n)
	}
	repos, err := s.db.ReposByDid("did:plc:repo3")
	if err != nil || len(repos) != 1 {
		t.Fatalf("expected repo3 intact, got %v err %v", repos, err)
	}
}

type destroyRecorder struct {
	models.Engine
	destroyed []models.WorkflowId
}

func (e *destroyRecorder) DestroyWorkflow(_ context.Context, wid models.WorkflowId) error {
	e.destroyed = append(e.destroyed, wid)
	return nil
}

func TestWipeDestroysFinishedWorkflowStillHoldingItsVM(t *testing.T) {
	s := newWipeTestSpindle(t)
	repoDid := syntax.DID("did:plc:repo1")
	seedWipeRepo(t, s, repoDid, "did:plc:owner1")
	eng := &destroyRecorder{}
	s.engs["microvm"] = eng

	wid := models.WorkflowId{PipelineId: models.PipelineId("p1"), Name: "w1"}
	if err := s.db.StatusFailed(wid, "User step error: exited with code 1", 1, s.n); err != nil {
		t.Fatal(err)
	}

	if _, err := s.cancelRepoWorkflows(context.Background(), repoDid, "banned"); err != nil {
		t.Fatal(err)
	}
	if len(eng.destroyed) != 1 || eng.destroyed[0] != wid {
		t.Fatalf("destroyed = %v, want the held workflow", eng.destroyed)
	}
	if st, err := s.db.GetStatus(wid); err != nil || st != models.StatusKindFailed {
		t.Fatalf("status = %q (err %v), want the failure kept", st, err)
	}
}

func ptr[T any](v T) *T { return &v }

func TestIngestAccountTakedownWipes(t *testing.T) {
	s := newWipeTestSpindle(t)
	owner := syntax.DID("did:plc:owner1")

	seedWipeRepo(t, s, "did:plc:repo1", owner)

	status := "takendown"
	err := s.ingest()(context.Background(), &jetstream.Event{
		Kind:    jetstream.EventKindAccount,
		Did:     owner.String(),
		Account: &comatproto.SyncSubscribeRepos_Account{Did: owner.String(), Active: false, Status: &status},
	})
	if err != nil {
		t.Fatalf("ingest account event: %v", err)
	}

	var n int
	if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	if n != 0 {
		t.Fatalf("expected repos wiped after takedown, got %d", n)
	}

	seedWipeRepo(t, s, "did:plc:repo1", owner)
	err = s.ingest()(context.Background(), &jetstream.Event{
		Kind:    jetstream.EventKindAccount,
		Did:     owner.String(),
		Account: &comatproto.SyncSubscribeRepos_Account{Did: owner.String(), Active: true},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	if n != 1 {
		t.Fatalf("expected active account untouched, got %d", n)
	}
}