package deploy

import (
	"database/sql"
	"errors"
	"os"
	"path/filepath"
	"strconv"
	"testing"
	"time"

	"github.com/iceBear67/simplepages/api"
	"github.com/iceBear67/simplepages/internal/cas"
	"github.com/iceBear67/simplepages/internal/store"
)

// These tests set BlobGrace negative. The grace exists so that a request which
// has resolved a digest and is about to open it cannot lose the file underneath
// it, and the default hour is far longer than any test wants to wait. Timestamps
// are whole seconds, so a grace of zero would not collect a blob dereferenced in
// the same second either — negative is the only value that means "now".
const collectNow = -time.Minute

// deployReady publishes a finished, inactive deployment holding contents.
func (e *env) deployReady(t *testing.T, contents map[string]string) *store.Deployment {
	t.Helper()
	dep := e.create(t)
	files := manifest(contents)
	if _, _, err := e.svc.SetManifest(t.Context(), dep, files); err != nil {
		t.Fatalf("SetManifest: %v", err)
	}
	paths := make([]string, 0, len(contents))
	for p := range contents {
		paths = append(paths, p)
	}
	e.upload(t, contents, paths...)
	dep, err := e.svc.Finalize(t.Context(), e.p, dep)
	if err != nil {
		t.Fatalf("Finalize: %v", err)
	}
	return dep
}

// version publishes a deployment whose single file identifies it, which is
// enough for retention tests: what matters is how many survive and which.
func (e *env) version(t *testing.T, n int) *store.Deployment {
	t.Helper()
	return e.deployReady(t, map[string]string{"index.html": "v" + strconv.Itoa(n)})
}

func (e *env) activate(t *testing.T, dep *store.Deployment) {
	t.Helper()
	if _, err := e.svc.Activate(t.Context(), e.p, dep); err != nil {
		t.Fatalf("Activate %s: %v", dep.PublicID, err)
	}
}

// exec runs a statement the service has no method for. Retention tests need to
// backdate rows, because the alternative is a test that sleeps for an hour.
func (e *env) exec(t *testing.T, query string, args ...any) {
	t.Helper()
	err := e.db.Tx(t.Context(), func(tx *sql.Tx) error {
		_, err := tx.ExecContext(t.Context(), query, args...)
		return err
	})
	if err != nil {
		t.Fatalf("%s: %v", query, err)
	}
}

// setRetention rewrites the project's policy and re-reads it, because the
// collector reads the row and not the struct the test is holding.
func (e *env) setRetention(t *testing.T, count int, graceS int64) {
	t.Helper()
	e.exec(t, `UPDATE projects SET retention_count = ?, retention_grace_s = ? WHERE id = ?`,
		count, graceS, e.p.ID)
	p, err := e.db.ProjectByID(t.Context(), e.p.ID)
	if err != nil {
		t.Fatal(err)
	}
	e.p = p
}

// alive reports whether a deployment still has a row and a tree.
func (e *env) alive(t *testing.T, dep *store.Deployment) (row, tree bool) {
	t.Helper()
	_, err := e.db.DeploymentByPublicID(t.Context(), e.p.ID, dep.PublicID)
	switch {
	case err == nil:
		row = true
	case errors.Is(err, store.ErrNotFound):
	default:
		t.Fatalf("DeploymentByPublicID: %v", err)
	}
	_, err = os.Stat(DeploymentDir(e.dir, e.p.ID, dep.PublicID))
	switch {
	case err == nil:
		tree = true
	case os.IsNotExist(err):
	default:
		t.Fatalf("stat deployment tree: %v", err)
	}
	return
}

func (e *env) mustSurvive(t *testing.T, dep *store.Deployment, why string) {
	t.Helper()
	row, tree := e.alive(t, dep)
	if !row || !tree {
		t.Errorf("%s (%s) was collected: row=%v tree=%v", why, dep.PublicID, row, tree)
	}
}

func (e *env) mustBeGone(t *testing.T, dep *store.Deployment, why string) {
	t.Helper()
	row, tree := e.alive(t, dep)
	if row || tree {
		t.Errorf("%s (%s) survived: row=%v tree=%v", why, dep.PublicID, row, tree)
	}
}

