package interaction

import (
	"context"
	"errors"
	"io/fs"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"

	migrationtest "github.com/codex-k8s/kodex/libs/go/migrationtest"
	outboxlib "github.com/codex-k8s/kodex/libs/go/outbox"
	interactionevents "github.com/codex-k8s/kodex/libs/go/platformevents/interaction"
	"github.com/codex-k8s/kodex/services/internal/interaction-hub/internal/domain/errs"
	"github.com/codex-k8s/kodex/services/internal/interaction-hub/internal/domain/types/entity"
	"github.com/codex-k8s/kodex/services/internal/interaction-hub/internal/domain/types/enum"
	"github.com/codex-k8s/kodex/services/internal/interaction-hub/internal/domain/types/query"
	"github.com/codex-k8s/kodex/services/internal/interaction-hub/internal/domain/types/value"
)

var sqlHeaderPattern = regexp.MustCompile(`^-- name: ([a-z0-9_]+__[a-z0-9_]+) :(one|many|exec)$`)

func TestSQLFilesHaveNamedHeaders(t *testing.T) {
	t.Parallel()

	files, err := fs.Glob(SQLFiles, "sql/*.sql")
	if err != nil {
		t.Fatalf("glob sql files: %v", err)
	}
	if len(files) == 0 {
		t.Fatal("expected embedded SQL files")
	}
	for _, file := range files {
		contentBytes, err := SQLFiles.ReadFile(file)
		if err != nil {
			t.Fatalf("read %s: %v", file, err)
		}
		firstLine, _, _ := strings.Cut(string(contentBytes), "\n")
		match := sqlHeaderPattern.FindStringSubmatch(firstLine)
		if match == nil {
			t.Fatalf("%s has invalid named query header: %q", file, firstLine)
		}
		queryName := strings.TrimSuffix(filepath.Base(file), ".sql")
		if match[1] != queryName {
			t.Fatalf("%s header query name = %s, want %s", file, match[1], queryName)
		}
	}
}

func TestRepositoryLoadsEverySQLFile(t *testing.T) {
	t.Parallel()

	files, err := fs.Glob(SQLFiles, "sql/*.sql")
	if err != nil {
		t.Fatalf("glob sql files: %v", err)
	}
	for _, file := range files {
		queryName := strings.TrimSuffix(filepath.Base(file), ".sql")
		query, err := loadQuery(queryName)
		if err != nil {
			t.Fatalf("load query %s: %v", queryName, err)
		}
		if strings.TrimSpace(query) == "" {
			t.Fatalf("query %s is empty", queryName)
		}
	}
}

func TestWrapErrorMapsPostgresErrors(t *testing.T) {
	t.Parallel()

	cases := []struct {
		name string
		err  error
		want error
	}{
		{name: "not found", err: pgx.ErrNoRows, want: errs.ErrNotFound},
		{name: "unique", err: &pgconn.PgError{Code: "23505"}, want: errs.ErrAlreadyExists},
		{name: "check", err: &pgconn.PgError{Code: "23514"}, want: errs.ErrInvalidArgument},
		{name: "serialization", err: &pgconn.PgError{Code: "40001"}, want: errs.ErrConflict},
		{name: "deadlock", err: &pgconn.PgError{Code: "40P01"}, want: errs.ErrConflict},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			if got := wrapError("test operation", tc.err); !errors.Is(got, tc.want) {
				t.Fatalf("wrapError() = %v, want %v", got, tc.want)
			}
		})
	}
}

