package connector_workqueue

import (
	"context"
	"database/sql"
	"sort"
	"strings"
)

// AdmitSourceProjectionWorkItems is the batch form of ordinary fresh admission.
// The caller owns the admitted pack transaction and has inserted the immutable
// Events. Duplicates and explicit replay remain separate entrypoints.
func (r *Repository) AdmitSourceProjectionWorkItems(ctx context.Context, inputs []SourceProjectionWorkItemInput, deferLane bool) error {
	if _, ok := r.db.(*sql.Tx); !ok {
		return ErrInvalidWorkItem
	}
	_, err := r.admitProjectionBatch(ctx, inputs, deferLane, "", false)
	return err
}

type projectionWrite struct {
	input  SourceProjectionWorkItemInput
	workID string
	active bool
	insert bool
}

func (r *Repository) admitProjectionBatch(ctx context.Context, raw []SourceProjectionWorkItemInput, deferLane bool, replayOf string, replay bool) ([]bool, error) {
	inputs := append([]SourceProjectionWorkItemInput(nil), raw...)
	seen := map[string]bool{}
	for i := range inputs {
		in := &inputs[i]
		in.WorkItem = r.defaultCreateWorkItemInput(in.WorkItem)
		w := in.WorkItem
		if validateCreateWorkItemInput(w) != nil || !IsSourceProjectionWorkID(w.WorkID) || w.SourceLaneKey == "" || in.InputStartOrdinal <= 0 || w.LaneOrdinal < in.InputStartOrdinal || seen[w.WorkID] {
			return nil, ErrInvalidWorkItem
		}
		seen[w.WorkID] = true
	}
	sort.Slice(inputs, func(i, j int) bool { return inputs[i].WorkItem.WorkID < inputs[j].WorkItem.WorkID })
	created := make([]bool, 0, len(inputs))
	var lanes []CreateWorkItemInput
	for offset := 0; offset < len(inputs); offset += 128 {
		chunk := inputs[offset:min(offset+128, len(inputs))]
		if err := r.validateProjectionInputs(ctx, chunk); err != nil {
			return nil, err
		}
		keys := make([]string, len(chunk))
		for i, in := range chunk {
			keys[i] = in.WorkItem.WorkID
		}
		active, err := r.loadActiveProjectionBatch(ctx, keys)
		if err != nil {
			return nil, err
		}
		var frozen []string
		var terminal []string
		for _, in := range chunk {
			old := active[in.WorkItem.WorkID]
			if old == nil {
				continue
			}
			if err := validateSourceProjectionWorkItemRow(old, in.WorkItem, in.WorkItem.WorkID); err != nil {
				return nil, err
			}
			if isTerminalWorkState(old.state) {
				terminal = append(terminal, old.workID)
				delete(active, in.WorkItem.WorkID)
				continue
			}
			if replay {
				return nil, ErrReplayBlocked
			}
			if projectionGenerationFrozen(old) {
				frozen = append(frozen, in.WorkItem.WorkID)
			}
		}
		successors, err := r.loadProjectionSuccessorsBatch(ctx, frozen)
		if err != nil {
			return nil, err
		}
		writes := make([]projectionWrite, 0, len(chunk))
		for _, in := range chunk {
			key := in.WorkItem.WorkID
			old := active[key]
			isActive := old == nil || !projectionGenerationFrozen(old)
			if !isActive {
				old = successors[key]
			}
			if old != nil {
				if err := validateSourceProjectionWorkItemRow(old, in.WorkItem, key); err != nil {
					return nil, err
				}
				if err := requireProjectionRangeExtended(old, in.WorkItem, in.InputStartOrdinal); err != nil {
					return nil, err
				}
			}
			w := projectionWrite{input: in, active: isActive, insert: old == nil}
			if old == nil {
				w.workID = r.nextSourceProjectionWorkID(key)
			} else {
				w.workID = old.workID
			}
			if replayOf != "" && replayOf == w.workID {
				return nil, ErrInvalidWorkItem
			}
			if !isActive || deferLane || isReadyWorkState(in.WorkItem.State) {
				w.input.WorkItem.State = blockedStateForReadyState(in.WorkItem.State)
			}
			writes = append(writes, w)
			created = append(created, w.insert)
			if !deferLane && isActive && isReadyWorkState(in.WorkItem.State) {
				lanes = append(lanes, in.WorkItem)
			}
		}
		if len(terminal) > 0 {
			args := appendStrings(nil, terminal)
			res, err := r.db.ExecContext(ctx, `UPDATE _System_ConnectorWorkItem SET ProjectionActiveKey=NULL,
 TerminalAt=COALESCE(TerminalAt,LastModifiedDate) WHERE WorkId IN `+inSQL(terminal)+`
 AND State IN ('processed','skipped','dead')`, args...)
			if err != nil {
				return nil, err
			}
			if err := requireProjectionWrites(res, len(terminal)); err != nil {
				return nil, err
			}
		}
		if err := r.writeProjectionBatch(ctx, writes, replayOf); err != nil {
			return nil, err
		}
	}
	if err := r.ReconcileLaneReadyStates(ctx, lanes); err != nil {
		return nil, err
	}
	return created, nil
}