// blobCount is how many blobs the database knows about, which is the number the
// milestone's manual check watches drop.
func blobCount(t *testing.T, e *env) int64 {
	t.Helper()
	counts, err := e.db.Counts(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	return counts.Blobs
}

func hasContent(t *testing.T, e *env, content string) bool {
	t.Helper()
	ok, err := e.cas.Has(cas.Sum([]byte(content)))
	if err != nil {
		t.Fatalf("cas.Has: %v", err)
	}
	return ok
}

// The headline retention rule from the milestone: deploy repeatedly, keep
// retention_count of them plus whichever one is being served, and watch the
// content of the rest go away.
func TestCollectKeepsRetentionCountPlusTheActiveOne(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 10, 0)

	var deps []*store.Deployment
	for i := 1; i <= 15; i++ {
		dep := e.version(t, i)
		deps = append(deps, dep)
		e.activate(t, dep)
	}
	// Roll back to the oldest one, so the deployment being served is also the
	// one retention would otherwise drop first. Nothing may collect it.
	e.activate(t, deps[0])

	before := blobCount(t, e)
	stats, err := e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}

	// 15 deployments, 10 kept by retention plus the active one: 4 collected.
	if stats.DeploymentsDeleted != 4 {
		t.Errorf("deleted %d deployments, want 4", stats.DeploymentsDeleted)
	}
	e.mustSurvive(t, deps[0], "the active deployment")
	for _, dep := range deps[5:] {
		e.mustSurvive(t, dep, "a deployment inside the retention window")
	}
	for _, dep := range deps[1:5] {
		e.mustBeGone(t, dep, "a deployment past the retention window")
	}

	if after := blobCount(t, e); after != before-4 {
		t.Errorf("blob count went from %d to %d, want %d", before, after, before-4)
	}
	if stats.BlobsDeleted != 4 {
		t.Errorf("collected %d blobs, want the 4 the deleted deployments held", stats.BlobsDeleted)
	}
	if stats.BytesFreed <= 0 {
		t.Errorf("BytesFreed = %d, want the size of what was removed", stats.BytesFreed)
	}
	for i := 2; i <= 5; i++ {
		if hasContent(t, e, "v"+strconv.Itoa(i)) {
			t.Errorf("content of the collected deployment v%d is still in the CAS", i)
		}
	}
	// What the survivors reference is untouched, which is what makes a rollback
	// to any of them still work.
	if !hasContent(t, e, "v1") || !hasContent(t, e, "v15") {
		t.Error("content a surviving deployment references was collected")
	}

	// Nothing left to do on a second pass.
	stats, err = e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	if stats.DeploymentsDeleted != 0 || stats.BlobsDeleted != 0 {
		t.Errorf("a second pass collected %+v, want nothing", stats)
	}
}

// The grace period is measured from when a deployment stopped being served, so
// a rollback that was a mistake can be undone for a while afterwards.
func TestCollectHonoursTheRetentionGrace(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 0, 3600)

	old := e.version(t, 1)
	e.activate(t, old)
	current := e.version(t, 2)
	e.activate(t, current)

	// retention_count is zero, so only the grace is protecting it.
	stats, err := e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	if stats.DeploymentsDeleted != 0 {
		t.Errorf("deleted %d deployments inside the grace period, want 0", stats.DeploymentsDeleted)
	}
	e.mustSurvive(t, old, "a deployment retired seconds ago")

	// Backdate the deactivation past the grace and it becomes collectable.
	e.exec(t, `UPDATE deployments SET deactivated_at = ? WHERE id = ?`,
		time.Now().Add(-2*time.Hour).Unix(), old.ID)
	if _, err := e.svc.Collect(t.Context(), false); err != nil {
		t.Fatal(err)
	}
	e.mustBeGone(t, old, "a deployment retired before the grace period")
	e.mustSurvive(t, current, "the active deployment")
}

