story/internal/orchestration/service.go

package orchestration

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/conversation"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/embeddingprovider"
	"data-terminal-cloud/internal/idempotency"
	"data-terminal-cloud/internal/memorytool"
	"data-terminal-cloud/internal/outbox"
	"data-terminal-cloud/internal/rtc"
	"data-terminal-cloud/internal/vikingmemory"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

type ServiceOptions struct {
	RTCAppID                string
	VoiceProvider           string
	EngineMode              string
	AllocationTTL           time.Duration
	TokenTTL                time.Duration
	HeartbeatInterval       int
	RequirePublishedRouting bool
	HardwareBotID           string
	HardwareTransport       string
	SubtitleCallbackURL     string
	SubtitleCallbackSecret  []byte
	SubtitlePrimary         bool
}

type Service struct {
	viking                *vikingmemory.Service
	executor              *idempotency.Executor
	repository            *Repository
	audit                 *audit.Repository
	outbox                *outbox.Store
	tokens                rtc.TokenIssuer
	leases                LeaseCache
	options               ServiceOptions
	pool                  *pgxpool.Pool
	embedder              embeddingprovider.Provider
	hardwareMemoryTool    *memorytool.TokenSigner
	hardwareMemoryToolURL string
	hardwareMemoryToolTTL time.Duration
	now                   func() time.Time
}

func (s *Service) ConfigureVikingMemory(memory *vikingmemory.Service) { s.viking = memory }

func (s *Service) ConfigureMemoryEmbedding(pool *pgxpool.Pool, provider embeddingprovider.Provider) error {
	if pool == nil || provider == nil {
		return errors.New("memory embedding pool and provider are required")
	}
	s.pool = pool
	s.embedder = provider
	return nil
}

func (s *Service) ConfigureHardwareMemoryTool(signer *memorytool.TokenSigner, endpoint string, ttl time.Duration) error {
	if signer == nil || strings.TrimSpace(endpoint) == "" || ttl <= 0 {
		return errors.New("hardware memory tool signer, endpoint and TTL are required")
	}
	s.hardwareMemoryTool = signer
	s.hardwareMemoryToolURL = strings.TrimSpace(endpoint)
	s.hardwareMemoryToolTTL = ttl
	return nil
}

func NewService(
	executor *idempotency.Executor,
	repository *Repository,
	auditRepository *audit.Repository,
	outboxStore *outbox.Store,
	tokens rtc.TokenIssuer,
	leases LeaseCache,
	options ServiceOptions,
) (*Service, error) {
	if executor == nil || repository == nil || auditRepository == nil || outboxStore == nil || tokens == nil || leases == nil {
		return nil, errors.New("orchestration dependencies are required")
	}
	if options.RTCAppID == "" || options.EngineMode == "" || options.AllocationTTL <= 0 || options.TokenTTL <= 0 || options.HeartbeatInterval <= 0 {
		return nil, errors.New("valid orchestration options are required")
	}
	if options.VoiceProvider == "" {
		options.VoiceProvider = "mock"
	}
	if options.HardwareTransport == "" {
		options.HardwareTransport = "high_quality"
	}
	if options.HardwareTransport != "high_quality" && options.HardwareTransport != "low_load" {
		return nil, errors.New("hardware agent transport must be high_quality or low_load")
	}
	return &Service{executor: executor, repository: repository, audit: auditRepository, outbox: outboxStore,
		tokens: tokens, leases: leases, options: options, now: time.Now}, nil
}

