package automation

import (
	"database/sql"
	"errors"
	"fmt"
	"reflect"
	"testing"
)

// A conversion commits every owner's identity together. Losing the caller's
// response must permit the same conversion to return success without changing
// durable authority or fabricating a second target link. The oracle reads the
// real catalog-backed rows, independently of the transfer's lookup helpers.
func TestConnectorRemoteIdentityMultiOwnerCommittedReplayOnTiDB(t *testing.T) {
	for _, sourceKind := range []string{"links", "evidence", "v1-v2"} {
		t.Run(sourceKind, func(t *testing.T) {
			db, cleanup := openTiDB(t)
			defer cleanup()
			createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
			transfer := seedConnectorMultiOwnerTransfer(t, db, sourceKind)
			commitConnectorMultiOwnerTransfer(t, db, transfer)

			var targetLinks, targetOwners, activeSources int
			if err := db.QueryRowContext(t.Context(), `SELECT COUNT(*), COUNT(DISTINCT ConnectorOwnerKey)
 FROM _System_ConnectorRemoteLink WHERE SourceObject = ? AND SourceRecordId = ? AND Status = 'finalized'`,
				transfer.TargetObject, transfer.TargetRecordId).Scan(&targetLinks, &targetOwners); err != nil {
				t.Fatal(err)
			}
			if err := db.QueryRowContext(t.Context(), `SELECT
 (SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE SourceObject = ? AND SourceRecordId = ? AND Status <> 'retired') +
 (SELECT COUNT(*) FROM _System_ConnectorInboundRemoteIDEvidence WHERE SourceObject = ? AND SourceRecordId = ? AND EvidenceStatus = 'authored')`,
				transfer.SourceObject, transfer.SourceRecordId, transfer.SourceObject, transfer.SourceRecordId).Scan(&activeSources); err != nil {
				t.Fatal(err)
			}
			if targetLinks != 2 || targetOwners != 2 || activeSources != 0 {
				t.Fatalf("committed transfer: target links=%d, target owners=%d, active source authorities=%d; want 2, 2, 0",
					targetLinks, targetOwners, activeSources)
			}

			if sourceKind == "v1-v2" {
				// A later Integration result may add a scoped representation of
				// the same target owner. It is still one logical authority.
				admission := transfer.admissions[0]
				linkKey := seedConnectorIntegrationTransferLink(t, db, admission.TargetIdentity,
					admission.TargetIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, true)
				seedConnectorIntegrationTransferCommand(t, db, "multi-owner-target-result", linkKey, "succeeded",
					admission.TargetIdentity, admission.TargetIntegrationScope, transfer.RemoteId)
				seedConnectorIntegrationTransferLink(t, db, admission.TargetIdentity,
					admission.TargetIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, false)
			}

			committed := readConnectorMultiOwnerAuthorityRows(t, db)
			transfer.TargetRemoteID = transfer.RemoteId // The atomic conversion also committed its mirror.
			// Discarding the first response models a lost response, not a process
			// crash. Each repeat uses a fresh store and transaction.
			for range 2 {
				commitConnectorMultiOwnerTransfer(t, db, transfer)
				if got := readConnectorMultiOwnerAuthorityRows(t, db); !reflect.DeepEqual(got, committed) {
					t.Fatalf("replayed committed transfer changed durable authority: before=%v after=%v", committed, got)
				}
			}
		})
	}
}

