package serving

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
)

const receiptAdmissionKeyColumns = `
SELECT Id, AdmissionKeyProtocolVersion, ServingScopeDigest,
       ConnectorAccountId, AccountServingGateId, IngressChannel,
       TransportIdempotencyKey, AuthorizedPayloadDigest, Disposition,
       ReceiptId, PredecessorRefProtocolVersion, PredecessorRef,
       PredecessorRefDigest
  FROM _System_ConnectorV2ReceiptAdmissionKey`

const receiptColumns = `
SELECT Id, ReceiptProtocolVersion, ServingScopeDigest, ConnectorAccountId,
       AccountServingGateId, AccountServingFenceToken,
       ServingEpochId, ServingEpochDigest, AdmissionKind,
       AdmissionAuthorityId, PayloadCiphertextProtocolVersion,
       PayloadCiphertext, PayloadDigest, PayloadSizeBytes,
       ReceiptComponentCount
  FROM _System_ConnectorV2Receipt`

// IngressReceiptRepository owns the source-free immutable V2Receipt/ReceiptAdmissionKey
// read path. It deliberately does not expose component,
// execution, replay, gate-transition, or provider-I/O APIs.
type IngressReceiptRepository struct {
	db *sql.DB
}

func NewIngressReceiptRepository(db *sql.DB) (*IngressReceiptRepository, error) {
	if db == nil {
		return nil, fmt.Errorf("%w: database is nil", ErrIngressReceiptDependencyMissing)
	}
	return &IngressReceiptRepository{db: db}, nil
}

func (repository *IngressReceiptRepository) VerifySchema(ctx context.Context) error {
	if repository == nil || repository.db == nil {
		return fmt.Errorf("%w: database is nil", ErrIngressReceiptDependencyMissing)
	}
	if ctx == nil {
		return fmt.Errorf("%w: context is nil", ErrIngressReceiptDependencyMissing)
	}
	return verifyIngressReceiptAuthoritySchema(ctx, repository.db)
}

// ResolveTransport reads the sole admission authority for one exact transport
// identity. A retained V1 tombstone is returned as lineage evidence only and is
// never materialized as a V2Receipt.
func (repository *IngressReceiptRepository) ResolveTransport(
	ctx context.Context,
	identity TransportIdentity,
) (TransportAdmission, bool, error) {
	if repository == nil || repository.db == nil {
		return TransportAdmission{}, false, fmt.Errorf("%w: database is nil", ErrIngressReceiptDependencyMissing)
	}
	if ctx == nil {
		return TransportAdmission{}, false, fmt.Errorf("%w: context is nil", ErrIngressReceiptDependencyMissing)
	}
	if err := identity.validate(); err != nil {
		return TransportAdmission{}, false, err
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return TransportAdmission{}, false, err
	}

	admission, found, err := loadTransportAdmissionByIdentity(ctx, repository.db, identity, false)
	if err != nil {
		return TransportAdmission{}, false, err
	}
	if !found {
		return TransportAdmission{}, false, nil
	}
	return admission.copy(), true, nil
}

// PublishTransport atomically admits one immutable V2 transport V2Receipt. The
// transaction first locks the exact account gate and its exact active
// ServingEpoch, then inserts V2Receipt and ReceiptAdmissionKey. Exact retries
// converge only after the colliding transaction is fully rolled back and a
// fresh pessimistic transaction lock-reads the complete durable graph.
type ingressReceiptQueryer interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

