package service

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/gitstore"
	"github.com/pingcap/agent-git-service/internal/wikicatalog"
)

func TestWikiReconcileWorkerConsumesDurableRequestOnStartup(t *testing.T) {
	svc, database, ctx, cleanup := newWikiV2InternalTestService(t)
	serverCtx, cancel := context.WithCancel(svc.ServerCtx())
	svc.serverCtx = serverCtx
	defer func() { cancel(); svc.Wg.Wait(); cleanup() }()
	setupRepoForInternalTest(t, svc, database, ctx, "worker", "wiki")
	full := "worker/wiki"
	page, err := svc.PutWikiPage(ctx, full, "home", "old", "seed", "")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := svc.ReconcileWikiV2(ctx, full); err != nil {
		t.Fatal(err)
	}
	svc.testWikiPreparedPublishFailure = func(string, string) error { return errors.New("interrupted") }
	if _, err := svc.PutWikiPage(ctx, full, "home", "new", "update", page.SHA); err == nil {
		t.Fatal("expected interrupted publish")
	}
	svc.testWikiPreparedPublishFailure = nil
	if _, err := svc.KickWikiV2Reconcile(ctx, full); err != nil {
		t.Fatal(err)
	}
	if err := svc.StartWikiReconcileWorker(ctx); err != nil {
		t.Fatal(err)
	}
	if err := svc.StartWikiReconcileWorker(ctx); err != nil {
		t.Fatal(err)
	}
	deadline := time.Now().Add(10 * time.Second)
	for {
		state, err := svc.GetWikiV2State(ctx, full)
		if err != nil {
			t.Fatal(err)
		}
		if state.ReconcileRequestedAt == nil && state.ReconcilerLeaseUntil == nil {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("request not consumed: %+v", state)
		}
		time.Sleep(20 * time.Millisecond)
	}
	got, err := svc.GetWikiPage(ctx, full, "home")
	if err != nil || got.Body != "new" {
		t.Fatalf("page=%+v err=%v", got, err)
	}
}

func TestWikiReconcileRecoversPublicationsWithoutRequest(t *testing.T) {
	svc, database, ctx, cleanup := newWikiV2InternalTestService(t)
	defer cleanup()
	setupRepoForInternalTest(t, svc, database, ctx, "discovery", "wiki")
	full := "discovery/wiki"
	repo, err := svc.GetRepo(ctx, full)
	if err != nil {
		t.Fatal(err)
	}
	catalog, err := svc.WikiCatalogForCtx(ctx)
	if err != nil {
		t.Fatal(err)
	}
	hook := catalog.OnChangeSetCommitted
	catalog.OnChangeSetCommitted = nil
	defer func() { catalog.OnChangeSetCommitted = hook }()
	// Recover without a Git branch, explicit request, or in-memory job.
	if _, err := catalog.ApplyChangeSet(ctx, wikicatalog.ChangeSetRequest{RepositoryID: repo.ID, Source: wikicatalog.SourceREST, Changes: []wikicatalog.Change{{Op: wikicatalog.OpUpsert, Slug: "home", Body: []byte("recovered")}}}); err != nil {
		t.Fatal(err)
	}
	catalog.OnChangeSetCommitted = hook
	if err := svc.recoverWikiPublications(ctx); err != nil {
		t.Fatal(err)
	}
	git, err := svc.GitForCtx(ctx)
	if err != nil {
		t.Fatal(err)
	}
	head, err := git.HeadSHA(ctx, full+".wiki", "master")
	if err != nil {
		t.Fatal(err)
	}
	if err := svc.recoverWikiPublications(ctx); err != nil {
		t.Fatal(err)
	}
	after, err := git.HeadSHA(ctx, full+".wiki", "master")
	if err != nil || after != head {
		t.Fatalf("non-idempotent recovery: %s -> %s, %v", head, after, err)
	}
	body, err := git.ReadFileAtRef(ctx, full+".wiki", "home.md", head)
	if err != nil || string(body) != "recovered" {
		t.Fatalf("body=%q err=%v", body, err)
	}
}

