package controlplane

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

	bootstate "github.com/nexuscrm/db/pkg/bootstate"
)

const migrationIdBackfillConnectorEventSourceLanes = "20260622_backfill_connector_event_source_lanes"

const connectorEventSourceLaneBackfillUpdateSQL = `
UPDATE _System_ConnectorEvent
   SET SourceObjectKind = NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.objectKind')), ''),
       SourceObjectID = NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.sourceID')), ''),
       SourceLaneKey = SHA2(CONCAT(
         LENGTH(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.objectKind')), '')), ':',
         NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.objectKind')), ''), ':',
         LENGTH(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.sourceID')), '')), ':',
         NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.sourceID')), '')
       ), 256)
 WHERE Payload IS NOT NULL
   AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.objectKind')), '') IS NOT NULL
   AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.sourceID')), '') IS NOT NULL
   AND (
        SourceObjectKind IS NULL OR SourceObjectKind = ''
        OR SourceObjectID IS NULL OR SourceObjectID = ''
        OR SourceLaneKey IS NULL OR SourceLaneKey = ''
   )`

// BackfillConnectorEventSourceLanes materializes queue-envelope objectKind and
// sourceID into indexed connector-event columns before the inbound worker reads
// from the backlog. Runtime enqueue writes these columns for new rows; this
// migration bridges warm-upgrade rows that only carried the values in Payload.
func BackfillConnectorEventSourceLanes(ctx context.Context, db *sql.DB) (int, error) {
	if db == nil {
		return 0, fmt.Errorf("BackfillConnectorEventSourceLanes requires a database")
	}
	applied, err := bootstate.MigrationApplied(ctx, db, migrationIdBackfillConnectorEventSourceLanes)
	if err != nil {
		return 0, err
	}
	if applied {
		remaining, err := bootstate.ConnectorEventSourceLanePayloadOnlyCount(ctx, db)
		if err != nil {
			return 0, err
		}
		if remaining == 0 {
			log.Println("   BackfillConnectorEventSourceLanes: already applied; skipping")
			return 0, nil
		}
		log.Printf("   BackfillConnectorEventSourceLanes: completed receipt exists but %d payload-only row(s) remain; repairing", remaining)
	}

	ready, err := bootstate.ConnectorEventSourceLaneColumnsReady(ctx, db)
	if err != nil {
		return 0, err
	}
	if !ready {
		return 0, fmt.Errorf("BackfillConnectorEventSourceLanes: required _System_ConnectorEvent columns missing after reconciliation")
	}

	res, err := db.ExecContext(ctx, connectorEventSourceLaneBackfillUpdateSQL)
	if err != nil {
		return 0, fmt.Errorf("BackfillConnectorEventSourceLanes: %w", err)
	}
	touched, err := res.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("BackfillConnectorEventSourceLanes rows affected: %w", err)
	}

	remaining, err := bootstate.ConnectorEventSourceLanePayloadOnlyCount(ctx, db)
	if err != nil {
		return int(touched), fmt.Errorf("BackfillConnectorEventSourceLanes closure check: %w", err)
	}
	if remaining != 0 {
		return int(touched), fmt.Errorf("BackfillConnectorEventSourceLanes: %d connector event row(s) still have envelope source lanes only in Payload", remaining)
	}

	if err := bootstate.RecordMigrationApplied(
		ctx, db,
		migrationIdBackfillConnectorEventSourceLanes,
		1,
		"Backfill connector event source-lane columns before indexed inbound claim cutover.",
	); err != nil {
		return int(touched), err
	}

	if touched > 0 {
		log.Printf("   BackfillConnectorEventSourceLanes: backfilled %d connector event row(s)", touched)
	}
	return int(touched), nil
}