func TestRepositoryIntegrationThreadMessageAndOutbox(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := openIntegrationPool(t, ctx)
	repository := NewRepository(pool)
	now := time.Date(2026, 5, 26, 12, 0, 0, 0, time.UTC)

	thread := testThread(now)
	createCommandID := uuid.New()
	createResult := testCommandResult(createCommandID, "thread-create", enum.OperationCreateConversationThread, interactionevents.AggregateThread, thread.ID, "create-fingerprint", now)
	createEvent := testOutboxEvent(interactionevents.EventThreadCreated, interactionevents.AggregateThread, thread.ID, now)
	if err := repository.CreateConversationThreadWithResult(ctx, thread, createResult, createEvent); err != nil {
		t.Fatalf("create thread with result: %v", err)
	}
	storedThread, err := repository.GetConversationThread(ctx, thread.ID)
	if err != nil {
		t.Fatalf("get thread: %v", err)
	}
	if storedThread.Scope.Type != enum.ScopeTypeService || storedThread.Version != 1 {
		t.Fatalf("stored thread = %+v, want service scope v1", storedThread)
	}
	replayCommandID := uuid.New()
	storedResult, err := repository.GetCommandResult(ctx, query.CommandIdentity{
		CommandID:      replayCommandID,
		IdempotencyKey: createResult.IdempotencyKey,
		ActorRef:       createResult.ActorRef,
		Operation:      createResult.Operation,
	})
	if err != nil {
		t.Fatalf("get command result by idempotency: %v", err)
	}
	if storedResult.CommandID != createCommandID || storedResult.RequestFingerprint != createResult.RequestFingerprint {
		t.Fatalf("stored result = %+v, want command %s", storedResult, createCommandID)
	}

	message := testMessage(thread.ID, now.Add(time.Minute))
	thread.LatestMessageID = &message.ID
	thread.Version = 2
	thread.UpdatedAt = message.CreatedAt
	messageResult := testCommandResult(uuid.New(), "message-create", enum.OperationRecordConversationMessage, interactionevents.AggregateMessage, message.ID, "message-fingerprint", message.CreatedAt)
	messageEvent := testOutboxEvent(interactionevents.EventMessageRecorded, interactionevents.AggregateMessage, message.ID, message.CreatedAt)
	if err := repository.CreateConversationMessageWithResult(ctx, message, thread, 99, messageResult, messageEvent); !errors.Is(err, errs.ErrConflict) {
		t.Fatalf("stale message create err = %v, want %v", err, errs.ErrConflict)
	}
	if err := repository.CreateConversationMessageWithResult(ctx, message, thread, 1, messageResult, messageEvent); err != nil {
		t.Fatalf("create message with result: %v", err)
	}
	storedMessage, err := repository.GetConversationMessage(ctx, message.ID)
	if err != nil {
		t.Fatalf("get message: %v", err)
	}
	if storedMessage.BodyObject.SizeBytes == nil || *storedMessage.BodyObject.SizeBytes != 512 || storedMessage.SafeMetadata["surface"] != "mcp" {
		t.Fatalf("stored message = %+v, want object ref and safe metadata", storedMessage)
	}
	updatedThread, err := repository.GetConversationThread(ctx, thread.ID)
	if err != nil {
		t.Fatalf("get updated thread: %v", err)
	}
	if updatedThread.LatestMessageID == nil || *updatedThread.LatestMessageID != message.ID || updatedThread.Version != 2 {
		t.Fatalf("updated thread = %+v, want latest message %s v2", updatedThread, message.ID)
	}
	messages, page, err := repository.ListConversationMessages(ctx, query.ConversationMessageFilter{ThreadID: thread.ID, Page: value.PageRequest{PageSize: 1}})
	if err != nil {
		t.Fatalf("list messages: %v", err)
	}
	if len(messages) != 1 || page.NextPageToken != "" || messages[0].ID != message.ID {
		t.Fatalf("messages = %+v page = %+v, want single message", messages, page)
	}

	claimedEvents, err := repository.ClaimOutboxEvents(ctx, 10, now.Add(2*time.Minute), now.Add(3*time.Minute))
	if err != nil {
		t.Fatalf("claim outbox events: %v", err)
	}
	if len(claimedEvents) != 2 || claimedEvents[0].AttemptCount != 1 {
		t.Fatalf("claimed events = %+v, want two leased events", claimedEvents)
	}
	if err := repository.MarkOutboxEventPublished(ctx, claimedEvents[0].ID, claimedEvents[0].AttemptCount, now.Add(4*time.Minute)); err != nil {
		t.Fatalf("mark first event published: %v", err)
	}
	if err := repository.MarkOutboxEventFailed(ctx, claimedEvents[1].ID, claimedEvents[1].AttemptCount, now.Add(5*time.Minute), "temporary"); err != nil {
		t.Fatalf("mark second event failed: %v", err)
	}
	reclaimedEvents, err := repository.ClaimOutboxEvents(ctx, 10, now.Add(6*time.Minute), now.Add(7*time.Minute))
	if err != nil {
		t.Fatalf("reclaim outbox events: %v", err)
	}
	if len(reclaimedEvents) != 1 || reclaimedEvents[0].ID != claimedEvents[1].ID || reclaimedEvents[0].AttemptCount != 2 {
		t.Fatalf("reclaimed events = %+v, want retry event attempt 2", reclaimedEvents)
	}
	if err := repository.MarkOutboxEventPermanentlyFailed(ctx, reclaimedEvents[0].ID, reclaimedEvents[0].AttemptCount, now.Add(8*time.Minute), "permanent"); err != nil {
		t.Fatalf("mark retry event permanently failed: %v", err)
	}
}