func loadTransportAdmissionByIdentity(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	identity TransportIdentity,
	forUpdate bool,
) (TransportAdmission, bool, error) {
	if !forUpdate {
		admission, found, err := loadTransportAdmissionByIdentityRaw(ctx, queryer, identity, false)
		if err != nil || !found {
			return admission, found, err
		}
		if err := validateTransportAdmissionParents(ctx, queryer, admission, false); err != nil {
			return TransportAdmission{}, false, err
		}
		return admission, true, nil
	}

	discovered, found, err := loadTransportAdmissionByIdentityRaw(ctx, queryer, identity, false)
	if err != nil || !found {
		return discovered, found, err
	}
	if err := validateTransportAdmissionParents(ctx, queryer, discovered, true); err != nil {
		return TransportAdmission{}, false, err
	}
	locked, found, err := loadTransportAdmissionByIdentityRaw(ctx, queryer, identity, true)
	if err != nil {
		return TransportAdmission{}, false, err
	}
	if !found || !discovered.equal(locked) {
		return TransportAdmission{}, false, storedIngressReceiptInvalid(
			"transport authority changed between discovery and locking read",
		)
	}
	return locked, true, nil
}

func loadTransportAdmissionByIdentityRaw(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	identity TransportIdentity,
	forUpdate bool,
) (TransportAdmission, bool, error) {
	query := receiptAdmissionKeyColumns + `
 WHERE ConnectorAccountId = ? AND BINARY ConnectorAccountId = BINARY ?
   AND IngressChannel = ? AND BINARY IngressChannel = BINARY ?
   AND TransportIdempotencyKey = ? AND BINARY TransportIdempotencyKey = BINARY ?
 ORDER BY BINARY Id`
	if forUpdate {
		query += " FOR UPDATE"
	}
	keys, err := scanReceiptAdmissionKeys(ctx, queryer, query,
		identity.connectorAccountID, identity.connectorAccountID,
		string(identity.ingressChannel), string(identity.ingressChannel),
		identity.transportIdempotencyKey, identity.transportIdempotencyKey,
	)
	if err != nil {
		return TransportAdmission{}, false, err
	}
	if len(keys) == 0 {
		if err := rejectIncompleteTransportAuthorityForAccount(ctx, queryer, identity.connectorAccountID, forUpdate); err != nil {
			return TransportAdmission{}, false, err
		}
		return TransportAdmission{}, false, nil
	}
	if len(keys) != 1 {
		return TransportAdmission{}, false, storedIngressReceiptInvalid("transport identity resolves to %d admission keys", len(keys))
	}
	if keys[0].identity != identity {
		return TransportAdmission{}, false, storedIngressReceiptInvalid("transport lookup returned a byte-distinct identity")
	}
	return loadTransportAdmissionForKey(ctx, queryer, keys[0], forUpdate)
}

func loadTransportAdmissionByCandidate(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	candidate TransportAdmission,
	forUpdate bool,
) (TransportAdmission, bool, error) {
	if !forUpdate {
		admission, found, err := loadTransportAdmissionByCandidateRaw(ctx, queryer, candidate, false)
		if err != nil || !found {
			return admission, found, err
		}
		if err := validateTransportAdmissionParents(ctx, queryer, admission, false); err != nil {
			return TransportAdmission{}, false, err
		}
		return admission, true, nil
	}

	discovered, found, err := loadTransportAdmissionByCandidateRaw(ctx, queryer, candidate, false)
	if err != nil || !found {
		return discovered, found, err
	}
	if err := validateTransportAdmissionParents(ctx, queryer, discovered, true); err != nil {
		return TransportAdmission{}, false, err
	}
	locked, found, err := loadTransportAdmissionByCandidateRaw(ctx, queryer, candidate, true)
	if err != nil {
		return TransportAdmission{}, false, err
	}
	if !found || !discovered.equal(locked) {
		return TransportAdmission{}, false, storedIngressReceiptInvalid(
			"candidate authority changed between discovery and locking read",
		)
	}
	return locked, true, nil
}

