package persistence

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

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

func finishManagedRowMutation(
	ctx context.Context,
	exec Executor,
	schema *models.ObjectMetadata,
	id string,
	action string,
	policy managedRowMutationProtection,
	result sql.Result,
) error {
	if len(policy.fields) == 0 {
		return nil
	}
	if result == nil {
		return apperrors.NewInternalError("generic row mutation returned no result", nil)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return apperrors.NewInternalError("inspect generic row mutation result", err)
	}
	if affected > 0 {
		return nil
	}
	return assertManagedRowMutable(ctx, exec, schema, id, action, policy)
}

func assertManagedRowMutable(
	ctx context.Context,
	exec Executor,
	schema *models.ObjectMetadata,
	id string,
	action string,
	policy managedRowMutationProtection,
) error {
	if len(policy.fields) == 0 {
		return nil
	}
	fields := append([]string{constants.FieldID}, policy.physicalFieldNames()...)
	q := query.From(policy.storageTableName).
		WithMetadata(schema).
		Select(fields).
		Where(constants.FieldID, id).
		WhereRaw("("+policy.protectedPredicate()+")", nil).
		Limit(1).
		Build()
	// Guarded DML is a current read. Under REPEATABLE READ, diagnosing its
	// zero-row result through a plain SELECT could observe the transaction's
	// older snapshot and falsely report success after a concurrent authority
	// claim. Keep the diagnosis on the same executor and make it a current read.
	q.SQL += " FOR UPDATE"
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		return apperrors.NewInternalError("inspect generic row mutation predicate miss", err)
	}
	defer rows.Close()
	records, err := query.ScanRowsToSObjects(rows)
	if err != nil {
		return apperrors.NewInternalError("scan generic row mutation predicate miss", err)
	}
	if len(records) == 0 {
		return nil
	}
	switch policy.classify(records[0]) {
	case managedRowMutationAuthorityOwned:
		return apperrors.NewInvalidFieldForWriteError(
			[]string{policy.fieldNames()[0]},
			fmt.Sprintf(
				"record cannot be %s because a catalog authority marker is present; generic row mutation is not permitted",
				action,
			),
		)
	case managedRowMutationCorrupt:
		return apperrors.NewInternalError(
			"catalog row mutation authority markers have a partial or unreadable assignment",
			nil,
		)
	default:
		return nil
	}
}

func rejectManagedRowMutationBulkDelete(
	tableName string,
	policy managedRowMutationProtection,
) error {
	if len(policy.fields) == 0 {
		return nil
	}
	return apperrors.NewInvalidFieldForWriteError(
		[]string{policy.fieldNames()[0]},
		fmt.Sprintf(
			"generic multi-row delete is not permitted for %s because the catalog declares row mutation protection",
			tableName,
		),
	)
}