story/internal/orchestration/repository_integration_test.go

package orchestration

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"testing"
	"time"

	"data-terminal-cloud/internal/database"
	"data-terminal-cloud/internal/deviceauth"

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

func TestRepositoryCreatesOnlyRoomIssuedStateBeforeActivation(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := database.Open(ctx, databaseURL, "orchestration-repository-test")
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	tx, err := pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		t.Fatal(err)
	}
	defer tx.Rollback(context.Background())
	now := time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC)
	identity := deviceauth.Identity{
		ID: "10000000-0000-0000-0000-000000000501", PublicID: "10000000-0000-0000-0000-000000000511",
		TenantID: "10000000-0000-0000-0000-000000000001", Status: "registered",
	}
	resume := true
	if _, err := tx.Exec(ctx, `UPDATE consent_snapshot SET effective_consents='{"voice_chat":true,"transcript_storage":true}'::jsonb
		WHERE tenant_id=$1::uuid AND subject_id=(SELECT subject_id FROM device_binding WHERE device_id=$2::uuid AND unbound_at IS NULL LIMIT 1)`, identity.TenantID, identity.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := tx.Exec(ctx, `UPDATE device SET minimum_firmware_version='9.0.0' WHERE id=$1::uuid`, identity.ID); err != nil {
		t.Fatal(err)
	}
	if _, _, firmwareErr := NewRepository().Create(ctx, tx, identity, CreateInput{RequestedGoal: "open_conversation"}, AllocationOptions{
		Now: now, AllocationExpires: now.Add(30 * time.Second), TokenExpires: now.Add(5 * time.Minute),
		RTCAppID: "test-app", EngineMode: "s2s_hybrid", HeartbeatInterval: 15,
	}); !errors.Is(firmwareErr, ErrFirmwareUpdateRequired) {
		t.Fatalf("minimum firmware error=%v", firmwareErr)
	}
	if _, err := tx.Exec(ctx, `UPDATE device SET minimum_firmware_version='0.1.0-sim' WHERE id=$1::uuid`, identity.ID); err != nil {
		t.Fatal(err)
	}
	if _, _, strictErr := NewRepository().Create(ctx, tx, identity, CreateInput{RequestedGoal: "open_conversation"}, AllocationOptions{
		Now: now, AllocationExpires: now.Add(30 * time.Second), TokenExpires: now.Add(5 * time.Minute),
		RTCAppID: "test-app", EngineMode: "s2s_hybrid", HeartbeatInterval: 15, RequirePublishedRouting: true,
	}); !errors.Is(strictErr, ErrRuntimeConfigUnavailable) {
		t.Fatalf("strict production routing error=%v", strictErr)
	}
	response, internalID, err := NewRepository().Create(ctx, tx, identity, CreateInput{
		DeviceTime: now, Capabilities: Capabilities{RTC: true, AudioTracks: []string{"near", "far"}, UploadResume: &resume},
		Network: Network{Type: "test"}, RequestedGoal: "open_conversation",
	}, AllocationOptions{
		Now: now, AllocationExpires: now.Add(30 * time.Second), TokenExpires: now.Add(5 * time.Minute),
		RTCAppID: "test-app", EngineMode: "s2s_hybrid", HeartbeatInterval: 15,
	})
	if err != nil {
		t.Fatal(err)
	}
	if response.Status != "ROOM_ISSUED" || response.RTC.Token != "" || response.Privacy.RetentionMode != "transcript_only" {
		t.Fatalf("response = %#v", response)
	}
	var status, taskStatus string
	var credentialIssuedAt, credentialExpiresAt time.Time
	var historyCount int
	if err := tx.QueryRow(ctx, "SELECT status FROM conversation WHERE id=$1::uuid", internalID).Scan(&status); err != nil {
		t.Fatal(err)
	}
	if err := tx.QueryRow(ctx, "SELECT status,credential_issued_at,credential_expires_at FROM rtc_task WHERE conversation_id=$1::uuid", internalID).Scan(&taskStatus, &credentialIssuedAt, &credentialExpiresAt); err != nil {
		t.Fatal(err)
	}
	if err := tx.QueryRow(ctx, "SELECT count(*) FROM conversation_state_history WHERE conversation_id=$1::uuid", internalID).Scan(&historyCount); err != nil {
		t.Fatal(err)
	}
	if status != "ROOM_ISSUED" || taskStatus != "ALLOCATED" || historyCount != 2 {
		t.Fatalf("status=%s task=%s history=%d", status, taskStatus, historyCount)
	}
	if !credentialIssuedAt.Equal(now) || !credentialExpiresAt.Equal(now.Add(5*time.Minute)) {
		t.Fatalf("credential metadata issued=%s expires=%s", credentialIssuedAt, credentialExpiresAt)
	}
	var desiredRevision int64
	if err := tx.QueryRow(ctx, `SELECT desired_revision FROM device_twin WHERE device_id=$1::uuid`, identity.ID).Scan(&desiredRevision); err != nil {
		t.Fatal(err)
	}
	activation, err := NewRepository().Activate(ctx, tx, identity, response.ConversationID, "request-activate", ActivateInput{
		JoinedAt: now.Add(time.Second), RoomID: response.RTC.RoomID, RTCUserID: response.RTC.UserID,
		RTCJoinResult: "success", ConfigRevision: desiredRevision,
	}, now.Add(2*time.Second))
	if err != nil || !activation.StartRequested || activation.Response.Status != "STARTING_AGENT" {
		t.Fatalf("activation=%#v err=%v", activation, err)
	}
	duplicate, err := NewRepository().Activate(ctx, tx, identity, response.ConversationID, "request-activate-duplicate", ActivateInput{
		JoinedAt: now.Add(time.Second), RoomID: response.RTC.RoomID, RTCUserID: response.RTC.UserID,
		RTCJoinResult: "success", ConfigRevision: desiredRevision,
	}, now.Add(3*time.Second))
	if err != nil || duplicate.StartRequested || duplicate.Response.Status != "STARTING_AGENT" {
		t.Fatalf("duplicate activation=%#v err=%v", duplicate, err)
	}
	if err := tx.QueryRow(ctx, "SELECT status FROM rtc_task WHERE conversation_id=$1::uuid", internalID).Scan(&taskStatus); err != nil {
		t.Fatal(err)
	}
	if err := tx.QueryRow(ctx, "SELECT count(*) FROM conversation_state_history WHERE conversation_id=$1::uuid", internalID).Scan(&historyCount); err != nil {
		t.Fatal(err)
	}
	if taskStatus != "STARTING" || historyCount != 3 {
		t.Fatalf("activated task=%s history=%d", taskStatus, historyCount)
	}
	if _, err := tx.Exec(ctx, "UPDATE conversation SET status='ACTIVE' WHERE id=$1::uuid", internalID); err != nil {
		t.Fatal(err)
	}
	if _, err := tx.Exec(ctx, "UPDATE rtc_task SET status='ACTIVE' WHERE conversation_id=$1::uuid", internalID); err != nil {
		t.Fatal(err)
	}
	monotonicMS := int64(9000)
	heartbeatInput := HeartbeatInput{
		SequenceNo: 2, DeviceTime: now.Add(4 * time.Second), MonotonicMS: &monotonicMS,
		ConversationStatus: "ACTIVE", RTC: RTCHeartbeat{Joined: true, Uplink: true, Downlink: true},
		Network: Network{Type: "wifi"}, FirmwareVersion: "1.0.0", ReportedConfigRevision: desiredRevision,
		RecordingQueue: RecordingQueue{PendingSegments: 2}, CommandResults: []json.RawMessage{},
	}
	heartbeat, err := NewRepository().Heartbeat(ctx, tx, identity, response.ConversationID, heartbeatInput, now.Add(4*time.Second), time.Minute, 5*time.Minute, time.Minute)
	if err != nil || !heartbeat.Updated || heartbeat.Response.LeaseExpiresAt == nil || heartbeat.Response.Actions[0].Type != "continue" {
		t.Fatalf("heartbeat=%#v err=%v", heartbeat, err)
	}
	originalLease := *heartbeat.Response.LeaseExpiresAt
	heartbeatInput.SequenceNo = 1
	stale, err := NewRepository().Heartbeat(ctx, tx, identity, response.ConversationID, heartbeatInput, now.Add(30*time.Second), time.Minute, 5*time.Minute, time.Minute)
	if err != nil || stale.Updated || stale.Response.LeaseExpiresAt == nil || !stale.Response.LeaseExpiresAt.Equal(originalLease) {
		t.Fatalf("stale heartbeat=%#v err=%v", stale, err)
	}
	heartbeatInput.SequenceNo = 3
	heartbeatInput.ReportedConfigRevision = desiredRevision + 1
	if _, err := NewRepository().Heartbeat(ctx, tx, identity, response.ConversationID, heartbeatInput, now.Add(31*time.Second), time.Minute, 5*time.Minute, time.Minute); !errors.Is(err, ErrConfigVersionConflict) {
		t.Fatalf("future configuration revision error=%v", err)
	}
	var storedSequence int64
	if err := tx.QueryRow(ctx, "SELECT last_heartbeat_seq FROM conversation WHERE id=$1::uuid", internalID).Scan(&storedSequence); err != nil {
		t.Fatal(err)
	}
	if storedSequence != 2 {
		t.Fatalf("last heartbeat sequence=%d, want 2", storedSequence)
	}
	if _, _, err := NewRepository().Create(ctx, tx, identity, CreateInput{RequestedGoal: "open_conversation"}, AllocationOptions{
		Now: now, AllocationExpires: now.Add(time.Minute), TokenExpires: now.Add(time.Minute), RTCAppID: "test-app", EngineMode: "s2s_hybrid", HeartbeatInterval: 15,
	}); !errors.Is(err, ErrConversationAlreadyActive) {
		t.Fatalf("second active conversation error = %v", err)
	}
}