package invocation

import (
	"database/sql"
	"fmt"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	domain "github.com/tidbcloud/AgentTeam/internal/workflow/invocation"
)

type rowScanner interface {
	Scan(...any) error
}

func invocationSelectSQL() string {
	return `
SELECT i.id, run.id, i.workspace_id, i.binding_id,
       i.binding_revision_id, i.binding_revision_number,
       i.binding_source_id, i.binding_activation_number,
       i.plan_id, i.plan_version_id, i.workflow_name_snapshot,
       i.source_kind, i.source_occurrence_ref, i.source_occurred_at,
       i.input_json, i.input_digest, i.source_provenance_json,
       i.owner_user_id_snapshot, i.owner_member_id_snapshot,
       i.principal_member_id_snapshot, i.author_member_id_snapshot,
       i.invoked_by_member_id, i.invoked_by_actor_type, i.invoked_by_actor_id,
       i.idempotency_key, i.request_digest, i.accepted_at
FROM workflow_invocations i
JOIN workflow_run_instances run
  ON run.workspace_id = i.workspace_id AND run.invocation_id = i.id`
}

// invocationLockSelectSQL deliberately reads only the Invocation row. A
// transactional replay must use three explicit current point reads in the
// stable order Invocation -> Run -> accepted Event. Joining Run into the first
// SELECT ... FOR UPDATE is unsafe under TiDB repeatable-read: the locking join
// can see a winner that committed after this transaction's startTS, while the
// following ordinary Run/Event reads still use the older snapshot and miss
// rows committed by that same winner.
func invocationLockSelectSQL() string {
	return `
SELECT i.id, i.workspace_id, i.binding_id,
       i.binding_revision_id, i.binding_revision_number,
       i.binding_source_id, i.binding_activation_number,
       i.plan_id, i.plan_version_id, i.workflow_name_snapshot,
       i.source_kind, i.source_occurrence_ref, i.source_occurred_at,
       i.input_json, i.input_digest, i.source_provenance_json,
       i.owner_user_id_snapshot, i.owner_member_id_snapshot,
       i.principal_member_id_snapshot, i.author_member_id_snapshot,
       i.invoked_by_member_id, i.invoked_by_actor_type, i.invoked_by_actor_id,
       i.idempotency_key, i.request_digest, i.accepted_at
FROM workflow_invocations i`
}

func invocationListSelectSQL() string {
	return `
SELECT i.id, run.id, run.status, i.workspace_id, i.binding_id,
       i.binding_revision_id, i.binding_revision_number,
       i.binding_source_id, i.binding_activation_number,
       i.plan_id, i.plan_version_id, i.workflow_name_snapshot,
       i.source_kind, i.source_occurrence_ref, i.source_occurred_at,
       i.input_digest,
       i.owner_user_id_snapshot, i.owner_member_id_snapshot,
       i.principal_member_id_snapshot, i.author_member_id_snapshot,
       i.invoked_by_member_id, i.invoked_by_actor_type, i.invoked_by_actor_id,
       i.accepted_at
FROM workflow_invocations i
JOIN workflow_run_instances run
  ON run.workspace_id = i.workspace_id AND run.invocation_id = i.id`
}

func scanInvocationListItem(scanner rowScanner) (ListItem, error) {
	var item ListItem
	value := &item.Invocation
	var invokedMember, invokedType, invokedID sql.NullString
	var bindingID, bindingRevisionID, bindingSourceID sql.NullString
	var bindingRevision, bindingActivation sql.NullInt64
	if err := scanner.Scan(
		&value.ID, &value.RunID, &item.RunStatus, &value.WorkspaceID, &bindingID,
		&bindingRevisionID, &bindingRevision,
		&bindingSourceID, &bindingActivation,
		&value.PlanID, &value.PlanVersionID, &value.WorkflowNameSnapshot,
		&value.SourceKind, &value.SourceOccurrenceRef, &value.SourceOccurredAt,
		&value.InputDigest,
		&value.Identity.OwnerUserID, &value.Identity.OwnerMemberID,
		&value.Identity.PrincipalMemberID, &value.Identity.AuthorMemberID,
		&invokedMember, &invokedType, &invokedID,
		&value.AcceptedAt,
	); err != nil {
		return ListItem{}, err
	}
	if err := assignBindingSnapshot(value, bindingID, bindingRevisionID, bindingRevision, bindingSourceID, bindingActivation); err != nil {
		return ListItem{}, err
	}
	if invokedMember.Valid || invokedType.Valid || invokedID.Valid {
		if !invokedMember.Valid || !invokedType.Valid || !invokedID.Valid {
			return ListItem{}, fmt.Errorf("workflow invocation %s has inconsistent invokedBy provenance", value.ID)
		}
		value.InvokedBy = Actor{MemberID: invokedMember.String, Type: invokedType.String, ID: invokedID.String}
	}
	value.WorkflowNameSnapshot = strings.TrimSpace(value.WorkflowNameSnapshot)
	if value.WorkflowNameSnapshot == "" || len(value.WorkflowNameSnapshot) > 255 {
		return ListItem{}, fmt.Errorf("workflow invocation %s has invalid workflow name snapshot", value.ID)
	}
	value.SourceOccurredAt = value.SourceOccurredAt.UTC()
	value.AcceptedAt = value.AcceptedAt.UTC()
	return item, nil
}

