package automation

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/automation/dsl"
	"github.com/tidbcloud/AgentTeam/internal/contentpayload"
	"github.com/tidbcloud/AgentTeam/internal/identity/trusted"
	"github.com/tidbcloud/AgentTeam/internal/service/automation/runtime"
	"github.com/tidbcloud/AgentTeam/internal/service/content"
	automationstore "github.com/tidbcloud/AgentTeam/internal/store/automation"
	"gorm.io/gorm"
)

const automationSourceKindContentVersion = "content_version"

type AutomationContentVersionReader interface {
	GetContent(context.Context, string, string) (content.Content, error)
	OpenFile(context.Context, string, string, string, string, *contentpayload.ByteRange) (io.ReadCloser, contentpayload.BlobMeta, error)
}

// ContentVersionAutomationConsumer starts runs only from the automation-owned
// exact publication projection. Missing projections fail delivery so the
// Content Store outbox lease/retry owner can replay safely.
type ContentVersionAutomationConsumer struct {
	content     AutomationContentVersionReader
	automations *AutomationsService
}

func NewContentVersionAutomationConsumer(content AutomationContentVersionReader, service *AutomationsService) (*ContentVersionAutomationConsumer, error) {
	if content == nil || service == nil || service.store == nil {
		return nil, fmt.Errorf("content version automation dependencies are required")
	}
	return &ContentVersionAutomationConsumer{content: content, automations: service}, nil
}

func (consumer *ContentVersionAutomationConsumer) ConsumeVersionCommitted(ctx context.Context, event content.VersionCommittedEvent) error {
	if consumer == nil || consumer.content == nil || consumer.automations == nil {
		return fmt.Errorf("content version automation consumer is not configured")
	}
	ctx = contentVersionAutomationEventerContext(ctx, event.WorkspaceID)
	content, err := consumer.content.GetContent(ctx, event.WorkspaceID, event.ContentID)
	if err != nil {
		return fmt.Errorf("load automation content lineage: %w", err)
	}
	if content.Domain != "automation" {
		return nil
	}
	publication, err := consumer.automations.store.GetContentPublication(ctx, event.WorkspaceID, event.ContentID, event.VersionID)
	if err != nil {
		return fmt.Errorf("load automation content publication: %w", err)
	}
	definitions, err := consumer.automations.store.ListContentDefinitions(ctx, event.WorkspaceID, event.ContentID, event.VersionID, automationstore.MaxAutomationDefinitionsPerVersion)
	if err != nil {
		return fmt.Errorf("list automation content definitions: %w", err)
	}
	changedPaths, err := decodeAutomationChangedPaths(publication.ChangedPathsJSON)
	if err != nil {
		return err
	}
	for _, definition := range definitions {
		source, err := consumer.readDefinition(ctx, event, definition)
		if err != nil {
			return err
		}
		_, err = consumer.automations.startContentVersionAutomationRun(ctx, event, publication, definition.DefinitionPath, source, changedPaths)
		if errors.Is(err, errAutomationTriggerNotMatched) {
			continue
		}
		if err != nil {
			return fmt.Errorf("start content version automation %q: %w", definition.DefinitionPath, err)
		}
	}
	return nil
}

func (consumer *ContentVersionAutomationConsumer) readDefinition(ctx context.Context, event content.VersionCommittedEvent, definition automationstore.AutomationContentDefinition) (string, error) {
	path := definition.DefinitionPath
	maxBytes := int64(consumer.automations.automationLimits().MaxAutomationSourceBytes)
	if definition.SizeBytes < 0 || definition.SizeBytes > maxBytes {
		return "", fmt.Errorf("automation definition %q exceeds source limit", path)
	}
	reader, meta, err := consumer.content.OpenFile(ctx, event.WorkspaceID, event.ContentID, event.VersionID, path, nil)
	if err != nil {
		return "", fmt.Errorf("open automation definition %q: %w", path, err)
	}
	defer reader.Close()
	bytes, err := io.ReadAll(io.LimitReader(reader, maxBytes+1))
	if err != nil {
		return "", fmt.Errorf("read automation definition %q: %w", path, err)
	}
	digest := sha256.Sum256(bytes)
	if int64(len(bytes)) != definition.SizeBytes || meta.SizeBytes != definition.SizeBytes ||
		string(meta.Digest) != definition.Digest || hex.EncodeToString(digest[:]) != definition.Digest {
		return "", fmt.Errorf("automation definition %q projection mismatch", path)
	}
	return string(bytes), nil
}

