package catalogpublicationv2store

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

	catalogpublicationv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogpublicationv2"
)

type rowQuerier interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

// PutExact validates and canonically hashes one complete v2 binding, then
// inserts it immutably into an already verified fixed table. An exact replay is
// idempotent; no operation updates or reclassifies an existing row.
func PutExact(
	ctx context.Context,
	table *VerifiedTable,
	binding catalogpublicationv2.Binding,
) (string, error) {
	db, identity, err := requireVerifiedTable(table)
	if err != nil {
		return "", err
	}
	digest, err := catalogpublicationv2.BindingDigest(binding)
	if err != nil {
		return "", fmt.Errorf("catalog publication v2 store: invalid binding: %w", err)
	}

	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return "", err
	}
	defer rollbackTx(tx)

	existing, err := readExactRow(ctx, tx, identity, binding.OrgID, digest, true)
	if err == nil {
		if existing != binding {
			return "", fmt.Errorf("%w: digest %q names different content", ErrCorrupt, digest)
		}
		if err := tx.Commit(); err != nil {
			return recoverExact(ctx, table, binding, digest, err)
		}
		return digest, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return "", err
	}

	statement := `INSERT INTO ` + identity.name + `
(PublicationDigest, OrgId, CatalogRevision, CatalogSnapshotDigest,
SchemaGeneration, EngineGeneration, LayoutValidationViewDigest,
LayoutBundleOwnerVerificationReceiptSetDigest, LayoutActivationSetDigest,
SecurityEpoch, LayoutRoutingSecurityViewDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`
	result, err := tx.ExecContext(
		ctx,
		statement,
		digest,
		binding.OrgID,
		decimalText(binding.CatalogRevision),
		binding.CatalogSnapshotDigest,
		decimalText(binding.SchemaGeneration),
		decimalText(binding.EngineGeneration),
		binding.LayoutValidationViewDigest,
		binding.LayoutBundleOwnerVerificationReceiptSetDigest,
		binding.LayoutActivationSetDigest,
		decimalText(binding.SecurityEpoch),
		binding.LayoutRoutingSecurityViewDigest,
	)
	if err != nil {
		_ = tx.Rollback()
		return recoverExact(ctx, table, binding, digest, err)
	}
	if err := requireOne(result); err != nil {
		return "", err
	}
	inserted, err := readExactRow(ctx, tx, identity, binding.OrgID, digest, true)
	if err != nil {
		return "", err
	}
	if inserted != binding {
		return "", fmt.Errorf("%w: inserted digest %q does not reproduce its binding", ErrCorrupt, digest)
	}
	if err := tx.Commit(); err != nil {
		return recoverExact(ctx, table, binding, digest, err)
	}
	return digest, nil
}

// ReadExact loads one immutable v2 binding by explicit org and canonical
// digest. It reconstructs and rehashes all ten fields before returning.
func ReadExact(
	ctx context.Context,
	table *VerifiedTable,
	orgID string,
	digest string,
) (catalogpublicationv2.Binding, error) {
	db, identity, err := requireVerifiedTable(table)
	if err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	defer rollbackTx(tx)
	binding, err := readExactRow(ctx, tx, identity, orgID, digest, false)
	if errors.Is(err, sql.ErrNoRows) {
		return catalogpublicationv2.Binding{}, fmt.Errorf(
			"%w: publication %q for org %q",
			ErrNotFound,
			digest,
			orgID,
		)
	}
	if err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	if err := tx.Commit(); err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	return binding, nil
}