func scanInvocation(scanner rowScanner) (Invocation, error) {
	return scanInvocationFields(scanner, true)
}

// scanLockedInvocation scans the Invocation-only point read used by
// transactional replay. RunID is populated only after the Run row has been
// acquired with its own current locking read.
func scanLockedInvocation(scanner rowScanner) (Invocation, error) {
	return scanInvocationFields(scanner, false)
}

func scanInvocationFields(scanner rowScanner, includesRunID bool) (Invocation, error) {
	var value Invocation
	var invokedMember, invokedType, invokedID sql.NullString
	var bindingID, bindingRevisionID, bindingSourceID sql.NullString
	var bindingRevision, bindingActivation sql.NullInt64
	targets := []any{&value.ID}
	if includesRunID {
		targets = append(targets, &value.RunID)
	}
	targets = append(targets,
		&value.WorkspaceID, &bindingID,
		&bindingRevisionID, &bindingRevision,
		&bindingSourceID, &bindingActivation,
		&value.PlanID, &value.PlanVersionID, &value.WorkflowNameSnapshot,
		&value.SourceKind, &value.SourceOccurrenceRef, &value.SourceOccurredAt,
		&value.InputJSON, &value.InputDigest, &value.SourceProvenanceJSON,
		&value.Identity.OwnerUserID, &value.Identity.OwnerMemberID,
		&value.Identity.PrincipalMemberID, &value.Identity.AuthorMemberID,
		&invokedMember, &invokedType, &invokedID,
		&value.IdempotencyKey, &value.RequestDigest, &value.AcceptedAt,
	)
	if err := scanner.Scan(targets...); err != nil {
		return Invocation{}, err
	}
	if err := assignBindingSnapshot(&value, bindingID, bindingRevisionID, bindingRevision, bindingSourceID, bindingActivation); err != nil {
		return Invocation{}, err
	}
	if invokedMember.Valid || invokedType.Valid || invokedID.Valid {
		if !invokedMember.Valid || !invokedType.Valid || !invokedID.Valid {
			return Invocation{}, fmt.Errorf("workflow invocation %s has inconsistent invokedBy provenance", value.ID)
		}
		value.InvokedBy = Actor{MemberID: invokedMember.String, Type: invokedType.String, ID: invokedID.String}
	}
	input, err := domain.DeterministicJSONObject("persisted workflow Invocation input", value.InputJSON, domain.MaxInputBytes)
	if err != nil {
		return Invocation{}, fmt.Errorf("workflow invocation %s has invalid input: %w", value.ID, err)
	}
	if value.InputDigest != domain.InputDigest(input) {
		return Invocation{}, fmt.Errorf("workflow invocation %s input digest does not match canonical input", value.ID)
	}
	provenance, err := domain.DeterministicSourceProvenanceJSONObject(
		"persisted workflow Invocation source provenance", value.SourceProvenanceJSON, domain.MaxSourceProvenanceBytes,
	)
	if err != nil {
		return Invocation{}, fmt.Errorf("workflow invocation %s has invalid source provenance: %w", value.ID, err)
	}
	value.InputJSON = input
	value.SourceProvenanceJSON = provenance
	value.WorkflowNameSnapshot = strings.TrimSpace(value.WorkflowNameSnapshot)
	if value.WorkflowNameSnapshot == "" || len(value.WorkflowNameSnapshot) > 255 {
		return Invocation{}, fmt.Errorf("workflow invocation %s has invalid workflow name snapshot", value.ID)
	}
	value.SourceOccurredAt = value.SourceOccurredAt.UTC()
	value.AcceptedAt = value.AcceptedAt.UTC()
	return value, nil
}