func loadTransportAdmissionByCandidateRaw(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	candidate TransportAdmission,
	forUpdate bool,
) (TransportAdmission, bool, error) {
	identity := candidate.key.identity
	query := receiptAdmissionKeyColumns + `
 WHERE (Id = ? AND BINARY Id = BINARY ?)
    OR (ConnectorAccountId = ? AND BINARY ConnectorAccountId = BINARY ?
    AND IngressChannel = ? AND BINARY IngressChannel = BINARY ?
    AND TransportIdempotencyKey = ? AND BINARY TransportIdempotencyKey = BINARY ?)
 ORDER BY BINARY Id`
	if forUpdate {
		query += " FOR UPDATE"
	}
	keys, err := scanReceiptAdmissionKeys(ctx, queryer, query,
		candidate.key.id, candidate.key.id,
		identity.connectorAccountID, identity.connectorAccountID,
		string(identity.ingressChannel), string(identity.ingressChannel),
		identity.transportIdempotencyKey, identity.transportIdempotencyKey,
	)
	if err != nil {
		return TransportAdmission{}, false, err
	}
	if len(keys) > 1 {
		return TransportAdmission{}, false, fmt.Errorf(
			"%w: receipt admission-key ID and transport identity name distinct rows",
			ErrIngressReceiptIdentityConflict,
		)
	}
	if len(keys) == 0 {
		if err := rejectIncompleteTransportAuthorityForAccount(ctx, queryer, identity.connectorAccountID, forUpdate); err != nil {
			return TransportAdmission{}, false, err
		}
		// A V2Receipt collision without its matching key is not an exact duplicate.
		if candidate.receipt != nil {
			receipts, receiptErr := loadReceiptsForCoordinates(ctx, queryer, candidate.receipt.id, candidate.key, forUpdate)
			if receiptErr != nil {
				return TransportAdmission{}, false, receiptErr
			}
			if len(receipts) != 0 {
				return TransportAdmission{}, false, fmt.Errorf(
					"%w: V2Receipt identity exists without the matching admission key",
					ErrIngressReceiptIdentityConflict,
				)
			}
		}
		return TransportAdmission{}, false, nil
	}
	return loadTransportAdmissionForKey(ctx, queryer, keys[0], forUpdate)
}

// rejectIncompleteTransportAuthorityForAccount proves both directions of the
// immutable V2 relationship before an absent lookup can authorize publication:
// every transport Receipt has its exact AdmissionKey, and every v2_receipt key
// has its exact Receipt. V1 tombstones deliberately carry a NULL ReceiptId and
// are excluded by their distinct disposition.
func rejectIncompleteTransportAuthorityForAccount(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	connectorAccountID string,
	forUpdate bool,
) error {
	if err := rejectOrphanedTransportReceiptForAccount(
		ctx,
		queryer,
		connectorAccountID,
		forUpdate,
	); err != nil {
		return err
	}
	return rejectOrphanedV2ReceiptAdmissionKeyForAccount(
		ctx,
		queryer,
		connectorAccountID,
		forUpdate,
	)
}

func rejectOrphanedTransportReceiptForAccount(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	connectorAccountID string,
	forUpdate bool,
) error {
	query := `SELECT receipt.Id
  FROM _System_ConnectorV2Receipt AS receipt
 WHERE receipt.ConnectorAccountId = ? AND BINARY receipt.ConnectorAccountId = BINARY ?
   AND receipt.AdmissionKind = ? AND BINARY receipt.AdmissionKind = BINARY ?
   AND NOT EXISTS (
       SELECT 1
         FROM _System_ConnectorV2ReceiptAdmissionKey AS admission_key
        WHERE admission_key.Id = receipt.AdmissionAuthorityId
          AND BINARY admission_key.Id = BINARY receipt.AdmissionAuthorityId
          AND admission_key.ReceiptId = receipt.Id
          AND BINARY admission_key.ReceiptId = BINARY receipt.Id
          AND admission_key.ServingScopeDigest = receipt.ServingScopeDigest
          AND BINARY admission_key.ServingScopeDigest = BINARY receipt.ServingScopeDigest
          AND admission_key.ConnectorAccountId = receipt.ConnectorAccountId
          AND BINARY admission_key.ConnectorAccountId = BINARY receipt.ConnectorAccountId
          AND admission_key.AccountServingGateId = receipt.AccountServingGateId
          AND BINARY admission_key.AccountServingGateId = BINARY receipt.AccountServingGateId
          AND admission_key.AuthorizedPayloadDigest = receipt.PayloadDigest
          AND BINARY admission_key.AuthorizedPayloadDigest = BINARY receipt.PayloadDigest
          AND admission_key.AdmissionKeyProtocolVersion = ?
          AND BINARY admission_key.AdmissionKeyProtocolVersion = BINARY ?
          AND admission_key.Disposition = ?
          AND BINARY admission_key.Disposition = BINARY ?
   )
 ORDER BY BINARY receipt.Id
 LIMIT 1`
	if forUpdate {
		query += " FOR UPDATE"
	}
	var orphanedReceiptID string
	err := queryer.QueryRowContext(
		ctx,
		query,
		connectorAccountID,
		connectorAccountID,
		string(ReceiptAdmissionKindTransport),
		string(ReceiptAdmissionKindTransport),
		string(ReceiptAdmissionKeyProtocolV1),
		string(ReceiptAdmissionKeyProtocolV1),
		string(ReceiptAdmissionDispositionV2Receipt),
		string(ReceiptAdmissionDispositionV2Receipt),
	).Scan(&orphanedReceiptID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("inspect connector account transport receipt completeness: %w", err)
	}
	return storedIngressReceiptInvalid(
		"connector account has transport receipt %q without its exact admission key",
		orphanedReceiptID,
	)
}