func TestWikiReconcileRejectsCandidateWhenCatalogAdvancedBeforeGit(t *testing.T) {
	svc, database, ctx, cleanup := newWikiV2InternalTestService(t)
	defer cleanup()
	setupRepoForInternalTest(t, svc, database, ctx, "candidate", "wiki")
	full := "candidate/wiki"
	old, err := svc.PutWikiPage(ctx, full, "home", "old", "seed", "")
	if err != nil {
		t.Fatal(err)
	}
	initial, err := svc.ReconcileWikiV2(ctx, full)
	if err != nil {
		t.Fatal(err)
	}
	svc.testWikiPreparedPublishFailure = func(string, string) error { return errors.New("interrupted") }
	defer func() { svc.testWikiPreparedPublishFailure = nil }()
	if _, err := svc.PutWikiPage(ctx, full, "home", "new", "update", old.SHA); err == nil {
		t.Fatal("expected interrupted publish")
	}
	repo, err := svc.GetRepo(ctx, full)
	if err != nil {
		t.Fatal(err)
	}
	result, err := svc.replaceWikiV2Snapshot(ctx, full+".wiki", repo.ID, initial.IndexedCommitSHA, nil, nil, nil, time.Now())
	if err != nil || result.Applied {
		t.Fatalf("admitted catalog-stale candidate: %+v %v", result, err)
	}
	assertWikiV2StateUnchanged(t, database, repo.ID, initial.IndexedCommitSHA, 1)
}

func TestWikiReconcileResumesPartiallyPreparedLegacyChain(t *testing.T) {
	svc, database, ctx, cleanup := newWikiV2InternalTestService(t)
	defer cleanup()
	setupRepoForInternalTest(t, svc, database, ctx, "partial", "wiki")
	full := "partial/wiki"
	if _, err := svc.PutWikiPage(ctx, full, "home", "base", "seed", ""); err != nil {
		t.Fatal(err)
	}
	repo, err := svc.GetRepo(ctx, full)
	if err != nil {
		t.Fatal(err)
	}
	git, err := svc.GitForCtx(ctx)
	if err != nil {
		t.Fatal(err)
	}
	base, err := git.HeadSHA(ctx, full+".wiki", "master")
	if err != nil {
		t.Fatal(err)
	}
	catalog, err := svc.WikiCatalogForCtx(ctx)
	if err != nil {
		t.Fatal(err)
	}
	hook := catalog.OnChangeSetCommitted
	catalog.OnChangeSetCommitted = nil
	defer func() { catalog.OnChangeSetCommitted = hook }()
	var first wikicatalog.ChangeSetResult
	for i, body := range []string{"first", "second"} {
		result, err := catalog.ApplyChangeSet(ctx, wikicatalog.ChangeSetRequest{RepositoryID: repo.ID, Source: wikicatalog.SourceREST, Message: body, Changes: []wikicatalog.Change{{Op: wikicatalog.OpUpsert, Slug: "home", Body: []byte(body)}}})
		if err != nil {
			t.Fatal(err)
		}
		if i == 0 {
			first = result
		}
	}
	// Simulate a crash after recording the first exact recovery commit, but
	// before publishing it. Only the child remains format 0 on the next scan.
	prepared, err := git.BuildCommitFilesAtParent(ctx, full+".wiki", "master", base, first.Message, []gitstore.FileMutation{{Path: "home.md", Content: []byte("first")}}, first.CommittedAt)
	if err != nil {
		t.Fatal(err)
	}
	if err := git.PersistPreparedCommit(ctx, prepared); err != nil {
		t.Fatal(err)
	}
	if err := svc.markWikiChangesetMaterialized(ctx, first.ChangesetID, prepared.SHA); err != nil {
		t.Fatal(err)
	}
	catalog.OnChangeSetCommitted = hook
	if err := svc.recoverWikiPublications(ctx); err != nil {
		t.Fatal(err)
	}
	body, err := git.ReadFileAtRef(ctx, full+".wiki", "home.md", "master")
	if err != nil || string(body) != "second" {
		t.Fatalf("head body=%q err=%v", body, err)
	}
	firstBody, err := git.ReadFileAtRef(ctx, full+".wiki", "home.md", prepared.SHA)
	if err != nil || string(firstBody) != "first" {
		t.Fatalf("first body=%q err=%v", firstBody, err)
	}
}

