package catalogwriter

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

	"github.com/go-sql-driver/mysql"
)

var ErrConflict = errors.New("catalog org writer: conflict")
var ErrLineageExhausted = errors.New("catalog org writer: lineage exhausted")

// State is the locked scalar lineage and writer fence for one org. The active
// command identity is stored in the legacy ActiveOperationId column until an
// explicit schema migration can rename it safely.
type State struct {
	OrgID            string
	RuntimeAudience  string
	CatalogRevision  uint64
	SecurityEpoch    uint64
	ActiveCommandID  string
	WriterGeneration uint64
}

// NextCatalogRevision allocates the next Catalog axis without reading or
// changing SecurityEpoch.
func NextCatalogRevision(state State) (uint64, error) {
	return nextLineage(state.CatalogRevision)
}

// NextSecurityEpoch allocates the next security axis without reading or
// changing CatalogRevision.
func NextSecurityEpoch(state State) (uint64, error) {
	return nextLineage(state.SecurityEpoch)
}

// LockOrCreate locks one org writer row, creating its zero-lineage seed only
// when no row exists. It never reserves the writer or allocates a revision.
func LockOrCreate(ctx context.Context, tx *sql.Tx, orgID, runtimeAudience string) (State, error) {
	if strings.TrimSpace(orgID) == "" || strings.TrimSpace(runtimeAudience) == "" {
		return State{}, fmt.Errorf("catalog org writer: org and Runtime audience are required")
	}
	state, err := Lock(ctx, tx, orgID)
	if err == nil {
		if state.RuntimeAudience != runtimeAudience {
			return State{}, ErrConflict
		}
		return state, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return State{}, err
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogOrgState
(OrgId, RuntimeAudience, CatalogRevision, LastSecurityEpoch, ActiveOperationId, WriterGeneration, CreatedDate, LastModifiedDate)
VALUES (?, ?, 0, 0, NULL, 0, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`, orgID, runtimeAudience)
	if isDuplicate(err) {
		return State{}, ErrConflict
	}
	if err != nil {
		return State{}, err
	}
	return State{OrgID: orgID, RuntimeAudience: runtimeAudience}, nil
}

// Lock returns one existing org writer row under SELECT FOR UPDATE.
func Lock(ctx context.Context, tx *sql.Tx, orgID string) (State, error) {
	var state State
	var storedOrgID string
	var rawCatalog, rawSecurity, rawWriterGeneration []byte
	var active sql.NullString
	err := tx.QueryRowContext(ctx, `SELECT OrgId, RuntimeAudience, CatalogRevision, LastSecurityEpoch, ActiveOperationId, WriterGeneration
FROM _System_CatalogOrgState WHERE OrgId = ? FOR UPDATE`, orgID).Scan(
		&storedOrgID, &state.RuntimeAudience, &rawCatalog, &rawSecurity, &active, &rawWriterGeneration)
	if err != nil {
		return State{}, err
	}
	if storedOrgID != orgID {
		return State{}, ErrConflict
	}
	state.OrgID = storedOrgID
	state.CatalogRevision, err = parseUint64("CatalogRevision", rawCatalog)
	if err != nil {
		return State{}, err
	}
	state.SecurityEpoch, err = parseUint64("LastSecurityEpoch", rawSecurity)
	if err != nil {
		return State{}, err
	}
	state.WriterGeneration, err = parseUint64("WriterGeneration", rawWriterGeneration)
	if err != nil {
		return State{}, err
	}
	state.ActiveCommandID = active.String
	return state, nil
}

// Reserve installs commandID as the sole org writer fence without advancing
// either lineage axis. The caller must hold state from LockOrCreate in the same
// transaction.
func Reserve(ctx context.Context, tx *sql.Tx, state State, commandID string) error {
	_, err := ReserveGeneration(ctx, tx, state, commandID)
	return err
}

// ReserveGeneration installs commandID as the sole org writer fence and
// advances the durable reservation generation exactly once. The returned
// generation distinguishes a recovered live reservation from an older use of
// the same command identity.
func ReserveGeneration(ctx context.Context, tx *sql.Tx, state State, commandID string) (uint64, error) {
	if state.OrgID == "" || state.ActiveCommandID != "" || strings.TrimSpace(commandID) == "" {
		return 0, ErrConflict
	}
	if state.WriterGeneration == ^uint64(0) {
		return 0, ErrLineageExhausted
	}
	next := state.WriterGeneration + 1
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogOrgState
SET ActiveOperationId = ?, WriterGeneration = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND ActiveOperationId IS NULL AND CatalogRevision = ? AND LastSecurityEpoch = ? AND WriterGeneration = ?`,
		commandID, uintText(next), state.OrgID, uintText(state.CatalogRevision),
		uintText(state.SecurityEpoch), uintText(state.WriterGeneration))
	if err != nil {
		return 0, err
	}
	if err := requireOneRow(result); err != nil {
		return 0, err
	}
	return next, nil
}

// Release clears the exact command fence without advancing either lineage
// axis. It is used when the owning command cannot commit its transition.
func Release(ctx context.Context, tx *sql.Tx, orgID, commandID string) error {
	if strings.TrimSpace(orgID) == "" || strings.TrimSpace(commandID) == "" {
		return fmt.Errorf("catalog org writer: org and command identity are required")
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogOrgState
SET ActiveOperationId = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND ActiveOperationId = ?`, orgID, commandID)
	if err != nil {
		return err
	}
	return requireOneRow(result)
}

// ReleaseGeneration clears only the exact command reservation incarnation.
// The caller must hold the writer row locked in the same transaction.
func ReleaseGeneration(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
	commandID string,
	writerGeneration uint64,
) error {
	if strings.TrimSpace(orgID) == "" ||
		strings.TrimSpace(commandID) == "" ||
		writerGeneration == 0 {
		return fmt.Errorf(
			"catalog org writer: org, command identity, and writer generation are required",
		)
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogOrgState
SET ActiveOperationId = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND ActiveOperationId = ? AND WriterGeneration = ?`,
		orgID,
		commandID,
		uintText(writerGeneration),
	)
	if err != nil {
		return err
	}
	return requireOneRow(result)
}

// Commit advances the scalar org lineage and releases the exact command fence.
// It defines no transition coupling between CatalogRevision and SecurityEpoch;
// the owning command validates its transition before calling this primitive.
func Commit(ctx context.Context, tx *sql.Tx, orgID, commandID string, catalogRevision, securityEpoch uint64) error {
	if strings.TrimSpace(orgID) == "" || strings.TrimSpace(commandID) == "" || catalogRevision == 0 || securityEpoch == 0 {
		return fmt.Errorf("catalog org writer: complete lineage and command identity are required")
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogOrgState
SET CatalogRevision = ?, LastSecurityEpoch = ?, ActiveOperationId = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND ActiveOperationId = ?`, uintText(catalogRevision), uintText(securityEpoch), orgID, commandID)
	if err != nil {
		return err
	}
	return requireOneRow(result)
}

func requireOneRow(result sql.Result) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return ErrConflict
	}
	return nil
}

func nextLineage(value uint64) (uint64, error) {
	if value == ^uint64(0) {
		return 0, ErrLineageExhausted
	}
	return value + 1, nil
}

func parseUint64(label string, raw []byte) (uint64, error) {
	value, err := strconv.ParseUint(string(raw), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("catalog org writer: invalid %s: %w", label, err)
	}
	return value, nil
}

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

func isDuplicate(err error) bool {
	var mysqlError *mysql.MySQLError
	return errors.As(err, &mysqlError) && mysqlError.Number == 1062
}