package connector_outboundintent

import (
	"context"
	"database/sql/driver"
	"errors"
	"net/http"
	"reflect"
	"testing"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/testing/connectorfixture"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type lifecycleCommitterFunc func(context.Context, outboundintentv1.CommitResultRequest) (outboundintentv1.CommitResultResponse, error)

func (f lifecycleCommitterFunc) CommitOutboundResult(ctx context.Context, request outboundintentv1.CommitResultRequest) (outboundintentv1.CommitResultResponse, error) {
	return f(ctx, request)
}

type unavailableFinalizer struct{ *Repository }

func (s unavailableFinalizer) finalizeWritebackWithCompletion(context.Context, Claim, writebackCompletion) (Command, error) {
	return Command{}, driver.ErrBadConn
}

func TestWorkerNoEffectWritebackSurvivesRestartAndLostRuntimeResponseAgainstTiDB(t *testing.T) {
	db := connectorfixture.OpenTiDB(t)
	connectorfixture.CreateSystemTables(t, db, "_System_IntegrationOutboundCommand", "_System_ConnectorRemoteLink")
	scope := outboundIntentTestScope()
	repository := NewRepository(db)
	accepted, _, err := repository.Accept(t.Context(), scope, outboundIntentTestIntent(scope.CatalogRevision, "no-effect", "never-dispatched"))
	if err != nil {
		t.Fatal(err)
	}
	worker, err := NewWorker(repository, failingPreparer{err: &PreparationError{Code: "invalid_execution_binding"}}, &scriptedCommitter{}, scope, "before-restart")
	if err != nil {
		t.Fatal(err)
	}
	if worked, err := worker.ProcessOne(t.Context()); !worked || err != nil {
		t.Fatalf("rejection checkpoint worked=%t err=%v", worked, err)
	}
	checkpointed, err := repository.loadByID(t.Context(), accepted.CommandID)
	if err != nil || checkpointed.State != StateWriteback || checkpointed.AttemptCount != 0 || checkpointed.RemoteID != "" {
		t.Fatalf("no-effect checkpoint=%#v err=%v", checkpointed, err)
	}
	var original outboundintentv1.CommitResultRequest
	calls := 0
	committer := lifecycleCommitterFunc(func(ctx context.Context, request outboundintentv1.CommitResultRequest) (outboundintentv1.CommitResultResponse, error) {
		calls++
		if calls == 1 {
			original = request
			// Runtime committed this exact result but its response was lost.
			return outboundintentv1.CommitResultResponse{}, driver.ErrBadConn
		}
		if !reflect.DeepEqual(original, request) {
			t.Fatal("writeback retry changed immutable terminal result")
		}
		return (&scriptedCommitter{}).CommitOutboundResult(ctx, request)
	})
	preparer := &countingPreparer{}
	for i := 0; i < 2; i++ {
		// Recreate Repository and Worker: no in-memory no-effect or completion
		// evidence survives the process boundary.
		restarted := NewRepository(db)
		worker, err := NewWorker(restarted, preparer, committer, scope, "after-restart")
		if err != nil {
			t.Fatal(err)
		}
		if worked, err := worker.ProcessOne(t.Context()); !worked || err != nil {
			t.Fatalf("writeback %d worked=%t err=%v", i, worked, err)
		}
		if i == 0 {
			if _, err := db.Exec(`UPDATE _System_IntegrationOutboundCommand SET NextAttemptAt=? WHERE Id=?`, time.Now().UTC().Add(-time.Second), accepted.CommandID); err != nil {
				t.Fatal(err)
			}
		}
	}
	completed, err := repository.loadByID(t.Context(), accepted.CommandID)
	if err != nil || completed.State != StateRejected || completed.AttemptCount != 0 || completed.ClaimToken != "" || calls != 2 || preparer.dispatches != 0 {
		t.Fatalf("completed=%#v calls=%d dispatches=%d err=%v", completed, calls, preparer.dispatches, err)
	}
	decoded, err := outboundintentv1.DecodeResourceNoEffectResultV1(original.CanonicalResult)
	if err != nil || decoded.Reason != outboundintentv1.ResourceNoEffectRejectedBeforeDispatch {
		t.Fatalf("no-effect result=%#v err=%v", decoded, err)
	}
	var links int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorRemoteLink`).Scan(&links); err != nil || links != 0 {
		t.Fatalf("no-effect result created %d links err=%v", links, err)
	}
}

func TestWorkerCompletionReceiptSurvivesOwnerChangeAndRestartAgainstTiDB(t *testing.T) {
	for _, status := range []string{"retired", "owner_deleted_pending_recovery", "missing"} {
		t.Run(status, func(t *testing.T) {
			db := connectorfixture.OpenTiDB(t)
			connectorfixture.CreateSystemTables(t, db, "_System_IntegrationOutboundCommand", "_System_ConnectorRemoteLink")
			repository := NewRepository(db)
			command := claimWritebackFinalizationFixture(t, db, repository)
			calls := 0
			committer := lifecycleCommitterFunc(func(ctx context.Context, request outboundintentv1.CommitResultRequest) (outboundintentv1.CommitResultResponse, error) {
				calls++
				if calls == 1 {
					// Owner retirement/deletion becomes legal after Runtime's
					// completion transaction, before Integration's finalizer.
					var err error
					if status == "missing" {
						_, err = db.Exec(`DELETE FROM _System_ConnectorRemoteLink WHERE LinkKey=?`, command.LinkKey)
					} else {
						_, err = db.Exec(`UPDATE _System_ConnectorRemoteLink SET Status=?, RemoteLocatorActiveKey=NULL WHERE LinkKey=?`, status, command.LinkKey)
					}
					if err != nil {
						t.Fatal(err)
					}
				}
				return (&scriptedCommitter{}).CommitOutboundResult(ctx, request)
			})
			preparer := &countingPreparer{}
			worker, err := NewWorker(unavailableFinalizer{repository}, preparer, committer, command.Scope, "before-restart")
			if err != nil {
				t.Fatal(err)
			}
			if err := worker.commitWriteback(t.Context(), command); !errors.Is(err, driver.ErrBadConn) {
				t.Fatalf("lost finalizer error=%v", err)
			}
			if _, err := db.Exec(`UPDATE _System_IntegrationOutboundCommand SET LeaseExpiresAt=? WHERE Id=?`, time.Now().UTC().Add(-time.Second), command.CommandID); err != nil {
				t.Fatal(err)
			}
			restarted := NewRepository(db)
			worker, err = NewWorker(restarted, preparer, committer, command.Scope, "after-restart")
			if err != nil {
				t.Fatal(err)
			}
			if worked, err := worker.ProcessOne(t.Context()); !worked || err != nil {
				t.Fatalf("restarted finalizer worked=%t err=%v", worked, err)
			}
			completed, err := repository.loadByID(t.Context(), command.CommandID)
			if err != nil || completed.State != StateSucceeded || completed.AttemptCount != 1 || completed.RemoteID != command.RemoteID || preparer.dispatches != 0 || calls != 2 {
				t.Fatalf("completed=%#v dispatches=%d acknowledgements=%d err=%v", completed, preparer.dispatches, calls, err)
			}
			if status == "missing" {
				var count int
				if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorRemoteLink`).Scan(&count); err != nil || count != 0 {
					t.Fatalf("finalizer recreated missing snapshot: count=%d err=%v", count, err)
				}
			} else {
				var storedStatus, mirror string
				if err := db.QueryRow(`SELECT Status, CompatibilityMirrorStatus FROM _System_ConnectorRemoteLink WHERE LinkKey=?`, command.LinkKey).Scan(&storedStatus, &mirror); err != nil || storedStatus != status || mirror != remoteLinkMirrorPending {
					t.Fatalf("finalizer changed owner's snapshot: status=%s mirror=%s err=%v", storedStatus, mirror, err)
				}
			}
		})
	}
}