func TestRepositoryIntegrationInteractionRequestResponseLifecycle(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := openIntegrationPool(t, ctx)
	repository := NewRepository(pool)
	now := time.Date(2026, 5, 26, 12, 0, 0, 0, time.UTC)

	request := testInteractionRequest(now)
	createResult := testCommandResult(uuid.New(), "request-create", enum.OperationRequestApproval, interactionevents.AggregateRequest, request.ID, "request-fingerprint", now)
	createEvent := testOutboxEvent(interactionevents.EventApprovalRequested, interactionevents.AggregateRequest, request.ID, now)
	if err := repository.CreateInteractionRequestWithResult(ctx, request, createResult, createEvent); err != nil {
		t.Fatalf("create request with result: %v", err)
	}
	storedRequest, err := repository.GetInteractionRequest(ctx, request.ID)
	if err != nil {
		t.Fatalf("get request: %v", err)
	}
	if storedRequest.Scope.Type != enum.ScopeTypeService || storedRequest.ReminderPolicyRef != "policy:standard" || len(storedRequest.TargetRefs) != 1 {
		t.Fatalf("stored request = %+v, want service request with refs", storedRequest)
	}
	listed, _, err := repository.ListInteractionRequests(ctx, query.InteractionRequestFilter{Scope: request.Scope, Status: enum.InteractionRequestStatusWaiting, Page: value.PageRequest{PageSize: 10}})
	if err != nil {
		t.Fatalf("list requests: %v", err)
	}
	if len(listed) != 1 || listed[0].ID != request.ID {
		t.Fatalf("listed requests = %+v, want created request", listed)
	}

	response := testInteractionResponse(request.ID, now.Add(time.Minute))
	response.ResponseAction = enum.InteractionResponseActionRequestChanges
	response.ResponseSummary = "needs changes"
	answered := storedRequest
	answered.Status = enum.InteractionRequestStatusAnswered
	answered.Version = 2
	answered.UpdatedAt = response.CreatedAt
	answered.ResolvedAt = &response.CreatedAt
	responseResult := testCommandResult(uuid.New(), "response-create", enum.OperationRecordInteractionResponse, "response", response.ID, "response-fingerprint", response.CreatedAt)
	responseEvent := testOutboxEvent(interactionevents.EventRequestResponseRecorded, interactionevents.AggregateRequest, request.ID, response.CreatedAt)
	if err := repository.CreateInteractionResponseWithResult(ctx, response, answered, 99, responseResult, responseEvent); !errors.Is(err, errs.ErrConflict) {
		t.Fatalf("stale response create err = %v, want ErrConflict", err)
	}
	if err := repository.CreateInteractionResponseWithResult(ctx, response, answered, 1, responseResult, responseEvent); err != nil {
		t.Fatalf("create response with result: %v", err)
	}
	storedResponse, err := repository.GetInteractionResponse(ctx, response.ID)
	if err != nil {
		t.Fatalf("get response: %v", err)
	}
	if storedResponse.ResponseAction != enum.InteractionResponseActionRequestChanges ||
		storedResponse.ResponseObject.SizeBytes == nil ||
		*storedResponse.ResponseObject.SizeBytes != 256 ||
		storedResponse.OwnerDecisionRef != "decision:1" {
		t.Fatalf("stored response = %+v, want request_changes with object ref and decision ref", storedResponse)
	}
	updatedRequest, err := repository.GetInteractionRequest(ctx, request.ID)
	if err != nil {
		t.Fatalf("get answered request: %v", err)
	}
	if updatedRequest.Status != enum.InteractionRequestStatusAnswered || updatedRequest.Version != 2 || updatedRequest.ResolvedAt == nil {
		t.Fatalf("updated request = %+v, want answered v2", updatedRequest)
	}

	expiringRequest := testInteractionRequest(now.Add(2 * time.Minute))
	expiringRequest.ID = uuid.New()
	expireDeadline := now.Add(-time.Minute)
	expiringRequest.DeadlineAt = &expireDeadline
	expireCreateResult := testCommandResult(uuid.New(), "request-expiring", enum.OperationRequestApproval, interactionevents.AggregateRequest, expiringRequest.ID, "request-expiring-fingerprint", now)
	expireCreateEvent := testOutboxEvent(interactionevents.EventApprovalRequested, interactionevents.AggregateRequest, expiringRequest.ID, now)
	if err := repository.CreateInteractionRequestWithResult(ctx, expiringRequest, expireCreateResult, expireCreateEvent); err != nil {
		t.Fatalf("create expiring request: %v", err)
	}
	candidates, err := repository.ListExpirableInteractionRequests(ctx, expiringRequest.Scope, now, 10)
	if err != nil {
		t.Fatalf("list expirable requests: %v", err)
	}
	if len(candidates) != 1 || candidates[0].ID != expiringRequest.ID {
		t.Fatalf("candidates = %+v, want expiring request", candidates)
	}
	expired := candidates[0]
	previousVersions := map[uuid.UUID]int64{expired.ID: expired.Version}
	expired.Status = enum.InteractionRequestStatusExpired
	expired.Version = 2
	expired.UpdatedAt = now.Add(3 * time.Minute)
	expired.ResolvedAt = &expired.UpdatedAt
	expireResult := testCommandResult(uuid.New(), "expire", enum.OperationExpireInteractionRequests, interactionevents.AggregateRequest, uuid.Nil, "expire-fingerprint", expired.UpdatedAt)
	expireEvent := testOutboxEvent(interactionevents.EventRequestExpired, interactionevents.AggregateRequest, expired.ID, expired.UpdatedAt)
	if err := repository.UpdateInteractionRequestsWithResult(ctx, []entity.InteractionRequest{expired}, previousVersions, expireResult, []entity.OutboxEvent{expireEvent}); err != nil {
		t.Fatalf("expire request with result: %v", err)
	}
	storedExpired, err := repository.GetInteractionRequest(ctx, expired.ID)
	if err != nil {
		t.Fatalf("get expired request: %v", err)
	}
	if storedExpired.Status != enum.InteractionRequestStatusExpired || storedExpired.Version != 2 {
		t.Fatalf("stored expired = %+v, want expired v2", storedExpired)
	}
}

