package persistence

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"testing"
	"time"

	database "github.com/nexuscrm/db"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

func TestRuntimeOutboundOwnerLifecycleNoLinkRequiresTerminalReceiptOnTiDB(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	// The canonical fixture starts with no remote-link rows and never creates
	// Integration command tables. Owner protection begins with the first intent.
	createRuntimeOutboundOwnerLifecycleTables(t, db)
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	repository := NewRuntimeOutboundIntentRepository(db, transactions)
	scope := runtimeOutboundOwnerLifecycleScope()
	for _, tc := range []struct {
		name, state string
		operation   outboundintentv1.Operation
		aged        bool
		noEffect    bool
		blocked     bool
	}{
		{name: "pending", state: RuntimeOutboundIntentPending, blocked: true},
		{name: "accepted", state: RuntimeOutboundIntentAccepted, blocked: true},
		{name: "pending_delete", state: RuntimeOutboundIntentPending, operation: outboundintentv1.OperationDelete, blocked: true},
		{name: "unknown_state", state: "RECONCILING", blocked: true},
		{name: "aged_pending", state: RuntimeOutboundIntentPending, aged: true, blocked: true},
		{name: "aged_accepted", state: RuntimeOutboundIntentAccepted, aged: true, blocked: true},
		{name: "completed_resource", state: RuntimeOutboundIntentCompleted},
		{name: "completed_no_effect", state: RuntimeOutboundIntentCompleted, noEffect: true},
		{name: "admission_rejected", state: RuntimeOutboundIntentRejected},
	} {
		t.Run(tc.name, func(t *testing.T) {
			intent := runtimeOutboundOwnerLifecycleIntent(tc.name)
			if tc.operation != "" {
				setRuntimeOutboundOwnerLifecycleOperation(&intent, tc.operation)
			}
			if err := enqueueRuntimeOutboundOwnerLifecycle(t.Context(), transactions, repository, scope, intent); err != nil {
				t.Fatal(err)
			}
			if tc.state != RuntimeOutboundIntentPending {
				ack := runtimeOutboundOwnerLifecycleAcknowledgement(intent)
				if tc.state == RuntimeOutboundIntentRejected {
					ack.Outcome, ack.IntegrationCommandID, ack.RejectionCode = outboundintentv1.AdmissionRejected, "", "INVALID_BINDING"
				}
				if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
					_, _, err := repository.Acknowledge(ctx, scope, ack)
					return err
				}); err != nil {
					t.Fatal(err)
				}
				if tc.state == RuntimeOutboundIntentCompleted {
					request := runtimeOutboundOwnerLifecycleResult(t, intent, tc.noEffect)
					if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
						_, err := repository.CommitResultWithProjection(ctx, scope, request, func(context.Context, RuntimeOutboundIntentRow) error { return nil })
						return err
					}); err != nil {
						t.Fatal(err)
					}
				} else if tc.state == "RECONCILING" {
					// A retained state from an unknown/newer worker is not a receipt.
					if _, err := db.Exec(`UPDATE _System_RuntimeOutboundIntent SET AdmissionState = ? WHERE IntentId = ?`, tc.state, intent.IntentID); err != nil {
						t.Fatal(err)
					}
				}
			}
			if tc.aged {
				if _, err := db.Exec(`UPDATE _System_RuntimeOutboundIntent SET CreatedDate = '2000-01-01 00:00:00', LastModifiedDate = '2000-01-01 00:00:00' WHERE IntentId = ?`, intent.IntentID); err != nil {
					t.Fatal(err)
				}
			}
			err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
				return CheckRuntimeOutboundOwnerQuiescentTx(ctx, transactions.ExtractTx(ctx).(*sql.Tx), intent.SourceObject, intent.SourceRecordID)
			})
			if tc.blocked {
				requireRuntimeOutboundOwnerRetryableConflict(t, err)
			} else if err != nil {
				t.Fatalf("terminal receipt did not release owner protection: %v", err)
			}
			// A different record is independently usable even while this one is busy.
			if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
				return CheckRuntimeOutboundOwnerQuiescentTx(ctx, transactions.ExtractTx(ctx).(*sql.Tx), intent.SourceObject, "unrelated-record")
			}); err != nil {
				t.Fatalf("unrelated record was blocked: %v", err)
			}
		})
	}
}