func TestRepositoryNoEffectCannotSettleExpiredOrProviderCheckpointedCommandAgainstTiDB(t *testing.T) {
	for _, scenario := range []string{"expired", "provider-checkpoint", "wrong-phase"} {
		t.Run(scenario, func(t *testing.T) {
			db := connectorfixture.OpenTiDB(t)
			connectorfixture.CreateSystemTables(t, db, "_System_IntegrationOutboundCommand", "_System_ConnectorRemoteLink")
			repository := NewRepository(db)
			scope := outboundIntentTestScope()
			if _, _, err := repository.Accept(t.Context(), scope, outboundIntentTestIntent(scope.CatalogRevision, "no-effect-negative", "negative")); err != nil {
				t.Fatal(err)
			}
			command, found, err := repository.ClaimNext(t.Context(), scope, "worker", "token", time.Minute)
			if err != nil || !found {
				t.Fatalf("claim found=%t err=%v", found, err)
			}
			if scenario == "expired" {
				_, err = db.Exec(`UPDATE _System_IntegrationOutboundCommand SET LeaseExpiresAt=? WHERE Id=?`, time.Now().UTC().Add(-time.Second), command.CommandID)
			} else {
				command, err = repository.MarkDispatching(t.Context(), Claim{Command: command, Token: command.ClaimToken})
				if err == nil && scenario == "provider-checkpoint" {
					command, err = repository.CheckpointProviderSuccess(t.Context(), Claim{Command: command, Token: command.ClaimToken}, http.StatusCreated,
						outboundIntentTestFinalization(command, "already-created"), "association_create_outcome_uncertain")
				}
			}
			if err != nil {
				t.Fatal(err)
			}
			reason := outboundintentv1.ResourceNoEffectRejectedBeforeDispatch
			if scenario == "provider-checkpoint" {
				reason = outboundintentv1.ResourceNoEffectDefinitiveNoEffect
			}
			if _, err := repository.recordNoEffectResult(t.Context(), Claim{Command: command, Token: command.ClaimToken}, 0, "invalid_request", reason); err == nil {
				t.Fatal("unsafe command converted to no-effect completion")
			}
			var results int
			if err := db.QueryRow(`SELECT COUNT(*) FROM _System_IntegrationOutboundCommand WHERE ResultFormatVersion IS NOT NULL`).Scan(&results); err != nil || results != 0 {
				t.Fatalf("unsafe completion persisted results=%d err=%v", results, err)
			}
		})
	}
}

