package persistence

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

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

// --- Salesforce-Compatible Flow Methods (FlowDefinition, FlowVersion, FlowInterview) ---

// flowDefinitionSelectColumns is the canonical column list for FlowDefinition
// reads. Kept as a package-level constant to keep SELECTs in lockstep.
const flowDefinitionSelectColumns = `Id, DeveloperName, MasterLabel, Description, ProcessType, TriggerType,
	TriggerObjectOrEvent, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized, OwnerPack, CreatedDate, LastModifiedDate`

const flowVersionSelectColumns = `Id, DefinitionId, MasterLabel, VersionNumber, Status, Description, ProcessType, Metadata,
	Source, IsDeleted, CreatedDate, LastModifiedDate`

const (
	flowVersionSourceStandard = "standard"
	flowVersionSourceCustom   = "custom"
)

// scanFlowDefinition applies a row scan into a FlowDefinition, handling
// nullable columns. Shared by Get/GetByID/GetByName to keep the column
// binding in one place.
func scanFlowDefinition(scan func(...interface{}) error) (*models.FlowDefinition, error) {
	var def models.FlowDefinition
	var desc, triggerType, triggerObj, activeVer, latestVer sql.NullString
	var ownerPack sql.NullString
	var isDeleted, isCustomized sql.NullBool
	if err := scan(&def.Id, &def.DeveloperName, &def.MasterLabel, &desc, &def.ProcessType,
		&triggerType, &triggerObj, &activeVer, &latestVer, &isDeleted, &isCustomized,
		&ownerPack, &def.CreatedDate, &def.LastModifiedDate); err != nil {
		return nil, err
	}
	if desc.Valid {
		def.Description = desc.String
	}
	if triggerType.Valid {
		def.TriggerType = triggerType.String
	}
	if triggerObj.Valid {
		def.TriggerObjectOrEvent = triggerObj.String
	}
	if activeVer.Valid {
		def.ActiveVersionId = activeVer.String
	}
	if latestVer.Valid {
		def.LatestVersionId = latestVer.String
	}
	if isDeleted.Valid {
		def.IsDeleted = isDeleted.Bool
	}
	if isCustomized.Valid {
		def.IsCustomized = isCustomized.Bool
	}
	if ownerPack.Valid {
		def.OwnerPack = ownerPack.String
	}
	return &def, nil
}

func nullableString(v string) interface{} {
	if v == "" {
		return nil
	}
	return v
}

func scanFlowVersion(scan func(...interface{}) error) (*models.FlowVersion, error) {
	var v models.FlowVersion
	var masterLabel, desc, metadata, source sql.NullString
	var isDeleted sql.NullBool
	var createdDate, lastModifiedDate sql.NullString
	if err := scan(&v.Id, &v.DefinitionId, &masterLabel, &v.VersionNumber, &v.Status, &desc, &v.ProcessType,
		&metadata, &source, &isDeleted, &createdDate, &lastModifiedDate); err != nil {
		return nil, err
	}
	if masterLabel.Valid {
		v.MasterLabel = masterLabel.String
	}
	if desc.Valid {
		v.Description = desc.String
	}
	if metadata.Valid {
		v.MetadataJSON = metadata.String
	}
	if source.Valid {
		v.Source = source.String
	}
	if isDeleted.Valid {
		v.IsDeleted = isDeleted.Bool
	}
	if createdDate.Valid {
		v.CreatedDate = createdDate.String
	}
	if lastModifiedDate.Valid {
		v.LastModifiedDate = lastModifiedDate.String
	}
	v.HydrateMetadata()
	return &v, nil
}

// GetAllFlowDefinitions returns all flow definitions.
func (r *MetadataRepository) GetAllFlowDefinitions(ctx context.Context) ([]*models.FlowDefinition, error) {
	query := `SELECT ` + flowDefinitionSelectColumns + `
		FROM FlowDefinition WHERE IsDeleted = false OR IsDeleted IS NULL`

	rows, err := r.GetExecutor(ctx).QueryContext(ctx, query)
	if err != nil {
		if IsTableNotExistsError(err) {
			log.Printf("FlowDefinition table does not exist, returning empty list")
			return []*models.FlowDefinition{}, nil
		}
		return nil, err
	}
	defer rows.Close()

	defs := make([]*models.FlowDefinition, 0)
	for rows.Next() {
		def, err := scanFlowDefinition(rows.Scan)
		if err != nil {
			log.Printf("Warning: Failed to scan flow definition: %v", err)
			continue
		}
		defs = append(defs, def)
	}
	return defs, rows.Err()
}