func TestConnectorRemoteIdentityMultiOwnerReplayRejectsChangedAuthorityOnTiDB(t *testing.T) {
	for _, change := range []string{
		"omitted admission", "unauthorized target owner", "different target identity",
		"different remote id", "different namespace", "different source mirror", "different target mirror",
		"suspended target owner", "quarantined target owner", "source and target authority", "ambiguous target owner",
	} {
		t.Run(change, func(t *testing.T) {
			db, cleanup := openTiDB(t)
			defer cleanup()
			createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
			transfer := seedConnectorMultiOwnerTransfer(t, db, "links")
			commitConnectorMultiOwnerTransfer(t, db, transfer)
			transfer.TargetRemoteID = transfer.RemoteId
			second := transfer.admissions[1]
			targetKey, err := second.TargetIdentity.LinkKey()
			if err != nil {
				t.Fatal(err)
			}
			exec := func(query string, args ...any) {
				t.Helper()
				if _, err := db.ExecContext(t.Context(), query, args...); err != nil {
					t.Fatal(err)
				}
			}
			switch change {
			case "omitted admission":
				transfer.admissions = transfer.admissions[:1]
			case "unauthorized target owner":
				foreign := second.TargetIdentity
				foreign.ConnectorOwnerKey = "unadmitted-owner"
				seedConnectorIntegrationTransferLink(t, db, foreign, second.TargetIntegrationScope,
					transfer.RemoteIDField, transfer.RemoteId, false)
			case "different target identity":
				transfer.admissions[1].TargetIdentity.ActionSlotID = "other-target-slot"
			case "different remote id":
				exec(`UPDATE _System_ConnectorRemoteLink SET RemoteId = 'different' WHERE LinkKey = ?`, targetKey)
			case "different namespace":
				exec(`UPDATE _System_ConnectorRemoteLink SET ConnectorNamespace = 'different' WHERE LinkKey = ?`, targetKey)
			case "different source mirror":
				transfer.RemoteId = "different"
			case "different target mirror":
				transfer.TargetRemoteID = "different"
			case "suspended target owner":
				exec(`UPDATE _System_ConnectorRemoteLink SET Status = ? WHERE LinkKey = ?`, ConnectorPushLinkStatusSuspendedDrift, targetKey)
			case "quarantined target owner":
				exec(`UPDATE _System_ConnectorRemoteLink SET Status = ? WHERE LinkKey = ?`, ConnectorPushLinkStatusOwnerDeletedPendingRecovery, targetKey)
			case "source and target authority", "ambiguous target owner":
				identity := second.SourceIdentity
				if change == "ambiguous target owner" {
					identity = second.TargetIdentity
				}
				if err := NewSQLConnectorPushDurableStore(db).InsertAuthoredEvidence(t.Context(), ConnectorInboundRemoteIDEvidence{
					ID: "new-authored-evidence", ConnectorOwnerKey: identity.ConnectorOwnerKey, ConnectorNamespace: second.ConnectorNamespace,
					SourceObject: identity.SourceObject, SourceRecordId: identity.SourceRecordID,
					RemoteIDField: transfer.RemoteIDField, RemoteId: transfer.RemoteId,
				}); err != nil {
					t.Fatal(err)
				}
			}
			before := readConnectorMultiOwnerAuthorityRows(t, db)
			tx, store := beginConnectorIntegrationTransfer(t, db)
			if got, err := store.TransferInboundRemoteIdentity(t.Context(), transfer); !errors.Is(err, ErrConnectorPushDurableConflict) {
				t.Fatalf("changed authority replay returned %q, %v; want durable conflict", got, err)
			}
			if err := tx.Rollback(); err != nil {
				t.Fatal(err)
			}
			if after := readConnectorMultiOwnerAuthorityRows(t, db); !reflect.DeepEqual(before, after) {
				t.Fatalf("rejected replay changed durable authority: before=%v after=%v", before, after)
			}
		})
	}
}

func TestConnectorRemoteIdentityMultiOwnerTransferRollbackOnTiDB(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	createConnectorPushDurableCatalogTables(t, db, "_System_ConnectorInboundRemoteIDEvidence")
	transfer := seedConnectorMultiOwnerTransfer(t, db, "evidence")
	before := readConnectorMultiOwnerAuthorityRows(t, db)
	tx, store := beginConnectorIntegrationTransfer(t, db)
	if got, err := store.TransferInboundRemoteIdentity(t.Context(), transfer); err != nil || got != transfer.RemoteId {
		t.Fatalf("transfer before caller rollback = %q, %v", got, err)
	}
	// Failure in a later atomic-flow action rolls back all owners, including
	// source evidence retirement and target publication.
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	if after := readConnectorMultiOwnerAuthorityRows(t, db); !reflect.DeepEqual(before, after) {
		t.Fatalf("caller rollback left a partial transfer: before=%v after=%v", before, after)
	}
	commitConnectorMultiOwnerTransfer(t, db, transfer)
}