// A deployment that never finished uploading is not retention's business: it
// has no deactivated_at to measure from and someone may still be pushing to it.
func TestCollectLeavesUnfinishedUploadsToTheExpiry(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 0, 0)

	contents := map[string]string{"index.html": "in progress"}
	dep := e.create(t)
	if _, _, err := e.svc.SetManifest(t.Context(), dep, manifest(contents)); err != nil {
		t.Fatal(err)
	}
	e.upload(t, contents, "index.html")

	if _, err := e.svc.Collect(t.Context(), false); err != nil {
		t.Fatal(err)
	}
	got, err := e.db.DeploymentByPublicID(t.Context(), e.p.ID, dep.PublicID)
	if err != nil {
		t.Fatalf("an in-progress upload was collected: %v", err)
	}
	if got.State != store.StateUploading {
		t.Errorf("state = %q, want it left alone as %q", got.State, store.StateUploading)
	}
	// Its content is protected too, by the manifest rows that already reference
	// it — which is the whole reason the manifest is written before the upload.
	if !hasContent(t, e, "in progress") {
		t.Error("the content of an in-progress upload was collected")
	}

	// Once it is old enough, one pass does the whole job: the expiry fails it
	// and drops its manifest, retention finds a failed deployment older than it
	// keeps failures for, and the blob pass then reaches what only it
	// referenced. That the three run in that order is why it takes one pass and
	// not three.
	e.exec(t, `UPDATE deployments SET created_at = ? WHERE id = ?`,
		time.Now().Add(-48*time.Hour).Unix(), dep.ID)
	stats, err := e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	if stats.DeploymentsDeleted != 1 || stats.BlobsDeleted != 1 {
		t.Errorf("collected %+v, want the abandoned upload and its content", stats)
	}
	if _, err := e.db.DeploymentByPublicID(t.Context(), e.p.ID, dep.PublicID); !errors.Is(err, store.ErrNotFound) {
		t.Errorf("the abandoned upload survived: %v", err)
	}
	if hasContent(t, e, "in progress") {
		t.Error("content nothing references any more survived the sweep")
	}
}

// A deployment that failed while being finalized keeps its row for a day, so an
// operator looking into a broken CI job can still see that it failed and why.
func TestCollectKeepsRecentFailuresForInspection(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 0, 0)

	dep := e.create(t)
	if err := e.db.MarkDeploymentFailed(t.Context(), dep.ID, "assembly failed"); err != nil {
		t.Fatal(err)
	}

	if _, err := e.svc.Collect(t.Context(), false); err != nil {
		t.Fatal(err)
	}
	got, err := e.db.DeploymentByPublicID(t.Context(), e.p.ID, dep.PublicID)
	if err != nil {
		t.Fatalf("a deployment that failed moments ago was collected: %v", err)
	}
	if got.Error != "assembly failed" {
		t.Errorf("error = %q, want the reason still readable", got.Error)
	}

	e.exec(t, `UPDATE deployments SET created_at = ? WHERE id = ?`,
		time.Now().Add(-48*time.Hour).Unix(), dep.ID)
	if _, err := e.svc.Collect(t.Context(), false); err != nil {
		t.Fatal(err)
	}
	if _, err := e.db.DeploymentByPublicID(t.Context(), e.p.ID, dep.PublicID); !errors.Is(err, store.ErrNotFound) {
		t.Errorf("a failed row older than the retention survived: %v", err)
	}
}

// Content two deployments share outlives the first of them. This is what makes
// cross-deployment deduplication safe to rely on.
func TestCollectKeepsSharedContent(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 0, 0)

	shared := "console.log(1)"
	old := e.deployReady(t, map[string]string{"index.html": "v1", "app.js": shared})
	e.activate(t, old)
	current := e.deployReady(t, map[string]string{"index.html": "v2", "app.js": shared})
	e.activate(t, current)

	if _, err := e.svc.Collect(t.Context(), false); err != nil {
		t.Fatal(err)
	}
	e.mustBeGone(t, old, "the superseded deployment")
	if hasContent(t, e, "v1") {
		t.Error("content only the collected deployment referenced survived")
	}
	if !hasContent(t, e, shared) {
		t.Fatal("content the active deployment still references was collected")
	}
	// And it is still readable through the deployment that survived, which is
	// the property the assembled tree shares an inode for.
	body, err := os.ReadFile(filepath.Join(DeploymentDir(e.dir, e.p.ID, current.PublicID), "app.js"))
	if err != nil || string(body) != shared {
		t.Errorf("reading shared content from the surviving tree = %q, %v", body, err)
	}
}