// GetFlowDefinition returns a flow definition by ID.
func (r *MetadataRepository) GetFlowDefinition(ctx context.Context, id string) (*models.FlowDefinition, error) {
	query := `SELECT ` + flowDefinitionSelectColumns + `
		FROM FlowDefinition WHERE Id = ?`

	def, err := scanFlowDefinition(r.GetExecutor(ctx).QueryRowContext(ctx, query, id).Scan)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	return def, nil
}

// GetFlowDefinitionByName returns a flow definition by DeveloperName.
func (r *MetadataRepository) GetFlowDefinitionByName(ctx context.Context, developerName string) (*models.FlowDefinition, error) {
	query := `SELECT ` + flowDefinitionSelectColumns + `
		FROM FlowDefinition WHERE DeveloperName = ?`

	def, err := scanFlowDefinition(r.GetExecutor(ctx).QueryRowContext(ctx, query, developerName).Scan)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	return def, nil
}

// CreateFlowDefinition creates a new flow definition.
func (r *MetadataRepository) CreateFlowDefinition(ctx context.Context, def *models.FlowDefinition) error {
	now := time.Now().Format(time.RFC3339)
	query := `INSERT INTO FlowDefinition (Id, DeveloperName, MasterLabel, Description, ProcessType,
		TriggerType, TriggerObjectOrEvent, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized,
		CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, def.Id, def.DeveloperName, def.MasterLabel, def.Description,
		def.ProcessType, def.TriggerType, def.TriggerObjectOrEvent, nullableString(def.ActiveVersionId), nullableString(def.LatestVersionId),
		def.IsDeleted, def.IsCustomized, now, now)
	return err
}

// UpdateFlowDefinition replaces the FlowDefinition projection, including
// reconcile-owned fields such as tombstones. Control-plane writers use this
// path; tooling PATCH must use PatchFlowDefinition instead.
func (r *MetadataRepository) UpdateFlowDefinition(ctx context.Context, id string, def *models.FlowDefinition) error {
	now := time.Now().Format(time.RFC3339)
	query := `UPDATE FlowDefinition SET DeveloperName = ?, MasterLabel = ?, Description = ?, ProcessType = ?,
		TriggerType = ?, TriggerObjectOrEvent = ?, ActiveVersionId = ?, LatestVersionId = ?, IsDeleted = ?, LastModifiedDate = ?
		WHERE Id = ?`
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, def.DeveloperName, def.MasterLabel, def.Description, def.ProcessType,
		def.TriggerType, def.TriggerObjectOrEvent, nullableString(def.ActiveVersionId), nullableString(def.LatestVersionId), def.IsDeleted, now, id)
	return err
}

// PatchFlowDefinition applies sparse admin-authored Tooling updates without
// touching internal reconcile state such as IsDeleted/IsCustomized.
func (r *MetadataRepository) PatchFlowDefinition(ctx context.Context, id string, patch *models.FlowDefinitionPatch) error {
	if patch == nil || patch.Empty() {
		return nil
	}

	now := time.Now().Format(time.RFC3339)
	sets := make([]string, 0, 9)
	args := make([]interface{}, 0, 10)

	if patch.DeveloperName != nil {
		sets = append(sets, "DeveloperName = ?")
		args = append(args, *patch.DeveloperName)
	}
	if patch.MasterLabel != nil {
		sets = append(sets, "MasterLabel = ?")
		args = append(args, *patch.MasterLabel)
	}
	if patch.Description != nil {
		sets = append(sets, "Description = ?")
		args = append(args, *patch.Description)
	}
	if patch.ProcessType != nil {
		sets = append(sets, "ProcessType = ?")
		args = append(args, *patch.ProcessType)
	}
	if patch.TriggerType != nil {
		sets = append(sets, "TriggerType = ?")
		args = append(args, *patch.TriggerType)
	}
	if patch.TriggerObjectOrEvent != nil {
		sets = append(sets, "TriggerObjectOrEvent = ?")
		args = append(args, *patch.TriggerObjectOrEvent)
	}
	if patch.ActiveVersionId != nil {
		sets = append(sets, "ActiveVersionId = ?")
		args = append(args, nullableString(*patch.ActiveVersionId))
	}
	if patch.LatestVersionId != nil {
		sets = append(sets, "LatestVersionId = ?")
		args = append(args, nullableString(*patch.LatestVersionId))
	}

	sets = append(sets, "LastModifiedDate = ?")
	args = append(args, now, id)

	query := `UPDATE FlowDefinition SET ` + strings.Join(sets, ", ") + ` WHERE Id = ?`
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, args...)
	return err
}

// MarkFlowDefinitionCustomized sets IsCustomized=1 on a flow definition so
// future apply passes preserve admin ownership of the row.
func (r *MetadataRepository) MarkFlowDefinitionCustomized(ctx context.Context, id string) error {
	now := time.Now().Format(time.RFC3339)
	_, err := r.GetExecutor(ctx).ExecContext(
		ctx,
		`UPDATE FlowDefinition SET IsCustomized = 1, LastModifiedDate = ? WHERE Id = ?`,
		now, id,
	)
	return err
}

// SetLatestVersionId updates ONLY the LatestVersionId field on a FlowDefinition.
func (r *MetadataRepository) SetLatestVersionId(ctx context.Context, definitionId string, versionId string) error {
	now := time.Now().Format(time.RFC3339)
	query := `UPDATE FlowDefinition SET LatestVersionId = ?, LastModifiedDate = ? WHERE Id = ?`
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, versionId, now, definitionId)
	return err
}

// SetTriggerInfo updates the TriggerType and TriggerObjectOrEvent fields on a FlowDefinition.
func (r *MetadataRepository) SetTriggerInfo(ctx context.Context, definitionId string, triggerType string, objectOrEvent string) error {
	now := time.Now().Format(time.RFC3339)
	query := `UPDATE FlowDefinition SET TriggerType = ?, TriggerObjectOrEvent = ?, LastModifiedDate = ? WHERE Id = ?`
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, triggerType, objectOrEvent, now, definitionId)
	return err
}

// DeleteFlowDefinition deletes a flow definition.
func (r *MetadataRepository) DeleteFlowDefinition(ctx context.Context, id string) error {
	now := time.Now().Format(time.RFC3339)
	query := `UPDATE FlowDefinition
		SET IsDeleted = 1, IsCustomized = 1, ActiveVersionId = NULL, LastModifiedDate = ?
		WHERE Id = ?`
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, now, id)
	return err
}

// LockFlowDefinitionForUpdate acquires a pessimistic FOR UPDATE lock on the
// FlowDefinition row by Id. Returns *NotFoundError if the row does not
// exist. Existing internal callers (ActivateFlowVersion, etc.) continue to
// use lockFlowDefinition/lockFlowDefinitionActiveVersion directly.
func (r *MetadataRepository) LockFlowDefinitionForUpdate(ctx context.Context, definitionId string) error {
	_, err := lockFlowDefinitionActiveVersion(ctx, r.GetExecutor(ctx), definitionId)
	if err == sql.ErrNoRows {
		return apperrors.NewNotFoundError("FlowDefinition", definitionId)
	}
	return err
}

// LockFlowDefinitionByDeveloperNameForUpdate acquires a pessimistic
// FOR UPDATE lock on the FlowDefinition row by DeveloperName and returns
// the authoritative row state observed under that lock.
func (r *MetadataRepository) LockFlowDefinitionByDeveloperNameForUpdate(ctx context.Context, developerName string) (*models.FlowDefinition, error) {
	query := `SELECT ` + flowDefinitionSelectColumns + `
		FROM FlowDefinition WHERE DeveloperName = ? FOR UPDATE`

	def, err := scanFlowDefinition(r.GetExecutor(ctx).QueryRowContext(ctx, query, developerName).Scan)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil, apperrors.NewNotFoundError("FlowDefinition", developerName)
		}
		return nil, err
	}
	return def, nil
}

// EnsureFlowDefinitionByDeveloperName performs an INSERT IGNORE stake-out
// followed by a SELECT FOR UPDATE to guarantee exactly one FlowDefinition
// row for the given DeveloperName. Returns the authoritative locked row
// (either the input def for a first-create or the existing row for a collision).
// Used by tenant-overlay first-create to prevent concurrent INSERT races
// on FlowDefinition.UNIQUE(DeveloperName).
func (r *MetadataRepository) EnsureFlowDefinitionByDeveloperName(ctx context.Context, def *models.FlowDefinition) (*models.FlowDefinition, error) {
	exec := r.GetExecutor(ctx)
	now := time.Now().Format(time.RFC3339)
	if _, err := exec.ExecContext(ctx, `INSERT IGNORE INTO FlowDefinition
		(Id, DeveloperName, MasterLabel, Description, ProcessType,
		 TriggerType, TriggerObjectOrEvent, ActiveVersionId, LatestVersionId,
		 IsDeleted, IsCustomized, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		def.Id, def.DeveloperName, def.MasterLabel, def.Description,
		def.ProcessType, def.TriggerType, def.TriggerObjectOrEvent,
		nullableString(def.ActiveVersionId), nullableString(def.LatestVersionId),
		def.IsDeleted, def.IsCustomized, now, now); err != nil {
		return nil, err
	}
	return r.LockFlowDefinitionByDeveloperNameForUpdate(ctx, def.DeveloperName)
}

