package connector_workqueue

import (
	"context"
	"strings"
)

type ordinaryWorkItemMaterializationMode uint8

const (
	ordinaryWorkItemDirect ordinaryWorkItemMaterializationMode = iota
	ordinaryWorkItemReconcileLane
	ordinaryWorkItemDeferLane
)

type persistedOrdinaryWorkItemIdentity struct {
	workID                string
	eventID               string
	installedPackID       string
	workType              string
	queueClass            string
	sourceLaneKey         string
	targetMutationLaneKey string
	laneOrdinal           int64
	generation            int64
	replayOfWorkID        string
	projectionKey         string
	projectionActiveKey   string
}

type workItemIdentityReadSchema struct {
	replayOfWorkID      bool
	projectionKey       bool
	projectionActiveKey bool
}

func allWorkItemIdentityColumns() workItemIdentityReadSchema {
	return workItemIdentityReadSchema{
		replayOfWorkID:      true,
		projectionKey:       true,
		projectionActiveKey: true,
	}
}

// materializeOrdinaryWorkItem owns idempotent convergence for non-replay,
// non-projection WorkItems. Fresh materialization takes the optimistic insert
// path and does not depend on optional rolling-schema columns. After a 1062,
// both durable identities are resolved by one locking current read in
// primary-key order. Existing optional identity columns remain fail-closed;
// only a specifically missing column is removed and retried.
func (r *Repository) materializeOrdinaryWorkItem(
	ctx context.Context,
	raw CreateWorkItemInput,
	mode ordinaryWorkItemMaterializationMode,
) (string, bool, error) {
	in := r.defaultCreateWorkItemInput(raw)
	if err := validateCreateWorkItemInput(in); err != nil {
		return "", false, err
	}

	reconcileLane := mode == ordinaryWorkItemReconcileLane &&
		isReadyWorkState(in.State) && workItemLaneKey(in) != ""
	if mode != ordinaryWorkItemDirect && workItemLaneKey(in) != "" {
		in.State = blockedStateForReadyState(in.State)
	}

	if err := r.insertWorkItem(ctx, in, ""); err != nil {
		if !isDuplicateKey(err) {
			return "", false, err
		}
		existing, loadErr := r.loadCurrentOrdinaryWorkItemIdentityForUpdate(ctx, in)
		if loadErr != nil {
			return "", false, loadErr
		}
		if existing == nil {
			return "", false, newWorkItemIdentityConflict(
				in,
				"",
				"duplicate key was not visible through the current identity read",
			)
		}
		return existing.workID, false, nil
	}

	if reconcileLane {
		if err := r.reconcileLaneReadyState(ctx, in); err != nil {
			return "", false, err
		}
	}
	return in.WorkID, true, nil
}

func (r *Repository) loadCurrentOrdinaryWorkItemIdentityForUpdate(
	ctx context.Context,
	in CreateWorkItemInput,
) (*persistedOrdinaryWorkItemIdentity, error) {
	schema := allWorkItemIdentityColumns()
	for {
		rows, err := r.db.QueryContext(ctx, ordinaryWorkItemIdentityQuery(schema), in.WorkID, in.EventID, in.WorkType)
		if err != nil {
			switch {
			case schema.replayOfWorkID && replayFenceColumnMissing(err, "ReplayOfWorkId"):
				schema.replayOfWorkID = false
				continue
			case schema.projectionKey && replayFenceColumnMissing(err, "ProjectionKey"):
				schema.projectionKey = false
				continue
			case schema.projectionActiveKey && replayFenceColumnMissing(err, "ProjectionActiveKey"):
				schema.projectionActiveKey = false
				continue
			default:
				return nil, err
			}
		}

		row, scanErr := resolveCurrentOrdinaryWorkItemIdentity(rows, schema, in)
		closeErr := rows.Close()
		if scanErr != nil {
			return nil, scanErr
		}
		if closeErr != nil {
			return nil, closeErr
		}
		return row, nil
	}
}

