package connector_workqueue

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"
	"strings"
)

const ProjectionRangeProtocolVersion = 1

var (
	ErrProjectionProtocolRequired = errors.New("connector projection range protocol is not ready")
	ErrProjectionCoverageMissing  = errors.New("connector projection input has no durable work range")
)

// SourceProjectionWorkItemInput is only for Events newly inserted by the
// caller's admitted pack transaction. Existing Events use the duplicate-only
// EnsureSourceProjectionWorkItem entrypoint. Keep CreateWorkItemInput's
// positional public shape unchanged.
type SourceProjectionWorkItemInput struct {
	WorkItem          CreateWorkItemInput
	InputStartOrdinal int64
}

// SourceProjectionIdentity contains immutable Event facts, not live route
// metadata. Workqueue owns the sole projection-key encoding.
type SourceProjectionIdentity struct {
	InstalledPackID, EventKind                     string
	RouteVersion                                   int64
	CompanionPackInstallID, CompanionFlowVersionID string
	WorkType, SourceLaneKey                        string
}

func SourceProjectionKey(in SourceProjectionIdentity) string {
	seed := strings.Join([]string{strings.TrimSpace(in.InstalledPackID), strings.ToLower(strings.TrimSpace(in.EventKind)),
		strconv.FormatInt(in.RouteVersion, 10), strings.TrimSpace(in.CompanionPackInstallID), strings.TrimSpace(in.CompanionFlowVersionID),
		strings.TrimSpace(in.WorkType), strings.TrimSpace(in.SourceLaneKey)}, "\x00")
	sum := sha256.Sum256([]byte(seed))
	return SourceProjectionWorkIDPrefix + hex.EncodeToString(sum[:])
}

// SourceProjectionRange is the persisted membership fact. It deliberately does
// not add fields to the published, positionally constructed WorkItem shape.
type SourceProjectionRange struct {
	WorkID, EventID, ProjectionKey     string
	Identity                           SourceProjectionIdentity
	InputStartOrdinal, InputEndOrdinal int64
}

// ProjectionLegacyMaxOrdinal reads the immutable controlplane handoff fact.
// Call once per pack admission/activation, never once per Event in a batch.
func (r *Repository) ProjectionLegacyMaxOrdinal(ctx context.Context, packID string) (int64, error) {
	var version int
	var cutoff int64
	err := r.db.QueryRowContext(ctx, `SELECT ProtocolVersion, LegacyMaxOrdinal
 FROM _System_ConnectorProjectionProtocol WHERE InstalledPackId = ?`, packID).Scan(&version, &cutoff)
	if err != nil {
		return 0, fmt.Errorf("%w: pack %q: %v", ErrProjectionProtocolRequired, packID, err)
	}
	if version != ProjectionRangeProtocolVersion || cutoff < 0 {
		return 0, fmt.Errorf("%w: pack %q has invalid version or cutoff", ErrProjectionProtocolRequired, packID)
	}
	return cutoff, nil
}

func (r *Repository) GetSourceProjectionRange(ctx context.Context, workID string) (*SourceProjectionRange, error) {
	var out SourceProjectionRange
	var selectedOrdinal int64
	err := r.db.QueryRowContext(ctx, `SELECT w.WorkId,w.EventId,COALESCE(w.ProjectionKey,''),
 w.InstalledPackId,e.EventKind,COALESCE(e.RouteVersion,0),COALESCE(e.CompanionPackInstallId,''),
 COALESCE(e.CompanionFlowVersionId,''),w.WorkType,COALESCE(w.SourceLaneKey,''),
 COALESCE(w.InputStartOrdinal,0),w.LaneOrdinal,e.EnqueueOrdinal
 FROM _System_ConnectorWorkItem w JOIN _System_ConnectorEvent e ON e.Id=w.EventId
 WHERE w.WorkId=? AND e.InstalledPackId=w.InstalledPackId AND e.SourceLaneKey=w.SourceLaneKey`, workID).Scan(
		&out.WorkID, &out.EventID, &out.ProjectionKey, &out.Identity.InstalledPackID, &out.Identity.EventKind, &out.Identity.RouteVersion,
		&out.Identity.CompanionPackInstallID, &out.Identity.CompanionFlowVersionID, &out.Identity.WorkType, &out.Identity.SourceLaneKey,
		&out.InputStartOrdinal, &out.InputEndOrdinal, &selectedOrdinal)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrInvalidWorkItem
	}
	if err != nil {
		return nil, err
	}
	if out.InputStartOrdinal <= 0 || out.InputEndOrdinal < out.InputStartOrdinal || selectedOrdinal != out.InputEndOrdinal ||
		out.ProjectionKey != SourceProjectionKey(out.Identity) || !IsSourceProjectionWorkID(out.WorkID) {
		return nil, ErrInvalidWorkItem
	}
	return &out, nil
}

