package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
)

// readFlow reads the latest (by VersionNumber DESC) Flow version row for a
// given FlowDefinition.DeveloperName along with the definition's current
// ProcessType. The active version is what the engine dispatches to, but
// since Phase 0 just set ActiveVersionId='' for the DROP list, we walk
// by most-recent version instead.
func readFlow(ctx context.Context, db *sql.DB, developerName string) (*FlowRecord, error) {
	const q = `
		SELECT fd.Id, fd.ProcessType, f.Id, f.Metadata
		FROM FlowDefinition fd
		JOIN flow f ON f.DefinitionId = fd.Id
		WHERE fd.DeveloperName = ?
		ORDER BY f.VersionNumber DESC
		LIMIT 1`

	var (
		fdID, fvID, processType, metadata string
	)
	err := db.QueryRowContext(ctx, q, developerName).Scan(&fdID, &processType, &fvID, &metadata)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, fmt.Errorf("flow %q not found", developerName)
		}
		return nil, fmt.Errorf("query flow %q: %w", developerName, err)
	}

	var parsed map[string]any
	if err := json.Unmarshal([]byte(metadata), &parsed); err != nil {
		return nil, fmt.Errorf("parse metadata for %q: %w", developerName, err)
	}

	return &FlowRecord{
		DeveloperName:    developerName,
		FlowDefinitionID: fdID,
		FlowVersionID:    fvID,
		ProcessType:      processType,
		Metadata:         parsed,
		MetadataRaw:      metadata,
	}, nil
}