// GetFlowVersions returns all versions for a flow definition.
func (r *MetadataRepository) GetFlowVersions(ctx context.Context, definitionId string) ([]*models.FlowVersion, error) {
	return r.getFlowVersions(ctx, definitionId, false)
}

// GetFlowVersionsForUpdate returns all versions for a flow definition using
// SELECT ... FOR UPDATE. TiDB pessimistic transactions treat plain SELECT as
// snapshot reads, so write paths that allocate VersionNumber must use this
// current-read variant after locking the parent FlowDefinition row.
func (r *MetadataRepository) GetFlowVersionsForUpdate(ctx context.Context, definitionId string) ([]*models.FlowVersion, error) {
	return r.getFlowVersions(ctx, definitionId, true)
}

func (r *MetadataRepository) getFlowVersions(ctx context.Context, definitionId string, forUpdate bool) ([]*models.FlowVersion, error) {
	query := `SELECT ` + flowVersionSelectColumns + `
		FROM Flow WHERE DefinitionId = ? ORDER BY VersionNumber DESC`
	if forUpdate {
		query += ` FOR UPDATE`
	}

	rows, err := r.GetExecutor(ctx).QueryContext(ctx, query, definitionId)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	versions := make([]*models.FlowVersion, 0)
	for rows.Next() {
		v, err := scanFlowVersion(rows.Scan)
		if err != nil {
			log.Printf("Warning: Failed to scan flow version: %v", err)
			continue
		}
		versions = append(versions, v)
	}
	return versions, rows.Err()
}