func readExactRow(
	ctx context.Context,
	query rowQuerier,
	identity *tableIdentity,
	orgID string,
	digest string,
	forUpdate bool,
) (catalogpublicationv2.Binding, error) {
	statement := `SELECT PublicationDigest, OrgId, CatalogRevision,
CatalogSnapshotDigest, SchemaGeneration, EngineGeneration,
LayoutValidationViewDigest, LayoutBundleOwnerVerificationReceiptSetDigest,
LayoutActivationSetDigest, SecurityEpoch, LayoutRoutingSecurityViewDigest
FROM ` + identity.name + ` WHERE OrgId = ? AND PublicationDigest = ?`
	if forUpdate {
		statement += " FOR UPDATE"
	}
	var storedDigest string
	var binding catalogpublicationv2.Binding
	var rawCatalogRevision, rawSchemaGeneration, rawEngineGeneration, rawSecurityEpoch []byte
	err := query.QueryRowContext(ctx, statement, orgID, digest).Scan(
		&storedDigest,
		&binding.OrgID,
		&rawCatalogRevision,
		&binding.CatalogSnapshotDigest,
		&rawSchemaGeneration,
		&rawEngineGeneration,
		&binding.LayoutValidationViewDigest,
		&binding.LayoutBundleOwnerVerificationReceiptSetDigest,
		&binding.LayoutActivationSetDigest,
		&rawSecurityEpoch,
		&binding.LayoutRoutingSecurityViewDigest,
	)
	if err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	if binding.OrgID != orgID {
		return catalogpublicationv2.Binding{}, fmt.Errorf(
			"%w: stored org %q does not equal requested org %q",
			ErrCorrupt,
			binding.OrgID,
			orgID,
		)
	}
	if storedDigest != digest {
		return catalogpublicationv2.Binding{}, fmt.Errorf(
			"%w: stored digest %q does not equal requested digest %q",
			ErrCorrupt,
			storedDigest,
			digest,
		)
	}
	binding.CatalogRevision, err = parseDurableUint64("CatalogRevision", rawCatalogRevision)
	if err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	binding.SchemaGeneration, err = parseDurableUint64("SchemaGeneration", rawSchemaGeneration)
	if err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	binding.EngineGeneration, err = parseDurableUint64("EngineGeneration", rawEngineGeneration)
	if err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	binding.SecurityEpoch, err = parseDurableUint64("SecurityEpoch", rawSecurityEpoch)
	if err != nil {
		return catalogpublicationv2.Binding{}, err
	}
	computed, err := catalogpublicationv2.BindingDigest(binding)
	if err != nil {
		return catalogpublicationv2.Binding{}, fmt.Errorf(
			"%w: invalid stored publication %q: %v",
			ErrCorrupt,
			digest,
			err,
		)
	}
	if computed != storedDigest {
		return catalogpublicationv2.Binding{}, fmt.Errorf(
			"%w: canonical digest is %q, stored %q",
			ErrCorrupt,
			computed,
			storedDigest,
		)
	}
	return binding, nil
}

func recoverExact(
	ctx context.Context,
	table *VerifiedTable,
	binding catalogpublicationv2.Binding,
	digest string,
	originalErr error,
) (string, error) {
	recovered, err := ReadExact(ctx, table, binding.OrgID, digest)
	if err == nil {
		if recovered != binding {
			return "", fmt.Errorf("%w: recovered digest %q names different content", ErrCorrupt, digest)
		}
		return digest, nil
	}
	if !errors.Is(err, ErrNotFound) {
		return "", err
	}
	return "", originalErr
}

func parseDurableUint64(label string, raw []byte) (uint64, error) {
	if len(raw) == 0 {
		return 0, fmt.Errorf("%w: %s is empty", ErrCorrupt, label)
	}
	for _, character := range raw {
		if character < '0' || character > '9' {
			return 0, fmt.Errorf("%w: %s is not an unsigned decimal", ErrCorrupt, label)
		}
	}
	value, err := strconv.ParseUint(string(raw), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("%w: %s exceeds uint64: %v", ErrCorrupt, label, err)
	}
	return value, nil
}

func decimalText(value uint64) string {
	return strconv.FormatUint(value, 10)
}

func requireOne(result sql.Result) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return fmt.Errorf("catalog publication v2 store: insert affected %d rows", affected)
	}
	return nil
}

func rollbackTx(tx *sql.Tx) {
	_ = tx.Rollback()
}