package persistence

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"reflect"
	"strings"
	"testing"
	"time"

	database "github.com/nexuscrm/db"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
)

func TestRuntimePlatformApplicationResultTiDBOwnsAtomicTerminalFact(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	createRuntimePlatformCommandTestTable(t, db, "MEDIUMBLOB")
	createRuntimePlatformCommandOwnershipTable(t, db)
	if _, err := db.Exec(`CREATE TABLE ApplicationResultEffect (
		CommandId VARCHAR(255) PRIMARY KEY,
		Value VARCHAR(255) NOT NULL
	)`); err != nil {
		t.Fatal(err)
	}

	control, err := NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID:         "activate-application-result-owner",
		WorkFamily:          RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "application-result-owner",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("a", 64),
	})
	if err != nil {
		t.Fatal(err)
	}
	scope := RuntimeWorkScope{
		OrgID: "org-1", AppInstallationID: "installation-1",
		RuntimeCapabilityID: commandsemantic.CapabilityName,
	}
	trustedScope := RuntimePlatformCommandTrustedScope{
		OrgID: scope.OrgID, AppInstallationID: scope.AppInstallationID,
		CallerWorkloadID: "integration-1", RuntimeCapabilityID: commandsemantic.CapabilityName,
	}
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	store, err := NewRuntimePlatformCommandExecutionStore(
		db,
		runtimePlatformCommandIntegrationTransactionOwner{transactions},
		owner,
		scope,
		30*time.Second,
		3,
	)
	if err != nil {
		t.Fatal(err)
	}
	repository := NewRuntimePlatformCommandRepository(db)

	appliedRequest := acceptRuntimePlatformApplicationCommandForTest(
		t, repository, scope, trustedScope, "application-command-applied", "execution-applied",
	)
	if err := store.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	appliedLease, found, err := store.Claim(t.Context())
	if err != nil || !found || appliedLease.Row.CommandID != appliedRequest.CommandID {
		t.Fatalf("Claim() = %#v found=%t err=%v", appliedLease, found, err)
	}
	appliedCompletion := RuntimePlatformCommandApplicationCompletion{
		Outcome: applicationresultsemantic.OutcomeApplied,
		Body:    json.RawMessage(`{"executionId":"execution-applied","outputs":{"recordId":"record-1"}}`),
	}
	if err := store.ExecuteApplication(t.Context(), appliedLease, func(txCtx context.Context) (RuntimePlatformCommandApplicationCompletion, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok || tx == nil {
			return RuntimePlatformCommandApplicationCompletion{}, errors.New("application transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO ApplicationResultEffect (CommandId, Value) VALUES (?, ?)`,
			appliedLease.Row.CommandID,
			"committed",
		); err != nil {
			return RuntimePlatformCommandApplicationCompletion{}, err
		}
		return appliedCompletion, nil
	}); err != nil {
		t.Fatal(err)
	}

	readCtx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
	defer cancel()
	applied, err := repository.GetApplicationResult(readCtx, trustedScope, appliedRequest)
	if err != nil || applied.State != applicationresultsemantic.ResultTerminal ||
		applied.Result == nil || applied.Result.Outcome != applicationresultsemantic.OutcomeApplied {
		t.Fatalf("GetApplicationResult() = %#v, %v", applied, err)
	}
	encodedApplied, err := encodeRuntimePlatformCommandApplicationCompletion(appliedCompletion)
	if err != nil {
		t.Fatal(err)
	}
	wantResultDigest, err := applicationresultsemantic.ComputeResultDigest(encodedApplied)
	if err != nil {
		t.Fatal(err)
	}
	if applied.Result.ResultDigest != wantResultDigest ||
		applicationresultsemantic.ValidateApplicationResult(*applied.Result) != nil {
		t.Fatalf("applied result = %#v, want digest %q", applied.Result, wantResultDigest)
	}
	restarted := NewRuntimePlatformCommandRepository(db)
	again, err := restarted.GetApplicationResult(readCtx, trustedScope, appliedRequest)
	if err != nil || !reflect.DeepEqual(again, applied) {
		t.Fatalf("restart result = %#v, %v; want %#v", again, err, applied)
	}
	wrongScope := trustedScope
	wrongScope.AppInstallationID = "installation-other"
	if _, err := restarted.GetApplicationResult(readCtx, wrongScope, appliedRequest); !runtimePlatformApplicationResultFailureIs(err, commonsemantic.FailureMissing) {
		t.Fatalf("wrong-scope result error = %v, want MISSING", err)
	}

	rejectedRequest := acceptRuntimePlatformApplicationCommandForTest(
		t, repository, scope, trustedScope, "application-command-rejected", "execution-rejected",
	)
	rejectedLease, found, err := store.Claim(t.Context())
	if err != nil || !found || rejectedLease.Row.CommandID != rejectedRequest.CommandID {
		t.Fatalf("rejected Claim() = %#v found=%t err=%v", rejectedLease, found, err)
	}
	if err := store.ExecuteApplication(t.Context(), rejectedLease, func(txCtx context.Context) (RuntimePlatformCommandApplicationCompletion, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok || tx == nil {
			return RuntimePlatformCommandApplicationCompletion{}, errors.New("application transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO ApplicationResultEffect (CommandId, Value) VALUES (?, ?)`,
			rejectedLease.Row.CommandID,
			"must-rollback",
		); err != nil {
			return RuntimePlatformCommandApplicationCompletion{}, err
		}
		return RuntimePlatformCommandApplicationCompletion{
			Outcome:     applicationresultsemantic.OutcomeRejected,
			FailureCode: applicationresultsemantic.FailureApplicationRejected,
			Body:        json.RawMessage(`{"rejected":true}`),
		}, nil
	}); err != nil {
		t.Fatal(err)
	}
	var rejectedEffects int
	if err := db.QueryRow(`SELECT COUNT(*) FROM ApplicationResultEffect WHERE CommandId = ?`, rejectedRequest.CommandID).
		Scan(&rejectedEffects); err != nil {
		t.Fatal(err)
	}
	if rejectedEffects != 0 {
		t.Fatal("rejected application effect committed before its terminal result")
	}
	rejected, err := repository.GetApplicationResult(readCtx, trustedScope, rejectedRequest)
	if err != nil || rejected.Result == nil ||
		rejected.Result.Outcome != applicationresultsemantic.OutcomeRejected ||
		rejected.Result.FailureCode != applicationresultsemantic.FailureApplicationRejected {
		t.Fatalf("rejected result = %#v, %v", rejected, err)
	}

	pendingRequest := acceptRuntimePlatformApplicationCommandForTest(
		t, repository, scope, trustedScope, "application-command-pending", "execution-pending",
	)
	pendingLease, found, err := store.Claim(t.Context())
	if err != nil || !found || pendingLease.Row.CommandID != pendingRequest.CommandID {
		t.Fatalf("pending Claim() = %#v found=%t err=%v", pendingLease, found, err)
	}
	effectErr := errors.New("transient application failure")
	err = store.ExecuteApplication(t.Context(), pendingLease, func(txCtx context.Context) (RuntimePlatformCommandApplicationCompletion, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok || tx == nil {
			return RuntimePlatformCommandApplicationCompletion{}, errors.New("application transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO ApplicationResultEffect (CommandId, Value) VALUES (?, ?)`,
			pendingLease.Row.CommandID,
			"must-rollback",
		); err != nil {
			return RuntimePlatformCommandApplicationCompletion{}, err
		}
		return RuntimePlatformCommandApplicationCompletion{}, effectErr
	})
	if !errors.Is(err, effectErr) {
		t.Fatalf("pending ExecuteApplication() error = %v, want %v", err, effectErr)
	}
	var pendingEffects int
	if err := db.QueryRow(`SELECT COUNT(*) FROM ApplicationResultEffect WHERE CommandId = ?`, pendingRequest.CommandID).
		Scan(&pendingEffects); err != nil {
		t.Fatal(err)
	}
	if pendingEffects != 0 {
		t.Fatal("ordinary application error committed a business effect")
	}
	pending, err := repository.GetApplicationResult(readCtx, trustedScope, pendingRequest)
	if err != nil || pending.State != applicationresultsemantic.ResultPending || pending.Result != nil {
		t.Fatalf("pending result = %#v, %v", pending, err)
	}

	if err := store.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand SET ResultPayload = ? WHERE CommandId = ?`,
		encodeRuntimePlatformCommandResult([]byte(`{}`)),
		appliedRequest.CommandID,
	); err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(t.Context()); err == nil {
		t.Fatal("CheckReady() accepted a malformed committed v3 terminal envelope")
	}
}

func acceptRuntimePlatformApplicationCommandForTest(
	t *testing.T,
	repository *RuntimePlatformCommandRepository,
	workScope RuntimeWorkScope,
	trustedScope RuntimePlatformCommandTrustedScope,
	commandID string,
	executionID string,
) applicationresultsemantic.GetResultRequest {
	t.Helper()
	canonicalInputs := []byte(`{"recordId":"record-1"}`)
	inputDigest, err := flowcommandsemanticv3.ComputeInputDigest(
		canonicalInputs,
		flowcommandsemanticv3.SuppressExternalEffects,
	)
	if err != nil {
		t.Fatal(err)
	}
	ref := applicationresultsemantic.ExecutionRef{
		ExecutionID: executionID, PackGenerationID: "pack-generation-1",
		ServingEpochDigest: "sha256:" + strings.Repeat("1", 64),
		DefinitionRef:      "flow-definition-1", DefinitionVersion: "flow-version-1",
		InputDigest: inputDigest,
	}
	payload, err := flowcommandsemanticv3.Encode(flowcommandsemanticv3.Payload{
		ExecutionRef: ref, CanonicalInputs: canonicalInputs,
		ExternalEffectsMode: flowcommandsemanticv3.SuppressExternalEffects,
	})
	if err != nil {
		t.Fatal(err)
	}
	executionScopeDigest, err := workScope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	trustedScopeDigest, err := trustedScope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	_, duplicate, err := repository.Accept(t.Context(), RuntimePlatformCommandAcceptance{
		RowID: commandID + "-row", OrgID: workScope.OrgID,
		AppInstallationID:    workScope.AppInstallationID,
		CallerWorkloadID:     trustedScope.CallerWorkloadID,
		RuntimeCapabilityID:  workScope.RuntimeCapabilityID,
		MetadataBinding:      `{"binding":"exact"}`,
		ExecutionScopeDigest: executionScopeDigest,
		ScopeDigest:          trustedScopeDigest,
		CommandID:            commandID,
		IdempotencyKey:       commandID + "-idempotency",
		CommandKind:          flowcommandsemanticv3.CommandKind,
		PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
		RequestDigest:        "sha256:" + strings.Repeat("2", 64),
		Payload:              payload,
		Receipts:             []byte(`{"catalogRevision":7,"securityEpoch":11}`),
	})
	if err != nil || duplicate {
		t.Fatalf("Accept(%s) duplicate=%t err=%v", commandID, duplicate, err)
	}
	refDigest, err := applicationresultsemantic.ExecutionRefDigest(ref)
	if err != nil {
		t.Fatal(err)
	}
	return applicationresultsemantic.GetResultRequest{
		CommandID: commandID, ExecutionRefDigest: refDigest,
	}
}

func runtimePlatformApplicationResultFailureIs(
	err error,
	category commonsemantic.FailureCategory,
) bool {
	var failure *applicationresultsemantic.Failure
	return errors.As(err, &failure) && failure.Category == category
}