func TestRepositoryIntegrationNotificationSubscriptionLifecycle(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := openIntegrationPool(t, ctx)
	repository := NewRepository(pool)
	now := time.Date(2026, 5, 26, 12, 0, 0, 0, time.UTC)

	subscription := testSubscription(now)
	createResult := testCommandResult(uuid.New(), "subscription-create", enum.OperationUpsertSubscription, interactionevents.AggregateSubscription, subscription.ID, "subscription-create-fingerprint", now)
	createEvent := testOutboxEvent(interactionevents.EventSubscriptionUpdated, interactionevents.AggregateSubscription, subscription.ID, now)
	if err := repository.CreateSubscriptionWithResult(ctx, subscription, createResult, createEvent); err != nil {
		t.Fatalf("create subscription with result: %v", err)
	}
	storedSubscription, err := repository.GetSubscription(ctx, subscription.ID)
	if err != nil {
		t.Fatalf("get subscription: %v", err)
	}
	if storedSubscription.SourceOwner.Kind != enum.SourceOwnerKindAgentManager || storedSubscription.SubscriptionPolicyRef != "policy:ops-notifications" || len(storedSubscription.ChannelHintRefs) != 1 {
		t.Fatalf("stored subscription = %+v, want source owner, channel hints and policy ref", storedSubscription)
	}
	listedSubscriptions, _, err := repository.ListSubscriptions(ctx, query.SubscriptionFilter{
		Scope:         subscription.Scope,
		SubscriberRef: subscription.SubscriberRef.String(),
		Status:        enum.SubscriptionStatusActive,
		Page:          value.PageRequest{PageSize: 10},
	})
	if err != nil {
		t.Fatalf("list subscriptions: %v", err)
	}
	if len(listedSubscriptions) != 1 || listedSubscriptions[0].ID != subscription.ID {
		t.Fatalf("listed subscriptions = %+v, want created subscription", listedSubscriptions)
	}

	updatedSubscription := storedSubscription
	updatedSubscription.Status = enum.SubscriptionStatusPaused
	updatedSubscription.Version = 2
	updatedSubscription.UpdatedAt = now.Add(time.Minute)
	updateResult := testCommandResult(uuid.New(), "subscription-update", enum.OperationUpsertSubscription, interactionevents.AggregateSubscription, subscription.ID, "subscription-update-fingerprint", updatedSubscription.UpdatedAt)
	updateEvent := testOutboxEvent(interactionevents.EventSubscriptionUpdated, interactionevents.AggregateSubscription, subscription.ID, updatedSubscription.UpdatedAt)
	if err := repository.UpdateSubscriptionWithResult(ctx, updatedSubscription, 99, updateResult, updateEvent); !errors.Is(err, errs.ErrConflict) {
		t.Fatalf("stale subscription update err = %v, want ErrConflict", err)
	}
	if err := repository.UpdateSubscriptionWithResult(ctx, updatedSubscription, 1, updateResult, updateEvent); err != nil {
		t.Fatalf("update subscription with result: %v", err)
	}
	storedUpdatedSubscription, err := repository.GetSubscription(ctx, subscription.ID)
	if err != nil {
		t.Fatalf("get updated subscription: %v", err)
	}
	if storedUpdatedSubscription.Status != enum.SubscriptionStatusPaused || storedUpdatedSubscription.Version != 2 {
		t.Fatalf("updated subscription = %+v, want paused v2", storedUpdatedSubscription)
	}

	notification := testNotification(now.Add(2 * time.Minute))
	notification.SubscriptionID = &subscription.ID
	notificationResult := testCommandResult(uuid.New(), "notification-create", enum.OperationRequestNotification, interactionevents.AggregateNotification, notification.ID, "notification-create-fingerprint", notification.CreatedAt)
	notificationEvent := testOutboxEvent(interactionevents.EventNotificationRequested, interactionevents.AggregateNotification, notification.ID, notification.CreatedAt)
	if err := repository.CreateNotificationWithResult(ctx, notification, notificationResult, notificationEvent); err != nil {
		t.Fatalf("create notification with result: %v", err)
	}
	storedNotification, err := repository.GetNotification(ctx, notification.ID)
	if err != nil {
		t.Fatalf("get notification: %v", err)
	}
	if storedNotification.SubscriptionID == nil || *storedNotification.SubscriptionID != subscription.ID || storedNotification.MessageTitle != "Safe title" || storedNotification.NotificationPolicyRef != "policy:notify-standard" {
		t.Fatalf("stored notification = %+v, want subscription ref and safe policy fields", storedNotification)
	}
}