// sourceProjectionCoversEvent never materializes or replays an existing input.
// A retained dead generation owns its range just as a successful one does.
func (r *Repository) sourceProjectionCoversEvent(ctx context.Context, in CreateWorkItemInput) (bool, error) {
	cutoff, err := r.ProjectionLegacyMaxOrdinal(ctx, in.InstalledPackID)
	if err != nil {
		return false, err
	}
	// Close old receipt retries before probing retained generations: legacy
	// zero-ordinal rows with no range must never turn a duplicate into a scan.
	if in.LaneOrdinal <= cutoff {
		return true, nil
	}
	var workID string
	err = r.db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem
 WHERE ProjectionKey=? AND LaneOrdinal>=? AND InputStartOrdinal IS NOT NULL
 ORDER BY LaneOrdinal,EventId,WorkId LIMIT 1 FOR UPDATE`, in.WorkID, in.LaneOrdinal).Scan(&workID)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return false, err
	}
	if err == nil {
		bounds, err := r.GetSourceProjectionRange(ctx, workID)
		if err != nil {
			return false, err
		}
		if bounds.InputStartOrdinal <= in.LaneOrdinal {
			if err := r.validateProjectionInput(ctx, in, in.LaneOrdinal); err != nil {
				return false, err
			}
			return true, nil
		}
	}
	return false, fmt.Errorf("%w: Event %q", ErrProjectionCoverageMissing, in.EventID)
}

func (r *Repository) validateProjectionInput(ctx context.Context, in CreateWorkItemInput, first int64) error {
	if first <= 0 || in.LaneOrdinal < first {
		return ErrInvalidWorkItem
	}
	var identity SourceProjectionIdentity
	var ordinal int64
	err := r.db.QueryRowContext(ctx, `SELECT head.InstalledPackId,head.EventKind,COALESCE(head.RouteVersion,0),
 COALESCE(head.CompanionPackInstallId,''),COALESCE(head.CompanionFlowVersionId,''),head.SourceLaneKey,head.EnqueueOrdinal
 FROM _System_ConnectorEvent head WHERE head.Id=? AND EXISTS (
 SELECT 1 FROM _System_ConnectorEvent first
 WHERE first.InstalledPackId=head.InstalledPackId AND first.SourceLaneKey=head.SourceLaneKey AND first.EnqueueOrdinal=?
 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,''))`, in.EventID, first).Scan(
		&identity.InstalledPackID, &identity.EventKind, &identity.RouteVersion, &identity.CompanionPackInstallID,
		&identity.CompanionFlowVersionID, &identity.SourceLaneKey, &ordinal)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrInvalidWorkItem
	}
	if err != nil {
		return err
	}
	identity.WorkType = in.WorkType
	if identity.InstalledPackID != in.InstalledPackID || identity.SourceLaneKey != in.SourceLaneKey || ordinal != in.LaneOrdinal || SourceProjectionKey(identity) != in.WorkID {
		return ErrInvalidWorkItem
	}
	return nil
}

func requireProjectionRangeExtended(old *sourceProjectionWorkItemRow, in CreateWorkItemInput, first int64) error {
	if old.inputStartOrdinal <= 0 || old.laneOrdinal < old.inputStartOrdinal || first <= old.laneOrdinal || in.LaneOrdinal < first {
		return fmt.Errorf("%w: projection input range is not a monotonic append", ErrInvalidWorkItem)
	}
	return nil
}