func TestWikiReconcileWorkerRetriesExpiredLease(t *testing.T) {
	svc, database, ctx, cleanup := newWikiV2InternalTestService(t)
	defer cleanup()
	setupRepoForInternalTest(t, svc, database, ctx, "expired", "wiki")
	full := "expired/wiki"
	if _, err := svc.PutWikiPage(ctx, full, "home", "body", "seed", ""); err != nil {
		t.Fatal(err)
	}
	kick, err := svc.KickWikiV2Reconcile(ctx, full)
	if err != nil {
		t.Fatal(err)
	}
	future := time.Now().UTC().Add(time.Hour)
	if err := database.Model(&db.WikiIndexState{}).Where("repository_id = ?", kick.RepositoryID).UpdateColumn("reconciler_lease_until", future).Error; err != nil {
		t.Fatal(err)
	}
	if err := svc.recoverPendingWikiReconciles(ctx); err != nil {
		t.Fatal(err)
	}
	state, err := svc.GetWikiV2State(ctx, full)
	if err != nil || state.ReconcileRequestedAt == nil {
		t.Fatalf("active lease was stolen: %+v %v", state, err)
	}
	if err := database.Model(&db.WikiIndexState{}).Where("repository_id = ?", kick.RepositoryID).UpdateColumn("reconciler_lease_until", time.Now().Add(-time.Minute)).Error; err != nil {
		t.Fatal(err)
	}
	if err := svc.recoverPendingWikiReconciles(ctx); err != nil {
		t.Fatal(err)
	}
	state, err = svc.GetWikiV2State(ctx, full)
	if err != nil || state.ReconcileRequestedAt != nil || state.ReconcilerLeaseUntil != nil {
		t.Fatalf("expired work not recovered: %+v %v", state, err)
	}
}

func TestWikiReconcileSnapshotPreservesNewRequestAndRejectsLostLease(t *testing.T) {
	svc, database, ctx, cleanup := newWikiV2InternalTestService(t)
	defer cleanup()
	setupRepoForInternalTest(t, svc, database, ctx, "fence", "wiki")
	full := "fence/wiki"
	if _, err := svc.PutWikiPage(ctx, full, "home", "body", "seed", ""); err != nil {
		t.Fatal(err)
	}
	initial, err := svc.ReconcileWikiV2(ctx, full)
	if err != nil {
		t.Fatal(err)
	}
	started := time.Now().Add(-time.Minute)
	kick, err := svc.KickWikiV2Reconcile(ctx, full)
	if err != nil {
		t.Fatal(err)
	}
	var rows []db.WikiPageIndex
	if err := database.Where("repository_id = ?", kick.RepositoryID).Find(&rows).Error; err != nil {
		t.Fatal(err)
	}
	result, err := svc.replaceWikiV2Snapshot(ctx, full+".wiki", kick.RepositoryID, initial.IndexedCommitSHA, rows, nil, nil, started)
	if err != nil || !result.Applied {
		t.Fatalf("replace: %+v %v", result, err)
	}
	state, err := svc.GetWikiV2State(ctx, full)
	if err != nil || state.ReconcileRequestedAt == nil {
		t.Fatalf("newer request was cleared: %+v %v", state, err)
	}
	// A later request can share the SQL timestamp's millisecond with the start.
	// Keep it conservatively instead of acknowledging work we did not observe.
	sameMillisecond := time.Now().UTC().Truncate(time.Millisecond)
	if err := database.Model(&db.WikiIndexState{}).Where("repository_id = ?", kick.RepositoryID).UpdateColumn("reconcile_requested_at", sameMillisecond).Error; err != nil {
		t.Fatal(err)
	}
	if _, err := svc.replaceWikiV2Snapshot(ctx, full+".wiki", kick.RepositoryID, initial.IndexedCommitSHA, rows, nil, nil, sameMillisecond.Add(100*time.Microsecond)); err != nil {
		t.Fatal(err)
	}
	state, err = svc.GetWikiV2State(ctx, full)
	if err != nil || state.ReconcileRequestedAt == nil {
		t.Fatalf("same-millisecond request cleared: %+v %v", state, err)
	}
	oldLease := time.Now().UTC().Add(time.Minute).Truncate(time.Millisecond)
	newLease := oldLease.Add(time.Minute)
	if err := database.Model(&db.WikiIndexState{}).Where("repository_id = ?", kick.RepositoryID).UpdateColumn("reconciler_lease_until", newLease).Error; err != nil {
		t.Fatal(err)
	}
	lostCtx := context.WithValue(ctx, wikiReconcileLeaseKey{}, oldLease)
	_, err = svc.replaceWikiV2Snapshot(lostCtx, full+".wiki", kick.RepositoryID, initial.IndexedCommitSHA, nil, nil, nil, time.Now())
	if !errors.Is(err, ErrConflict) {
		t.Fatalf("lost lease error=%v", err)
	}
	assertWikiV2StateUnchanged(t, database, kick.RepositoryID, initial.IndexedCommitSHA, 1)
}

