story/internal/orchestration/transcript_batch.go

package orchestration

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

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/idempotency"
	"data-terminal-cloud/internal/outbox"

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

var ErrTranscriptStorageNotAllowed = errors.New("transcript storage is not allowed")

type TranscriptTurnInput struct {
	TurnKey     string    `json:"turn_key"`
	SequenceNo  int64     `json:"sequence_no"`
	Speaker     string    `json:"speaker"`
	StartAt     time.Time `json:"start_at"`
	EndAt       time.Time `json:"end_at"`
	FinalText   string    `json:"final_text"`
	Interrupted bool      `json:"interrupted"`
}

type TranscriptBatchInput struct {
	Turns []TranscriptTurnInput `json:"turns"`
}

type TranscriptConflict struct {
	TurnKey    string `json:"turn_key"`
	SequenceNo int64  `json:"sequence_no"`
	Reason     string `json:"reason"`
}

type TranscriptConflictError struct {
	Conflicts []TranscriptConflict
}

func (e *TranscriptConflictError) Error() string {
	return "device transcript conflicts with stored final turns"
}

type TranscriptBatchResponse struct {
	ConversationID string `json:"conversation_id"`
	AcceptedCount  int    `json:"accepted_count"`
	DuplicateCount int    `json:"duplicate_count"`
}

type TranscriptBatchService struct {
	pool     *pgxpool.Pool
	executor *idempotency.Executor
	audit    *audit.Repository
	outbox   *outbox.Store
}

func NewTranscriptBatchService(pool *pgxpool.Pool, executor *idempotency.Executor, auditRepository *audit.Repository, outboxStore *outbox.Store) (*TranscriptBatchService, error) {
	if pool == nil || executor == nil || auditRepository == nil || outboxStore == nil {
		return nil, errors.New("transcript batch dependencies are required")
	}
	return &TranscriptBatchService{pool: pool, executor: executor, audit: auditRepository, outbox: outboxStore}, nil
}

func (s *TranscriptBatchService) Submit(ctx context.Context, identity deviceauth.Identity, conversationID, idempotencyKey, requestID string, rawRequest []byte, input TranscriptBatchInput) (TranscriptBatchResponse, bool, error) {
	tenantID := identity.TenantID
	scope := "POST /device/v2/conversations/" + conversationID + "/turns:batch"
	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) {
		response, internalID, submitErr := s.submit(operationContext, tx, identity, conversationID, input)
		if submitErr != nil {
			return idempotency.Result{}, submitErr
		}
		body, marshalErr := json.Marshal(response)
		if marshalErr != nil {
			return idempotency.Result{}, marshalErr
		}
		actorID := identity.PublicID
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{
			TenantID: &tenantID, ActorType: "device", ActorPublicID: &actorID,
			Action: "conversation.transcript_batch.submit", ResourceType: "conversation", ResourceID: &internalID,
			RequestID: requestID, Result: "success", Detail: map[string]any{
				"accepted_count": response.AcceptedCount, "duplicate_count": response.DuplicateCount,
			},
		}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		if response.AcceptedCount > 0 {
			var generationVersion int64
			if versionErr := tx.QueryRow(operationContext, `SELECT COALESCE(max(sequence_no),1) FROM conversation_turn WHERE conversation_id=$1::uuid`, internalID).Scan(&generationVersion); versionErr != nil {
				return idempotency.Result{}, versionErr
			}
			deduplicationKey := "content-extraction-v1:" + internalID + ":device:" + idempotency.Hash(rawRequest)
			payload, marshalErr := json.Marshal(map[string]int64{"generation_version": generationVersion})
			if marshalErr != nil {
				return idempotency.Result{}, marshalErr
			}
			if _, enqueueErr := s.outbox.Enqueue(operationContext, tx, outbox.NewEvent{
				TenantID: &tenantID, AggregateType: "conversation", AggregateID: internalID,
				EventType: "conversation.content_extraction_requested", DeduplicationKey: &deduplicationKey,
				Payload: payload,
			}); enqueueErr != nil {
				return idempotency.Result{}, enqueueErr
			}
		}
		return idempotency.Result{StatusCode: http.StatusOK, Body: body, ResourceID: &internalID}, nil
	})
	if err != nil {
		return TranscriptBatchResponse{}, false, err
	}
	var response TranscriptBatchResponse
	if err := json.Unmarshal(result.Body, &response); err != nil {
		return TranscriptBatchResponse{}, result.Replayed, fmt.Errorf("decode transcript batch response: %w", err)
	}
	return response, result.Replayed, nil
}

