package connector_outboundintent

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"sort"
	"strings"

	connectorv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

// preparedLegacyRemoteLink is a process-local proof of an installed-authority
// read, not a new durable authority. V1 remains the retirement/deletion fence
// until provider success can atomically publish the scoped V2 link.
type preparedLegacyRemoteLink struct {
	scope                    Scope
	commandID, payloadDigest string
	link                     remoteLinkIdentity
	locator                  legacyRemoteLocator
}

type legacyRemoteLocator struct{ namespace, hash, activeKey string }

func legacyRemoteLinkKey(command Command) string {
	i := command.Intent
	return "cpush_link_v1:" + digestStrings(i.ConnectorOwnerKey, i.ActionSlotID,
		connectorPushActionType, i.SourceObject, i.SourceRecordID, i.DestinationSlotID)
}

func readLegacyRemoteLocator(ctx context.Context, db interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}, key string, lock bool) (legacyRemoteLocator, error) {
	var locator legacyRemoteLocator
	query := `SELECT COALESCE(ConnectorNamespace, ''),
 COALESCE(RemoteLocatorHash, ''), COALESCE(RemoteLocatorActiveKey, '')
 FROM _System_ConnectorRemoteLink WHERE LinkKey = ? LIMIT 1`
	if lock {
		query += " FOR UPDATE"
	}
	err := db.QueryRowContext(ctx, query, key).
		Scan(&locator.namespace, &locator.hash, &locator.activeKey)
	return locator, err
}

func (p *ProviderPreparer) prepareLegacyRemoteLink(ctx context.Context, command Command, link remoteLinkIdentity, resolved ResolvedAuthority) (*preparedLegacyRemoteLink, error) {
	if p.authority.Scope() != command.Scope || command.LinkKey != outboundLinkKey(command.Scope, command.Intent) ||
		link.LinkKey != legacyRemoteLinkKey(command) || link.RemoteID == "" || link.RemoteID != strings.TrimSpace(link.RemoteID) ||
		link.CreatedPackInstallationID == "" || link.LastValidatedPackInstallationID != resolved.Profile.InstalledPackID {
		return nil, ErrIdentityConflict
	}
	mappingFingerprint, bodyFingerprint, err := legacyExecutionFingerprints(ctx, p.db, resolved)
	if err != nil {
		return nil, err
	}
	// V1 hashes the actual canonical mapping rows and the authored body shape,
	// whereas V2 hashes sealed binding references. Compare in the V1 codec;
	// never reinterpret or overwrite a stored fingerprint to make it match.
	legacyResolved := resolved
	legacyResolved.MappingFingerprint, legacyResolved.BodyFingerprint = mappingFingerprint, bodyFingerprint
	legacyCommand := command
	legacyCommand.LinkKey = link.LinkKey
	if err := validateRemoteLinkExecutionSnapshot(link, legacyCommand, resolved.Contract, legacyResolved); err != nil {
		return nil, err
	}
	locator, err := readLegacyRemoteLocator(ctx, p.db, link.LinkKey, false)
	if err != nil {
		return nil, err
	}
	expectedLocator := digestStrings(command.Intent.ConnectorOwnerKey, command.Intent.DestinationSlotID, link.RemoteID)
	if locator.namespace != command.Intent.ConnectorOwnerKey || locator.hash != expectedLocator || locator.activeKey != expectedLocator {
		return nil, ErrIdentityConflict
	}
	if err := rejectForeignLegacyCommands(ctx, p.db, command); err != nil {
		return nil, err
	}
	return &preparedLegacyRemoteLink{scope: command.Scope, commandID: command.CommandID,
		payloadDigest: command.Intent.PayloadDigest, link: link, locator: locator}, nil
}

// V1 has no scope axis. A different scoped command for the same logical
// identity is denial evidence, never authority to adopt or retarget this link.
// Dispatch and first acceptance serialize Command -> V1 Link. The post-link
// current read observes committed foreign commands, but NOWAIT makes the V1
// holder abort instead of introducing a V1 -> locked foreign Command wait edge.
func rejectForeignLegacyCommands(ctx context.Context, db rowQuerier, command Command) error {
	var present int
	i := command.Intent
	err := db.QueryRowContext(ctx, `SELECT 1 FROM _System_IntegrationOutboundCommand
 WHERE ConnectorOwnerKey = ? AND ActionSlotId = ? AND SourceObject = ?
 AND SourceRecordId = ? AND DestinationSlotId = ? AND Operation = 'UPSERT'
 AND NOT (BINARY LinkKey = BINARY ? AND BINARY OrgId = BINARY ?
 AND BINARY AppInstallationId = BINARY ? AND BINARY AuthorizedWorkloadId = BINARY ?)
 LIMIT 1 FOR UPDATE NOWAIT`, i.ConnectorOwnerKey, i.ActionSlotID, i.SourceObject, i.SourceRecordID, i.DestinationSlotID,
		command.LinkKey, command.Scope.OrgID, command.Scope.AppInstallationID, command.Scope.AuthorizedWorkloadID).Scan(&present)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	return newDurableIdentityConflict(command, "legacy_remote_link_fence", "ForeignScopeHistory")
}