// applyOneFlow reads the flow, migrates it in memory, and — inside a
// single transaction — updates FlowDefinition (ProcessType, TriggerType,
// TriggerObjectOrEvent) and the active Flow version's Metadata +
// ProcessType. Already-migrated flows are a no-op (returns 0, 0, nil).
//
// The FOR UPDATE row lock on FlowDefinition prevents a concurrent writer
// from resurrecting a legacy version between our read and write.
func applyOneFlow(ctx context.Context, db *sql.DB, out io.Writer, developerName string) (fdAffected, fvAffected int64, err error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return 0, 0, fmt.Errorf("begin tx: %w", err)
	}
	defer func() {
		if err != nil {
			_ = tx.Rollback()
		}
	}()

	// Lock + read FlowDefinition first. We need its Id to scope the
	// subsequent flow-row lookup.
	var fdID, oldProcessType, oldTriggerType, oldTriggerObject string
	err = tx.QueryRowContext(ctx,
		"SELECT Id, ProcessType, COALESCE(TriggerType,''), COALESCE(TriggerObjectOrEvent,'') "+
			"FROM FlowDefinition WHERE DeveloperName = ? FOR UPDATE",
		developerName,
	).Scan(&fdID, &oldProcessType, &oldTriggerType, &oldTriggerObject)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return 0, 0, fmt.Errorf("FlowDefinition %q not found", developerName)
		}
		return 0, 0, fmt.Errorf("lock FlowDefinition %q: %w", developerName, err)
	}

	// Lock + read the active / latest Flow version.
	var fvID, fvMetadata string
	err = tx.QueryRowContext(ctx,
		"SELECT Id, Metadata FROM flow WHERE DefinitionId = ? "+
			"ORDER BY VersionNumber DESC LIMIT 1 FOR UPDATE",
		fdID,
	).Scan(&fvID, &fvMetadata)
	if err != nil {
		return 0, 0, fmt.Errorf("lock flow row for %q: %w", developerName, err)
	}

	// Pre-change state as SQL comment so `tee` doubles as a rollback log.
	fmt.Fprintf(out, "-- %s: FlowDefinition.Id=%s pre-change ProcessType=%q TriggerType=%q TriggerObjectOrEvent=%q\n",
		developerName, fdID, oldProcessType, oldTriggerType, oldTriggerObject)
	fmt.Fprintf(out, "-- %s: flow.Id=%s metadata bytes=%d\n", developerName, fvID, len(fvMetadata))

	// Parse + migrate.
	var parsed map[string]any
	if err = json.Unmarshal([]byte(fvMetadata), &parsed); err != nil {
		return 0, 0, fmt.Errorf("parse metadata for %q: %w", developerName, err)
	}
	rec := &FlowRecord{
		DeveloperName:    developerName,
		FlowDefinitionID: fdID,
		FlowVersionID:    fvID,
		ProcessType:      oldProcessType,
		Metadata:         parsed,
		MetadataRaw:      fvMetadata,
	}
	result, migrateErr := Migrate(rec)
	if migrateErr != nil {
		return 0, 0, fmt.Errorf("migrate %q: %w", developerName, migrateErr)
	}

	if result.AlreadyMigrated {
		// Metadata is in post-migration shape but the FlowDefinition row
		// may still be out of sync (e.g. someone ran an older version
		// of this script, or the DB was partially hand-patched). Still
		// reconcile FlowDefinition so the trigger listener sees a
		// consistent view. The guarded WHERE clause makes this a true
		// no-op when everything already matches.
		res, reconcileErr := tx.ExecContext(ctx,
			"UPDATE FlowDefinition SET ProcessType = ?, TriggerType = ?, TriggerObjectOrEvent = ?, ActiveVersionId = '', LastModifiedDate = NOW() "+
				"WHERE Id = ? AND ("+
				"ProcessType <> ? OR COALESCE(TriggerType,'') <> ? OR COALESCE(TriggerObjectOrEvent,'') <> ? OR COALESCE(ActiveVersionId,'') <> ''"+
				")",
			result.NewProcessType, result.NewTriggerType, result.NewTriggerObject, fdID,
			result.NewProcessType, result.NewTriggerType, result.NewTriggerObject,
		)
		if reconcileErr != nil {
			return 0, 0, fmt.Errorf("reconcile FlowDefinition: %w", reconcileErr)
		}
		fdAffected, _ = res.RowsAffected()
		if commitErr := tx.Commit(); commitErr != nil {
			err = commitErr
			return 0, 0, fmt.Errorf("commit reconcile tx: %w", commitErr)
		}
		if fdAffected > 0 {
			fmt.Fprintf(out, "%s: metadata already migrated; FlowDefinition reconciled (rows=%d)\n", developerName, fdAffected)
		} else {
			fmt.Fprintf(out, "%s: already migrated (idempotent no-op)\n", developerName)
		}
		return fdAffected, 0, nil
	}

	// Phase 1 leaves the migrated flow INERT — Phase 4 re-activates in
	// waves. Clearing ActiveVersionId here means the trigger listener
	// stops dispatching to the flow until a deliberate activation
	// happens (Wave 1 / 2 / 3 per the design doc). Without this, a
	// legacy flow whose ActiveVersionId already points at the latest
	// version would become immediately live under the new format the
	// moment Lanes A/B/C merge.
	//
	// Guarded UPDATE FlowDefinition — only touch rows that actually change.
	res, err := tx.ExecContext(ctx,
		"UPDATE FlowDefinition SET ProcessType = ?, TriggerType = ?, TriggerObjectOrEvent = ?, ActiveVersionId = '', LastModifiedDate = NOW() "+
			"WHERE Id = ? AND ("+
			"ProcessType <> ? OR COALESCE(TriggerType,'') <> ? OR COALESCE(TriggerObjectOrEvent,'') <> ? OR COALESCE(ActiveVersionId,'') <> ''"+
			")",
		result.NewProcessType, result.NewTriggerType, result.NewTriggerObject, fdID,
		result.NewProcessType, result.NewTriggerType, result.NewTriggerObject,
	)
	if err != nil {
		return 0, 0, fmt.Errorf("update FlowDefinition: %w", err)
	}
	fdAffected, _ = res.RowsAffected()

	// Guarded UPDATE flow — only rewrite Metadata / ProcessType when they
	// actually differ. Comparing Metadata byte-for-byte would be fragile
	// (SF re-renders keys in different orders), so we instead compare
	// ProcessType and trust our idempotency gate (isAlreadyMigrated) above
	// to have detected "no changes needed" before reaching this branch.
	res, err = tx.ExecContext(ctx,
		"UPDATE flow SET Metadata = ?, ProcessType = ?, LastModifiedDate = NOW() "+
			"WHERE Id = ?",
		string(result.MigratedCompact), result.NewProcessType, fvID,
	)
	if err != nil {
		return 0, 0, fmt.Errorf("update flow: %w", err)
	}
	fvAffected, _ = res.RowsAffected()

	if err = tx.Commit(); err != nil {
		return 0, 0, fmt.Errorf("commit: %w", err)
	}

	fmt.Fprintf(out, "%s: FlowDefinition rows=%d, flow rows=%d (processType=%s, triggerType=%s, object=%s, rewrites=%d)\n",
		developerName, fdAffected, fvAffected,
		result.NewProcessType, result.NewTriggerType, result.NewTriggerObject, result.RewriteCount)
	return fdAffected, fvAffected, nil
}