package persistence

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

	"github.com/go-sql-driver/mysql"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

var (
	ErrRuntimeOutboundOwnerBusy   = errors.New("record has an unsettled outbound intent")
	ErrRuntimeOutboundOwnerClosed = errors.New("outbound intent source owner is deleted")
)

// RuntimeOutboundIdentity identifies the local source whose Runtime-owned
// receipts are examined before retiring or quarantining its remote identity.
type RuntimeOutboundIdentity struct {
	SourceObject, SourceRecordID, ConnectorOwnerKey, ActionSlotID, DestinationSlotID string
}

func ConnectorRemoteIdentityOwnerFenceHash(object, recordID string) (string, error) {
	object, recordID = strings.ToLower(strings.TrimSpace(object)), strings.TrimSpace(recordID)
	if object == "" || recordID == "" {
		return "", fmt.Errorf("connector remote identity owner fence requires object and record id")
	}
	sum := sha256.Sum256([]byte(strings.Join([]string{"connector-remote-identity-owner-fence-v1", object, recordID}, "\x00")))
	return hex.EncodeToString(sum[:]), nil
}

func lockRuntimeOutboundLane(ctx context.Context, tx *sql.Tx, key string) error {
	if tx == nil || ctx == nil {
		return ErrRuntimeOutboundIntentTransactionRequired
	}
	_, err := tx.ExecContext(ctx, `INSERT INTO _System_ConnectorRemoteIdentityLane
 (RemoteLocatorHash, CreatedDate, LastModifiedDate) VALUES (?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
 ON DUPLICATE KEY UPDATE RemoteLocatorHash = VALUES(RemoteLocatorHash)`, key)
	return err
}

// LockConnectorRemoteIdentityOwnerFenceTx is shared by record deletion and
// immutable outbound enqueue. Holding the existing row covers the absence of
// any RemoteLink or intent; a missing result row is never used as a gap lock.
func LockConnectorRemoteIdentityOwnerFenceTx(ctx context.Context, tx *sql.Tx, object, recordID string) (bool, error) {
	key, err := ConnectorRemoteIdentityOwnerFenceHash(object, recordID)
	if err != nil {
		return false, err
	}
	if err := lockRuntimeOutboundLane(ctx, tx, key); err != nil {
		return false, err
	}
	var deleted bool
	err = tx.QueryRowContext(ctx, `SELECT OwnerDeleted FROM _System_ConnectorRemoteIdentityLane WHERE RemoteLocatorHash = ? FOR UPDATE`, key).Scan(&deleted)
	return deleted, err
}

func SetConnectorRemoteIdentityOwnerDeletedTx(ctx context.Context, tx *sql.Tx, object, recordID string, deleted bool) error {
	if _, err := LockConnectorRemoteIdentityOwnerFenceTx(ctx, tx, object, recordID); err != nil {
		return err
	}
	key, err := ConnectorRemoteIdentityOwnerFenceHash(object, recordID)
	if err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, `UPDATE _System_ConnectorRemoteIdentityLane SET OwnerDeleted = ?, LastModifiedDate = CURRENT_TIMESTAMP WHERE RemoteLocatorHash = ?`, deleted, key)
	return err
}

func validateRuntimeOutboundIdentity(identity RuntimeOutboundIdentity) error {
	for _, value := range []string{identity.SourceObject, identity.SourceRecordID, identity.ConnectorOwnerKey, identity.ActionSlotID, identity.DestinationSlotID} {
		if value == "" || strings.TrimSpace(value) != value {
			return ErrRuntimeOutboundIntentConflict
		}
	}
	return nil
}

func guardRuntimeOutboundEnqueue(ctx context.Context, tx *sql.Tx, intent outboundintentv1.Intent) error {
	if intent.Operation != outboundintentv1.OperationUpsert && intent.Operation != outboundintentv1.OperationDelete {
		return nil
	}
	deleted, err := LockConnectorRemoteIdentityOwnerFenceTx(ctx, tx, intent.SourceObject, intent.SourceRecordID)
	if err != nil {
		return err
	}
	if deleted {
		return ErrRuntimeOutboundOwnerClosed
	}

	return nil
}

func runtimeOutboundLifecycleReadError(err error) error {
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) && mysqlErr.Number == 3572 {
		return apperrors.NewLockContentionError(fmt.Errorf("%w: result or admission is committing", ErrRuntimeOutboundOwnerBusy))
	}
	return err
}

// CheckRuntimeOutboundOwnerQuiescentTx performs a current read while the
// caller owns the record fence. NOWAIT avoids a reverse wait against result
// commit, which locks its intent before projecting onto the local record.
// Pending also blocks: Integration commits acceptance before acknowledging it.
func CheckRuntimeOutboundOwnerQuiescentTx(ctx context.Context, tx *sql.Tx, object, recordID string) error {
	if _, err := LockConnectorRemoteIdentityOwnerFenceTx(ctx, tx, object, recordID); err != nil {
		return err
	}
	var id string
	err := tx.QueryRowContext(ctx, `SELECT Id FROM _System_RuntimeOutboundIntent
 WHERE SourceObject = ? AND SourceRecordId = ? AND BINARY SourceRecordId = BINARY ?
 AND Operation IN ('UPSERT', 'DELETE') AND AdmissionState NOT IN ('COMPLETED', 'REJECTED')
 LIMIT 1 FOR UPDATE NOWAIT`, object, recordID, recordID).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return runtimeOutboundLifecycleReadError(err)
	}
	return apperrors.NewLockContentionError(fmt.Errorf("%w: %s", ErrRuntimeOutboundOwnerBusy, id))
}

// RuntimeOutboundIdentityReceiptsTx reads only Runtime-owned immutable facts.
// The caller holds the owner fence; locked terminal receipts are stable proof
// even after Integration's private delivery history has been pruned.
func RuntimeOutboundIdentityReceiptsTx(ctx context.Context, tx *sql.Tx, identity RuntimeOutboundIdentity) ([]RuntimeOutboundIntentRow, error) {
	if err := validateRuntimeOutboundIdentity(identity); err != nil {
		return nil, err
	}
	rows, err := tx.QueryContext(ctx, runtimeOutboundIntentSelect+`
 WHERE SourceObject = ? AND BINARY SourceObject = BINARY ?
 AND SourceRecordId = ? AND BINARY SourceRecordId = BINARY ?
 AND ConnectorOwnerKey = ? AND BINARY ConnectorOwnerKey = BINARY ?
 AND ActionSlotId = ? AND BINARY ActionSlotId = BINARY ?
 AND DestinationSlotId = ? AND BINARY DestinationSlotId = BINARY ?
 AND Operation IN ('UPSERT', 'DELETE') ORDER BY Id FOR UPDATE NOWAIT`,
		identity.SourceObject, identity.SourceObject, identity.SourceRecordID, identity.SourceRecordID,
		identity.ConnectorOwnerKey, identity.ConnectorOwnerKey, identity.ActionSlotID, identity.ActionSlotID,
		identity.DestinationSlotID, identity.DestinationSlotID)
	if err != nil {
		return nil, runtimeOutboundLifecycleReadError(err)
	}
	defer rows.Close()
	var result []RuntimeOutboundIntentRow
	for rows.Next() {
		row, err := scanRuntimeOutboundIntent(rows)
		if err != nil {
			return nil, err
		}
		result = append(result, row)
	}
	return result, runtimeOutboundLifecycleReadError(rows.Err())
}