// An acceptance that reaches V1 after dispatch published its durable in-flight
// state must not become new foreign-scope evidence during provider I/O. Pending
// foreign work accepted first remains legal evidence that dispatch will reject.
func rejectForeignLegacyDispatches(ctx context.Context, db rowQuerier, command Command) error {
	var present int
	i := command.Intent
	err := db.QueryRowContext(ctx, `SELECT 1 FROM _System_IntegrationOutboundCommand
 WHERE ConnectorOwnerKey = ? AND ActionSlotId = ? AND SourceObject = ?
 AND SourceRecordId = ? AND DestinationSlotId = ? AND Operation = 'UPSERT'
 AND State = 'dispatching'
 AND NOT (BINARY LinkKey = BINARY ? AND BINARY OrgId = BINARY ?
 AND BINARY AppInstallationId = BINARY ? AND BINARY AuthorizedWorkloadId = BINARY ?)
 LIMIT 1 FOR UPDATE NOWAIT`, i.ConnectorOwnerKey, i.ActionSlotID, i.SourceObject, i.SourceRecordID, i.DestinationSlotID,
		command.LinkKey, command.Scope.OrgID, command.Scope.AppInstallationID, command.Scope.AuthorizedWorkloadID).Scan(&present)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	return newDurableIdentityConflict(command, "legacy_remote_link_acceptance", "ForeignScopeDispatch")
}

func lockPreparedLegacyRemoteLink(ctx context.Context, tx *sql.Tx, command Command, proof *preparedLegacyRemoteLink) error {
	if proof == nil || proof.scope != command.Scope || proof.commandID != command.CommandID ||
		proof.payloadDigest != command.Intent.PayloadDigest || command.Intent.Operation != outboundintentv1.OperationUpsert ||
		proof.link.LinkKey != legacyRemoteLinkKey(command) || command.LinkKey != outboundLinkKey(command.Scope, command.Intent) {
		return ErrIdentityConflict
	}
	link, found, err := lockRemoteLink(ctx, tx, proof.link.LinkKey)
	if err != nil {
		return err
	}
	if !found || link != proof.link || link.Status != remoteLinkStatusFinalized {
		return newDurableIdentityConflict(command, "legacy_remote_link_fence", "PreparedRemoteLink")
	}
	locator, err := readLegacyRemoteLocator(ctx, tx, link.LinkKey, true)
	if err != nil {
		return err
	}
	if locator != proof.locator {
		return newDurableIdentityConflict(command, "legacy_remote_link_fence", "PreparedRemoteLocator")
	}
	return nil
}

func legacyExecutionFingerprints(ctx context.Context, db *sql.DB, resolved ResolvedAuthority) (string, string, error) {
	contract := resolved.Contract
	var mappings []connectorv1.ConnectorFieldMappingInput
	declarations := map[string][]persistedRuntimeRow{}
	seen := map[string]bool{}
	for _, binding := range resolved.MappingBindings {
		generation := strings.Join([]string{binding.OwnerPack, binding.PackInstallationID, binding.PackPayloadDigest}, "\x00")
		rows, found := declarations[generation]
		if !found {
			var err error
			rows, err = loadExactRuntimeRows(ctx, db, binding)
			if err != nil {
				return "", "", err
			}
			declarations[generation] = rows
		}
		projection, err := exactRuntimeProjection(rows, "connector_field_mapping", binding.BusinessKey)
		if err != nil {
			return "", "", err
		}
		var decoded []connectorv1.ConnectorFieldMappingInput
		if decodeStrictJSON([]byte(projection), &decoded) != nil || len(decoded) == 0 {
			return "", "", ErrIdentityConflict
		}
		for _, mapping := range decoded {
			if mapping.Source != binding.OwnerPack {
				return "", "", ErrIdentityConflict
			}
			if mapping.Source != contract.ConnectorOwnerKey || !strings.EqualFold(mapping.SourceObject, contract.SourceObject) ||
				!strings.EqualFold(mapping.TargetObject, contract.TargetObject) || !strings.EqualFold(mapping.ActionType, connectorPushActionType) ||
				(mapping.Direction != connectorv1.DirectionOutbound && mapping.Direction != connectorv1.DirectionBidirectional) {
				continue
			}
			source, target := "source:"+strings.ToLower(strings.TrimSpace(mapping.SourceField)), "target:"+strings.ToLower(strings.TrimSpace(mapping.TargetField))
			if source == "source:" || target == "target:" || seen[source] || seen[target] {
				return "", "", ErrIdentityConflict
			}
			seen[source], seen[target] = true, true
			mappings = append(mappings, mapping)
		}
	}
	if len(mappings) == 0 {
		return "", "", ErrIdentityConflict
	}
	// This is the released Catalog finalization-authority order, shared by
	// the V1 producer: direction, authored order, folded source/target names.
	sort.Slice(mappings, func(i, j int) bool {
		a, b := mappings[i], mappings[j]
		if a.Direction != b.Direction {
			return a.Direction == connectorv1.DirectionOutbound
		}
		if a.SortOrder != b.SortOrder {
			return a.SortOrder < b.SortOrder
		}
		if strings.ToLower(a.SourceField) != strings.ToLower(b.SourceField) {
			return strings.ToLower(a.SourceField) < strings.ToLower(b.SourceField)
		}
		return strings.ToLower(a.TargetField) < strings.ToLower(b.TargetField)
	})
	encoded, err := json.Marshal(mappings)
	if err != nil {
		return "", "", err
	}
	associations := append([]connectorv1.ConnectorPushAssociationContract(nil), contract.Associations...)
	for i := range associations {
		associations[i].TargetPreviousActionSlotIDs = nil
		associations[i].TargetProfileOwnerPack, associations[i].TargetProfileName = "", ""
	}
	body, err := json.Marshal(struct {
		BodyRoot     string                                         `json:"bodyRoot"`
		Mappings     []connectorv1.ConnectorFieldMappingInput       `json:"mappings"`
		Associations []connectorv1.ConnectorPushAssociationContract `json:"associations,omitempty"`
	}{strings.TrimSpace(contract.BodyRoot), mappings, associations})
	if err != nil {
		return "", "", err
	}
	return digestStrings(string(encoded)), digestStrings(string(body)), nil
}