func ordinaryWorkItemIdentityQuery(schema workItemIdentityReadSchema) string {
	columns := []string{
		"WorkId",
		"EventId",
		"InstalledPackId",
		"WorkType",
		"QueueClass",
		"COALESCE(SourceLaneKey, '')",
		"COALESCE(TargetMutationLaneKey, '')",
		"LaneOrdinal",
		"Generation",
	}
	if schema.replayOfWorkID {
		columns = append(columns, "COALESCE(ReplayOfWorkId, '')")
	}
	if schema.projectionKey {
		columns = append(columns, "COALESCE(ProjectionKey, '')")
	}
	if schema.projectionActiveKey {
		columns = append(columns, "COALESCE(ProjectionActiveKey, '')")
	}
	return `
SELECT ` + strings.Join(columns, ", ") + `
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
    OR (EventId = ? AND WorkType = ?)
 ORDER BY WorkId
 LIMIT 2
 FOR UPDATE`
}

type workItemIdentityRows interface {
	Next() bool
	Scan(dest ...any) error
	Err() error
}

func resolveCurrentOrdinaryWorkItemIdentity(
	rows workItemIdentityRows,
	schema workItemIdentityReadSchema,
	in CreateWorkItemInput,
) (*persistedOrdinaryWorkItemIdentity, error) {
	var byWorkID *persistedOrdinaryWorkItemIdentity
	var byBusinessKey *persistedOrdinaryWorkItemIdentity
	for rows.Next() {
		row := &persistedOrdinaryWorkItemIdentity{}
		dest := []any{
			&row.workID,
			&row.eventID,
			&row.installedPackID,
			&row.workType,
			&row.queueClass,
			&row.sourceLaneKey,
			&row.targetMutationLaneKey,
			&row.laneOrdinal,
			&row.generation,
		}
		if schema.replayOfWorkID {
			dest = append(dest, &row.replayOfWorkID)
		}
		if schema.projectionKey {
			dest = append(dest, &row.projectionKey)
		}
		if schema.projectionActiveKey {
			dest = append(dest, &row.projectionActiveKey)
		}
		if err := rows.Scan(dest...); err != nil {
			return nil, err
		}
		if row.workID == in.WorkID {
			if byWorkID != nil && byWorkID.workID != row.workID {
				return nil, newWorkItemIdentityConflict(in, row.workID, "multiple rows resolved the requested WorkId")
			}
			byWorkID = row
		}
		if row.eventID == in.EventID && row.workType == in.WorkType {
			if byBusinessKey != nil && byBusinessKey.workID != row.workID {
				return nil, newWorkItemIdentityConflict(in, row.workID, "multiple rows resolved the requested EventId and WorkType")
			}
			byBusinessKey = row
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if byWorkID != nil && byBusinessKey != nil && byWorkID.workID != byBusinessKey.workID {
		return nil, newWorkItemIdentityConflict(
			in,
			byBusinessKey.workID,
			"WorkId and EventId/WorkType resolve to different persisted rows",
		)
	}

	row := byWorkID
	if row == nil {
		row = byBusinessKey
	}
	if row == nil {
		return nil, nil
	}
	if !ordinaryWorkItemIdentityMatches(row, in) {
		return nil, newWorkItemIdentityConflict(in, row.workID, "persisted immutable fields differ")
	}
	return row, nil
}

func ordinaryWorkItemIdentityMatches(row *persistedOrdinaryWorkItemIdentity, in CreateWorkItemInput) bool {
	return row != nil &&
		row.eventID == in.EventID &&
		row.installedPackID == in.InstalledPackID &&
		row.workType == in.WorkType &&
		row.queueClass == in.QueueClass &&
		row.sourceLaneKey == in.SourceLaneKey &&
		row.targetMutationLaneKey == in.TargetMutationLaneKey &&
		row.laneOrdinal == in.LaneOrdinal &&
		row.generation == in.Generation &&
		row.replayOfWorkID == "" &&
		row.projectionKey == "" &&
		row.projectionActiveKey == ""
}

func newWorkItemIdentityConflict(in CreateWorkItemInput, existingWorkID, reason string) *WorkItemIdentityConflictError {
	return &WorkItemIdentityConflictError{
		RequestedWorkID: in.WorkID,
		EventID:         in.EventID,
		WorkType:        in.WorkType,
		ExistingWorkID:  existingWorkID,
		Reason:          reason,
	}
}