package automation

import (
	"bytes"
	"context"
	"errors"
	"testing"

	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type rejectNoEffectSourceAccess struct {
	*stubPersistenceCapture
	reads, writes int
}

func (p *rejectNoEffectSourceAccess) GetRaw(context.Context, string, string) (models.SObject, error) {
	p.reads++
	return nil, errors.New("no-effect completion must not read its missing source")
}

func (p *rejectNoEffectSourceAccess) Update(context.Context, string, string, models.SObject, *models.UserSession) error {
	p.writes++
	return errors.New("no-effect completion must not update its source")
}

// A conclusive no-effect outcome settles an accepted request even when there
// is no local record to project onto. This tests the Core committer boundary;
// ordinary successful results must still use the projection tested below.
func TestRuntimeOutboundResultCommitterTiDBNoEffectClosesWithoutSourceProjection(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorRuntimeOwnerCatalogTables(t, db)
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	repository := persistence.NewRuntimeOutboundIntentRepository(db, transactions)
	view := runtimeOutboundIntentTestView(t)
	intent := runtimeOutboundIntentTestIntent(view)
	scope := persistence.RuntimeOutboundIntentScope{
		OrgID: view.OrgID(), AppInstallationID: view.AppInstallationID(), AuthorizedWorkloadID: "integration-service",
	}
	canonical, digest, err := outboundintentv1.CanonicalResourceNoEffectResultV1(outboundintentv1.ResourceNoEffectResultV1{
		Version: 1, Outcome: outboundintentv1.ResourceOutcomeNoEffect, Reason: outboundintentv1.ResourceNoEffectDefinitiveNoEffect,
	})
	if err != nil {
		t.Fatal(err)
	}
	request := outboundintentv1.CommitResultRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
		IntegrationCommandID: "no-effect-command", ResultFormatVersion: outboundintentv1.ResourceNoEffectResultFormatV1,
		ResultDigest: digest, CanonicalResult: canonical,
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		if _, _, err := repository.Enqueue(ctx, scope, intent); err != nil {
			return err
		}
		_, _, err := repository.Acknowledge(ctx, scope, outboundintentv1.AcknowledgeRequest{
			IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
			Outcome: outboundintentv1.AdmissionAccepted, IntegrationCommandID: request.IntegrationCommandID,
		})
		return err
	}); err != nil {
		t.Fatal(err)
	}
	projection := &rejectNoEffectSourceAccess{stubPersistenceCapture: &stubPersistenceCapture{}}
	committer, err := NewRuntimeOutboundResultCommitter(&FlowExecutor{persistence: projection}, repository)
	if err != nil {
		t.Fatal(err)
	}
	for _, wantDuplicate := range []bool{false, true} {
		if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
			duplicate, err := committer.Commit(ctx, view, scope, request)
			if err == nil && duplicate != wantDuplicate {
				return errors.New("no-effect completion returned incorrect duplicate status")
			}
			return err
		}); err != nil {
			t.Fatal(err)
		}
	}
	if projection.reads != 0 || projection.writes != 0 {
		t.Fatalf("no-effect completion accessed source: reads=%d writes=%d", projection.reads, projection.writes)
	}
	var state, remoteID, gotDigest string
	var format uint32
	var result []byte
	if err := db.QueryRowContext(t.Context(), `SELECT AdmissionState, COALESCE(ResultRemoteId, ''), ResultFormatVersion, ResultDigest, CanonicalResult
 FROM _System_RuntimeOutboundIntent WHERE Id = ?`, intent.IntentID).Scan(&state, &remoteID, &format, &gotDigest, &result); err != nil {
		t.Fatal(err)
	}
	if state != persistence.RuntimeOutboundIntentCompleted || remoteID != "" || format != request.ResultFormatVersion || gotDigest != digest || !bytes.Equal(result, canonical) {
		t.Fatalf("no-effect completion receipt = %q, remote=%q, format=%d, digest=%q, result=%q", state, remoteID, format, gotDigest, result)
	}
}