// The grace is what makes the read path safe without per-request reference
// counting, so it has to actually hold content back.
func TestCollectHoldsRecentlyDereferencedBlobs(t *testing.T) {
	e := newEnv(t)
	e.setRetention(t, 0, 0) // BlobGrace left at its default hour.

	old := e.version(t, 1)
	e.activate(t, old)
	e.activate(t, e.version(t, 2))

	stats, err := e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	e.mustBeGone(t, old, "the superseded deployment")
	if stats.BlobsDeleted != 0 {
		t.Errorf("collected %d blobs, want them held by the grace period", stats.BlobsDeleted)
	}
	if !hasContent(t, e, "v1") {
		t.Error("content dereferenced moments ago was removed inside the grace period")
	}
}

func TestCollectDryRunChangesNothing(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 0, 0)

	old := e.version(t, 1)
	e.activate(t, old)
	current := e.version(t, 2)
	e.activate(t, current)

	stats, err := e.svc.Collect(t.Context(), true)
	if err != nil {
		t.Fatal(err)
	}
	if !stats.DryRun {
		t.Error("the report does not say it was a dry run")
	}
	if stats.DeploymentsDeleted != 1 {
		t.Errorf("reported %d deployments, want the 1 that would be deleted", stats.DeploymentsDeleted)
	}
	e.mustSurvive(t, old, "a deployment a dry run only reported on")
	e.mustSurvive(t, current, "the active deployment")
	if !hasContent(t, e, "v1") {
		t.Error("a dry run removed content")
	}

	// Blobs the reported deletions would free are not counted: nothing was
	// deleted, so they are all still referenced. The number is a floor.
	if stats.BlobsDeleted != 0 {
		t.Errorf("a dry run reported %d collectable blobs, want 0 while everything is referenced",
			stats.BlobsDeleted)
	}

	// The real pass then does what the dry run said it would.
	stats, err = e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	if stats.DryRun {
		t.Error("a real pass reported itself as a dry run")
	}
	if stats.DeploymentsDeleted != 1 {
		t.Errorf("deleted %d deployments, want 1", stats.DeploymentsDeleted)
	}
	e.mustBeGone(t, old, "the superseded deployment")
}

func TestDeleteRemovesRowAndTree(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow

	dep := e.version(t, 1)
	current := e.version(t, 2)
	e.activate(t, current)

	if err := e.svc.Delete(t.Context(), e.p, dep); err != nil {
		t.Fatalf("Delete: %v", err)
	}
	e.mustBeGone(t, dep, "the deleted deployment")

	// Deleting it dropped the manifest, so the next collection reaches what only
	// it referenced.
	if _, err := e.svc.Collect(t.Context(), false); err != nil {
		t.Fatal(err)
	}
	if hasContent(t, e, "v1") {
		t.Error("content the deleted deployment held survived collection")
	}

	if err := e.svc.Delete(t.Context(), e.p, dep); apiCode(err) != api.CodeNotFound {
		t.Errorf("deleting it again = %v, want %q", err, api.CodeNotFound)
	}
}

// Deleting what a project is serving is a conflict, not a permission problem:
// the client is told to activate something else first.
func TestDeleteRefusesTheActiveDeployment(t *testing.T) {
	e := newEnv(t)
	dep := e.version(t, 1)
	e.activate(t, dep)

	err := e.svc.Delete(t.Context(), e.p, dep)
	wantCode(t, err, api.CodeDeploymentActive)
	e.mustSurvive(t, dep, "the active deployment")

	// It becomes deletable the moment something else is being served, which is
	// the sequence the error message describes.
	e.activate(t, e.version(t, 2))
	if err := e.svc.Delete(t.Context(), e.p, dep); err != nil {
		t.Fatalf("Delete after activating another deployment: %v", err)
	}
	e.mustBeGone(t, dep, "the deployment that was superseded and then deleted")
}

func TestRemoveProjectTrees(t *testing.T) {
	e := newEnv(t)
	dep := e.version(t, 1)
	e.activate(t, dep)

	other := store.DefaultProject("other")
	if err := e.db.CreateProject(t.Context(), other); err != nil {
		t.Fatal(err)
	}
	otherDir := DeploymentDir(e.dir, other.ID, "dpl_0000000000000000")
	if err := os.MkdirAll(otherDir, 0o755); err != nil {
		t.Fatal(err)
	}

	if err := e.svc.RemoveProjectTrees(e.p.ID); err != nil {
		t.Fatalf("RemoveProjectTrees: %v", err)
	}
	if _, err := os.Stat(filepath.Join(e.dir, strconv.FormatInt(e.p.ID, 10))); !os.IsNotExist(err) {
		t.Errorf("the project's directory survived: %v", err)
	}
	if _, err := os.Stat(otherDir); err != nil {
		t.Errorf("another project's directory was removed: %v", err)
	}

	// Idempotent: recovery may run it again after a crash partway through.
	if err := e.svc.RemoveProjectTrees(e.p.ID); err != nil {
		t.Errorf("a second removal: %v", err)
	}
}