func (s *TranscriptBatchService) submit(ctx context.Context, tx pgx.Tx, identity deviceauth.Identity, conversationID string, input TranscriptBatchInput) (TranscriptBatchResponse, string, error) {
	var internalID, subjectID, retentionMode, status string
	err := tx.QueryRow(ctx, `
		SELECT c.id::text,c.subject_id::text,c.retention_mode,c.status
		FROM conversation c
		WHERE c.public_id=$1::uuid AND c.tenant_id=$2::uuid AND c.device_id=$3::uuid
		FOR UPDATE`, conversationID, identity.TenantID, identity.ID).Scan(&internalID, &subjectID, &retentionMode, &status)
	if errors.Is(err, pgx.ErrNoRows) {
		return TranscriptBatchResponse{}, "", ErrConversationNotFound
	}
	if err != nil {
		return TranscriptBatchResponse{}, "", fmt.Errorf("read transcript conversation: %w", err)
	}
	if status != "ACTIVE" && status != "STOPPING" && status != "COMPLETED" && status != "FAILED" {
		return TranscriptBatchResponse{}, "", ErrConversationStateConflict
	}
	if retentionMode == "none" {
		return TranscriptBatchResponse{}, "", ErrTranscriptStorageNotAllowed
	}
	var nativePrimary bool
	if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM hardware_agent_session WHERE conversation_id=$1::uuid AND tenant_id=$2::uuid AND capability_snapshot->>'native_subtitle_primary'='true')`, internalID, identity.TenantID).Scan(&nativePrimary); err != nil {
		return TranscriptBatchResponse{}, "", err
	}
	if nativePrimary {
		return TranscriptBatchResponse{}, "", ErrConversationStateConflict
	}
	var hasConsentHistory, hasActiveConsent bool
	if err := tx.QueryRow(ctx, `
		SELECT
			EXISTS(SELECT 1 FROM consent_record WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND consent_type='transcript_storage'),
			EXISTS(SELECT 1 FROM consent_record WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND consent_type='transcript_storage' AND withdrawn_at IS NULL AND (expires_at IS NULL OR expires_at>now()))`,
		identity.TenantID, subjectID).Scan(&hasConsentHistory, &hasActiveConsent); err != nil {
		return TranscriptBatchResponse{}, "", err
	}
	if hasConsentHistory && !hasActiveConsent {
		return TranscriptBatchResponse{}, "", ErrTranscriptStorageNotAllowed
	}

	response := TranscriptBatchResponse{ConversationID: conversationID}
	conflicts := make([]TranscriptConflict, 0)
	for _, turn := range input.Turns {
		rows, queryErr := tx.Query(ctx, `
			SELECT turn_key,sequence_no,speaker,start_at,end_at,final_text,interrupted
			FROM conversation_turn
			WHERE conversation_id=$1::uuid AND (turn_key=$2 OR sequence_no=$3)
			FOR UPDATE`, internalID, turn.TurnKey, turn.SequenceNo)
		if queryErr != nil {
			return TranscriptBatchResponse{}, "", queryErr
		}
		matches := 0
		exact := false
		for rows.Next() {
			matches++
			var stored TranscriptTurnInput
			if scanErr := rows.Scan(&stored.TurnKey, &stored.SequenceNo, &stored.Speaker, &stored.StartAt, &stored.EndAt, &stored.FinalText, &stored.Interrupted); scanErr != nil {
				rows.Close()
				return TranscriptBatchResponse{}, "", scanErr
			}
			if sameTranscriptTurn(stored, turn) {
				exact = true
			}
		}
		rows.Close()
		if rows.Err() != nil {
			return TranscriptBatchResponse{}, "", rows.Err()
		}
		if matches > 0 {
			if matches == 1 && exact {
				response.DuplicateCount++
				continue
			}
			conflicts = append(conflicts, TranscriptConflict{TurnKey: turn.TurnKey, SequenceNo: turn.SequenceNo, Reason: "TURN_KEY_OR_SEQUENCE_ALREADY_USED"})
			continue
		}
		if _, insertErr := tx.Exec(ctx, `
			INSERT INTO conversation_turn(
				tenant_id,subject_id,conversation_id,turn_key,sequence_no,speaker,start_at,end_at,final_text,interrupted,source_type
			) VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,$6,$7,$8,$9,$10,'device')`,
			identity.TenantID, subjectID, internalID, turn.TurnKey, turn.SequenceNo, turn.Speaker,
			turn.StartAt, turn.EndAt, turn.FinalText, turn.Interrupted); insertErr != nil {
			return TranscriptBatchResponse{}, "", insertErr
		}
		response.AcceptedCount++
	}
	if len(conflicts) > 0 {
		return TranscriptBatchResponse{}, "", &TranscriptConflictError{Conflicts: conflicts}
	}
	return response, internalID, nil
}

func sameTranscriptTurn(left, right TranscriptTurnInput) bool {
	return left.TurnKey == right.TurnKey && left.SequenceNo == right.SequenceNo && left.Speaker == right.Speaker &&
		left.StartAt.Equal(right.StartAt) && left.EndAt.Equal(right.EndAt) && left.FinalText == right.FinalText &&
		left.Interrupted == right.Interrupted
}