package connector_workqueue

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"time"
)

// ReplayReservation describes which mutable owner reserved replay for a dead
// Event. WorkItemOwned=false is the explicit legacy-generation fallback. A
// zero-attempt compatibility mirror can still contribute WorkID and
// ProjectionKey so replay preserves lineage and projection conflict checks.
type ReplayReservation struct {
	WorkItemOwned bool
	Reserved      bool
	WorkID        string
	ProjectionKey string
}

// ReserveDeadWorkItemReplay reserves replay on the terminal WorkItem generation
// for one Event-backed work type. Both event-backed WorkTypes are inspected
// because compatibility data can predate the workflow-action type split. A
// zero-attempt compatibility mirror has never acquired WorkItem execution
// authority and remains on the legacy Event reservation path, while still
// supplying lineage and projection identity to the new generation.
func (r *Repository) ReserveDeadWorkItemReplay(ctx context.Context, eventID, workType string, now time.Time) (ReplayReservation, error) {
	eventID = strings.TrimSpace(eventID)
	workType = strings.TrimSpace(workType)
	if eventID == "" ||
		(workType != WorkTypeConnectorEvent && workType != WorkTypeWorkflowActionApply) ||
		now.IsZero() {
		return ReplayReservation{}, ErrInvalidWorkItem
	}

	rows, err := r.db.QueryContext(ctx, `
	SELECT w.WorkId, COALESCE(w.ProjectionKey, ''), w.State, w.ReplayOfWorkId, w.ReplayedAt, w.Attempt
	  FROM _System_ConnectorWorkItem w
	 WHERE w.EventId = ?
	   AND w.WorkType IN (?, ?)
 ORDER BY CASE WHEN w.WorkType = ? THEN 0 ELSE 1 END, w.WorkId
 LIMIT 2
 FOR UPDATE`,
		eventID, WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, workType)
	if isDeadWorkItemSchemaMissing(err) {
		// Preserve the identity of a compatibility mirror even while optional
		// replay/projection columns are rolling out. Returning an empty
		// reservation here would let Event replay bypass the durable fence and
		// commit an Event-only child alongside a still-claimable WorkItem.
		rows, err = r.db.QueryContext(ctx, `
SELECT w.WorkId, w.State, w.Attempt
  FROM _System_ConnectorWorkItem w
 WHERE w.EventId = ?
   AND w.WorkType IN (?, ?)
 ORDER BY CASE WHEN w.WorkType = ? THEN 0 ELSE 1 END, w.WorkId
 LIMIT 2
 FOR UPDATE`,
			eventID, WorkTypeConnectorEvent, WorkTypeWorkflowActionApply, workType)
		if connectorWorkItemTableMissing(err) {
			return ReplayReservation{}, nil
		}
		if err != nil {
			return ReplayReservation{}, err
		}
		defer rows.Close()

		var reservation ReplayReservation
		var state string
		var attempt int
		if !rows.Next() {
			return ReplayReservation{}, rows.Err()
		}
		if err := rows.Scan(&reservation.WorkID, &state, &attempt); err != nil {
			return ReplayReservation{}, err
		}
		if rows.Next() {
			return ReplayReservation{}, ErrReplayBlocked
		}
		if err := rows.Err(); err != nil {
			return ReplayReservation{}, err
		}
		reservation.WorkItemOwned = attempt > 0
		// The incomplete schema cannot safely reserve a WorkItem-owned replay.
		// Returning its identity unreserved makes the caller fail closed. An
		// attempt-zero mirror proceeds to the fence, and any unavailable child
		// materialization rolls the entire Event reservation back.
		return reservation, nil
	}
	if err != nil {
		return ReplayReservation{}, err
	}
	defer rows.Close()

	var reservation ReplayReservation
	var state string
	var replayOfWorkID sql.NullString
	var replayedAt sql.NullTime
	var attempt int
	if !rows.Next() {
		return ReplayReservation{}, rows.Err()
	}
	if err := rows.Scan(
		&reservation.WorkID,
		&reservation.ProjectionKey,
		&state,
		&replayOfWorkID,
		&replayedAt,
		&attempt,
	); err != nil {
		return ReplayReservation{}, err
	}
	if rows.Next() {
		// Multiple event-backed WorkItems for one Event make execution ownership
		// ambiguous. Fail closed instead of reserving or replaying either row.
		return ReplayReservation{}, ErrReplayBlocked
	}
	if err := rows.Err(); err != nil {
		return ReplayReservation{}, err
	}

	reservation.WorkItemOwned = attempt > 0
	if reservation.WorkItemOwned && (state != WorkStateDead || replayedAt.Valid) {
		return reservation, nil
	}
	if reservation.ProjectionKey != "" {
		var activeWorkID string
		err = r.db.QueryRowContext(ctx, `
SELECT WorkId
  FROM _System_ConnectorWorkItem
 WHERE ProjectionActiveKey = ?
   AND WorkId <> ?
   AND State NOT IN (?, ?, ?)
 LIMIT 1
 FOR UPDATE`,
			reservation.ProjectionKey, reservation.WorkID,
			WorkStateProcessed, WorkStateSkipped, WorkStateDead).Scan(&activeWorkID)
		if err == nil {
			return ReplayReservation{}, ErrReplayBlocked
		}
		if replayFenceColumnMissing(err, "ProjectionActiveKey") {
			// Keep the original identity so Event replay cannot downgrade to
			// an Event-only child. Attempt-zero callers will reach the partial
			// schema fence; WorkItem-owned callers remain unreserved and fail
			// closed.
			return reservation, nil
		}
		if !errors.Is(err, sql.ErrNoRows) {
			return ReplayReservation{}, err
		}
	}
	if !reservation.WorkItemOwned {
		return reservation, nil
	}

	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET ReplayedAt = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND State = ?
   AND ReplayedAt IS NULL`,
		now, now, reservation.WorkID, eventID, WorkStateDead)
	if err != nil {
		return ReplayReservation{}, err
	}
	reservation.Reserved, err = rowsAffected(res)
	return reservation, err
}

// FenceZeroAttemptReplayMirror terminalizes a compatibility WorkItem after the
// Event-owned replay reservation succeeds. This prevents the original
// zero-attempt row from becoming executable when WorkItem readiness stops
// consulting Event lifecycle state.
func (r *Repository) FenceZeroAttemptReplayMirror(ctx context.Context, eventID, workID string, now time.Time) (bool, error) {
	eventID = strings.TrimSpace(eventID)
	workID = strings.TrimSpace(workID)
	if eventID == "" || workID == "" || now.IsZero() {
		return false, ErrInvalidWorkItem
	}
	includeTerminalAt := true
	includeProjectionActiveKey := true
	for {
		setClauses := []string{
			"State = ?",
			"LeaseOwner = NULL",
			"LeaseToken = NULL",
			"LeaseUntil = NULL",
		}
		args := []any{WorkStateDead}
		if includeTerminalAt {
			setClauses = append(setClauses, "TerminalAt = COALESCE(TerminalAt, ?)")
			args = append(args, now)
		}
		setClauses = append(setClauses, "ReplayedAt = ?")
		args = append(args, now)
		if includeProjectionActiveKey {
			setClauses = append(setClauses, "ProjectionActiveKey = NULL")
		}
		setClauses = append(setClauses, "LastModifiedDate = ?")
		args = append(args, now,
			workID, eventID,
			WorkStatePending, WorkStateRetry, WorkStateBlocked, WorkStateBlockedRetry, WorkStateDead,
		)

		res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET `+strings.Join(setClauses, ",\n       ")+`
 WHERE WorkId = ?
   AND EventId = ?
   AND Attempt = 0
   AND State IN (?, ?, ?, ?, ?)
   AND ReplayedAt IS NULL`, args...)
		if err == nil {
			return rowsAffected(res)
		}
		if includeTerminalAt && replayFenceColumnMissing(err, "TerminalAt") {
			includeTerminalAt = false
			continue
		}
		if includeProjectionActiveKey && replayFenceColumnMissing(err, "ProjectionActiveKey") {
			includeProjectionActiveKey = false
			continue
		}
		return false, err
	}
}

func replayFenceColumnMissing(err error, column string) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, strings.ToLower(column)) &&
		(strings.Contains(msg, "1054") ||
			strings.Contains(msg, "unknown column") ||
			strings.Contains(msg, "no such column"))
}

func connectorWorkItemTableMissing(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "_system_connectorworkitem") &&
		(strings.Contains(msg, "1146") ||
			strings.Contains(msg, "unknown table") ||
			strings.Contains(msg, "doesn't exist") ||
			strings.Contains(msg, "does not exist") ||
			strings.Contains(msg, "no such table"))
}