func TestRepositoryIntegrationDeliveryAttemptLifecycle(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	pool := openIntegrationPool(t, ctx)
	repository := NewRepository(pool)
	now := time.Date(2026, 5, 26, 12, 0, 0, 0, time.UTC)

	request := testInteractionRequest(now)
	createResult := testCommandResult(uuid.New(), "request-delivery", enum.OperationRequestApproval, interactionevents.AggregateRequest, request.ID, "request-delivery-fingerprint", now)
	createEvent := testOutboxEvent(interactionevents.EventApprovalRequested, interactionevents.AggregateRequest, request.ID, now)
	if err := repository.CreateInteractionRequestWithResult(ctx, request, createResult, createEvent); err != nil {
		t.Fatalf("create request: %v", err)
	}
	route := testDeliveryRoute(now)
	insertDeliveryRoute(t, ctx, pool, route)
	storedRoute, err := repository.GetDeliveryRoute(ctx, route.ID)
	if err != nil {
		t.Fatalf("get route: %v", err)
	}
	if storedRoute.Scope != route.Scope || storedRoute.RoutingPolicyRef != "policy:route-standard" || storedRoute.PackageVersionRef != route.PackageVersionRef || storedRoute.CallbackRouteRef != route.CallbackRouteRef || storedRoute.RuntimeRef != route.RuntimeRef {
		t.Fatalf("route = %+v, want stored route", storedRoute)
	}
	activeRoute, err := repository.FindActiveDeliveryRoute(ctx, route.Scope)
	if err != nil {
		t.Fatalf("find active route: %v", err)
	}
	if activeRoute.ID != route.ID {
		t.Fatalf("active route = %+v, want %s", activeRoute, route.ID)
	}

	attempt := testDeliveryAttempt(request.ID, route.ID, now.Add(time.Minute))
	attemptResult := testCommandResult(uuid.New(), "delivery-plan", enum.OperationPlanDelivery, interactionevents.AggregateDelivery, attempt.ID, "delivery-plan-fingerprint", attempt.CreatedAt)
	attemptEvent := testOutboxEvent(interactionevents.EventDeliveryRequested, interactionevents.AggregateDelivery, attempt.ID, attempt.CreatedAt)
	if err := repository.CreateDeliveryAttemptWithResult(ctx, attempt, attemptResult, attemptEvent); err != nil {
		t.Fatalf("create delivery attempt: %v", err)
	}
	storedAttempt, err := repository.GetDeliveryAttemptByDeliveryID(ctx, attempt.DeliveryID)
	if err != nil {
		t.Fatalf("get attempt by delivery id: %v", err)
	}
	if storedAttempt.Target.ID != request.ID || storedAttempt.Status != enum.DeliveryAttemptStatusQueued || storedAttempt.PayloadDigest != attempt.PayloadDigest || storedAttempt.DeliveryCommandRef != attempt.DeliveryCommandRef {
		t.Fatalf("stored attempt = %+v, want queued request attempt", storedAttempt)
	}
	listedAttempts, err := repository.ListDeliveryAttempts(ctx, query.DeliveryAttemptFilter{Target: attempt.Target, Limit: 10})
	if err != nil {
		t.Fatalf("list attempts: %v", err)
	}
	if len(listedAttempts) != 1 || listedAttempts[0].ID != attempt.ID {
		t.Fatalf("listed attempts = %+v, want created attempt", listedAttempts)
	}

	accepted := storedAttempt
	sentAt := now.Add(2 * time.Minute)
	accepted.Status = enum.DeliveryAttemptStatusAccepted
	accepted.ChannelMessageRef = "channel:message-1"
	accepted.ResultFingerprint = "sha256:" + strings.Repeat("f", 64)
	accepted.SentAt = &sentAt
	accepted.UpdatedAt = sentAt
	acceptedResult := testCommandResult(uuid.New(), "delivery-result", enum.OperationRecordDeliveryResult, interactionevents.AggregateDelivery, accepted.ID, "delivery-result-fingerprint", sentAt)
	acceptedEvent := testOutboxEvent(interactionevents.EventDeliveryAccepted, interactionevents.AggregateDelivery, accepted.ID, sentAt)
	if err := repository.UpdateDeliveryAttemptWithResult(ctx, accepted, acceptedResult, acceptedEvent); err != nil {
		t.Fatalf("update delivery attempt: %v", err)
	}
	storedAccepted, err := repository.GetDeliveryAttempt(ctx, accepted.ID)
	if err != nil {
		t.Fatalf("get accepted attempt: %v", err)
	}
	if storedAccepted.Status != enum.DeliveryAttemptStatusAccepted || storedAccepted.ChannelMessageRef != "channel:message-1" || storedAccepted.ResultFingerprint != accepted.ResultFingerprint || storedAccepted.SentAt == nil {
		t.Fatalf("accepted attempt = %+v, want accepted with channel ref", storedAccepted)
	}

	callback := testChannelCallback(storedAccepted, request.ID, sentAt.Add(time.Minute))
	responseObjectSize := int64(128)
	callbackResult := testCommandResult(uuid.New(), "callback-record", enum.OperationRecordChannelCallback, interactionevents.AggregateCallback, callback.ID, "callback-record-fingerprint", callback.CreatedAt)
	callbackEvent := testOutboxEvent(interactionevents.EventCallbackReceived, interactionevents.AggregateCallback, callback.ID, callback.CreatedAt)
	callbackResponse := entity.InteractionResponse{
		ID:                  uuid.New(),
		RequestID:           request.ID,
		ResponseAction:      enum.InteractionResponseActionApprove,
		RespondedByActorRef: callback.ActorRef,
		ResponseSummary:     callback.CallbackSummary,
		ResponseObject:      value.ObjectRef{URI: "s3://kodex-interactions/responses/callback-1", Digest: "sha256:callback-response", SizeBytes: &responseObjectSize},
		SourceKind:          enum.InteractionResponseSourceKindChannelCallback,
		SourceRef:           callback.ID.String(),
		CreatedAt:           callback.CreatedAt,
	}
	answeredRequest := request
	answeredRequest.Status = enum.InteractionRequestStatusAnswered
	answeredRequest.Version = 2
	answeredRequest.UpdatedAt = callback.CreatedAt
	answeredRequest.ResolvedAt = &callback.CreatedAt
	responseEvent := testOutboxEvent(interactionevents.EventRequestResponseRecorded, interactionevents.AggregateRequest, request.ID, callback.CreatedAt)
	if err := repository.CreateChannelCallbackResponseWithResult(ctx, callback, callbackResponse, answeredRequest, 1, callbackResult, []entity.OutboxEvent{callbackEvent, responseEvent}); err != nil {
		t.Fatalf("create channel callback response: %v", err)
	}
	storedCallback, err := repository.GetChannelCallbackByCallbackID(ctx, callback.CallbackID)
	if err != nil {
		t.Fatalf("get callback by callback id: %v", err)
	}
	if storedCallback.DeliveryAttemptID == nil || *storedCallback.DeliveryAttemptID != storedAccepted.ID || storedCallback.CallbackRouteRef != callback.CallbackRouteRef || storedCallback.CallbackFingerprint != callback.CallbackFingerprint {
		t.Fatalf("stored callback = %+v, want linked callback refs", storedCallback)
	}
	storedCallbackResponse, err := repository.GetInteractionResponseBySource(ctx, enum.InteractionResponseSourceKindChannelCallback, callback.ID.String())
	if err != nil {
		t.Fatalf("get callback response by source: %v", err)
	}
	if storedCallbackResponse.ID != callbackResponse.ID || storedCallbackResponse.RequestID != request.ID {
		t.Fatalf("stored callback response = %+v, want %s", storedCallbackResponse, callbackResponse.ID)
	}
	storedAnsweredRequest, err := repository.GetInteractionRequest(ctx, request.ID)
	if err != nil {
		t.Fatalf("get answered callback request: %v", err)
	}
	if storedAnsweredRequest.Status != enum.InteractionRequestStatusAnswered || storedAnsweredRequest.Version != 2 {
		t.Fatalf("answered request = %+v, want answered v2", storedAnsweredRequest)
	}
	latestCallback, err := repository.GetLatestChannelCallback(ctx, query.ChannelCallbackFilter{DeliveryAttemptIDs: []uuid.UUID{storedAccepted.ID}})
	if err != nil {
		t.Fatalf("get latest callback: %v", err)
	}
	if latestCallback.ID != callback.ID {
		t.Fatalf("latest callback = %+v, want %s", latestCallback, callback.ID)
	}
	inboxItems, _, err := repository.ListOwnerInboxItems(ctx, query.OwnerInboxFilter{
		Scope:          request.Scope,
		Statuses:       []enum.InteractionRequestStatus{enum.InteractionRequestStatusAnswered},
		AssigneeRef:    value.ActorRef{Kind: "user", Ref: "approver-1"},
		ActorRef:       callback.ActorRef,
		CorrelationRef: value.ExternalRef{Kind: "agent_run", Ref: "run:123"},
		CorrelationID:  callback.CorrelationID,
		Page:           value.PageRequest{PageSize: 10},
	})
	if err != nil {
		t.Fatalf("list owner inbox items: %v", err)
	}
	if len(inboxItems) != 1 {
		t.Fatalf("owner inbox items = %+v, want one item", inboxItems)
	}
	inboxItem := inboxItems[0]
	if inboxItem.Request.ID != request.ID || inboxItem.DeliverySummary.AttemptCount != 1 || inboxItem.DeliverySummary.LatestDeliveryID != storedAccepted.DeliveryID {
		t.Fatalf("owner inbox item = %+v, want request and delivery summary", inboxItem)
	}
	if len(inboxItem.Request.AllowedActions) == 0 || inboxItem.Request.AllowedActions[0].ActionKey != "approve" {
		t.Fatalf("owner inbox allowed actions = %+v, want response actions for UI detail", inboxItem.Request.AllowedActions)
	}
	if inboxItem.LatestCallback == nil || inboxItem.LatestCallback.ID != callback.ID || inboxItem.LatestResponse == nil || inboxItem.LatestResponse.ID != callbackResponse.ID {
		t.Fatalf("owner inbox callback=%+v response=%+v, want latest safe refs", inboxItem.LatestCallback, inboxItem.LatestResponse)
	}
	if inboxItem.LatestResponse.ResponseSummary != callback.CallbackSummary || inboxItem.LatestResponse.ResponseSummaryDigest == "" {
		t.Fatalf("owner inbox response summary=%q digest=%q, want safe summary and digest", inboxItem.LatestResponse.ResponseSummary, inboxItem.LatestResponse.ResponseSummaryDigest)
	}
	if inboxItem.LatestResponse.ResponseObject.Digest != "sha256:callback-response" || inboxItem.LatestResponse.ResponseObject.SizeBytes == nil || *inboxItem.LatestResponse.ResponseObject.SizeBytes != responseObjectSize {
		t.Fatalf("owner inbox response object = %+v, want safe object ref", inboxItem.LatestResponse.ResponseObject)
	}
	if inboxItem.LatestCallback.CallbackSummary != "" {
		t.Fatalf("owner inbox leaked callback summary: %q", inboxItem.LatestCallback.CallbackSummary)
	}
	detailItems, _, err := repository.ListOwnerInboxItems(ctx, query.OwnerInboxFilter{
		RequestID:   request.ID,
		Scope:       request.Scope,
		Statuses:    []enum.InteractionRequestStatus{enum.InteractionRequestStatusAnswered},
		AssigneeRef: value.ActorRef{Kind: "user", Ref: "approver-1"},
		Page:        value.PageRequest{PageSize: 1},
	})
	if err != nil {
		t.Fatalf("get owner inbox item via request filter: %v", err)
	}
	if len(detailItems) != 1 || detailItems[0].Request.ID != request.ID {
		t.Fatalf("detail items = %+v, want request-filtered owner inbox item", detailItems)
	}
	mismatchItems, _, err := repository.ListOwnerInboxItems(ctx, query.OwnerInboxFilter{
		RequestID:   request.ID,
		Scope:       request.Scope,
		Statuses:    []enum.InteractionRequestStatus{enum.InteractionRequestStatusAnswered},
		AssigneeRef: value.ActorRef{Kind: "user", Ref: "someone-else"},
		Page:        value.PageRequest{PageSize: 1},
	})
	if err != nil {
		t.Fatalf("owner inbox item with mismatched assignee: %v", err)
	}
	if len(mismatchItems) != 0 {
		t.Fatalf("mismatch items = %+v, want no owner inbox item for wrong assignee", mismatchItems)
	}
}

