package persistence

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

	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

func TestRuntimeOutboundIntentTiDBClosesAcceptanceAndResultIdempotently(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createRuntimeOutboundOwnerLifecycleTables(t, db)
	if err := VerifyRuntimeOutboundIntentSchema(t.Context(), db); err != nil {
		t.Fatal(err)
	}
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	repository := NewRuntimeOutboundIntentRepository(db, transactions)
	scope := RuntimeOutboundIntentScope{OrgID: "org-1", AppInstallationID: "installation-1", AuthorizedWorkloadID: "integration-1"}
	intent := runtimeOutboundIntentIntegrationTestIntent()
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, duplicate, err := repository.Enqueue(ctx, scope, intent)
		if err == nil && duplicate {
			t.Fatal("first enqueue reported duplicate")
		}
		return err
	}); err != nil {
		t.Fatal(err)
	}
	if generation, err := repository.ReadGeneration(t.Context(), scope); err != nil || generation != 1 {
		t.Fatalf("generation after first enqueue = %d, %v, want 1", generation, err)
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, duplicate, err := repository.Enqueue(ctx, scope, intent)
		if err == nil && !duplicate {
			return errors.New("duplicate enqueue reported a new intent")
		}
		return err
	}); err != nil {
		t.Fatal(err)
	}
	if generation, err := repository.ReadGeneration(t.Context(), scope); err != nil || generation != 1 {
		t.Fatalf("generation after duplicate enqueue = %d, %v, want 1", generation, err)
	}
	conflict := runtimeOutboundIntentIntegrationTestIntent()
	conflict.IntentID = "runtime-intent-conflict"
	conflict.IdempotencyKey = intent.IdempotencyKey
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, duplicate, enqueueErr := repository.Enqueue(ctx, scope, conflict)
		if !duplicate || !errors.Is(enqueueErr, ErrRuntimeOutboundIntentConflict) {
			return fmt.Errorf("idempotency conflict duplicate=%v err=%v", duplicate, enqueueErr)
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if generation, err := repository.ReadGeneration(t.Context(), scope); err != nil || generation != 1 {
		t.Fatalf("generation after conflicting duplicate = %d, %v, want 1", generation, err)
	}
	listed, err := repository.ListPending(t.Context(), scope, intent.CatalogRevision, 1)
	if err != nil || len(listed) != 1 || listed[0].Intent.IntentID != intent.IntentID {
		t.Fatalf("ListPending() = %#v, %v", listed, err)
	}
	acknowledgement := outboundintentv1.AcknowledgeRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
		Outcome: outboundintentv1.AdmissionAccepted, IntegrationCommandID: "integration-command-1",
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, _, err := repository.Acknowledge(ctx, scope, acknowledgement)
		return err
	}); err != nil {
		t.Fatal(err)
	}
	commit := outboundintentv1.CommitResultRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
		IntegrationCommandID: acknowledgement.IntegrationCommandID, RemoteID: "remote-1",
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, err := repository.CommitResultWithProjection(ctx, scope, commit, nil)
		return err
	}); err == nil {
		t.Fatal("direct completion without a projection was accepted")
	}
	var state string
	if err := db.QueryRow(`SELECT AdmissionState FROM _System_RuntimeOutboundIntent WHERE Id = ?`, intent.IntentID).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != RuntimeOutboundIntentAccepted {
		t.Fatalf("direct completion changed state to %q", state)
	}
	projectionCalls := 0
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		duplicate, err := repository.CommitResultWithProjection(ctx, scope, commit, func(context.Context, RuntimeOutboundIntentRow) error {
			projectionCalls++
			return nil
		})
		if err != nil || duplicate {
			return errors.Join(err, errors.New("first result preparation was not unique"))
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if projectionCalls != 1 {
		t.Fatalf("projection calls = %d, want 1", projectionCalls)
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		duplicate, err := repository.CommitResultWithProjection(ctx, scope, commit, func(context.Context, RuntimeOutboundIntentRow) error {
			projectionCalls++
			return errors.New("duplicate result invoked projection")
		})
		if err != nil || !duplicate {
			return errors.Join(err, errors.New("same result was not an idempotent duplicate"))
		}
		_, ackDuplicate, err := repository.Acknowledge(ctx, scope, acknowledgement)
		if err != nil || !ackDuplicate {
			return errors.Join(err, errors.New("accepted acknowledgement did not survive completion"))
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if projectionCalls != 1 {
		t.Fatalf("duplicate projection calls = %d, want 1", projectionCalls)
	}
	drifted := commit
	drifted.RemoteID = "remote-2"
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, err := repository.CommitResultWithProjection(ctx, scope, drifted, func(context.Context, RuntimeOutboundIntentRow) error {
			projectionCalls++
			return nil
		})
		return err
	}); !errors.Is(err, ErrRuntimeOutboundIntentConflict) {
		t.Fatalf("drifted result error = %v, want conflict", err)
	}

	mailIntent := runtimeOutboundMailIntegrationTestIntent()
	mailAcknowledgement := outboundintentv1.AcknowledgeRequest{
		IntentID: mailIntent.IntentID, PayloadDigest: mailIntent.PayloadDigest,
		CatalogRevision: mailIntent.CatalogRevision, Outcome: outboundintentv1.AdmissionAccepted,
		IntegrationCommandID: "integration-mail-command-1",
	}
	mailResult := []byte(`{"version":1,"outcome":"DELIVERED","remoteMessageId":"message-1","senderAddress":"sender@example.com"}`)
	mailResultSum := sha256.Sum256(mailResult)
	mailCommit := outboundintentv1.CommitResultRequest{
		IntentID: mailIntent.IntentID, PayloadDigest: mailIntent.PayloadDigest,
		CatalogRevision: mailIntent.CatalogRevision, IntegrationCommandID: mailAcknowledgement.IntegrationCommandID,
		ResultFormatVersion: outboundintentv1.MailResultFormatV1,
		ResultDigest:        "sha256:" + hex.EncodeToString(mailResultSum[:]),
		CanonicalResult:     mailResult,
	}
	mailProjectionCalls := 0
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		if _, duplicate, err := repository.Enqueue(ctx, scope, mailIntent); err != nil || duplicate {
			return errors.Join(err, errors.New("mail intent was not enqueued once"))
		}
		if _, duplicate, err := repository.Acknowledge(ctx, scope, mailAcknowledgement); err != nil || duplicate {
			return errors.Join(err, errors.New("mail acknowledgement was not recorded once"))
		}
		duplicate, err := repository.CommitResultWithProjection(ctx, scope, mailCommit, func(context.Context, RuntimeOutboundIntentRow) error {
			mailProjectionCalls++
			return nil
		})
		if err != nil || duplicate {
			return errors.Join(err, errors.New("typed mail result was not committed once"))
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	var storedVersion uint32
	var storedDigest string
	var storedResult []byte
	var storedRemoteID sql.NullString
	if err := db.QueryRow(`SELECT ResultFormatVersion, ResultDigest, CanonicalResult, ResultRemoteId
 FROM _System_RuntimeOutboundIntent WHERE Id = ?`, mailIntent.IntentID).
		Scan(&storedVersion, &storedDigest, &storedResult, &storedRemoteID); err != nil {
		t.Fatal(err)
	}
	if storedVersion != mailCommit.ResultFormatVersion || storedDigest != mailCommit.ResultDigest ||
		!bytes.Equal(storedResult, mailCommit.CanonicalResult) || storedRemoteID.Valid {
		t.Fatalf("stored typed result = %d %q %q remote=%#v", storedVersion, storedDigest, storedResult, storedRemoteID)
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		duplicate, err := repository.CommitResultWithProjection(ctx, scope, mailCommit, func(context.Context, RuntimeOutboundIntentRow) error {
			return errors.New("exact typed replay invoked projection")
		})
		if err != nil || !duplicate {
			return errors.Join(err, errors.New("exact typed result was not an idempotent replay"))
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	conflictingResult := []byte(`{"version":1,"outcome":"REJECTED","failureCode":"PROVIDER_REJECTED"}`)
	conflictingSum := sha256.Sum256(conflictingResult)
	conflictingCommit := mailCommit
	conflictingCommit.ResultDigest = "sha256:" + hex.EncodeToString(conflictingSum[:])
	conflictingCommit.CanonicalResult = conflictingResult
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, err := repository.CommitResultWithProjection(ctx, scope, conflictingCommit, func(context.Context, RuntimeOutboundIntentRow) error {
			mailProjectionCalls++
			return nil
		})
		return err
	}); !errors.Is(err, ErrRuntimeOutboundIntentConflict) {
		t.Fatalf("conflicting typed result error = %v, want conflict", err)
	}
	if mailProjectionCalls != 1 {
		t.Fatalf("mail projection calls = %d, want 1", mailProjectionCalls)
	}
}

func TestRuntimeOutboundIntentTiDBRequiresAndSharesCallerTransaction(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createRuntimeOutboundOwnerLifecycleTables(t, db)
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	repository := NewRuntimeOutboundIntentRepository(db, transactions)
	scope := RuntimeOutboundIntentScope{OrgID: "org-1", AppInstallationID: "installation-1", AuthorizedWorkloadID: "integration-1"}
	intent := runtimeOutboundIntentIntegrationTestIntent()

	if _, _, err := repository.Enqueue(t.Context(), scope, intent); !errors.Is(err, ErrRuntimeOutboundIntentTransactionRequired) {
		t.Fatalf("enqueue without caller transaction error = %v, want transaction-required", err)
	}

	rollbackCause := errors.New("subscriber failed after enqueue")
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		if _, duplicate, err := repository.Enqueue(ctx, scope, intent); err != nil || duplicate {
			return errors.Join(err, errors.New("rollback attempt did not enqueue one new intent"))
		}
		return rollbackCause
	}); !errors.Is(err, rollbackCause) {
		t.Fatalf("rollback transaction error = %v, want %v", err, rollbackCause)
	}
	rows, err := repository.ListPending(t.Context(), scope, intent.CatalogRevision, 1)
	if err != nil || len(rows) != 0 {
		t.Fatalf("rolled-back intent remained visible: rows=%#v err=%v", rows, err)
	}
	if generation, generationErr := repository.ReadGeneration(t.Context(), scope); generationErr != nil || generation != 0 {
		t.Fatalf("rolled-back generation = %d, %v, want 0", generation, generationErr)
	}

	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, duplicate, err := repository.Enqueue(ctx, scope, intent)
		if err == nil && duplicate {
			return errors.New("committed attempt saw rolled-back identity as a duplicate")
		}
		return err
	}); err != nil {
		t.Fatal(err)
	}
	rows, err = repository.ListPending(t.Context(), scope, intent.CatalogRevision, 1)
	if err != nil || len(rows) != 1 || rows[0].Intent.IntentID != intent.IntentID {
		t.Fatalf("committed intent = %#v, %v", rows, err)
	}
	if generation, generationErr := repository.ReadGeneration(t.Context(), scope); generationErr != nil || generation != 1 {
		t.Fatalf("committed generation = %d, %v, want 1", generation, generationErr)
	}
}