func seedConnectorMultiOwnerTransfer(t *testing.T, db *sql.DB, sourceKind string) ConnectorRemoteIdentityTransfer {
	t.Helper()
	transfer := connectorIntegrationTransferFixture("multi-owner")
	second := transfer.admissions[0]
	second.ConnectorNamespace = "second-owner"
	second.SourceIdentity.ConnectorOwnerKey = second.ConnectorNamespace
	second.TargetIdentity.ConnectorOwnerKey = second.ConnectorNamespace
	second.TargetSnapshot = connectorTransferTargetSnapshot(second.TargetIdentity, transfer.RemoteIDField, second.ConnectorNamespace)
	transfer.admissions = append(transfer.admissions, second)
	for index, admission := range transfer.admissions {
		if sourceKind == "evidence" && index == 1 {
			if err := NewSQLConnectorPushDurableStore(db).InsertAuthoredEvidence(t.Context(), ConnectorInboundRemoteIDEvidence{
				ID: "source-second-owner-evidence", ConnectorOwnerKey: admission.SourceIdentity.ConnectorOwnerKey,
				ConnectorNamespace: admission.ConnectorNamespace, SourceObject: transfer.SourceObject,
				SourceRecordId: transfer.SourceRecordId, RemoteIDField: transfer.RemoteIDField, RemoteId: transfer.RemoteId,
			}); err != nil {
				t.Fatal(err)
			}
			continue
		}
		if sourceKind == "v1-v2" && index == 0 {
			linkKey := seedConnectorIntegrationTransferLink(t, db, admission.SourceIdentity,
				admission.SourceIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, true)
			seedConnectorIntegrationTransferCommand(t, db, "multi-owner-source-result", linkKey, "succeeded",
				admission.SourceIdentity, admission.SourceIntegrationScope, transfer.RemoteId)
		}
		seedConnectorIntegrationTransferLink(t, db, admission.SourceIdentity,
			admission.SourceIntegrationScope, transfer.RemoteIDField, transfer.RemoteId, false)
	}
	return transfer
}

func commitConnectorMultiOwnerTransfer(t *testing.T, db *sql.DB, transfer ConnectorRemoteIdentityTransfer) {
	t.Helper()
	tx, store := beginConnectorIntegrationTransfer(t, db)
	if got, err := store.TransferInboundRemoteIdentity(t.Context(), transfer); err != nil || got != transfer.RemoteId {
		t.Fatalf("transfer = %q, %v; want %q", got, err, transfer.RemoteId)
	}
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}
}

func readConnectorMultiOwnerAuthorityRows(t *testing.T, db *sql.DB) map[string][][]string {
	t.Helper()
	result := make(map[string][][]string)
	for _, table := range []string{"_System_ConnectorRemoteLink", "_System_ConnectorInboundRemoteIDEvidence"} {
		rows, err := db.QueryContext(t.Context(), "SELECT * FROM "+table+" ORDER BY Id")
		if err != nil {
			t.Fatal(err)
		}
		columns, err := rows.Columns()
		if err != nil {
			_ = rows.Close()
			t.Fatal(err)
		}
		for rows.Next() {
			values := make([]any, len(columns))
			pointers := make([]any, len(columns))
			for i := range values {
				pointers[i] = &values[i]
			}
			if err := rows.Scan(pointers...); err != nil {
				_ = rows.Close()
				t.Fatal(err)
			}
			row := make([]string, len(columns))
			for i, value := range values {
				if bytes, ok := value.([]byte); ok {
					row[i] = string(bytes)
				} else {
					row[i] = fmt.Sprint(value)
				}
			}
			result[table] = append(result[table], row)
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			t.Fatal(err)
		}
		if err := rows.Close(); err != nil {
			t.Fatal(err)
		}
	}
	return result
}