func TestRuntimeOutboundOwnerLifecycleDeletedOwnerAdmissionAndRollbackOnTiDB(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createRuntimeOutboundOwnerLifecycleTables(t, db)
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	repository := NewRuntimeOutboundIntentRepository(db, transactions)
	scope := runtimeOutboundOwnerLifecycleScope()
	intent := runtimeOutboundOwnerLifecycleIntent("deleted-source")
	rollback := errors.New("roll back the containing record deletion")
	err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		tx := transactions.ExtractTx(ctx).(*sql.Tx)
		if err := SetConnectorRemoteIdentityOwnerDeletedTx(ctx, tx, intent.SourceObject, intent.SourceRecordID, true); err != nil {
			return err
		}
		if _, _, err := repository.Enqueue(ctx, scope, intent); !errors.Is(err, ErrRuntimeOutboundOwnerClosed) {
			return fmt.Errorf("transaction did not observe its own deleted-owner fence: %v", err)
		}
		return rollback
	})
	if !errors.Is(err, rollback) {
		t.Fatalf("deletion rollback: %v", err)
	}
	if err := enqueueRuntimeOutboundOwnerLifecycle(t.Context(), transactions, repository, scope, intent); err != nil {
		t.Fatalf("rolled-back delete permanently closed admission: %v", err)
	}
	deleted := runtimeOutboundOwnerLifecycleIntent("committed-deletion")
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		return SetConnectorRemoteIdentityOwnerDeletedTx(ctx, transactions.ExtractTx(ctx).(*sql.Tx), deleted.SourceObject, deleted.SourceRecordID, true)
	}); err != nil {
		t.Fatal(err)
	}
	for _, operation := range []outboundintentv1.Operation{outboundintentv1.OperationUpsert, outboundintentv1.OperationDelete} {
		other := deleted
		other.IntentID += string(operation)
		other.IdempotencyKey += string(operation)
		setRuntimeOutboundOwnerLifecycleOperation(&other, operation)
		other.ConnectorOwnerKey, other.ActionSlotID = "other-owner", "other-action"
		if err := enqueueRuntimeOutboundOwnerLifecycle(t.Context(), transactions, repository, scope, other); !errors.Is(err, ErrRuntimeOutboundOwnerClosed) {
			t.Fatalf("deleted owner admitted %s from another identity: %v", operation, err)
		}
	}
	otherRecord := deleted
	otherRecord.SourceRecordID += "-other"
	if err := enqueueRuntimeOutboundOwnerLifecycle(t.Context(), transactions, repository, scope, otherRecord); err != nil {
		t.Fatalf("deleted owner blocked another record: %v", err)
	}
}