func TestWikiReconcileWorkerRetainsFailedRequest(t *testing.T) {
	svc, database, ctx, cleanup := newWikiV2InternalTestService(t)
	defer cleanup()
	setupRepoForInternalTest(t, svc, database, ctx, "failure", "wiki")
	full := "failure/wiki"
	repo, err := svc.GetRepo(ctx, full)
	if err != nil {
		t.Fatal(err)
	}
	catalog, err := svc.WikiCatalogForCtx(ctx)
	if err != nil {
		t.Fatal(err)
	}
	hook := catalog.OnChangeSetCommitted
	catalog.OnChangeSetCommitted = nil
	defer func() { catalog.OnChangeSetCommitted = hook }()
	result, err := catalog.ApplyChangeSet(ctx, wikicatalog.ChangeSetRequest{RepositoryID: repo.ID, Source: wikicatalog.SourceREST, Changes: []wikicatalog.Change{{Op: wikicatalog.OpUpsert, Slug: "home", Body: []byte("missing")}}})
	if err != nil {
		t.Fatal(err)
	}
	if err := database.Model(&db.WikiPageRevision{}).Where("changeset_id = ?", result.ChangesetID).UpdateColumn("body_inline", nil).Error; err != nil {
		t.Fatal(err)
	}
	if _, err := svc.KickWikiV2Reconcile(ctx, full); err != nil {
		t.Fatal(err)
	}
	if err := svc.recoverPendingWikiReconciles(ctx); err == nil {
		t.Fatal("missing source body must fail recovery")
	}
	state, err := svc.GetWikiV2State(ctx, full)
	if err != nil || state.ReconcileRequestedAt == nil || state.ReconcilerLeaseUntil == nil || state.IndexedCommitSHA != "" {
		t.Fatalf("failure lost durable request: %+v %v", state, err)
	}
	if err := database.Model(&db.WikiPageRevision{}).Where("changeset_id = ?", result.ChangesetID).UpdateColumn("body_inline", []byte("missing")).Error; err != nil {
		t.Fatal(err)
	}
	if err := database.Model(&db.WikiIndexState{}).Where("repository_id = ?", repo.ID).UpdateColumn("reconciler_lease_until", nil).Error; err != nil {
		t.Fatal(err)
	}
	if err := svc.recoverPendingWikiReconciles(ctx); err != nil {
		t.Fatal(err)
	}
	got, err := svc.GetWikiPage(ctx, full, "home")
	if err != nil || got.Body != "missing" {
		t.Fatalf("retry=%+v %v", got, err)
	}
}