func rejectOrphanedV2ReceiptAdmissionKeyForAccount(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	connectorAccountID string,
	forUpdate bool,
) error {
	query := `SELECT admission_key.Id
  FROM _System_ConnectorV2ReceiptAdmissionKey AS admission_key
 WHERE admission_key.ConnectorAccountId = ? AND BINARY admission_key.ConnectorAccountId = BINARY ?
   AND admission_key.AdmissionKeyProtocolVersion = ?
   AND BINARY admission_key.AdmissionKeyProtocolVersion = BINARY ?
   AND admission_key.Disposition = ? AND BINARY admission_key.Disposition = BINARY ?
   AND (
       admission_key.ReceiptId IS NULL
       OR admission_key.PredecessorRefProtocolVersion IS NOT NULL
       OR admission_key.PredecessorRef IS NOT NULL
       OR admission_key.PredecessorRefDigest IS NOT NULL
       OR NOT EXISTS (
           SELECT 1
             FROM _System_ConnectorV2Receipt AS receipt
            WHERE receipt.Id = admission_key.ReceiptId
              AND BINARY receipt.Id = BINARY admission_key.ReceiptId
              AND receipt.ServingScopeDigest = admission_key.ServingScopeDigest
              AND BINARY receipt.ServingScopeDigest = BINARY admission_key.ServingScopeDigest
              AND receipt.ConnectorAccountId = admission_key.ConnectorAccountId
              AND BINARY receipt.ConnectorAccountId = BINARY admission_key.ConnectorAccountId
              AND receipt.AccountServingGateId = admission_key.AccountServingGateId
              AND BINARY receipt.AccountServingGateId = BINARY admission_key.AccountServingGateId
              AND receipt.AdmissionKind = ? AND BINARY receipt.AdmissionKind = BINARY ?
              AND receipt.AdmissionAuthorityId = admission_key.Id
              AND BINARY receipt.AdmissionAuthorityId = BINARY admission_key.Id
              AND receipt.PayloadDigest = admission_key.AuthorizedPayloadDigest
              AND BINARY receipt.PayloadDigest = BINARY admission_key.AuthorizedPayloadDigest
       )
   )
 ORDER BY BINARY admission_key.Id
 LIMIT 1`
	if forUpdate {
		query += " FOR UPDATE"
	}
	var orphanedKeyID string
	err := queryer.QueryRowContext(
		ctx,
		query,
		connectorAccountID,
		connectorAccountID,
		string(ReceiptAdmissionKeyProtocolV1),
		string(ReceiptAdmissionKeyProtocolV1),
		string(ReceiptAdmissionDispositionV2Receipt),
		string(ReceiptAdmissionDispositionV2Receipt),
		string(ReceiptAdmissionKindTransport),
		string(ReceiptAdmissionKindTransport),
	).Scan(&orphanedKeyID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("inspect connector account V2 receipt admission-key completeness: %w", err)
	}
	return storedIngressReceiptInvalid(
		"connector account has v2_receipt admission key %q without its exact receipt",
		orphanedKeyID,
	)
}