// GetFlowVersion returns a specific flow version.
func (r *MetadataRepository) GetFlowVersion(ctx context.Context, id string) (*models.FlowVersion, error) {
	query := `SELECT Id, DefinitionId, MasterLabel, VersionNumber, Status, Description, ProcessType, Metadata,
		Source, IsDeleted, CreatedDate, LastModifiedDate FROM Flow WHERE Id = ?`

	var v models.FlowVersion
	var masterLabel, desc, metadata, source sql.NullString
	var isDeleted sql.NullBool
	var createdDate, lastModifiedDate sql.NullString
	err := r.GetExecutor(ctx).QueryRowContext(ctx, query, id).Scan(&v.Id, &v.DefinitionId, &masterLabel, &v.VersionNumber, &v.Status,
		&desc, &v.ProcessType, &metadata, &source, &isDeleted, &createdDate, &lastModifiedDate)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	if masterLabel.Valid {
		v.MasterLabel = masterLabel.String
	}
	if desc.Valid {
		v.Description = desc.String
	}
	if metadata.Valid {
		v.MetadataJSON = metadata.String
	}
	if source.Valid {
		v.Source = source.String
	}
	if isDeleted.Valid {
		v.IsDeleted = isDeleted.Bool
	}
	if createdDate.Valid {
		v.CreatedDate = createdDate.String
	}
	if lastModifiedDate.Valid {
		v.LastModifiedDate = lastModifiedDate.String
	}
	v.HydrateMetadata()
	return &v, nil
}