func createRuntimeOutboundOwnerLifecycleTables(t *testing.T, db *sql.DB) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	names := map[string]bool{"_System_RuntimeOutboundIntent": true, "_System_RuntimeOutboundIntentGeneration": true, "_System_ConnectorRemoteIdentityLane": true}
	for _, definition := range definitions {
		if !names[definition.TableName] {
			continue
		}
		if err := NewSchemaRepository(db, nil).CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatal(err)
		}
		delete(names, definition.TableName)
	}
	if len(names) != 0 {
		t.Fatalf("missing canonical definitions: %v", names)
	}
}

func runtimeOutboundIntentIntegrationTestIntent() outboundintentv1.Intent {
	payload, _ := json.Marshal(outboundintentv1.UpsertDesiredStateV1{
		Version: 1, TargetObject: "remoteContact",
		MappedFields: map[string]json.RawMessage{"email": json.RawMessage(`"a@example.com"`)},
	})
	sum := sha256.Sum256(payload)
	return outboundintentv1.Intent{
		IntentID: "intent-1", IdempotencyKey: "intent-key-1", PayloadFormatVersion: 1,
		PayloadDigest: "sha256:" + hex.EncodeToString(sum[:]), CatalogRevision: 41,
		Operation: outboundintentv1.OperationUpsert, SourceObject: "Contact", SourceRecordID: "contact-1",
		ConnectorOwnerKey: "connector-a", FlowDeveloperName: "Outbound_Flow", FlowActionName: "Push",
		ActionSlotID: "action-slot", DestinationSlotID: "destination-slot",
		Bindings: []outboundintentv1.BindingReference{{
			Kind: catalogv1.BindingConnectorPushContract, OwnerPack: "connector-a", PackInstallationID: "installation-1",
			PackPayloadDigest: "nexus-pack-sha256-v1:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
			BusinessKey:       "connector-a/Outbound_Flow/Push/Contact/destination-slot", DefinitionVersion: 1,
		}},
		CanonicalPayload: payload,
	}
}