func TestRuntimeOutboundOwnerLifecycleCompletedDuplicateSurvivesOwnerClosureOnTiDB(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createRuntimeOutboundOwnerLifecycleTables(t, db)
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	repository := NewRuntimeOutboundIntentRepository(db, transactions)
	scope := runtimeOutboundOwnerLifecycleScope()
	intent := runtimeOutboundOwnerLifecycleIntent("completed-deleted")
	request := runtimeOutboundOwnerLifecycleResult(t, intent, false)
	if err := enqueueRuntimeOutboundOwnerLifecycle(t.Context(), transactions, repository, scope, intent); err != nil {
		t.Fatal(err)
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		if _, _, err := repository.Acknowledge(ctx, scope, runtimeOutboundOwnerLifecycleAcknowledgement(intent)); err != nil {
			return err
		}
		_, err := repository.CommitResultWithProjection(ctx, scope, request, func(context.Context, RuntimeOutboundIntentRow) error { return nil })
		return err
	}); err != nil {
		t.Fatal(err)
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		tx := transactions.ExtractTx(ctx).(*sql.Tx)
		if err := CheckRuntimeOutboundOwnerQuiescentTx(ctx, tx, intent.SourceObject, intent.SourceRecordID); err != nil {
			return err
		}
		return SetConnectorRemoteIdentityOwnerDeletedTx(ctx, tx, intent.SourceObject, intent.SourceRecordID, true)
	}); err != nil {
		t.Fatal(err)
	}
	beforeGeneration, err := repository.ReadGeneration(t.Context(), scope)
	if err != nil {
		t.Fatal(err)
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		row, duplicate, err := repository.Enqueue(ctx, scope, intent)
		if err != nil {
			return err
		}
		if !duplicate || row.AdmissionState != RuntimeOutboundIntentCompleted || row.ResultRemoteID != request.RemoteID {
			return fmt.Errorf("completed retry lost its receipt: duplicate=%t state=%s remote=%s", duplicate, row.AdmissionState, row.ResultRemoteID)
		}
		return nil
	}); err != nil {
		t.Fatalf("exact completed retry after owner deletion: %v", err)
	}
	changed := intent
	changed.CanonicalPayload = []byte(`{"changed":"payload"}`)
	sum := sha256.Sum256(changed.CanonicalPayload)
	changed.PayloadDigest = "sha256:" + hex.EncodeToString(sum[:])
	if err := enqueueRuntimeOutboundOwnerLifecycle(t.Context(), transactions, repository, scope, changed); err == nil {
		t.Fatal("closed owner accepted a different payload under the completed intent ID")
	}
	newIntent := intent
	newIntent.IntentID += "-new"
	newIntent.IdempotencyKey += "-new"
	if err := enqueueRuntimeOutboundOwnerLifecycle(t.Context(), transactions, repository, scope, newIntent); !errors.Is(err, ErrRuntimeOutboundOwnerClosed) {
		t.Fatalf("closed owner accepted new work: %v", err)
	}
	afterGeneration, err := repository.ReadGeneration(t.Context(), scope)
	if err != nil || afterGeneration != beforeGeneration {
		t.Fatalf("closed-owner retries authored new work: generation %d -> %d err=%v", beforeGeneration, afterGeneration, err)
	}
}