// GetFlowVersionsByIDs retrieves multiple flow versions by IDs in a single query.
func (r *MetadataRepository) GetFlowVersionsByIDs(ctx context.Context, ids []string) (map[string]*models.FlowVersion, error) {
	if len(ids) == 0 {
		return nil, nil
	}

	placeholders := make([]string, len(ids))
	args := make([]interface{}, len(ids))
	for i, id := range ids {
		placeholders[i] = "?"
		args[i] = id
	}

	query := fmt.Sprintf(
		`SELECT Id, DefinitionId, MasterLabel, VersionNumber, Status, Description, ProcessType, Metadata,
			Source, IsDeleted, CreatedDate, LastModifiedDate FROM Flow WHERE Id IN (%s)`,
		strings.Join(placeholders, ","),
	)

	rows, err := r.GetExecutor(ctx).QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	result := make(map[string]*models.FlowVersion, len(ids))
	for rows.Next() {
		var v models.FlowVersion
		var masterLabel, desc, metadata, source sql.NullString
		var isDeleted sql.NullBool
		var createdDate, lastModifiedDate sql.NullString
		if err := rows.Scan(&v.Id, &v.DefinitionId, &masterLabel, &v.VersionNumber, &v.Status,
			&desc, &v.ProcessType, &metadata, &source, &isDeleted, &createdDate, &lastModifiedDate); err != nil {
			return nil, err
		}
		if masterLabel.Valid {
			v.MasterLabel = masterLabel.String
		}
		if desc.Valid {
			v.Description = desc.String
		}
		if metadata.Valid {
			v.MetadataJSON = metadata.String
		}
		if source.Valid {
			v.Source = source.String
		}
		if isDeleted.Valid {
			v.IsDeleted = isDeleted.Bool
		}
		if createdDate.Valid {
			v.CreatedDate = createdDate.String
		}
		if lastModifiedDate.Valid {
			v.LastModifiedDate = lastModifiedDate.String
		}
		v.HydrateMetadata()
		result[v.Id] = &v
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return result, nil
}

// GetActiveFlowVersion returns the active version for a flow definition.
func (r *MetadataRepository) GetActiveFlowVersion(ctx context.Context, definitionId string) (*models.FlowVersion, error) {
	query := `SELECT Id, DefinitionId, MasterLabel, VersionNumber, Status, Description, ProcessType, Metadata,
		Source, IsDeleted, CreatedDate, LastModifiedDate FROM Flow WHERE DefinitionId = ? AND Status = 'Active' LIMIT 1`

	var v models.FlowVersion
	var masterLabel, desc, metadata, source sql.NullString
	var isDeleted sql.NullBool
	var createdDate, lastModifiedDate sql.NullString
	err := r.GetExecutor(ctx).QueryRowContext(ctx, query, definitionId).Scan(&v.Id, &v.DefinitionId, &masterLabel, &v.VersionNumber, &v.Status,
		&desc, &v.ProcessType, &metadata, &source, &isDeleted, &createdDate, &lastModifiedDate)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	if masterLabel.Valid {
		v.MasterLabel = masterLabel.String
	}
	if desc.Valid {
		v.Description = desc.String
	}
	if metadata.Valid {
		v.MetadataJSON = metadata.String
	}
	if source.Valid {
		v.Source = source.String
	}
	if isDeleted.Valid {
		v.IsDeleted = isDeleted.Bool
	}
	if createdDate.Valid {
		v.CreatedDate = createdDate.String
	}
	if lastModifiedDate.Valid {
		v.LastModifiedDate = lastModifiedDate.String
	}
	v.HydrateMetadata()
	return &v, nil
}

// CreateFlowVersion creates a new tooling-authored flow version. Tooling
// writes always mark the version as custom.
func (r *MetadataRepository) CreateFlowVersion(ctx context.Context, version *models.FlowVersion) error {
	return r.createFlowVersionWithSource(ctx, version, flowVersionSourceCustom)
}

// CreateSeededFlowVersion creates a pack-authored flow version. Control-plane
// seeders use this path so shipped versions stay Source='standard' and do not
// masquerade as admin-authored versions.
func (r *MetadataRepository) CreateSeededFlowVersion(ctx context.Context, version *models.FlowVersion) error {
	return r.createFlowVersionWithSource(ctx, version, flowVersionSourceStandard)
}

// CreateTenantFlowVersion creates a tenant-overlay flow version. Tenant overlay
// flows are org-authored runtime metadata, not protected platform-standard flows,
// so they are written with Source='custom' and can be executed by the generic
// automation path.
func (r *MetadataRepository) CreateTenantFlowVersion(ctx context.Context, version *models.FlowVersion) error {
	return r.createFlowVersionWithSource(ctx, version, flowVersionSourceCustom)
}

func (r *MetadataRepository) createFlowVersionWithSource(ctx context.Context, version *models.FlowVersion, source string) error {
	version.SerializeMetadata()
	now := time.Now().Format(time.RFC3339)
	query := fmt.Sprintf(`INSERT INTO Flow (Id, DefinitionId, VersionNumber, Status, Description, ProcessType, Metadata,
		Source, CreatedDate, LastModifiedDate) VALUES (?, ?, ?, ?, ?, ?, ?, '%s', ?, ?)`, source)
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, version.Id, version.DefinitionId, version.VersionNumber, version.Status,
		version.Description, version.ProcessType, version.MetadataJSON, now, now)
	return err
}