func runtimeOutboundMailIntegrationTestIntent() outboundintentv1.Intent {
	payload := []byte(`{"version":1,"authorizedActorId":"user-1","toAddresses":["recipient@example.com"],"subject":"Hello","textBody":"Plain text body"}`)
	sum := sha256.Sum256(payload)
	return outboundintentv1.Intent{
		IntentID: "mail-intent-1", IdempotencyKey: "mail-intent-key-1",
		PayloadFormatVersion: outboundintentv1.MailPayloadFormatV1,
		PayloadDigest:        "sha256:" + hex.EncodeToString(sum[:]), CatalogRevision: 41,
		Operation: outboundintentv1.OperationSendMail, SourceObject: "Contact", SourceRecordID: "contact-1",
		ConnectorOwnerKey: "mail-owner", FlowDeveloperName: "Outbound_Mail", FlowActionName: "Send",
		ActionSlotID: "mail-action-slot", DestinationSlotID: "mail-destination-slot",
		Bindings: []outboundintentv1.BindingReference{{
			Kind: catalogv1.BindingConnectorCalloutProfile, OwnerPack: "mail-owner", PackInstallationID: "installation-1",
			PackPayloadDigest: "nexus-pack-sha256-v1:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
			BusinessKey:       "mail-owner/delivery", DefinitionVersion: 1,
		}},
		CanonicalPayload: payload,
	}
}