func testThread(now time.Time) entity.ConversationThread {
	return entity.ConversationThread{
		ID:              uuid.New(),
		Scope:           value.ScopeRef{Type: enum.ScopeTypeService, Ref: "agent-manager"},
		ThreadKind:      enum.ConversationThreadKindUserDialog,
		PrimaryActorRef: "service:agent-manager",
		SourceKind:      enum.ConversationSourceKindService,
		SourceRef:       "run:123",
		Status:          enum.ConversationThreadStatusOpen,
		CorrelationID:   "trace-123",
		RetentionClass:  "standard",
		Version:         1,
		CreatedAt:       now,
		UpdatedAt:       now,
	}
}

func testMessage(threadID uuid.UUID, now time.Time) entity.ConversationMessage {
	size := int64(512)
	return entity.ConversationMessage{
		ID:          uuid.New(),
		ThreadID:    threadID,
		MessageKind: enum.ConversationMessageKindAgentText,
		AuthorRef:   "agent:codex",
		BodySummary: "safe summary",
		BodyObject: value.ObjectRef{
			URI:       "s3://kodex-interactions/messages/1",
			Digest:    "sha256:" + strings.Repeat("a", 64),
			SizeBytes: &size,
		},
		BodyDigest:   "sha256:" + strings.Repeat("b", 64),
		Locale:       "ru",
		SafeMetadata: map[string]string{"surface": "mcp"},
		CreatedAt:    now,
	}
}

