package persistence

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

	pkgerrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// MaxNumericFieldForParent returns the current MAX(targetField) for rows with
// the same parent. Callers use the result inside their create transaction.
func (r *RecordRepository) MaxNumericFieldForParent(ctx context.Context, schema *models.ObjectMetadata, targetField, parentField string, parentValue interface{}) (float64, error) {
	if schema == nil {
		return 0, pkgerrors.NewValidationError("object", "schema is required")
	}
	tableName := schema.APIName
	if !isValidFieldName(tableName) {
		return 0, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	if !isValidFieldName(targetField) {
		return 0, pkgerrors.NewValidationError("targetField", fmt.Sprintf("invalid field name: %s", targetField))
	}
	if !isValidFieldName(parentField) {
		return 0, pkgerrors.NewValidationError("parentField", fmt.Sprintf("invalid field name: %s", parentField))
	}

	targetColumn := resolvePhysicalName(schema, targetField)
	parentColumn := resolvePhysicalName(schema, parentField)
	if !isValidFieldName(targetColumn) {
		return 0, pkgerrors.NewValidationError("targetField", fmt.Sprintf("invalid physical field name: %s", targetColumn))
	}
	if !isValidFieldName(parentColumn) {
		return 0, pkgerrors.NewValidationError("parentField", fmt.Sprintf("invalid physical field name: %s", parentColumn))
	}

	whereDeleted := "1 = 1"
	if schemaHasIsDeleted(schema) {
		whereDeleted = fmt.Sprintf("(`%s` = 0 OR `%s` IS NULL)", constants.FieldIsDeleted, constants.FieldIsDeleted)
	}
	exec := r.GetExecutor(ctx)
	lockClause := ""
	if _, ok := exec.(*sql.Tx); ok {
		if err := r.lockSiblingMaxParent(ctx, schema, parentField, parentValue); err != nil {
			return 0, err
		}
		lockClause = " FOR UPDATE"
	}
	queryText := fmt.Sprintf(
		"SELECT COALESCE(MAX(`%s`), 0) FROM `%s` WHERE %s AND `%s` = ?",
		targetColumn,
		tableName,
		whereDeleted,
		parentColumn,
	)
	queryText += lockClause

	var max sql.NullFloat64
	if err := exec.QueryRowContext(ctx, queryText, parentValue).Scan(&max); err != nil {
		return 0, err
	}
	if !max.Valid {
		return 0, nil
	}
	return max.Float64, nil
}

func (r *RecordRepository) lockSiblingMaxParent(ctx context.Context, schema *models.ObjectMetadata, parentField string, parentValue interface{}) error {
	field := models.FindField(schema, parentField)
	if field == nil || len(field.ReferenceTo) != 1 {
		return nil
	}
	parentObject := strings.TrimSpace(field.ReferenceTo[0])
	if parentObject == "" {
		return nil
	}
	if !isValidFieldName(parentObject) {
		return pkgerrors.NewValidationError("parentObject", fmt.Sprintf("invalid parent object name: %s", parentObject))
	}
	parentID, ok := parentValue.(string)
	if !ok {
		return nil
	}
	parentID = strings.TrimSpace(parentID)
	if parentID == "" {
		return nil
	}
	key := recordlocks.ParentKey{ObjectName: parentObject, RecordID: parentID}
	if scope := recordlocks.ScopeFromContext(ctx); scope != nil && scope.HasAcquired(key) {
		return nil
	}
	return pkgerrors.NewInternalError("sibling-max parent lock requires a pre-admitted parent lock", nil)
}