story/internal/orchestration/native_subtitle.go

package orchestration

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

	"data-terminal-cloud/internal/outbox"
	"data-terminal-cloud/internal/volcsubtitle"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var ErrNativeSubtitleSpeaker = errors.New("native subtitle speaker is not mapped")

type NativeSubtitleResult struct {
	Mode                 string
	Accepted, Duplicates int
}
type NativeSubtitleService struct{ pool *pgxpool.Pool }

func NewNativeSubtitleService(pool *pgxpool.Pool) *NativeSubtitleService {
	return &NativeSubtitleService{pool: pool}
}

// Hardware low_load callbacks currently carry a device identity, but no session
// identity. Validate connectivity without guessing a conversation or storing text.
func (s *NativeSubtitleService) ProbeNativeSubtitleDevice(ctx context.Context, packet volcsubtitle.Packet) error {
	var found bool
	err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM hardware_agent_provider_credential WHERE product_key=$1 AND provider_device_name=$2 AND status='ACTIVE')`, packet.ProductKey, packet.DeviceName).Scan(&found)
	if err != nil {
		return err
	}
	if !found {
		return ErrConversationNotFound
	}
	return nil
}

func (s *NativeSubtitleService) ProcessNativeSubtitle(ctx context.Context, publicID string, packet volcsubtitle.Packet) (NativeSubtitleResult, error) {
	result := NativeSubtitleResult{Mode: "probe"}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return result, err
	}
	defer tx.Rollback(context.Background())
	result, err = s.processNativeSubtitleTx(ctx, tx, publicID, packet)
	if err != nil {
		return NativeSubtitleResult{}, err
	}
	return result, tx.Commit(ctx)
}

func (s *NativeSubtitleService) processNativeSubtitleTx(ctx context.Context, tx pgx.Tx, publicID string, packet volcsubtitle.Packet) (NativeSubtitleResult, error) {
	result := NativeSubtitleResult{Mode: "probe"}
	var id, tenantID, subjectID, retention, status, deviceUID, agentUID, deviceName, botID string
	var primary, allowed bool
	err := tx.QueryRow(ctx, `SELECT c.id::text,c.tenant_id::text,c.subject_id::text,c.retention_mode,c.status,
 COALESCE(hs.capability_snapshot->>'subtitle_device_user_id',''),COALESCE(hs.capability_snapshot->>'subtitle_agent_user_id',''),
 COALESCE(hs.capability_snapshot->>'subtitle_device_user_id',''),hs.bot_id,COALESCE((hs.capability_snapshot->>'native_subtitle_primary')::boolean,false),
 EXISTS(SELECT 1 FROM consent_record cr WHERE cr.tenant_id=c.tenant_id AND cr.subject_id=c.subject_id AND cr.consent_type='transcript_storage' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now()))
 FROM conversation c JOIN hardware_agent_session hs ON hs.conversation_id=c.id AND hs.tenant_id=c.tenant_id
 WHERE c.public_id=$1::uuid FOR UPDATE OF c`, publicID).Scan(&id, &tenantID, &subjectID, &retention, &status, &deviceUID, &agentUID, &deviceName, &botID, &primary, &allowed)
	if errors.Is(err, pgx.ErrNoRows) {
		return result, ErrConversationNotFound
	}
	if err != nil {
		return result, err
	}
	if retention == "none" || !allowed {
		return result, ErrTranscriptStorageNotAllowed
	}
	if status != "ROOM_ISSUED" && status != "ACTIVE" && status != "STOPPING" && status != "COMPLETED" && status != "FAILED" {
		return result, ErrConversationStateConflict
	}
	if !primary {
		return result, nil
	}
	result.Mode = "primary"
	for _, part := range packet.Data {
		// Server callbacks contain independent completed clauses; retaining only
		// paragraph=true would silently drop all preceding clauses of the reply.
		if !part.Definite || strings.TrimSpace(part.Text) == "" {
			continue
		}
		speaker, err := nativeSubtitleSpeaker(part.UserID, deviceUID, deviceName, agentUID, botID)
		if err != nil {
			return result, err
		}
		digest := sha256.Sum256([]byte(part.UserID))
		key := fmt.Sprintf("volc-subv-%s-%d-%d", hex.EncodeToString(digest[:8]), *part.RoundID, *part.Sequence)
		var stored string
		err = tx.QueryRow(ctx, `SELECT final_text FROM conversation_turn WHERE conversation_id=$1::uuid AND turn_key=$2`, id, key).Scan(&stored)
		if err == nil {
			if stored != part.Text {
				return result, ErrHardwareEventConflict
			}
			result.Duplicates++
			continue
		}
		if !errors.Is(err, pgx.ErrNoRows) {
			return result, err
		}
		_, err = 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)
   SELECT $1::uuid,$2::uuid,$3::uuid,$4,COALESCE(max(sequence_no),0)+1,$5,now(),now(),$6,false,'provider' FROM conversation_turn WHERE conversation_id=$3::uuid`, tenantID, subjectID, id, key, speaker, part.Text)
		if err != nil {
			return result, err
		}
		result.Accepted++
	}
	if result.Accepted > 0 && (status == "COMPLETED" || status == "FAILED") {
		if err = enqueueHardwareContentExtraction(ctx, tx, outbox.NewStore(s.pool), tenantID, id); err != nil {
			return result, err
		}
	}
	return result, nil
}

func nativeSubtitleSpeaker(uid, deviceUID, deviceName, agentUID, botID string) (string, error) {
	subject := uid != "" && (uid == deviceUID || uid == deviceName)
	agent := uid != "" && (uid == agentUID || uid == botID)
	if subject == agent {
		return "", ErrNativeSubtitleSpeaker
	}
	if subject {
		return "subject", nil
	}
	return "agent", nil
}