func testInteractionRequest(now time.Time) entity.InteractionRequest {
	deadline := now.Add(time.Hour)
	size := int64(1024)
	return entity.InteractionRequest{
		ID:            uuid.New(),
		RequestKind:   enum.InteractionRequestKindApproval,
		Scope:         value.ScopeRef{Type: enum.ScopeTypeService, Ref: "agent-manager"},
		SourceOwner:   value.SourceOwnerRef{Kind: enum.SourceOwnerKindAgentManager, Ref: "run:123"},
		Ingress:       value.IngressRef{Kind: enum.IngressKindDirectGRPC, Ref: "grpc:command-1"},
		DecisionOwner: value.DecisionOwnerRef{Kind: enum.DecisionOwnerKindGovernanceManager, OwnerRequestRef: "gate:req-1"},
		TargetRefs: []value.ActorRef{
			{Kind: "user", Ref: "approver-1"},
		},
		ContextRefs: []value.ExternalRef{
			{Kind: "agent_run", Ref: "run:123"},
			{Kind: "provider_operation", Ref: "provider:op-1"},
		},
		PromptSummary: "safe prompt summary",
		PromptObject: value.ObjectRef{
			URI:       "s3://kodex-interactions/prompts/1",
			Digest:    "sha256:" + strings.Repeat("c", 64),
			SizeBytes: &size,
		},
		AllowedActions: []value.InteractionAction{
			{ActionKey: string(enum.InteractionResponseActionApprove), LabelTemplateRef: "interaction.actions.approve", Terminal: true},
		},
		RiskClass:         enum.InteractionRiskClassHigh,
		Status:            enum.InteractionRequestStatusWaiting,
		DeadlineAt:        &deadline,
		ReminderPolicyRef: "policy:standard",
		Version:           1,
		CreatedAt:         now,
		UpdatedAt:         now,
	}
}

func testInteractionResponse(requestID uuid.UUID, now time.Time) entity.InteractionResponse {
	size := int64(256)
	return entity.InteractionResponse{
		ID:                  uuid.New(),
		RequestID:           requestID,
		ResponseAction:      enum.InteractionResponseActionApprove,
		RespondedByActorRef: "user:approver-1",
		ResponseSummary:     "approved",
		ResponseObject: value.ObjectRef{
			URI:       "s3://kodex-interactions/responses/1",
			Digest:    "sha256:" + strings.Repeat("d", 64),
			SizeBytes: &size,
		},
		SourceKind:       enum.InteractionResponseSourceKindMCP,
		SourceRef:        "mcp:command-1",
		OwnerDecisionRef: "decision:1",
		CreatedAt:        now,
	}
}

func testNotification(now time.Time) entity.Notification {
	expiresAt := now.Add(time.Hour)
	return entity.Notification{
		ID:                 uuid.New(),
		Scope:              value.ScopeRef{Type: enum.ScopeTypeService, Ref: "agent-manager"},
		NotificationKind:   enum.NotificationKindAttention,
		RecipientRefs:      []value.ActorRef{{Kind: "user", Ref: "owner-1"}},
		MessageTemplateRef: "interaction.notification.attention",
		MessageTitle:       "Safe title",
		MessageSummary:     "safe bounded summary",
		BodyPreview:        "safe bounded preview",
		Priority:           enum.NotificationPriorityHigh,
		Status:             enum.NotificationStatusCreated,
		SourceOwner:        value.SourceOwnerRef{Kind: enum.SourceOwnerKindAgentManager, Ref: "run:123"},
		Ingress:            value.IngressRef{Kind: enum.IngressKindDirectGRPC, Ref: "grpc:notify-1"},
		ContextRefs: []value.ExternalRef{
			{Kind: "agent_run", Ref: "run:123"},
		},
		ChannelHintRefs: []value.ExternalRef{
			{Kind: "surface", Ref: "web_console"},
		},
		NotificationPolicyRef: "policy:notify-standard",
		CreatedAt:             now,
		UpdatedAt:             now,
		ExpiresAt:             &expiresAt,
	}
}

func testSubscription(now time.Time) entity.Subscription {
	return entity.Subscription{
		ID:                      uuid.New(),
		Scope:                   value.ScopeRef{Type: enum.ScopeTypeService, Ref: "agent-manager"},
		SubscriberRef:           value.ActorRef{Kind: "user", Ref: "owner-1"},
		EventFilterJSON:         `{"event_kind":["run_waiting"],"severity":["high"]}`,
		DeliveryPreferencesJSON: `{"surfaces":["web_console"],"fallback_policy_ref":"policy:fallback"}`,
		Status:                  enum.SubscriptionStatusActive,
		Version:                 1,
		SourceOwner:             value.SourceOwnerRef{Kind: enum.SourceOwnerKindAgentManager, Ref: "run:123"},
		ChannelHintRefs: []value.ExternalRef{
			{Kind: "surface", Ref: "web_console"},
		},
		SubscriptionPolicyRef: "policy:ops-notifications",
		CreatedAt:             now,
		UpdatedAt:             now,
	}
}

func testDeliveryRoute(now time.Time) entity.DeliveryRoute {
	return entity.DeliveryRoute{
		ID:                     uuid.New(),
		Scope:                  value.ScopeRef{Type: enum.ScopeTypeService, Ref: "agent-manager"},
		SurfaceKind:            enum.DeliverySurfaceKindChannelPackage,
		ChannelCapabilityRef:   "capability:channel",
		PackageInstallationRef: "package-installation:channel-core",
		PackageVersionRef:      "package-version:channel-core:v1",
		RoutingPolicyRef:       "policy:route-standard",
		CallbackRouteRef:       "callback-route:interaction-channel",
		RuntimeRef:             "runtime:channel-core",
		Status:                 enum.DeliveryRouteStatusActive,
		CreatedAt:              now,
		UpdatedAt:              now,
	}
}