func validateTransportAdmissionParents(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	admission TransportAdmission,
	forUpdate bool,
) error {
	gate, found, err := loadAccountServingGateExact(
		ctx,
		queryer,
		admission.key.servingScopeDigest,
		admission.key.identity.connectorAccountID,
		forUpdate,
	)
	if err != nil {
		if errors.Is(err, errStoredAccountServingGateInvalid) {
			return errors.Join(ErrStoredIngressReceiptInvalid, err)
		}
		return fmt.Errorf("validate retained receipt account gate: %w", err)
	}
	if !found || gate.id != admission.key.accountServingGateID {
		return storedIngressReceiptInvalid(
			"admission key does not belong to its exact scope/account gate",
		)
	}
	if admission.receipt == nil {
		return nil
	}

	receipt := *admission.receipt
	if gate.fenceToken < receipt.accountServingFenceToken {
		return storedIngressReceiptInvalid(
			"account gate fence trails the fence frozen by its receipt",
		)
	}
	// A gate transition advances its fence before a successor epoch can become
	// authoritative. At the fence observed by this Receipt, the gate and Receipt
	// must therefore name the same epoch. A higher current fence is a legitimate
	// successor transition; the immutable Receipt remains bound to its retained
	// ServingEpoch below rather than to the gate's newer epoch.
	if gate.fenceToken == receipt.accountServingFenceToken &&
		(gate.servingEpochID == nil || gate.servingEpochDigest == nil ||
			*gate.servingEpochID != receipt.servingEpochID ||
			*gate.servingEpochDigest != receipt.servingEpochDigest) {
		return storedIngressReceiptInvalid(
			"account gate and receipt disagree on the serving epoch at the frozen fence",
		)
	}
	header, found, err := loadAccountServingEpochHeaderExact(
		ctx,
		queryer,
		receipt.servingEpochID,
		forUpdate,
	)
	if err != nil {
		if errors.Is(err, ErrStoredCandidateInvalid) {
			return errors.Join(ErrStoredIngressReceiptInvalid, err)
		}
		return fmt.Errorf("validate retained receipt serving epoch: %w", err)
	}
	if !found ||
		header.id != receipt.servingEpochID ||
		header.epochDigest != receipt.servingEpochDigest ||
		header.servingScopeDigest != receipt.servingScopeDigest {
		return storedIngressReceiptInvalid(
			"receipt does not belong to its exact serving epoch digest and scope",
		)
	}
	return nil
}

func loadTransportAdmissionForKey(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	key receiptAdmissionKey,
	forUpdate bool,
) (TransportAdmission, bool, error) {
	receiptID := ""
	if key.receiptID != nil {
		receiptID = *key.receiptID
	}
	receipts, err := loadReceiptsForCoordinates(ctx, queryer, receiptID, key, forUpdate)
	if err != nil {
		return TransportAdmission{}, false, err
	}
	if len(receipts) > 1 {
		return TransportAdmission{}, false, storedIngressReceiptInvalid("receipt identity and admission authority name distinct rows")
	}

	admission := TransportAdmission{key: key}
	if len(receipts) == 1 {
		receipt := receipts[0]
		admission.receipt = &receipt
	}
	if err := admission.validateStored(); err != nil {
		return TransportAdmission{}, false, err
	}
	return admission, true, nil
}