func TestRepositoryCompletionReceiptMustMatchExactCommandAgainstTiDB(t *testing.T) {
	db := connectorfixture.OpenTiDB(t)
	connectorfixture.CreateSystemTables(t, db, "_System_IntegrationOutboundCommand", "_System_ConnectorRemoteLink")
	repository := NewRepository(db)
	command := claimWritebackFinalizationFixture(t, db, repository)
	request := outboundintentv1.CommitResultRequest{
		IntentID: command.Intent.IntentID, PayloadDigest: command.Intent.PayloadDigest,
		CatalogRevision: command.Scope.CatalogRevision, IntegrationCommandID: command.CommandID,
		RemoteID: "different-command-result",
	}
	response, err := (&scriptedCommitter{}).CommitOutboundResult(t.Context(), request)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_ConnectorRemoteLink SET Status='retired', RemoteLocatorActiveKey=NULL WHERE LinkKey=?`, command.LinkKey); err != nil {
		t.Fatal(err)
	}
	claim := Claim{Command: command, Token: command.ClaimToken}
	if _, err := repository.finalizeWritebackWithCompletion(t.Context(), claim, writebackCompletion{request: request, response: response}); !errors.Is(err, ErrIdentityConflict) {
		t.Fatalf("foreign result receipt finalized command: %v", err)
	}
	stored, err := repository.loadByID(t.Context(), command.CommandID)
	if err != nil || stored.State != StateWriteback {
		t.Fatalf("foreign receipt changed state=%s err=%v", stored.State, err)
	}
}

func TestWorkerOnlyAffirmativeNoEffectCanSettleDispatchedCommandAgainstTiDB(t *testing.T) {
	for _, tc := range []struct {
		name    string
		outcome EffectOutcome
		state   string
	}{
		{"definitive-no-effect", EffectRejectedNoEffect, StateWriteback},
		{"generic-refusal", EffectRejected, StateRejected},
		{"uncertain-provider-outcome", EffectUncertain, StateUncertain},
	} {
		t.Run(tc.name, func(t *testing.T) {
			db := connectorfixture.OpenTiDB(t)
			connectorfixture.CreateSystemTables(t, db, "_System_IntegrationOutboundCommand", "_System_ConnectorRemoteLink")
			repository := NewRepository(db)
			scope := outboundIntentTestScope()
			accepted, _, err := repository.Accept(t.Context(), scope, outboundIntentTestIntent(scope.CatalogRevision, "effect-certainty", "record"))
			if err != nil {
				t.Fatal(err)
			}
			preparer := &countingPreparer{result: EffectResult{Outcome: tc.outcome, FailureCode: "effect_refused"}}
			committer := &scriptedCommitter{}
			worker, err := NewWorker(repository, preparer, committer, scope, "worker")
			if err != nil {
				t.Fatal(err)
			}
			if worked, err := worker.ProcessOne(t.Context()); !worked || err != nil {
				t.Fatalf("effect worked=%t err=%v", worked, err)
			}
			stored, err := repository.loadByID(t.Context(), accepted.CommandID)
			if err != nil || stored.State != tc.state || preparer.dispatches != 1 {
				t.Fatalf("state=%s dispatches=%d err=%v", stored.State, preparer.dispatches, err)
			}
			worked, err := worker.ProcessOne(t.Context())
			if err != nil || worked != (tc.outcome == EffectRejectedNoEffect) || preparer.dispatches != 1 {
				t.Fatalf("next turn worked=%t dispatches=%d err=%v", worked, preparer.dispatches, err)
			}
			if tc.outcome == EffectRejectedNoEffect {
				result, err := outboundintentv1.DecodeResourceNoEffectResultV1(committer.lastRequest.CanonicalResult)
				if err != nil || result.Reason != outboundintentv1.ResourceNoEffectDefinitiveNoEffect {
					t.Fatalf("wrong certainty receipt=%#v err=%v", result, err)
				}
			} else if committer.calls != 0 {
				t.Fatal("non-definitive provider outcome acknowledged as terminal no effect")
			}
		})
	}
}

func TestWorkerRejectionAfterProviderCheckpointPreservesUncertaintyAgainstTiDB(t *testing.T) {
	db := connectorfixture.OpenTiDB(t)
	connectorfixture.CreateSystemTables(t, db, "_System_IntegrationOutboundCommand", "_System_ConnectorRemoteLink")
	repository := NewRepository(db)
	scope := outboundIntentTestScope()
	accepted, _, err := repository.Accept(t.Context(), scope, outboundIntentTestIntent(scope.CatalogRevision, "continuation", "record"))
	if err != nil {
		t.Fatal(err)
	}
	command, found, err := repository.ClaimNext(t.Context(), scope, "worker", "dispatch-token", time.Minute)
	if err != nil || !found {
		t.Fatalf("claim found=%t err=%v", found, err)
	}
	command, err = repository.MarkDispatching(t.Context(), Claim{Command: command, Token: command.ClaimToken})
	if err != nil {
		t.Fatal(err)
	}
	_, err = repository.RecordProviderSuccessForRetry(t.Context(), Claim{Command: command, Token: command.ClaimToken}, http.StatusCreated,
		outboundIntentTestFinalization(command, "remote-created"), time.Now().UTC().Add(time.Second), "association_create_outcome_uncertain")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_IntegrationOutboundCommand SET NextAttemptAt=? WHERE Id=?`, time.Now().UTC().Add(-time.Second), accepted.CommandID); err != nil {
		t.Fatal(err)
	}
	committer := &scriptedCommitter{}
	worker, err := NewWorker(repository, failingPreparer{err: &PreparationError{Code: "invalid_execution_binding"}}, committer, scope, "continuation-worker")
	if err != nil {
		t.Fatal(err)
	}
	if worked, err := worker.ProcessOne(t.Context()); !worked || err != nil {
		t.Fatalf("preparation rejection worked=%t err=%v", worked, err)
	}
	stored, err := repository.loadByID(t.Context(), accepted.CommandID)
	if err != nil || stored.State != StateUncertain || stored.RemoteID != "remote-created" || stored.ResponseStatus != http.StatusCreated || committer.calls != 0 {
		t.Fatalf("partial provider effect lost evidence: %#v commits=%d err=%v", stored, committer.calls, err)
	}
	if _, _, err := repository.Accept(t.Context(), scope, outboundIntentTestIntent(scope.CatalogRevision, "later-intent", "record")); err != nil {
		t.Fatal(err)
	}
	if _, found, err := repository.ClaimNext(t.Context(), scope, "other-worker", "other-token", time.Minute); err != nil || found {
		t.Fatalf("unresolved partial effect released later lane: found=%t err=%v", found, err)
	}
}