type transactionCheckingOutboundProjection struct {
	*capturePushMirrorPersistence
	transactions  *persistence.TransactionManager
	inTransaction bool
}

func (p *transactionCheckingOutboundProjection) Update(
	ctx context.Context,
	objectName string,
	id string,
	updates models.SObject,
	user *models.UserSession,
) error {
	p.inTransaction = p.transactions != nil && p.transactions.IsPessimisticTx(ctx)
	return p.capturePushMirrorPersistence.Update(ctx, objectName, id, updates, user)
}

func TestRuntimeOutboundResultCommitterTiDBClosesAcceptedProjectionAtomically(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorRuntimeOwnerCatalogTables(t, db)

	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	repository := persistence.NewRuntimeOutboundIntentRepository(db, transactions)
	view := runtimeOutboundIntentTestView(t)
	intent := runtimeOutboundIntentTestIntent(view)
	scope := persistence.RuntimeOutboundIntentScope{
		OrgID: view.OrgID(), AppInstallationID: view.AppInstallationID(), AuthorizedWorkloadID: "integration-service",
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, duplicate, err := repository.Enqueue(ctx, scope, intent)
		if duplicate {
			return errors.New("first enqueue was a duplicate")
		}
		return err
	}); err != nil {
		t.Fatal(err)
	}

	projection := &transactionCheckingOutboundProjection{
		capturePushMirrorPersistence: &capturePushMirrorPersistence{stubPersistenceCapture: &stubPersistenceCapture{
			rawRecords: map[string]models.SObject{"Contact/003-test": {constants.FieldID: "003-test"}},
		}},
		transactions: transactions,
	}
	executor := &FlowExecutor{persistence: projection}
	committer, err := NewRuntimeOutboundResultCommitter(executor, repository)
	if err != nil {
		t.Fatal(err)
	}
	request := outboundintentv1.CommitResultRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
		IntegrationCommandID: "integration-command-1", RemoteID: "remote-42",
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, err := committer.Commit(ctx, view, scope, request)
		return err
	}); err == nil {
		t.Fatal("result committed before Integration durably accepted the intent")
	}
	if len(projection.calls) != 0 {
		t.Fatal("pre-acceptance result attempted a local projection")
	}

	acknowledgement := outboundintentv1.AcknowledgeRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest, CatalogRevision: intent.CatalogRevision,
		Outcome: outboundintentv1.AdmissionAccepted, IntegrationCommandID: request.IntegrationCommandID,
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, _, err := repository.Acknowledge(ctx, scope, acknowledgement)
		return err
	}); err != nil {
		t.Fatal(err)
	}

	projection.err = errors.New("projection failed")
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		_, err := committer.Commit(ctx, view, scope, request)
		return err
	}); err == nil {
		t.Fatal("projection failure did not abort the result transaction")
	}
	projection.err = nil
	projection.calls = nil
	projection.inTransaction = false
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		duplicate, err := committer.Commit(ctx, view, scope, request)
		if duplicate {
			return errors.New("first successful result commit was a duplicate")
		}
		return err
	}); err != nil {
		t.Fatal(err)
	}
	if !projection.inTransaction || len(projection.calls) != 1 || projection.calls[0].detached {
		t.Fatalf("projection transaction evidence: inTransaction=%v calls=%#v", projection.inTransaction, projection.calls)
	}
	if err := transactions.WithPessimisticTransaction(t.Context(), func(ctx context.Context) error {
		duplicate, err := committer.Commit(ctx, view, scope, request)
		if err == nil && !duplicate {
			return errors.New("replayed completed result was not a duplicate")
		}
		return err
	}); err != nil {
		t.Fatal(err)
	}
	if len(projection.calls) != 1 {
		t.Fatalf("replayed result performed %d projection writes, want 1", len(projection.calls))
	}
}