func runSelectSQL() string {
	return `
SELECT id, workspace_id, invocation_id,
       plan_id, plan_version_id, team_id, workflow_name_snapshot, status, input_json, input_digest,
       owner_user_id_snapshot, owner_member_id_snapshot,
       principal_member_id_snapshot, author_member_id_snapshot,
       queued_at, created_at, updated_at
FROM workflow_run_instances`
}

func scanRun(scanner rowScanner) (Run, error) {
	var value Run
	var teamID sql.NullString
	if err := scanner.Scan(
		&value.ID, &value.WorkspaceID, &value.InvocationID,
		&value.PlanID, &value.PlanVersionID, &teamID, &value.WorkflowNameSnapshot, &value.Status, &value.InputJSON, &value.InputDigest,
		&value.Identity.OwnerUserID, &value.Identity.OwnerMemberID,
		&value.Identity.PrincipalMemberID, &value.Identity.AuthorMemberID,
		&value.QueuedAt, &value.CreatedAt, &value.UpdatedAt,
	); err != nil {
		return Run{}, err
	}
	value.TeamID = teamID.String
	input, err := domain.DeterministicJSONObject("persisted workflow Run input", value.InputJSON, domain.MaxInputBytes)
	if err != nil {
		return Run{}, fmt.Errorf("workflow Run %s has invalid input: %w", value.ID, err)
	}
	if value.InputDigest != domain.InputDigest(input) {
		return Run{}, fmt.Errorf("workflow Run %s input digest does not match canonical input", value.ID)
	}
	value.InputJSON = input
	value.WorkflowNameSnapshot = strings.TrimSpace(value.WorkflowNameSnapshot)
	if value.WorkflowNameSnapshot == "" || len(value.WorkflowNameSnapshot) > 255 {
		return Run{}, fmt.Errorf("workflow Run %s has invalid workflow name snapshot", value.ID)
	}
	value.QueuedAt = value.QueuedAt.UTC()
	value.CreatedAt = value.CreatedAt.UTC()
	value.UpdatedAt = value.UpdatedAt.UTC()
	return value, nil
}

func assignBindingSnapshot(
	value *Invocation,
	bindingID sql.NullString,
	bindingRevisionID sql.NullString,
	bindingRevision sql.NullInt64,
	bindingSourceID sql.NullString,
	bindingActivation sql.NullInt64,
) error {
	allNull := !bindingID.Valid && !bindingRevisionID.Valid && !bindingRevision.Valid && !bindingSourceID.Valid && !bindingActivation.Valid
	allSet := bindingID.Valid && bindingRevisionID.Valid && bindingRevision.Valid && bindingSourceID.Valid && bindingActivation.Valid
	if !allNull && !allSet {
		return fmt.Errorf("workflow invocation %s has an incomplete Binding snapshot", value.ID)
	}
	if allSet {
		value.BindingID = bindingID.String
		value.BindingRevisionID = bindingRevisionID.String
		value.BindingRevision = bindingRevision.Int64
		value.BindingSourceID = bindingSourceID.String
		value.BindingActivation = bindingActivation.Int64
	}
	return nil
}

func runEventSelectSQL() string {
	return `
SELECT id, workspace_id, run_id, seq, event_type, payload_json, created_at
FROM workflow_run_instance_events`
}

func scanRunEvent(scanner rowScanner) (RunEvent, error) {
	var value RunEvent
	if err := scanner.Scan(
		&value.ID, &value.WorkspaceID, &value.RunID, &value.Seq,
		&value.Type, &value.PayloadJSON, &value.CreatedAt,
	); err != nil {
		return RunEvent{}, err
	}
	payload, err := domain.DeterministicJSONObject(
		"persisted workflow Run event payload", value.PayloadJSON, domain.MaxSourceProvenanceBytes,
	)
	if err != nil {
		return RunEvent{}, fmt.Errorf("workflow Run event %s has invalid payload: %w", value.ID, err)
	}
	value.PayloadJSON = payload
	value.CreatedAt = value.CreatedAt.UTC()
	return value, nil
}

func mapReadError(message string, err error) error {
	if coredb.IsNotFound(err) {
		return ErrNotFound
	}
	if err != nil {
		return fmt.Errorf("%s: %w", message, err)
	}
	return nil
}