package app

import (
	"context"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
)

// CommittedMessageFanout owns post-commit delivery for app-store workflows.
// Ordinary autonomous messages use DeliverCommittedMessage. Explicit
// no-attention transcript events use EmitCommittedMessageProjectionOnly and
// therefore never enter ordinary unread, notification, Agent wake, or outbox
// completion.
type CommittedMessageFanout interface {
	DeliverCommittedMessage(context.Context, string, message.AppMessage, message.MessageDelivery) error
	EmitCommittedMessageProjectionOnly(context.Context, string, message.AppMessage, message.MessageDelivery)
}

type committedMessageProducerID string

const (
	committedMessageProducerAppNotification                 committedMessageProducerID = "app_notification"
	committedMessageProducerExternalAppRuntimeReply         committedMessageProducerID = "external_app_runtime_reply"
	committedMessageProducerPlatformServiceAccountBroadcast committedMessageProducerID = "platform_service_account_broadcast"
	committedMessageProducerProviderEvent                   committedMessageProducerID = "provider_event"
	committedMessageProducerKnowledgeQA                     committedMessageProducerID = "knowledge_qa"
	committedMessagePostCommitTimeout                                                  = 5 * time.Second
)

func (producerID committedMessageProducerID) ownsRecoverablePostCommitFanout() bool {
	switch producerID {
	case committedMessageProducerAppNotification,
		committedMessageProducerExternalAppRuntimeReply,
		committedMessageProducerPlatformServiceAccountBroadcast:
		return true
	default:
		return false
	}
}

// postCommitMessage is an opaque handoff token outside package app. The
// canonical prepare function snapshots the same recoverable delivery that it
// persists in the message outbox. A source-level guardrail prevents production
// code from constructing this package-private token directly.
type postCommitMessage struct {
	workspaceID string
	producerID  committedMessageProducerID
	payload     []byte
}

// appServiceMessageContract is a closed producer registry at the shared
// Runtime App/App append seam. Each producer therefore declares its stable
// identity and first post-commit fanout owner at compile time; there is no
// permissive zero/default policy.
type appServiceMessageContract uint8

const (
	appServiceMessageAppNotification appServiceMessageContract = iota + 1
	appServiceMessageExternalAppRuntimeReply
	appServiceMessageProviderEvent
	appServiceMessageKnowledgeQA
)

type appServiceMessagePostCommitOwner uint8

const (
	appServiceMessageCallerOwned appServiceMessagePostCommitOwner = iota + 1
	appServiceMessageStoreOwnedRecoverable
)

func (contract appServiceMessageContract) descriptor() (committedMessageProducerID, appServiceMessagePostCommitOwner, bool) {
	switch contract {
	case appServiceMessageAppNotification:
		return committedMessageProducerAppNotification, appServiceMessageStoreOwnedRecoverable, true
	case appServiceMessageExternalAppRuntimeReply:
		return committedMessageProducerExternalAppRuntimeReply, appServiceMessageStoreOwnedRecoverable, true
	case appServiceMessageProviderEvent:
		return committedMessageProducerProviderEvent, appServiceMessageCallerOwned, true
	case appServiceMessageKnowledgeQA:
		return committedMessageProducerKnowledgeQA, appServiceMessageCallerOwned, true
	default:
		return "", 0, false
	}
}

// SetCommittedMessageFanout installs the process-level owner shared by every
// autonomous app-store message producer. Server wiring must use the same
// fanout instance as ordinary message APIs.
func (store *Store) SetCommittedMessageFanout(fanout CommittedMessageFanout) {
	if store == nil {
		return
	}
	if fanout == nil {
		panic("committed message fanout must not be nil")
	}
	store.committedMessageFanoutMu.Lock()
	defer store.committedMessageFanoutMu.Unlock()
	if store.committedMessageFanout != nil {
		panic("committed message fanout is already configured")
	}
	store.committedMessageFanout = fanout
}

// preparePostCommitMessage declares that the current transaction is the
// authoritative creator of a new ordinary message whose first Human fanout is
// owned by the app store. The durable replay may repair only the stable
// message.created projection; ordinary unread remains a first-attempt effect
// because its audience is not frozen.
//
// Callers must invoke this inside the same transaction that creates the
// message, retain the returned token only after that transaction commits, and
// pass it to emitPostCommitMessage. Idempotent replays intentionally return an
// empty token.
func (store *Store) preparePostCommitMessage(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID string,
	producerID committedMessageProducerID,
	delivery message.MessageDelivery,
) (message.MessageDelivery, postCommitMessage, error) {
	if delivery.IdempotentReplay {
		return delivery, postCommitMessage{}, nil
	}
	workspaceID = strings.TrimSpace(workspaceID)
	producerID = committedMessageProducerID(strings.TrimSpace(string(producerID)))
	if workspaceID == "" || !producerID.ownsRecoverablePostCommitFanout() {
		return message.MessageDelivery{}, postCommitMessage{}, fmt.Errorf("post-commit message requires a registered autonomous producer and workspace")
	}
	preparedDelivery, payload, err := encodeRecoverablePostCommitDelivery(delivery)
	if err != nil {
		return message.MessageDelivery{}, postCommitMessage{}, err
	}
	if err := message.NewScopeRepositories(tx).Outbox.SetPayloadForMessage(ctx, preparedDelivery.MessageID, payload); err != nil {
		return message.MessageDelivery{}, postCommitMessage{}, err
	}
	return preparedDelivery, postCommitMessage{
		workspaceID: workspaceID,
		producerID:  producerID,
		payload:     append([]byte(nil), payload...),
	}, nil
}