func testDeliveryAttempt(requestID uuid.UUID, routeID uuid.UUID, now time.Time) entity.DeliveryAttempt {
	return entity.DeliveryAttempt{
		ID:                     uuid.New(),
		Target:                 value.DeliveryTarget{Kind: value.DeliveryTargetKindRequest, ID: requestID},
		RouteID:                routeID,
		DeliveryID:             "delivery-" + uuid.NewString(),
		DeliveryKind:           enum.DeliveryKindApproval,
		Status:                 enum.DeliveryAttemptStatusQueued,
		AttemptNumber:          1,
		PayloadDigest:          "sha256:" + strings.Repeat("e", 64),
		ChannelCapabilityRef:   "capability:channel",
		PackageInstallationRef: "package-installation:channel-core",
		PackageVersionRef:      "package-version:channel-core:v1",
		DeliveryCommandRef:     "interaction.delivery_command:" + uuid.NewString(),
		CallbackRef:            "interaction.callback:" + uuid.NewString(),
		CallbackRouteRef:       "callback-route:interaction-channel",
		RuntimeRef:             "runtime:channel-core",
		RoutingPolicyRef:       "policy:route-standard",
		CreatedAt:              now,
		UpdatedAt:              now,
	}
}

func testChannelCallback(attempt entity.DeliveryAttempt, requestID uuid.UUID, now time.Time) entity.ChannelCallback {
	size := int64(256)
	return entity.ChannelCallback{
		ID:                  uuid.New(),
		CallbackID:          "callback-" + uuid.NewString(),
		DeliveryID:          attempt.DeliveryID,
		DeliveryAttemptID:   &attempt.ID,
		RequestID:           &requestID,
		SourceRouteID:       &attempt.RouteID,
		ActorRef:            "user:approver-1",
		Action:              string(enum.InteractionResponseActionApprove),
		CallbackSummary:     "safe callback summary",
		CallbackObject:      value.ObjectRef{URI: "s3://kodex-interactions/callbacks/1", Digest: "sha256:" + strings.Repeat("c", 64), SizeBytes: &size},
		SignatureStatus:     enum.CallbackSignatureStatusVerified,
		ProcessingStatus:    enum.CallbackProcessingStatusAccepted,
		ReceivedAt:          now,
		CreatedAt:           now,
		CallbackRouteRef:    attempt.CallbackRouteRef,
		GatewayRef:          "gateway:request-1",
		CorrelationID:       "trace-callback",
		CallbackFingerprint: "sha256:" + strings.Repeat("d", 64),
	}
}

func insertDeliveryRoute(t *testing.T, ctx context.Context, pool *pgxpool.Pool, route entity.DeliveryRoute) {
	t.Helper()

	_, err := pool.Exec(ctx, `
INSERT INTO interaction_hub_delivery_routes (
    id,
    scope_type,
    scope_ref,
    surface_kind,
    channel_capability_ref,
    package_installation_ref,
    routing_policy_ref,
    status,
    created_at,
    updated_at,
    package_version_ref,
    callback_route_ref,
    runtime_ref
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
`, route.ID, string(route.Scope.Type), route.Scope.Ref, string(route.SurfaceKind), route.ChannelCapabilityRef, route.PackageInstallationRef, route.RoutingPolicyRef, string(route.Status), route.CreatedAt, route.UpdatedAt, route.PackageVersionRef, route.CallbackRouteRef, route.RuntimeRef)
	if err != nil {
		t.Fatalf("insert delivery route: %v", err)
	}
}

func testCommandResult(commandID uuid.UUID, idempotencyKey string, operation enum.Operation, aggregateType string, aggregateID uuid.UUID, fingerprint string, now time.Time) entity.CommandResult {
	return entity.CommandResult{
		Key:                "command:" + commandID.String(),
		CommandID:          commandID,
		IdempotencyKey:     idempotencyKey,
		ActorRef:           "service:interaction-test",
		Operation:          operation,
		AggregateType:      aggregateType,
		AggregateID:        aggregateID,
		RequestFingerprint: fingerprint,
		ResultPayload:      []byte(`{}`),
		CreatedAt:          now,
	}
}

func testOutboxEvent(eventType string, aggregateType string, aggregateID uuid.UUID, now time.Time) entity.OutboxEvent {
	return entity.OutboxEvent{
		Event: outboxlib.NewEvent(uuid.New(), eventType, interactionevents.SchemaVersion, aggregateType, aggregateID, []byte(`{"version":1}`), now, 0),
	}
}

func openIntegrationPool(t *testing.T, ctx context.Context) *pgxpool.Pool {
	t.Helper()

	dsn := os.Getenv("KODEX_INTERACTION_HUB_TEST_DATABASE_DSN")
	if strings.TrimSpace(dsn) == "" {
		t.Skip("set KODEX_INTERACTION_HUB_TEST_DATABASE_DSN to run PostgreSQL repository integration tests")
	}
	adminPool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatalf("open admin pool: %v", err)
	}
	t.Cleanup(adminPool.Close)

	schema := "interaction_repo_test_" + strings.ReplaceAll(uuid.NewString(), "-", "_")
	quotedSchema := pgx.Identifier{schema}.Sanitize()
	if _, err := adminPool.Exec(ctx, "CREATE SCHEMA "+quotedSchema); err != nil {
		t.Fatalf("create test schema: %v", err)
	}
	t.Cleanup(func() {
		_, _ = adminPool.Exec(context.WithoutCancel(ctx), "DROP SCHEMA IF EXISTS "+quotedSchema+" CASCADE")
	})

	cfg, err := pgxpool.ParseConfig(dsn)
	if err != nil {
		t.Fatalf("parse pool config: %v", err)
	}
	cfg.ConnConfig.RuntimeParams["search_path"] = schema
	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		t.Fatalf("open test pool: %v", err)
	}
	t.Cleanup(pool.Close)
	applyMigrations(t, ctx, pool)
	return pool
}

func applyMigrations(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
	t.Helper()

	statements := migrationtest.GooseUpStatements(t, "../../../../cmd/cli/migrations")
	for _, statement := range statements {
		if _, err := pool.Exec(ctx, statement); err != nil {
			t.Fatalf("apply migration statement %q: %v", statement, err)
		}
	}
}