func TestRuntimeOutboundOwnerLifecycleResultProjectionReverseLockOnTiDB(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createRuntimeOutboundOwnerLifecycleTables(t, db)
	if _, err := db.Exec(`CREATE TABLE outbound_owner_projection_fixture (Id VARCHAR(255) PRIMARY KEY, Deleted BOOLEAN NOT NULL DEFAULT FALSE, RemoteId VARCHAR(255) NOT NULL DEFAULT '')`); err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
	defer cancel()
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	repository := NewRuntimeOutboundIntentRepository(db, transactions)
	scope := runtimeOutboundOwnerLifecycleScope()
	intent := runtimeOutboundOwnerLifecycleIntent("reverse-lock")
	if err := enqueueRuntimeOutboundOwnerLifecycle(ctx, transactions, repository, scope, intent); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO outbound_owner_projection_fixture (Id) VALUES (?)`, intent.SourceRecordID); err != nil {
		t.Fatal(err)
	}
	if err := transactions.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		_, _, err := repository.Acknowledge(txCtx, scope, runtimeOutboundOwnerLifecycleAcknowledgement(intent))
		return err
	}); err != nil {
		t.Fatal(err)
	}

	sourceLocked, projectionEntered := make(chan struct{}), make(chan struct{})
	deletionDone, resultDone := make(chan error, 1), make(chan error, 1)
	type observedCheck struct {
		err     error
		elapsed time.Duration
	}
	checked := make(chan observedCheck, 1)
	go func() {
		deletionDone <- transactions.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
			tx := transactions.ExtractTx(txCtx).(*sql.Tx)
			// Match record deletion's order: it already owns the source row
			// before checking outbound quiescence under the shared owner fence.
			if _, err := tx.ExecContext(txCtx, `UPDATE outbound_owner_projection_fixture SET Deleted = TRUE WHERE Id = ?`, intent.SourceRecordID); err != nil {
				return err
			}
			if err := SetConnectorRemoteIdentityOwnerDeletedTx(txCtx, tx, intent.SourceObject, intent.SourceRecordID, true); err != nil {
				return err
			}
			close(sourceLocked)
			select {
			case <-projectionEntered:
			case <-txCtx.Done():
				return txCtx.Err()
			}
			checkCtx, checkCancel := context.WithTimeout(txCtx, 2*time.Second)
			defer checkCancel()
			started := time.Now()
			err := CheckRuntimeOutboundOwnerQuiescentTx(checkCtx, tx, intent.SourceObject, intent.SourceRecordID)
			checked <- observedCheck{err: err, elapsed: time.Since(started)}
			return err
		})
	}()
	select {
	case <-sourceLocked:
	case err := <-deletionDone:
		t.Fatalf("deletion failed before source lock: %v", err)
	case <-ctx.Done():
		t.Fatal("deletion did not acquire its source row")
	}
	request := runtimeOutboundOwnerLifecycleResult(t, intent, false)
	go func() {
		resultDone <- transactions.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
			_, err := repository.CommitResultWithProjection(txCtx, scope, request, func(projectionCtx context.Context, row RuntimeOutboundIntentRow) error {
				// This callback can run only after the result transaction owns
				// the accepted intent. Its source update waits for deletion.
				close(projectionEntered)
				tx := transactions.ExtractTx(projectionCtx).(*sql.Tx)
				_, err := tx.ExecContext(projectionCtx, `UPDATE outbound_owner_projection_fixture SET RemoteId = ? WHERE Id = ?`, request.RemoteID, row.Intent.SourceRecordID)
				return err
			})
			return err
		})
	}()
	select {
	case <-projectionEntered:
	case err := <-resultDone:
		// A very fast NOWAIT rollback can let the result finish before this
		// observer is scheduled. Preserve its completion for the final checks.
		resultDone <- err
		select {
		case <-projectionEntered:
		default:
			t.Fatalf("result failed before locking its accepted intent: %v", err)
		}
	case <-ctx.Done():
		t.Fatal("result did not reach the controlled source projection")
	}
	var observation observedCheck
	select {
	case observation = <-checked:
	case <-ctx.Done():
		t.Fatal("owner check did not return while result projected onto locked source")
	}
	var deletionErr, resultErr error
	select {
	case deletionErr = <-deletionDone:
	case <-ctx.Done():
		t.Fatal("deletion did not roll back after owner conflict")
	}
	select {
	case resultErr = <-resultDone:
	case <-ctx.Done():
		t.Fatal("result projection did not recover after deletion rollback")
	}
	requireRuntimeOutboundOwnerRetryableConflict(t, observation.err)
	requireRuntimeOutboundOwnerRetryableConflict(t, deletionErr)
	if observation.elapsed >= time.Second {
		t.Fatalf("NOWAIT owner check blocked for %s", observation.elapsed)
	}
	if resultErr != nil {
		t.Fatalf("accepted result could not finish after deletion rollback: %v", resultErr)
	}
	var deleted bool
	var remoteID, state string
	if err := db.QueryRowContext(ctx, `SELECT Deleted, RemoteId FROM outbound_owner_projection_fixture WHERE Id = ?`, intent.SourceRecordID).Scan(&deleted, &remoteID); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRowContext(ctx, `SELECT AdmissionState FROM _System_RuntimeOutboundIntent WHERE IntentId = ?`, intent.IntentID).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if deleted || remoteID != request.RemoteID || state != RuntimeOutboundIntentCompleted {
		t.Fatalf("race left inconsistent state: deleted=%t remote=%q admission=%q", deleted, remoteID, state)
	}
	if err := transactions.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		tx := transactions.ExtractTx(txCtx).(*sql.Tx)
		ownerDeleted, err := LockConnectorRemoteIdentityOwnerFenceTx(txCtx, tx, intent.SourceObject, intent.SourceRecordID)
		if err != nil || ownerDeleted {
			return fmt.Errorf("deletion rollback did not restore owner fence: deleted=%t err=%v", ownerDeleted, err)
		}
		return CheckRuntimeOutboundOwnerQuiescentTx(txCtx, tx, intent.SourceObject, intent.SourceRecordID)
	}); err != nil {
		t.Fatalf("completed result did not release owner guard: %v", err)
	}
}

func runtimeOutboundOwnerLifecycleScope() RuntimeOutboundIntentScope {
	return RuntimeOutboundIntentScope{OrgID: "org-1", AppInstallationID: "installation-1", AuthorizedWorkloadID: "integration-1"}
}

func runtimeOutboundOwnerLifecycleIntent(suffix string) outboundintentv1.Intent {
	intent := runtimeOutboundIntentIntegrationTestIntent()
	intent.IntentID, intent.IdempotencyKey, intent.SourceRecordID = "intent-"+suffix, "key-"+suffix, "source-"+suffix
	return intent
}

func enqueueRuntimeOutboundOwnerLifecycle(ctx context.Context, transactions *TransactionManager, repository *RuntimeOutboundIntentRepository, scope RuntimeOutboundIntentScope, intent outboundintentv1.Intent) error {
	return transactions.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		_, _, err := repository.Enqueue(txCtx, scope, intent)
		return err
	})
}

func runtimeOutboundOwnerLifecycleAcknowledgement(intent outboundintentv1.Intent) outboundintentv1.AcknowledgeRequest {
	return outboundintentv1.AcknowledgeRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
		Outcome: outboundintentv1.AdmissionAccepted, IntegrationCommandID: "command-" + intent.IntentID,
	}
}

func runtimeOutboundOwnerLifecycleResult(t *testing.T, intent outboundintentv1.Intent, noEffect bool) outboundintentv1.CommitResultRequest {
	t.Helper()
	result := outboundintentv1.CommitResultRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
		IntegrationCommandID: "command-" + intent.IntentID, RemoteID: "remote-" + intent.IntentID,
	}
	if noEffect {
		raw, digest, err := outboundintentv1.CanonicalResourceNoEffectResultV1(outboundintentv1.ResourceNoEffectResultV1{
			Version: 1, Outcome: outboundintentv1.ResourceOutcomeNoEffect, Reason: outboundintentv1.ResourceNoEffectRejectedBeforeDispatch,
		})
		if err != nil {
			t.Fatal(err)
		}
		result.RemoteID, result.ResultFormatVersion, result.CanonicalResult, result.ResultDigest = "", outboundintentv1.ResourceNoEffectResultFormatV1, raw, digest
	}
	return result
}

func requireRuntimeOutboundOwnerRetryableConflict(t *testing.T, err error) {
	t.Helper()
	var conflict *apperrors.LockContentionError
	if !errors.As(err, &conflict) || !conflict.RetriableCondition() || conflict.HTTPStatus() != 409 {
		t.Fatalf("owner mutation error=%T %v; expected retryable lock contention", err, err)
	}
}

func setRuntimeOutboundOwnerLifecycleOperation(intent *outboundintentv1.Intent, operation outboundintentv1.Operation) {
	intent.Operation = operation
	if operation == outboundintentv1.OperationDelete {
		intent.CanonicalPayload = []byte(`{"version":1,"remoteId":"known-remote"}`)
		sum := sha256.Sum256(intent.CanonicalPayload)
		intent.PayloadDigest = "sha256:" + hex.EncodeToString(sum[:])
	}
}