func projectionGenerationFrozen(row *sourceProjectionWorkItemRow) bool {
	return row.attempt > 0 || row.state == WorkStateProcessing || row.state == WorkStateRetry || row.state == WorkStateBlockedRetry
}

func (r *Repository) validateProjectionInputs(ctx context.Context, inputs []SourceProjectionWorkItemInput) error {
	if len(inputs) == 1 {
		return r.validateProjectionInput(ctx, inputs[0].WorkItem, inputs[0].InputStartOrdinal)
	}
	var requests []string
	var args []any
	for i, in := range inputs {
		requests = append(requests, "SELECT ? AS n, ? AS head_id, ? AS first_ordinal")
		args = append(args, i, in.WorkItem.EventID, in.InputStartOrdinal)
	}
	rows, err := r.db.QueryContext(ctx, `SELECT requested.n,head.InstalledPackId,head.EventKind,COALESCE(head.RouteVersion,0),
 COALESCE(head.CompanionPackInstallId,''),COALESCE(head.CompanionFlowVersionId,''),head.SourceLaneKey,head.EnqueueOrdinal
 FROM (`+strings.Join(requests, " UNION ALL ")+`) requested
 JOIN _System_ConnectorEvent head ON head.Id=requested.head_id
 WHERE EXISTS (SELECT 1 FROM _System_ConnectorEvent first
 WHERE first.InstalledPackId=head.InstalledPackId AND first.SourceLaneKey=head.SourceLaneKey
 AND first.EnqueueOrdinal=requested.first_ordinal AND LOWER(TRIM(first.EventKind))=LOWER(TRIM(head.EventKind))
 AND COALESCE(first.RouteVersion,0)=COALESCE(head.RouteVersion,0)
 AND COALESCE(first.CompanionPackInstallId,'')=COALESCE(head.CompanionPackInstallId,'')
 AND COALESCE(first.CompanionFlowVersionId,'')=COALESCE(head.CompanionFlowVersionId,''))`, args...)
	if err != nil {
		return err
	}
	defer rows.Close()
	seen := map[int]bool{}
	for rows.Next() {
		var n int
		var identity SourceProjectionIdentity
		var ordinal int64
		if err := rows.Scan(&n, &identity.InstalledPackID, &identity.EventKind, &identity.RouteVersion, &identity.CompanionPackInstallID, &identity.CompanionFlowVersionID, &identity.SourceLaneKey, &ordinal); err != nil {
			return err
		}
		if n < 0 || n >= len(inputs) || seen[n] {
			return ErrInvalidWorkItem
		}
		in := inputs[n].WorkItem
		identity.WorkType = in.WorkType
		if identity.InstalledPackID != in.InstalledPackID || identity.SourceLaneKey != in.SourceLaneKey || ordinal != in.LaneOrdinal || SourceProjectionKey(identity) != in.WorkID {
			return ErrInvalidWorkItem
		}
		seen[n] = true
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if len(seen) != len(inputs) {
		return ErrInvalidWorkItem
	}
	return nil
}

const projectionRowColumns = `WorkId,EventId,InstalledPackId,WorkType,COALESCE(SourceLaneKey,''),State,
 COALESCE(ProjectionKey,''),COALESCE(ProjectionActiveKey,''),COALESCE(InputStartOrdinal,0),LaneOrdinal,Attempt`

func (r *Repository) loadProjectionRows(ctx context.Context, where string, args []any) (map[string]*sourceProjectionWorkItemRow, error) {
	rows, err := r.db.QueryContext(ctx, "SELECT "+projectionRowColumns+" FROM _System_ConnectorWorkItem "+where, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string]*sourceProjectionWorkItemRow{}
	for rows.Next() {
		row := new(sourceProjectionWorkItemRow)
		if err := rows.Scan(&row.workID, &row.eventID, &row.installedPackID, &row.workType, &row.sourceLaneKey, &row.state, &row.projectionKey, &row.activeKey, &row.inputStartOrdinal, &row.laneOrdinal, &row.attempt); err != nil {
			return nil, err
		}
		key := row.projectionKey
		if row.activeKey != "" {
			key = row.activeKey
		}
		if out[key] != nil {
			return nil, ErrInvalidWorkItem
		}
		out[key] = row
	}
	return out, rows.Err()
}

func (r *Repository) loadActiveProjectionBatch(ctx context.Context, keys []string) (map[string]*sourceProjectionWorkItemRow, error) {
	if len(keys) == 1 {
		row, err := r.loadActiveSourceProjectionWorkItemForUpdate(ctx, keys[0])
		out := map[string]*sourceProjectionWorkItemRow{}
		if row != nil {
			out[keys[0]] = row
		}
		return out, err
	}
	candidates, err := r.loadProjectionRows(ctx, "WHERE ProjectionActiveKey IN "+inSQL(keys), appendStrings(nil, keys))
	if err != nil || len(candidates) == 0 {
		return candidates, err
	}
	ids := make([]string, 0, len(candidates))
	for _, row := range candidates {
		ids = append(ids, row.workID)
	}
	sort.Strings(ids)
	args := appendStrings(appendStrings(nil, ids), keys)
	// Match the singleton's current primary-key read; completion may have
	// cleared the candidate pointer since the nonlocking discovery query.
	return r.loadProjectionRows(ctx, "WHERE WorkId IN "+inSQL(ids)+" AND ProjectionActiveKey IN "+inSQL(keys)+" ORDER BY WorkId FOR UPDATE", args)
}

func (r *Repository) loadProjectionSuccessorsBatch(ctx context.Context, keys []string) (map[string]*sourceProjectionWorkItemRow, error) {
	if len(keys) == 0 {
		return map[string]*sourceProjectionWorkItemRow{}, nil
	}
	// At most one queued successor is valid per requested projection.
	// Read one extra witness so truncation cannot turn ambiguity into success.
	args := append(appendStrings(nil, keys), len(keys)+1)
	rows, err := r.loadProjectionRows(ctx, "WHERE ProjectionKey IN "+inSQL(keys)+` AND ProjectionActiveKey IS NULL
 AND State IN ('pending','retry','blocked','blocked_retry') ORDER BY WorkId LIMIT ? FOR UPDATE`, args)
	if err != nil {
		return nil, err
	}
	requested := make(map[string]bool, len(keys))
	for _, key := range keys {
		requested[key] = true
	}
	// SQL equality can return a malformed alias with different bytes. Reject it
	// rather than counting it as a distinct valid successor or ignoring it.
	for key := range rows {
		if !requested[key] {
			return nil, ErrInvalidWorkItem
		}
	}
	return rows, nil
}

func requireProjectionWrites(result sql.Result, want int) error {
	n, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if n != int64(want) {
		return ErrInvalidWorkItem
	}
	return nil
}

func (r *Repository) writeProjectionBatch(ctx context.Context, writes []projectionWrite, replayOf string) error {
	var inserts []string
	var insertArgs []any
	for _, write := range writes {
		if !write.insert {
			continue
		}
		in := write.input.WorkItem
		var active, lineage any
		if write.active {
			active = in.WorkID
		}
		if replayOf != "" {
			lineage = replayOf
		}
		inserts = append(inserts, "(?,?,?,?,?,?,?,0,?,?,?,?,?,?,?,?,?,?)")
		insertArgs = append(insertArgs, write.workID, in.EventID, in.InstalledPackID, in.WorkType, in.QueueClass, in.State, in.RunAfter, in.SourceLaneKey, in.TargetMutationLaneKey, in.LaneOrdinal, write.input.InputStartOrdinal, in.Generation, lineage, in.WorkID, active, in.Now, in.Now)
	}
	if len(inserts) > 0 {
		res, err := r.db.ExecContext(ctx, `INSERT INTO _System_ConnectorWorkItem
 (WorkId,EventId,InstalledPackId,WorkType,QueueClass,State,RunAfter,Attempt,SourceLaneKey,TargetMutationLaneKey,
 LaneOrdinal,InputStartOrdinal,Generation,ReplayOfWorkId,ProjectionKey,ProjectionActiveKey,CreatedDate,LastModifiedDate)
 VALUES `+strings.Join(inserts, ","), insertArgs...)
		if err != nil {
			return err
		}
		if err := requireProjectionWrites(res, len(inserts)); err != nil {
			return err
		}
	}
	for _, active := range []bool{true, false} {
		var updates []string
		var args []any
		for _, write := range writes {
			if write.insert || write.active != active {
				continue
			}
			in := write.input.WorkItem
			updates = append(updates, "SELECT ? AS work_id,? AS event_id,? AS queue_class,? AS state,? AS run_after,? AS source_lane,? AS target_lane,? AS ordinal,? AS projection,? AS modified")
			args = append(args, write.workID, in.EventID, in.QueueClass, in.State, in.RunAfter, in.SourceLaneKey, in.TargetMutationLaneKey, in.LaneOrdinal, in.WorkID, in.Now)
		}
		if len(updates) == 0 {
			continue
		}
		guard := `w.ProjectionActiveKey IS NULL AND w.State IN ('pending','retry','blocked','blocked_retry')`
		if active {
			guard = `w.ProjectionActiveKey=v.projection AND w.State NOT IN ('processing','retry','blocked_retry','processed','skipped','dead')`
		}
		res, err := r.db.ExecContext(ctx, `UPDATE _System_ConnectorWorkItem w JOIN (`+strings.Join(updates, " UNION ALL ")+`) v ON w.WorkId=v.work_id
 SET w.EventId=v.event_id,w.QueueClass=v.queue_class,w.State=v.state,w.RunAfter=v.run_after,w.Attempt=0,
 w.LeaseOwner=NULL,w.LeaseToken=NULL,w.LeaseUntil=NULL,w.SourceLaneKey=v.source_lane,w.TargetMutationLaneKey=v.target_lane,
 w.LaneOrdinal=v.ordinal,w.LastModifiedDate=v.modified
 WHERE w.ProjectionKey=v.projection AND w.Attempt=0 AND `+guard, args...)
		if err != nil {
			return err
		}
		if err := requireProjectionWrites(res, len(updates)); err != nil {
			return err
		}
	}
	return nil
}