func TestRepositoryNoEffectFinalizationRequiresRuntimeAcknowledgementAgainstTiDB(t *testing.T) {
	db := connectorfixture.OpenTiDB(t)
	connectorfixture.CreateSystemTables(t, db, "_System_IntegrationOutboundCommand", "_System_ConnectorRemoteLink")
	repository := NewRepository(db)
	scope := outboundIntentTestScope()
	accepted, _, err := repository.Accept(t.Context(), scope, outboundIntentTestIntent(scope.CatalogRevision, "no-ack", "record"))
	if err != nil {
		t.Fatal(err)
	}
	command, found, err := repository.ClaimNext(t.Context(), scope, "worker", "token", time.Minute)
	if err != nil || !found {
		t.Fatalf("claim found=%t err=%v", found, err)
	}
	if _, err := repository.recordNoEffectResult(t.Context(), Claim{Command: command, Token: command.ClaimToken}, 0, "invalid_request", outboundintentv1.ResourceNoEffectRejectedBeforeDispatch); err != nil {
		t.Fatal(err)
	}
	command, found, err = repository.ClaimWriteback(t.Context(), scope, "worker", "writeback", time.Minute)
	if err != nil || !found {
		t.Fatalf("writeback found=%t err=%v", found, err)
	}
	if _, err := repository.FinalizeWriteback(t.Context(), Claim{Command: command, Token: command.ClaimToken}); !errors.Is(err, ErrIdentityConflict) {
		t.Fatalf("finalized no-effect result without Runtime receipt: %v", err)
	}
	stored, err := repository.loadByID(t.Context(), accepted.CommandID)
	if err != nil || stored.State != StateWriteback {
		t.Fatalf("missing receipt discarded durable retry: state=%s err=%v", stored.State, err)
	}
}