// UpdateFlowVersion updates a flow version and flips Source to 'custom'.
func (r *MetadataRepository) UpdateFlowVersion(ctx context.Context, id string, version *models.FlowVersion) error {
	version.SerializeMetadata()
	now := time.Now().Format(time.RFC3339)
	query := fmt.Sprintf(`UPDATE Flow SET Status = ?, Description = ?, Metadata = ?, Source = '%s', LastModifiedDate = ? WHERE Id = ?`, flowVersionSourceCustom)
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, version.Status, version.Description, version.MetadataJSON, now, id)
	return err
}

// ActivateFlowVersion activates a flow version and deactivates others.
func (r *MetadataRepository) ActivateFlowVersion(ctx context.Context, id string) error {
	exec := r.GetExecutor(ctx)

	var definitionId string
	err := exec.QueryRowContext(ctx, "SELECT DefinitionId FROM Flow WHERE Id = ?", id).Scan(&definitionId)
	if err != nil {
		return err
	}
	if err := lockFlowDefinition(ctx, exec, definitionId); err != nil {
		return err
	}
	if err := lockFlowVersionSiblings(ctx, exec, definitionId); err != nil {
		return err
	}
	if exists, err := flowVersionExists(ctx, exec, id); err != nil {
		return err
	} else if !exists {
		return sql.ErrNoRows
	}
	latestID, err := latestFlowVersionID(ctx, exec, definitionId)
	if err != nil {
		return err
	}

	_, err = exec.ExecContext(ctx, "UPDATE Flow SET Status = 'Obsolete' WHERE DefinitionId = ? AND Status = 'Active' AND Id <> ?", definitionId, id)
	if err != nil {
		return err
	}

	_, err = exec.ExecContext(ctx, "UPDATE Flow SET Status = 'Active', IsDeleted = 0 WHERE Id = ?", id)
	if err != nil {
		return err
	}

	if latestID == "" {
		latestID = id
	}
	_, err = exec.ExecContext(ctx, "UPDATE FlowDefinition SET ActiveVersionId = ?, LatestVersionId = ?, IsDeleted = 0 WHERE Id = ?", id, latestID, definitionId)
	return err
}