func encodeRecoverablePostCommitDelivery(delivery message.MessageDelivery) (message.MessageDelivery, []byte, error) {
	messageID := strings.TrimSpace(delivery.MessageID)
	if messageID == "" || delivery.AppMessage == nil {
		return message.MessageDelivery{}, nil, fmt.Errorf("post-commit message requires message and AppMessage")
	}
	if strings.TrimSpace(delivery.AppMessage.ID) != messageID {
		return message.MessageDelivery{}, nil, fmt.Errorf("post-commit message projection does not match message %s", messageID)
	}
	if delivery.HumanRecipientsTargeted && !isTaskTargetedRecoverableDelivery(delivery) {
		return message.MessageDelivery{}, nil, fmt.Errorf("targeted message %s must use its frozen delivery path", messageID)
	}
	if len(delivery.FollowupDeliveries) > 0 {
		return message.MessageDelivery{}, nil, fmt.Errorf("post-commit message %s must not contain caller-owned followup deliveries", messageID)
	}
	delivery.RecoverableStreamProjection = true
	// The encoded outbox payload is also the immutable post-commit handoff. It
	// avoids retaining caller-owned pointer, map, or slice fields and guarantees
	// first delivery and durable replay decode exactly the same snapshot.
	payload, err := message.MarshalScopeFanoutPayload(*delivery.AppMessage, delivery)
	if err != nil {
		return message.MessageDelivery{}, nil, err
	}
	return delivery, payload, nil
}

// isTaskTargetedRecoverableDelivery reports whether a targeted Human delivery
// belongs to a Task scope and therefore already uses the frozen lease-coordinated
// delivery path (deliverInlineTaskMessage) that the store-owned recoverable
// post-commit token routes to. Targeted non-Task deliveries have no such path
// and remain rejected by the recoverable snapshotter.
func isTaskTargetedRecoverableDelivery(delivery message.MessageDelivery) bool {
	scopeID := strings.TrimSpace(delivery.ThreadID)
	if scopeID == "" {
		scopeID = strings.TrimSpace(delivery.Message.TargetKey)
	}
	return strings.TrimSpace(delivery.Message.ChannelType) == message.ConversationKindTask ||
		(delivery.Conversation != nil && strings.TrimSpace(delivery.Conversation.Kind) == message.ConversationKindTask) ||
		strings.HasPrefix(scopeID, "tsk_")
}

// finalizeAppServiceMessageTx is called only by the shared Runtime App/App
// append entrypoint after it has built the complete browser projection.
// Requiring a closed owner choice at this seam makes a new producer declare who
// performs the first post-commit fanout without exposing a stage that a caller
// could return before finalization.
func (store *Store) finalizeAppServiceMessageTx(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID string,
	delivery message.MessageDelivery,
	contract appServiceMessageContract,
) (message.MessageDelivery, postCommitMessage, error) {
	producerID, owner, ok := contract.descriptor()
	if !ok {
		return message.MessageDelivery{}, postCommitMessage{}, fmt.Errorf("app service message producer contract is required")
	}
	switch owner {
	case appServiceMessageCallerOwned:
		if err := store.persistScopeFanoutPayloads(ctx, tx, delivery); err != nil {
			return message.MessageDelivery{}, postCommitMessage{}, err
		}
		return delivery, postCommitMessage{}, nil
	case appServiceMessageStoreOwnedRecoverable:
		preparedDelivery, committed, err := store.preparePostCommitMessage(ctx, tx, workspaceID, producerID, delivery)
		if err != nil {
			return message.MessageDelivery{}, postCommitMessage{}, err
		}
		return preparedDelivery, committed, nil
	default:
		return message.MessageDelivery{}, postCommitMessage{}, fmt.Errorf("app service message producer %s has an invalid post-commit owner", producerID)
	}
}

// emitPostCommitMessage consumes a token only after its creating transaction
// has committed. Projection is best-effort: the database commit is
// authoritative and the pending outbox remains the stable stream recovery
// path when the in-process fanout fails.
func (store *Store) emitPostCommitMessage(ctx context.Context, committed postCommitMessage) {
	if store == nil || strings.TrimSpace(committed.workspaceID) == "" ||
		!committed.producerID.ownsRecoverablePostCommitFanout() || len(committed.payload) == 0 {
		return
	}
	payload, err := message.UnmarshalScopeFanoutPayload(committed.payload)
	if err != nil {
		store.committedMessageLogger().Error(
			"decode committed autonomous message",
			"producer", committed.producerID,
			"workspaceID", committed.workspaceID,
			"error", err,
		)
		return
	}
	delivery := payload.Delivery
	delivery.AppMessage = &payload.AppMessage
	store.committedMessageFanoutMu.RLock()
	fanout := store.committedMessageFanout
	store.committedMessageFanoutMu.RUnlock()
	if fanout == nil {
		store.missingFanoutLogOnce.Do(func() {
			store.committedMessageLogger().Error(
				"committed message fanout is not configured",
				"producer", committed.producerID,
				"workspaceID", committed.workspaceID,
				"messageID", delivery.MessageID,
			)
		})
		return
	}
	fanoutCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), committedMessagePostCommitTimeout)
	defer cancel()
	if err := fanout.DeliverCommittedMessage(
		fanoutCtx,
		committed.workspaceID,
		payload.AppMessage,
		delivery,
	); err != nil {
		store.committedMessageLogger().Warn(
			"deliver committed autonomous message",
			"producer", committed.producerID,
			"workspaceID", committed.workspaceID,
			"messageID", delivery.MessageID,
			"error", err,
		)
	}
}

func (store *Store) committedMessageLogger() *slog.Logger {
	if store != nil && store.logger != nil {
		return store.logger
	}
	return slog.Default()
}