func (s *Service) Create(
	ctx context.Context,
	identity deviceauth.Identity,
	idempotencyKey, requestID string,
	rawRequest []byte,
	input CreateInput,
) (CreateResponse, bool, error) {
	tenantID := identity.TenantID
	request := idempotency.Request{
		TenantID: &tenantID, ActorKey: "device:" + identity.PublicID,
		Scope: "POST /device/v2/conversations", Key: idempotencyKey,
		RequestHash: idempotency.Hash(rawRequest), ExpiresAfter: 24 * time.Hour,
	}
	if cached, found, lookupErr := s.executor.Lookup(ctx, request); lookupErr != nil {
		return CreateResponse{}, false, lookupErr
	} else if found {
		return s.finalizeCreateResult(cached)
	}
	memoryVector, memoryPolicy, err := s.prepareMemoryQuery(ctx, identity, input.RequestedGoal)
	if err != nil {
		return CreateResponse{}, false, err
	}
	result, err := s.executor.Execute(ctx, request, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		now := s.now().UTC()
		response, internalID, createErr := s.repository.Create(operationContext, tx, identity, input, AllocationOptions{
			Now: now, AllocationExpires: now.Add(s.options.AllocationTTL), TokenExpires: now.Add(s.options.TokenTTL),
			RTCAppID: s.options.RTCAppID, VoiceProvider: s.options.VoiceProvider, EngineMode: s.options.EngineMode, HeartbeatInterval: s.options.HeartbeatInterval,
			MemoryQueryVector: memoryVector, MemoryPolicy: memoryPolicy, RequirePublishedRouting: s.options.RequirePublishedRouting,
		})
		if createErr != nil {
			return idempotency.Result{}, createErr
		}
		body, marshalErr := json.Marshal(response)
		if marshalErr != nil {
			return idempotency.Result{}, fmt.Errorf("encode create conversation result: %w", marshalErr)
		}
		actorPublicID := identity.PublicID
		auditResult := "success"
		if response.Status == conversation.StatusFailed {
			auditResult = "failure"
		}
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{
			TenantID: &tenantID, ActorType: "device", ActorPublicID: &actorPublicID,
			Action: "conversation.create", ResourceType: "conversation", ResourceID: &internalID,
			RequestID: requestID, Purpose: input.RequestedGoal, Result: auditResult,
			Detail: map[string]any{"status": response.Status, "failure_code": response.FailureCode},
		}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		statusCode := http.StatusCreated
		if response.Status == conversation.StatusFailed {
			statusCode = http.StatusUnprocessableEntity
		} else {
			deduplicationKey := response.ConversationID
			if _, outboxErr := s.outbox.Enqueue(operationContext, tx, outbox.NewEvent{
				TenantID: &tenantID, AggregateType: "conversation", AggregateID: internalID,
				EventType: "conversation.room_issued", DeduplicationKey: &deduplicationKey,
				Payload: json.RawMessage(fmt.Sprintf(`{"conversation_id":%q,"status":"ROOM_ISSUED"}`, response.ConversationID)),
			}); outboxErr != nil {
				return idempotency.Result{}, outboxErr
			}
		}
		return idempotency.Result{StatusCode: statusCode, Body: body, ResourceID: &internalID}, nil
	})
	if err != nil {
		return CreateResponse{}, false, err
	}
	return s.finalizeCreateResult(result)
}

func (s *Service) finalizeCreateResult(result idempotency.Result) (CreateResponse, bool, error) {
	var response CreateResponse
	if err := json.Unmarshal(result.Body, &response); err != nil {
		return CreateResponse{}, result.Replayed, fmt.Errorf("decode stored create conversation result: %w", err)
	}
	if response.Status == conversation.StatusFailed {
		return CreateResponse{}, result.Replayed, &CreateFailureError{
			ConversationID: response.ConversationID,
			FailureCode:    response.FailureCode,
		}
	}
	token, err := s.tokens.Issue(response.RTC.RoomID, response.RTC.UserID, response.RTC.ExpiresAt)
	if err != nil {
		return CreateResponse{}, result.Replayed, fmt.Errorf("issue RTC token: %w", err)
	}
	response.RTC.Token = token
	return response, result.Replayed, nil
}