// DeactivateFlowVersion deactivates a flow version.
func (r *MetadataRepository) DeactivateFlowVersion(ctx context.Context, id string) error {
	exec := r.GetExecutor(ctx)

	var definitionId string
	err := exec.QueryRowContext(ctx, "SELECT DefinitionId FROM Flow WHERE Id = ?", id).Scan(&definitionId)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil
		}
		return err
	}
	if err := lockFlowDefinition(ctx, exec, definitionId); err != nil {
		return err
	}
	if err := lockFlowVersionSiblings(ctx, exec, definitionId); err != nil {
		return err
	}
	if exists, err := flowVersionExists(ctx, exec, id); err != nil {
		return err
	} else if !exists {
		return nil
	}
	latestID, err := latestFlowVersionID(ctx, exec, definitionId)
	if err != nil {
		return err
	}

	_, err = exec.ExecContext(ctx, "UPDATE Flow SET Status = 'Draft' WHERE Id = ?", id)
	if err != nil {
		return err
	}
	_, err = exec.ExecContext(
		ctx,
		"UPDATE FlowDefinition SET ActiveVersionId = CASE WHEN ActiveVersionId = ? THEN NULL ELSE ActiveVersionId END, LatestVersionId = ? WHERE Id = ?",
		id, nullableString(latestID), definitionId,
	)
	return err
}

// DeleteFlowVersion deletes a flow version.
func (r *MetadataRepository) DeleteFlowVersion(ctx context.Context, id string) error {
	exec := r.GetExecutor(ctx)

	var definitionId string
	err := exec.QueryRowContext(ctx, "SELECT DefinitionId FROM Flow WHERE Id = ?", id).Scan(&definitionId)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil
		}
		return err
	}
	activeID, err := lockFlowDefinitionActiveVersion(ctx, exec, definitionId)
	if err != nil {
		return err
	}
	if err := lockFlowVersionSiblings(ctx, exec, definitionId); err != nil {
		return err
	}
	if exists, err := flowVersionExists(ctx, exec, id); err != nil {
		return err
	} else if !exists {
		return nil
	}

	if _, err := exec.ExecContext(ctx, `DELETE FROM Flow WHERE Id = ?`, id); err != nil {
		return err
	}
	latestID, err := latestFlowVersionID(ctx, exec, definitionId)
	if err != nil {
		return err
	}
	nextActiveID := activeID
	if activeID == id {
		nextActiveID = ""
	}
	_, err = exec.ExecContext(
		ctx,
		"UPDATE FlowDefinition SET ActiveVersionId = ?, LatestVersionId = ? WHERE Id = ?",
		nullableString(nextActiveID), nullableString(latestID), definitionId,
	)
	return err
}

func lockFlowDefinition(ctx context.Context, exec Executor, definitionId string) error {
	_, err := lockFlowDefinitionActiveVersion(ctx, exec, definitionId)
	return err
}

func lockFlowDefinitionActiveVersion(ctx context.Context, exec Executor, definitionId string) (string, error) {
	var active sql.NullString
	err := exec.QueryRowContext(ctx, "SELECT ActiveVersionId FROM FlowDefinition WHERE Id = ? FOR UPDATE", definitionId).Scan(&active)
	if err != nil {
		return "", err
	}
	if !active.Valid {
		return "", nil
	}
	return active.String, nil
}

func lockFlowVersionSiblings(ctx context.Context, exec Executor, definitionId string) error {
	rows, err := exec.QueryContext(ctx, "SELECT Id FROM Flow WHERE DefinitionId = ? FOR UPDATE", definitionId)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var ignored string
		if err := rows.Scan(&ignored); err != nil {
			return err
		}
	}
	return rows.Err()
}

func latestFlowVersionID(ctx context.Context, exec Executor, definitionId string) (string, error) {
	var latestID string
	err := exec.QueryRowContext(ctx, "SELECT Id FROM Flow WHERE DefinitionId = ? ORDER BY VersionNumber DESC, Id DESC LIMIT 1 FOR UPDATE", definitionId).Scan(&latestID)
	if err != nil {
		if err == sql.ErrNoRows {
			return "", nil
		}
		return "", err
	}
	return latestID, nil
}