// Deleting the project takes its deployments with it, and the collector then
// reclaims everything they referenced.
func TestCollectAfterProjectDeletion(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow

	dep := e.version(t, 1)
	e.activate(t, dep)

	if err := e.db.DeleteProject(t.Context(), e.p.ID); err != nil {
		t.Fatalf("DeleteProject: %v", err)
	}
	if err := e.svc.RemoveProjectTrees(e.p.ID); err != nil {
		t.Fatalf("RemoveProjectTrees: %v", err)
	}

	stats, err := e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	if stats.BlobsDeleted != 1 {
		t.Errorf("collected %d blobs, want the 1 the deleted project held", stats.BlobsDeleted)
	}
	if hasContent(t, e, "v1") {
		t.Error("content of a deleted project survived collection")
	}
	if n := blobCount(t, e); n != 0 {
		t.Errorf("%d blob rows left after the project was deleted", n)
	}
}

// A deployment claimed by a sweep that was interrupted is finished by the next
// one, whatever its retention would otherwise have said.
func TestCollectResumesAnInterruptedDeletion(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 10, 3600) // Generous: retention alone would keep it.

	dep := e.version(t, 1)
	e.activate(t, e.version(t, 2))
	if err := e.db.MarkDeploymentDeleting(t.Context(), dep.ID); err != nil {
		t.Fatal(err)
	}

	stats, err := e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	if stats.DeploymentsDeleted != 1 {
		t.Errorf("deleted %d deployments, want the claimed one", stats.DeploymentsDeleted)
	}
	e.mustBeGone(t, dep, "a deployment a previous sweep had claimed")
}

// The activation path and the collector both take the project lock, and the
// claim rechecks active = 0 under it. A deployment that becomes active between
// being listed and being claimed is therefore refused rather than deleted.
func TestCollectSkipsADeploymentActivatedUnderIt(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 0, 0)

	old := e.version(t, 1)
	e.activate(t, e.version(t, 2))

	// Stand in for the interleaving: retention has decided to drop `old`, and a
	// rollback activates it before the claim runs.
	e.activate(t, old)
	if err := e.svc.claim(t.Context(), e.p, old); !errors.Is(err, store.ErrConflict) {
		t.Fatalf("claiming a deployment that became active = %v, want ErrConflict", err)
	}
	e.mustSurvive(t, old, "a deployment activated during collection")

	stats, err := e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	if stats.DeploymentsDeleted != 1 {
		t.Errorf("deleted %d deployments, want only the one that is no longer served", stats.DeploymentsDeleted)
	}
	e.mustSurvive(t, old, "the deployment the rollback made active")
}

// Collection walks every project, not just the one a request happened to name.
func TestCollectSpansProjects(t *testing.T) {
	e := newEnv(t)
	e.svc.BlobGrace = collectNow
	e.setRetention(t, 0, 0)

	first := e.p
	firstOld := e.version(t, 1)
	e.activate(t, e.version(t, 2))

	second := store.DefaultProject("second")
	second.RetentionCount = 0
	second.RetentionGraceS = 0
	if err := e.db.CreateProject(t.Context(), second); err != nil {
		t.Fatal(err)
	}
	e.p = second
	secondOld := e.version(t, 3)
	e.activate(t, e.version(t, 4))
	e.p = first

	stats, err := e.svc.Collect(t.Context(), false)
	if err != nil {
		t.Fatal(err)
	}
	if stats.DeploymentsDeleted != 2 {
		t.Errorf("deleted %d deployments, want one from each project", stats.DeploymentsDeleted)
	}
	e.mustBeGone(t, firstOld, "the first project's superseded deployment")
	e.p = second
	e.mustBeGone(t, secondOld, "the second project's superseded deployment")
}