func (s *Service) prepareMemoryQuery(ctx context.Context, identity deviceauth.Identity, requestedGoal string) (string, string, error) {
	if s.pool == nil || s.embedder == nil {
		return "", "", nil
	}
	var subjectName string
	err := s.pool.QueryRow(ctx, `
		SELECT sp.preferred_name
		FROM device d
		JOIN device_binding b ON b.device_id=d.id AND b.tenant_id=d.tenant_id AND b.unbound_at IS NULL
		JOIN subject_profile sp ON sp.id=b.subject_id AND sp.tenant_id=b.tenant_id AND sp.status='active'
		WHERE d.id=$1::uuid AND d.tenant_id=$2::uuid AND d.status='registered'
			AND EXISTS(SELECT 1 FROM consent_snapshot cs WHERE cs.tenant_id=d.tenant_id AND cs.subject_id=sp.id
				AND cs.effective_consents @> '{"voice_chat":true}'::jsonb)
			AND EXISTS(SELECT 1 FROM consent_record cr WHERE cr.tenant_id=d.tenant_id AND cr.subject_id=sp.id
				AND cr.consent_type='personalized_memory' AND cr.withdrawn_at IS NULL
				AND (cr.expires_at IS NULL OR cr.expires_at>now()))`, identity.ID, identity.TenantID).Scan(&subjectName)
	if errors.Is(err, pgx.ErrNoRows) {
		return "", "pgvector-not-eligible-v1", nil
	}
	if err != nil {
		return "", "", fmt.Errorf("prepare memory retrieval query: %w", err)
	}
	result, err := s.embedder.Embed(ctx, strings.TrimSpace(subjectName+" "+requestedGoal))
	if err != nil {
		if outbox.IsRetryable(err) {
			return "", "pgvector-degraded-v1", nil
		}
		return "", "", fmt.Errorf("generate memory retrieval embedding: %w", err)
	}
	literal, err := embeddingprovider.Literal(result.Vector)
	if err != nil {
		return "", "", fmt.Errorf("encode memory retrieval embedding: %w", err)
	}
	return literal, "pgvector-" + result.ProviderRef + "-normal-v1", nil
}

func (s *Service) Activate(
	ctx context.Context,
	identity deviceauth.Identity,
	conversationID, idempotencyKey, requestID string,
	rawRequest []byte,
	input ActivateInput,
) (ActivateResponse, bool, error) {
	tenantID := identity.TenantID
	scope := "POST /device/v2/conversations/" + conversationID + "/activate"
	result, err := s.executor.Execute(ctx, idempotency.Request{
		TenantID: &tenantID, ActorKey: "device:" + identity.PublicID, Scope: scope,
		Key: idempotencyKey, RequestHash: idempotency.Hash(rawRequest), ExpiresAfter: 24 * time.Hour,
	}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		activation, activateErr := s.repository.Activate(
			operationContext, tx, identity, conversationID, requestID, input, s.now().UTC(),
		)
		if activateErr != nil {
			return idempotency.Result{}, activateErr
		}
		if activation.StartRequested {
			deduplicationKey := activation.ProviderTaskRef
			payload, _ := json.Marshal(map[string]string{
				"conversation_id": conversationID, "provider_task_ref": activation.ProviderTaskRef,
			})
			if _, enqueueErr := s.outbox.Enqueue(operationContext, tx, outbox.NewEvent{
				TenantID: &tenantID, AggregateType: "conversation", AggregateID: activation.InternalID,
				EventType: "conversation.agent_start_requested", DeduplicationKey: &deduplicationKey,
				Payload: payload,
			}); enqueueErr != nil {
				return idempotency.Result{}, enqueueErr
			}
		}
		actorPublicID := identity.PublicID
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{
			TenantID: &tenantID, ActorType: "device", ActorPublicID: &actorPublicID,
			Action: "conversation.activate", ResourceType: "conversation", ResourceID: &activation.InternalID,
			RequestID: requestID, Purpose: "confirm_rtc_join", Result: "success",
			Detail: map[string]any{"status": activation.Response.Status, "start_requested": activation.StartRequested},
		}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		body, marshalErr := json.Marshal(activation.Response)
		if marshalErr != nil {
			return idempotency.Result{}, marshalErr
		}
		return idempotency.Result{StatusCode: http.StatusAccepted, Body: body, ResourceID: &activation.InternalID}, nil
	})
	if err != nil {
		return ActivateResponse{}, false, err
	}
	var response ActivateResponse
	if err := json.Unmarshal(result.Body, &response); err != nil {
		return ActivateResponse{}, result.Replayed, fmt.Errorf("decode stored activation result: %w", err)
	}
	return response, result.Replayed, nil
}