func flowVersionExists(ctx context.Context, exec Executor, id string) (bool, error) {
	var found int
	err := exec.QueryRowContext(ctx, "SELECT 1 FROM Flow WHERE Id = ? LIMIT 1 FOR UPDATE", id).Scan(&found)
	if err != nil {
		if err == sql.ErrNoRows {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

// GetFlowInterviews returns all flow interviews.
func (r *MetadataRepository) GetFlowInterviews(ctx context.Context) ([]*models.FlowInterview, error) {
	query := `SELECT Id, FlowVersionId, InterviewLabel, InterviewStatus, CurrentElement, OwnerId,
		ContextData, CreatedDate, PausedDate FROM FlowInterview
		WHERE InterviewStatus IN ('Running', 'Paused') ORDER BY CreatedDate DESC`

	rows, err := r.GetExecutor(ctx).QueryContext(ctx, query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	interviews := make([]*models.FlowInterview, 0)
	for rows.Next() {
		var i models.FlowInterview
		var label, currentElement, contextData sql.NullString
		var pausedDate sql.NullTime
		if err := rows.Scan(&i.Id, &i.FlowVersionId, &label, &i.InterviewStatus, &currentElement,
			&i.OwnerId, &contextData, &i.CreatedDate, &pausedDate); err != nil {
			log.Printf("Warning: Failed to scan flow interview: %v", err)
			continue
		}
		if label.Valid {
			i.InterviewLabel = label.String
		}
		if currentElement.Valid {
			i.CurrentElement = currentElement.String
		}
		if contextData.Valid {
			i.ContextDataJSON = contextData.String
		}
		if pausedDate.Valid {
			i.PausedDate = &pausedDate.Time
		}
		interviews = append(interviews, &i)
	}
	return interviews, rows.Err()
}

// GetFlowInterview returns a flow interview by ID.
func (r *MetadataRepository) GetFlowInterview(ctx context.Context, id string) (*models.FlowInterview, error) {
	query := `SELECT Id, FlowVersionId, InterviewLabel, InterviewStatus, CurrentElement, OwnerId,
		ContextData, CreatedDate, PausedDate FROM FlowInterview WHERE Id = ?`

	var i models.FlowInterview
	var label, currentElement, contextData sql.NullString
	var pausedDate sql.NullTime
	err := r.GetExecutor(ctx).QueryRowContext(ctx, query, id).Scan(&i.Id, &i.FlowVersionId, &label, &i.InterviewStatus,
		&currentElement, &i.OwnerId, &contextData, &i.CreatedDate, &pausedDate)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	if label.Valid {
		i.InterviewLabel = label.String
	}
	if currentElement.Valid {
		i.CurrentElement = currentElement.String
	}
	if contextData.Valid {
		i.ContextDataJSON = contextData.String
	}
	if pausedDate.Valid {
		i.PausedDate = &pausedDate.Time
	}
	return &i, nil
}

// SaveFlowInterview creates or updates a flow interview.
func (r *MetadataRepository) SaveFlowInterview(ctx context.Context, interview *models.FlowInterview) error {
	query := `INSERT INTO FlowInterview (Id, FlowVersionId, InterviewLabel, InterviewStatus, CurrentElement,
		OwnerId, ContextData, CreatedDate, PausedDate) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE InterviewStatus = VALUES(InterviewStatus), CurrentElement = VALUES(CurrentElement),
		ContextData = VALUES(ContextData), PausedDate = VALUES(PausedDate)`
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, interview.Id, interview.FlowVersionId, interview.InterviewLabel,
		interview.InterviewStatus, interview.CurrentElement, interview.OwnerId, interview.ContextDataJSON,
		interview.CreatedDate, interview.PausedDate)
	return err
}

// DeleteFlowInterview deletes a flow interview.
func (r *MetadataRepository) DeleteFlowInterview(ctx context.Context, id string) error {
	query := `DELETE FROM FlowInterview WHERE Id = ?`
	_, err := r.GetExecutor(ctx).ExecContext(ctx, query, id)
	return err
}