func (service *AutomationsService) startContentVersionAutomationRun(ctx context.Context, event content.VersionCommittedEvent, publication automationstore.AutomationContentPublication, definitionPath, source string, changedPaths []string) (automationstore.AutomationRun, error) {
	idempotencyKey := contentVersionAutomationIdempotencyKey(event.ID, definitionPath)
	if existing, err := service.store.GetRunByIdempotencyKey(ctx, event.WorkspaceID, idempotencyKey); err == nil {
		if existing.ContentID != event.ContentID || existing.ContentVersionID != event.VersionID || existing.DefinitionPath != definitionPath {
			return automationstore.AutomationRun{}, conflict("Automation content event idempotency key conflicts")
		}
		service.signalRunnerWorkspace(existing.WorkspaceID)
		return existing, nil
	} else if !errors.Is(err, gorm.ErrRecordNotFound) {
		return automationstore.AutomationRun{}, err
	}
	spec, report := service.decodeAndValidateAutomationContent(source, definitionPath)
	if !report.Valid {
		return automationstore.AutomationRun{}, invalidArgument("Automation content version definition is invalid")
	}
	if !dsl.MatchesPushTrigger(spec.On, dsl.PushEvent{Ref: publication.Ref, ChangedPaths: changedPaths, ChangedPathsKnown: true}) {
		return automationstore.AutomationRun{}, errAutomationTriggerNotMatched
	}
	if err := service.validateAutomationSecretReferences(ctx, event.WorkspaceID, AutomationEventerPrincipalID, publication.RepositoryOwner, publication.RepositoryName, spec); err != nil {
		return automationstore.AutomationRun{}, err
	}
	runID, err := newAutomationRunID()
	if err != nil {
		return automationstore.AutomationRun{}, fmt.Errorf("generate automation run id: %w", err)
	}
	now := service.now()
	started := now
	repoFullName := publication.RepositoryOwner + "/" + publication.RepositoryName
	metadata, _ := json.Marshal(map[string]any{
		"contentId": event.ContentID, "contentVersionId": event.VersionID,
		"contentOutboxEventId": event.ID, "changedPaths": changedPaths,
	})
	run := automationstore.AutomationRun{
		ID: runID, WorkspaceID: event.WorkspaceID, IdempotencyKey: optionalStringPtr(idempotencyKey),
		SourceKind: automationSourceKindContentVersion, ContentID: event.ContentID, ContentVersionID: event.VersionID,
		RepositoryOwner: publication.RepositoryOwner, RepositoryName: publication.RepositoryName, RepositoryFullName: repoFullName,
		DefinitionPath: definitionPath, DefinitionID: "contentstore://" + event.ContentID + "/" + event.VersionID + "/" + definitionPath,
		Name: spec.Name, RunName: spec.RunName, Status: automationstore.AutomationRunStatusRunning,
		TriggerKind: "content_version", TriggerRef: event.ID, TargetKind: "content_version", TargetRef: event.VersionID,
		OwnerPrincipalID: publication.ActorID, CreatedByPrincipalID: publication.ActorID,
		Metadata: string(metadata), StartedAt: &started, CreatedAt: now, UpdatedAt: now,
	}
	runtimeState, err := runtime.BuildInitialRuntime(run, spec, now, newAutomationRuntimeID)
	if err != nil {
		return automationstore.AutomationRun{}, err
	}
	if len(runtimeState.Waits) > 0 && len(runtimeState.Jobs) == 0 {
		run.Status = automationstore.AutomationRunStatusWaiting
	}
	created, err := service.store.CreateRunWithRuntimeWithCapacity(ctx, service.automationLimits().MaxActiveRunsPerWorkspace, run, runtimeState.JobRuns, runtimeState.Events, runtimeState.Jobs, runtimeState.Waits)
	if errors.Is(err, automationstore.ErrActiveRunCapacity) {
		return automationstore.AutomationRun{}, conflict("Automation active run limit reached")
	}
	if err == nil {
		service.signalRunnerWorkspace(created.WorkspaceID)
	}
	return created, err
}

func contentVersionAutomationEventerContext(ctx context.Context, workspaceID string) context.Context {
	return trusted.WithIdentity(ctx, trusted.Identity{
		Subject: AutomationEventerPrincipalID,
		RawClaims: map[string]any{
			"sub": AutomationEventerPrincipalID, "workspace_ids": []string{strings.TrimSpace(workspaceID)},
			"automation_permissions": []string{"automation:eventer", "automation:service", "automation:secret:use"},
			"service_actor":          true,
		},
	})
}

func contentVersionAutomationIdempotencyKey(eventID, definitionPath string) string {
	hash := sha256.Sum256([]byte(strings.TrimSpace(eventID) + "\x00" + strings.TrimSpace(definitionPath)))
	return "content-version:" + hex.EncodeToString(hash[:])
}

func decodeAutomationChangedPaths(raw string) ([]string, error) {
	var paths []string
	if err := json.Unmarshal([]byte(strings.TrimSpace(raw)), &paths); err != nil {
		return nil, fmt.Errorf("decode automation content changed paths: %w", err)
	}
	if len(paths) > contentpayload.DefaultLimits.MaxFiles {
		return nil, fmt.Errorf("automation content changed paths exceed publication limit %d", contentpayload.DefaultLimits.MaxFiles)
	}
	return paths, nil
}