func (s *Service) GetDeviceState(ctx context.Context, identity deviceauth.Identity, conversationID string) (DeviceConversationState, error) {
	if s.pool == nil {
		return DeviceConversationState{}, errors.New("conversation state database is not configured")
	}
	var state DeviceConversationState
	err := s.pool.QueryRow(ctx, `
		SELECT c.public_id::text,c.status,COALESCE(c.failure_code,'')
		FROM conversation c
		WHERE c.public_id=$1::uuid AND c.device_id=$2::uuid AND c.tenant_id=$3::uuid`,
		conversationID, identity.ID, identity.TenantID,
	).Scan(&state.ConversationID, &state.Status, &state.FailureCode)
	if errors.Is(err, pgx.ErrNoRows) {
		return DeviceConversationState{}, ErrConversationNotFound
	}
	if err != nil {
		return DeviceConversationState{}, fmt.Errorf("read device conversation state: %w", err)
	}
	switch state.Status {
	case conversation.StatusActive:
		state.Action = "continue"
	case conversation.StatusStopping, conversation.StatusCompleted, conversation.StatusFailed, conversation.StatusExpired:
		state.Action = "leave_room"
	default:
		state.Action = "wait"
	}
	return state, nil
}

func (s *Service) Stop(
	ctx context.Context,
	identity deviceauth.Identity,
	conversationID, idempotencyKey, requestID string,
	rawRequest []byte,
	input StopInput,
) (StopResponse, bool, error) {
	tenantID := identity.TenantID
	scope := "POST /device/v2/conversations/" + conversationID + "/stop"
	result, err := s.executor.Execute(ctx, idempotency.Request{
		TenantID: &tenantID, ActorKey: "device:" + identity.PublicID, Scope: scope,
		Key: idempotencyKey, RequestHash: idempotency.Hash(rawRequest), ExpiresAfter: 24 * time.Hour,
	}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		stop, stopErr := s.repository.Stop(operationContext, tx, identity, conversationID, requestID, input, s.now().UTC())
		if stopErr != nil {
			return idempotency.Result{}, stopErr
		}
		if stop.StopRequested {
			deduplicationKey := stop.ProviderTaskRef
			payload, _ := json.Marshal(map[string]string{
				"conversation_id": conversationID, "provider_task_ref": stop.ProviderTaskRef, "reason": input.Reason,
			})
			if _, enqueueErr := s.outbox.Enqueue(operationContext, tx, outbox.NewEvent{
				TenantID: &tenantID, AggregateType: "conversation", AggregateID: stop.InternalID,
				EventType: "conversation.agent_stop_requested", DeduplicationKey: &deduplicationKey,
				Payload: payload,
			}); enqueueErr != nil {
				return idempotency.Result{}, enqueueErr
			}
		}
		actorPublicID := identity.PublicID
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{
			TenantID: &tenantID, ActorType: "device", ActorPublicID: &actorPublicID,
			Action: "conversation.stop", ResourceType: "conversation", ResourceID: &stop.InternalID,
			RequestID: requestID, Purpose: input.Reason, Result: "success",
			Detail: map[string]any{"status": stop.Response.Status, "stop_requested": stop.StopRequested},
		}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		body, marshalErr := json.Marshal(stop.Response)
		if marshalErr != nil {
			return idempotency.Result{}, marshalErr
		}
		return idempotency.Result{StatusCode: http.StatusAccepted, Body: body, ResourceID: &stop.InternalID}, nil
	})
	if err != nil {
		return StopResponse{}, false, err
	}
	var response StopResponse
	if err := json.Unmarshal(result.Body, &response); err != nil {
		return StopResponse{}, result.Replayed, fmt.Errorf("decode stored stop result: %w", err)
	}
	_ = s.leases.End(ctx, conversationID, identity.PublicID, s.now().UTC())
	return response, result.Replayed, nil
}