func loadReceiptsForCoordinates(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	receiptID string,
	key receiptAdmissionKey,
	forUpdate bool,
) ([]V2Receipt, error) {
	query := receiptColumns + `
 WHERE (Id = ? AND BINARY Id = BINARY ?)
    OR (ServingScopeDigest = ? AND BINARY ServingScopeDigest = BINARY ?
    AND ConnectorAccountId = ? AND BINARY ConnectorAccountId = BINARY ?
    AND AdmissionKind = ? AND BINARY AdmissionKind = BINARY ?
    AND AdmissionAuthorityId = ? AND BINARY AdmissionAuthorityId = BINARY ?)
 ORDER BY BINARY Id`
	if forUpdate {
		query += " FOR UPDATE"
	}
	return scanReceipts(ctx, queryer, query,
		receiptID, receiptID,
		key.servingScopeDigest, key.servingScopeDigest,
		key.identity.connectorAccountID, key.identity.connectorAccountID,
		string(ReceiptAdmissionKindTransport), string(ReceiptAdmissionKindTransport),
		key.id, key.id,
	)
}

func scanReceiptAdmissionKeys(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	query string,
	args ...any,
) ([]receiptAdmissionKey, error) {
	rows, err := queryer.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load receipt admission keys: %w", err)
	}
	defer rows.Close()

	keys := make([]receiptAdmissionKey, 0, 2)
	for rows.Next() {
		if len(keys) == 2 {
			return nil, storedIngressReceiptInvalid("receipt admission lookup returned more than two rows")
		}
		var (
			key                 receiptAdmissionKey
			channel             string
			receiptID           sql.NullString
			predecessorProtocol sql.NullString
			predecessorRef      []byte
			predecessorDigest   sql.NullString
		)
		if err := rows.Scan(
			&key.id,
			&key.protocolVersion,
			&key.servingScopeDigest,
			&key.identity.connectorAccountID,
			&key.accountServingGateID,
			&channel,
			&key.identity.transportIdempotencyKey,
			&key.authorizedPayloadDigest,
			&key.disposition,
			&receiptID,
			&predecessorProtocol,
			&predecessorRef,
			&predecessorDigest,
		); err != nil {
			return nil, fmt.Errorf("scan receipt admission key: %w", err)
		}
		key.identity.ingressChannel = IngressChannel(channel)
		if receiptID.Valid {
			key.receiptID = stringPointer(receiptID.String)
		}
		if predecessorProtocol.Valid {
			value := PredecessorReferenceProtocolVersion(predecessorProtocol.String)
			key.predecessorProtocolVersion = &value
		}
		if predecessorRef != nil {
			key.predecessorRef = append([]byte(nil), predecessorRef...)
		}
		if predecessorDigest.Valid {
			key.predecessorRefDigest = stringPointer(predecessorDigest.String)
		}
		keys = append(keys, key)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate receipt admission keys: %w", err)
	}
	return keys, nil
}

func scanReceipts(
	ctx context.Context,
	queryer ingressReceiptQueryer,
	query string,
	args ...any,
) ([]V2Receipt, error) {
	rows, err := queryer.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load V2 receipts: %w", err)
	}
	defer rows.Close()

	receipts := make([]V2Receipt, 0, 2)
	for rows.Next() {
		if len(receipts) == 2 {
			return nil, storedIngressReceiptInvalid("receipt lookup returned more than two rows")
		}
		var (
			receipt    V2Receipt
			ciphertext []byte
		)
		if err := rows.Scan(
			&receipt.id,
			&receipt.receiptProtocolVersion,
			&receipt.servingScopeDigest,
			&receipt.connectorAccountID,
			&receipt.accountServingGateID,
			&receipt.accountServingFenceToken,
			&receipt.servingEpochID,
			&receipt.servingEpochDigest,
			&receipt.admissionKind,
			&receipt.admissionAuthorityID,
			&receipt.payloadCiphertextProtocolVersion,
			&ciphertext,
			&receipt.payloadDigest,
			&receipt.payloadSizeBytes,
			&receipt.receiptComponentCount,
		); err != nil {
			return nil, fmt.Errorf("scan V2 receipt: %w", err)
		}
		receipt.payloadCiphertext = append([]byte(nil), ciphertext...)
		receipts = append(receipts, receipt)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate V2 receipts: %w", err)
	}
	return receipts, nil
}