package callmemory

// Direct chat keeps audio and subtitles off the device/platform data path:
// the device obtains SDK parameters; Volcano sends signed subtitles to us.
import (
	"context"
	"crypto/subtle"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"data-terminal-cloud/internal/billing"
	"data-terminal-cloud/internal/deviceactivation"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/volcsubtitle"
	"github.com/jackc/pgx/v5"
)

type DirectOptions struct {
	Prepare            func(context.Context, string, string, string, string) error
	PrepareWithWelcome func(context.Context, string, string, string, string, string) error
	Template           []byte
	BotID, CallbackURL string
	CallbackSecret     []byte
}
type DirectInput struct {
	DeviceID   string `json:"device_id,omitempty"`
	SessionID  string `json:"session_id"`
	BootID     string `json:"boot_id,omitempty"` // Accepted for old firmware only; no startup registration required.
	Reason     string `json:"reason,omitempty"`
	SDKStopped bool   `json:"sdk_stopped,omitempty"`
	RoomID     string `json:"room_id,omitempty"`
	RTCAppID   string `json:"rtc_app_id,omitempty"`
	RTCUserID  string `json:"rtc_user_id,omitempty"`
}

func (s *Service) ConfigureDirect(o DirectOptions) { s.direct = o }

func (s *Service) RegisterDirectRoutes(mux *http.ServeMux, auth DeviceAuth) {
	for _, op := range []string{"start", "end", "recording"} {
		mux.HandleFunc("POST /api/v1/device/chat/"+op, func(w http.ResponseWriter, r *http.Request) {
			d, _, raw, e := auth(r)
			if e != nil {
				s.write(w, r, nil, fault(401, "DEVICE_AUTHENTICATION_FAILED"))
				return
			}
			var in DirectInput
			if len(raw) > 2048 || StrictJSON(raw, &in) != nil {
				s.write(w, r, nil, fault(400, "INVALID_CHAT_REQUEST"))
				return
			}
			out, e := s.DirectRequest(r.Context(), d, op, in)
			s.write(w, r, out, e)
		})
	}
}

func (s *Service) DirectRequest(ctx context.Context, d deviceauth.Identity, op string, in DirectInput) (any, error) {
	if (op != "recording" && (in.RoomID != "" || in.RTCAppID != "" || in.RTCUserID != "")) || len(in.RoomID) > 128 || len(in.RTCAppID) > 128 || len(in.RTCUserID) > 128 {
		return nil, fault(400, "INVALID_CHAT_REQUEST")
	}
	// Legacy firmware already supplies its factory ID in the authenticated header.
	// New firmware also includes it explicitly in the minimal chat body.
	if in.DeviceID != "" && in.DeviceID != strings.TrimPrefix(d.CredentialRef, "onboarding://") {
		return nil, fault(400, "DEVICE_ID_MISMATCH")
	}
	if !validUUID(in.SessionID) || (op == "start" && ((in.BootID != "" && !validUUID(in.BootID)) || in.Reason != "" || in.SDKStopped)) || (op == "end" && !oneOf(in.Reason, "hangup", "start_failed", "disconnected", "reboot", "timeout")) {
		return nil, fault(400, "INVALID_CHAT_REQUEST")
	}
	ctx, cancel := context.WithTimeout(ctx, 12*time.Second)
	defer cancel()
	tx, e := s.pool.Begin(ctx)
	if e != nil {
		return nil, e
	}
	defer tx.Rollback(context.Background())
	if _, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,11))`, d.ID); e != nil {
		return nil, e
	}
	if op == "recording" {
		out, err := s.prepareMixedRecording(ctx, tx, d, in)
		if err != nil {
			return nil, err
		}
		return out, tx.Commit(ctx)
	}
	if op == "end" {
		// End is durable even when it wins a race against start. No media claim is
		// inferred from a lost HTTP response or a local SDK return code.
		_, e = tx.Exec(ctx, `INSERT INTO call_request_result(actor,request_id,path,body_digest,response) VALUES($1,$2::uuid,'direct-end',$3,'{}') ON CONFLICT DO NOTHING`, "device:"+d.ID, in.SessionID, hash(jsonBytes(in)))
		if e != nil {
			return nil, e
		}
		_, e = tx.Exec(ctx, `UPDATE call_session SET call_state=CASE WHEN $3='hangup' THEN 'ENDED' ELSE 'FAILED' END,manifest_status='DISCARDED',manifest_b64=NULL,local_audio_off=true,sdk_stopped=$4,ended_at=COALESCE(ended_at,now()),drain_started_at=COALESCE(drain_started_at,now()),transcript_state=CASE WHEN transcript_state='COLLECTING' THEN 'DRAINING' ELSE transcript_state END WHERE id=$1::uuid AND device_id=$2::uuid AND budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND ended_at IS NULL`, in.SessionID, d.ID, in.Reason, in.SDKStopped)
		if e != nil {
			return nil, e
		}
		if e = s.activateDirect(ctx, tx, in.SessionID); e != nil {
			return nil, e
		}
		s.stopMixedRecording(ctx, tx, in.SessionID, d.ID)
		return map[string]any{"session_id": in.SessionID, "ended": true}, tx.Commit(ctx)
	}
	if op != "start" {
		return nil, fault(404, "NOT_FOUND")
	}
	var cancelled bool
	if e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_request_result WHERE actor=$1 AND request_id=$2::uuid AND path='direct-end')`, "device:"+d.ID, in.SessionID).Scan(&cancelled); e != nil {
		return nil, e
	}
	if cancelled {
		return nil, fault(409, "CHAT_ENDED")
	}
	var subject, tenant, name, bot string
	var generation, epoch int64
	var bound bool
	e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding b JOIN device d ON d.id=b.device_id AND d.binding_generation=b.generation JOIN subject_profile sp ON sp.id=b.subject_id AND sp.status='active' WHERE d.id=$1::uuid AND b.unbound_at IS NULL)`, d.ID).Scan(&bound)
	if e != nil {
		return nil, e
	}
	if !bound {
		return nil, &Error{409, "DEVICE_UNBOUND", "绑定错误：设备没有绑定人员", "REBIND"}
	}
	// Registration belongs to the platform. Local SDK initialization is the
	// device's responsibility, not a second remote activation handshake.
	e = tx.QueryRow(ctx, `SELECT b.subject_id::text,b.tenant_id::text,d.binding_generation,j.provider_device_name,COALESCE(a.bot_id,'') FROM device d JOIN device_binding b ON b.device_id=d.id AND b.generation=d.binding_generation AND b.unbound_at IS NULL JOIN subject_profile sp ON sp.id=b.subject_id AND sp.status='active' JOIN household h ON h.id=b.household_id AND h.status='active' JOIN device_activation_job j ON j.device_id=d.id AND j.binding_generation=d.binding_generation LEFT JOIN subject_voice_agent a ON a.subject_id=sp.id AND a.status='READY' WHERE d.id=$1::uuid AND d.status='registered' AND d.activation_state NOT IN ('REVOKED','LOST') AND j.sdk_registered_at IS NOT NULL AND j.status IN ('REGISTERED_OBSERVED','READY')`, d.ID).Scan(&subject, &tenant, &generation, &name, &bot)
	if errors.Is(e, pgx.ErrNoRows) {
		return nil, fault(409, "DEVICE_NOT_READY")
	}
	if e != nil {
		return nil, e
	}
	if generation != d.BindingGeneration {
		return nil, fault(409, "DEVICE_BINDING_CHANGED")
	}
	if bot == "" {
		return nil, &Error{409, "VOICE_AGENT_NOT_READY", "该老人的智能体尚未创建成功，请重试创建", "NONE"}
	}
	if _, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,64))`, subject); e != nil {
		return nil, e
	}
	var allowed, memory, storyExtraction bool
	e = tx.QueryRow(ctx, `SELECT COALESCE((SELECT effective_consents @> '{"voice_chat":true}'::jsonb FROM consent_snapshot WHERE subject_id=$1::uuid ORDER BY snapshot_version DESC LIMIT 1),false) AND NOT EXISTS(SELECT 1 FROM privacy_request WHERE subject_id=$1::uuid AND request_type='DELETE'),call_memory_authorized($1::uuid)`, subject).Scan(&allowed, &memory)
	if e != nil {
		return nil, e
	}
	if !allowed {
		return nil, fault(403, "VOICE_CHAT_CONSENT_REQUIRED")
	}
	_, e = tx.Exec(ctx, `INSERT INTO memory_user_state(subject_id,tenant_id) VALUES($1::uuid,$2::uuid) ON CONFLICT DO NOTHING`, subject, tenant)
	if e != nil {
		return nil, e
	}
	if e = tx.QueryRow(ctx, `SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid FOR UPDATE`, subject).Scan(&epoch); e != nil {
		return nil, e
	}
	// Read again after acquiring the privacy lock, rather than retaining a
	// consent result obtained before a concurrent withdrawal committed.
	if e = tx.QueryRow(ctx, `SELECT NOT EXISTS(SELECT 1 FROM (VALUES ('transcript_storage'),('story_extraction')) required(kind)
 WHERE NOT EXISTS(SELECT 1 FROM consent_record cr WHERE cr.subject_id=$1::uuid AND cr.consent_type=required.kind AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now())))`, subject).Scan(&storyExtraction); e != nil {
		return nil, e
	}
	var frozen []byte
	var state, owner, protocol string
	var oldGeneration, oldEpoch int64
	var expiry, lease time.Time
	e = tx.QueryRow(ctx, `SELECT device_id::text,binding_generation,privacy_epoch,call_state,COALESCE(manifest_b64,''),expires_at,lease_expires_at,COALESCE(budget_snapshot->>'protocol','') FROM call_session WHERE id=$1::uuid FOR UPDATE`, in.SessionID).Scan(&owner, &oldGeneration, &oldEpoch, &state, &frozen, &expiry, &lease, &protocol)
	if e == nil {
		if owner != d.ID || protocol != "DIRECT-CHAT-1" {
			return nil, fault(404, "CHAT_NOT_FOUND")
		}
		if oldGeneration != generation || oldEpoch != epoch || len(frozen) == 0 || !oneOf(state, "STARTING", "TALKING") || !expiry.After(s.now()) || !lease.After(s.now()) {
			return nil, fault(409, "CHAT_ENDED")
		}
		var wasMemory bool
		if e = tx.QueryRow(ctx, `SELECT memory_mode<>'OFF' FROM call_session WHERE id=$1::uuid`, in.SessionID).Scan(&wasMemory); e != nil {
			return nil, e
		}
		if wasMemory && !memory {
			return nil, fault(403, "MEMORY_CONSENT_REVOKED")
		}
	} else if errors.Is(e, pgx.ErrNoRows) {
		var busy bool
		e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_session WHERE (device_id=$1::uuid OR subject_id=$2::uuid) AND manifest_status IN ('ISSUED','REVOKED') AND NOT(local_audio_off AND sdk_stopped AND remote_state IN ('NOT_STARTED','STOPPED_OBSERVED'))) OR EXISTS(SELECT 1 FROM conversation WHERE device_id=$1::uuid AND status IN ('ROOM_ISSUED','ACTIVE','STOPPING'))`, d.ID, subject).Scan(&busy)
		if e != nil {
			return nil, e
		}
		if busy {
			return nil, fault(409, "CHAT_ALREADY_ACTIVE")
		}
		budget, err := billing.CheckQuotaTx(ctx, tx, d, s.now())
		if errors.Is(err, billing.ErrQuotaExceeded) {
			return nil, fault(429, "CALL_QUOTA_EXCEEDED")
		}
		if err != nil {
			return nil, err
		}
		max := 3600
		if budget != nil && budget.MaxCallDurationSec < max {
			max = budget.MaxCallDurationSec
		}
		if max < 1 {
			return nil, fault(429, "CALL_QUOTA_EXCEEDED")
		}
		expiry = s.now().Add(time.Duration(max) * time.Second)
		mode := "OFF"
		contextText := ""
		story := storyContext{Version: storyContextVersion}
		if memory && (s.direct.Prepare != nil || s.direct.PrepareWithWelcome != nil) {
			mode = "NATIVE_VIKING"
		}
		welcome, opening, err := s.selectDirectOpening(ctx, tx, subject, bot, in.SessionID, d.ID, generation, epoch, memory)
		if err != nil {
			return nil, err
		}
		if opening != nil && s.direct.PrepareWithWelcome != nil {
			e = s.direct.PrepareWithWelcome(ctx, subject, bot, in.SessionID, contextText, welcome)
			if e != nil {
				return nil, &Error{503, "VOICE_AGENT_CONFIG_FAILED", "火山智能体配置暂时失败，请重试", "NONE"}
			}
		} else if s.direct.Prepare != nil {
			if e = s.direct.Prepare(ctx, subject, bot, in.SessionID, contextText); e != nil {
				return nil, &Error{503, "VOICE_AGENT_CONFIG_FAILED", "火山智能体配置暂时失败，请重试", "NONE"}
			}
		} else if s.direct.PrepareWithWelcome != nil {
			if e = s.direct.PrepareWithWelcome(ctx, subject, bot, in.SessionID, contextText, welcome); e != nil {
				return nil, &Error{503, "VOICE_AGENT_CONFIG_FAILED", "火山智能体配置暂时失败，请重试", "NONE"}
			}
		}
		frozen, e = s.directParameters(in.SessionID, bot, welcome)
		if e != nil {
			return nil, e
		}
		lease = s.now().Add(45 * time.Second)
		if lease.After(expiry) {
			lease = expiry
		}
		// The shared legacy table requires a boot_id. Use the session's own UUID
		// as an internal namespace without inventing a device startup event.
		snapshot := map[string]any{"protocol": "DIRECT-CHAT-1", "device_name": name, "bot_id": bot, "quota": budget, "story_extraction_allowed": storyExtraction}
		if memory {
			snapshot["story_context"] = story
		}
		if opening != nil {
			snapshot["opening"] = opening
		}
		_, e = tx.Exec(ctx, `INSERT INTO call_session(id,device_id,tenant_id,subject_id,boot_id,hook_event_id,call_generation,binding_generation,privacy_epoch,memory_mode,manifest_status,manifest_b64,expires_at,lease_expires_at,call_state,budget_snapshot,text_gap) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$1::uuid,$1,0,$5,$6,$7,'ISSUED',$8,$9,$10,'STARTING',$11,true)`, in.SessionID, d.ID, tenant, subject, generation, epoch, mode, string(frozen), expiry, lease, jsonBytes(snapshot))
		if e != nil {
			return nil, e
		}
	} else {
		return nil, e
	}
	lease = s.now().Add(45 * time.Second)
	if lease.After(expiry) {
		lease = expiry
	}
	if s.mixedRecorder != nil {
		var recordingRevoked bool
		e = tx.QueryRow(ctx, `SELECT COALESCE(budget_snapshot->>'mixed_recording','false')='true' AND NOT EXISTS(SELECT 1 FROM consent_record cr WHERE cr.subject_id=call_session.subject_id AND cr.tenant_id=call_session.tenant_id AND cr.consent_type='raw_audio_storage' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now())) FROM call_session WHERE id=$1::uuid`, in.SessionID).Scan(&recordingRevoked)
		if e != nil {
			return nil, e
		}
		if recordingRevoked {
			s.stopMixedRecording(ctx, tx, in.SessionID, d.ID)
		}
	}
	if _, e = tx.Exec(ctx, `UPDATE call_session SET lease_expires_at=$2,authorization_renewed_at=now() WHERE id=$1::uuid`, in.SessionID, lease); e != nil {
		return nil, e
	}
	var out map[string]any
	if json.Unmarshal(frozen, &out) != nil {
		return nil, fault(503, "CHAT_CONFIG_INVALID")
	}
	out["session_id"] = in.SessionID
	out["lease_expires_at_ms"] = lease.UnixMilli()
	out["expires_at_ms"] = expiry.UnixMilli()
	return out, tx.Commit(ctx)
}

func (s *Service) directParameters(sid, bot, welcome string) ([]byte, error) {
	u, e := url.Parse(s.direct.CallbackURL)
	if e != nil || u.Scheme != "https" || u.Host == "" || len(s.direct.CallbackSecret) < 32 || bot == "" {
		return nil, fault(503, "CHAT_CONFIG_REQUIRED")
	}
	// S2S credentials and model settings are configured on the elder's bot once.
	// Freeze the elder's welcome alongside the per-call callback address.
	u.Path = "/callbacks/v1/volcengine/chat"
	q := url.Values{}
	q.Set("session_id", sid)
	u.RawQuery = q.Encode()
	if strings.TrimSpace(welcome) == "" {
		welcome = "你好，今天想聊些什么？"
	}
	agent := map[string]any{"WelcomeMessage": welcome, "EnableConversationStateCallback": false, "ServerMessageURLForRTS": u.String(), "ServerMessageSignatureForRTS": volcsubtitle.Signature(s.direct.CallbackSecret, "direct:"+sid)}
	params := map[string]any{"AgentConfig": agent, "PlatformTaskID": sid}
	raw := jsonBytes(params)
	if len(raw) > 16384 {
		return nil, fault(503, "CHAT_CONFIG_TOO_LARGE")
	}
	return jsonBytes(map[string]any{"bot_id": bot, "params": params}), nil
}

func (s *Service) DirectCallback(w http.ResponseWriter, r *http.Request) {
	sid := r.URL.Query().Get("session_id")
	var envelope volcsubtitle.Envelope
	raw, e := io.ReadAll(http.MaxBytesReader(w, r.Body, 384*1024))
	if !validUUID(sid) || e != nil || json.Unmarshal(raw, &envelope) != nil {
		s.write(w, r, nil, fault(400, "INVALID_SUBTITLE"))
		return
	}
	expected := volcsubtitle.Signature(s.direct.CallbackSecret, "direct:"+sid)
	if len(s.direct.CallbackSecret) < 32 || subtle.ConstantTimeCompare([]byte(envelope.Signature), []byte(expected)) != 1 {
		s.write(w, r, nil, fault(401, "SUBTITLE_AUTHENTICATION_FAILED"))
		return
	}
	packet, e := volcsubtitle.Decode(envelope.Message)
	if e != nil {
		s.write(w, r, nil, fault(400, "INVALID_SUBTITLE"))
		return
	}
	e = s.DirectSubtitle(r.Context(), sid, packet)
	if e != nil {
		s.write(w, r, nil, e)
		return
	}
	w.Header().Set("Content-Type", "text/plain")
	_, _ = w.Write([]byte("ok"))
}

func (s *Service) DirectSubtitle(ctx context.Context, sid string, p volcsubtitle.Packet) error {
	tx, e := s.pool.Begin(ctx)
	if e != nil {
		return e
	}
	defer tx.Rollback(context.Background())
	var subject, name, bot string
	var epoch, current int64
	var expires time.Time
	var storyRaw []byte
	var created time.Time
	var storyAllowed bool
	e = tx.QueryRow(ctx, `SELECT c.subject_id::text,c.budget_snapshot->>'device_name',c.budget_snapshot->>'bot_id',c.privacy_epoch,us.privacy_epoch,c.expires_at,COALESCE(c.budget_snapshot->'story_context','{}'::jsonb),c.created_at,c.memory_mode<>'OFF' AND call_memory_authorized(c.subject_id) FROM call_session c JOIN memory_user_state us ON us.subject_id=c.subject_id WHERE c.id=$1::uuid AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1' FOR UPDATE OF c`, sid).Scan(&subject, &name, &bot, &epoch, &current, &expires, &storyRaw, &created, &storyAllowed)
	if errors.Is(e, pgx.ErrNoRows) {
		return fault(404, "CHAT_NOT_FOUND")
	}
	if e != nil {
		return e
	}
	allowed, e := transcriptPermission(ctx, tx, subject)
	if e != nil {
		return e
	}
	if !allowed || epoch != current {
		return fault(403, "TRANSCRIPT_STORAGE_REVOKED")
	}
	if s.now().After(expires.Add(24 * time.Hour)) {
		return fault(410, "SUBTITLE_EXPIRED")
	}
	if p.DeviceName != "" && p.DeviceName != name {
		return fault(409, "SUBTITLE_DEVICE_MISMATCH")
	}
	changed := false
	var story storyContext
	_ = json.Unmarshal(storyRaw, &story)
	storyAllowed = storyAllowed && story.Version == storyContextVersion
	for _, part := range p.Data {
		if !part.Definite || strings.TrimSpace(part.Text) == "" {
			continue
		}
		speaker := "UNKNOWN"
		if part.UserID == "input" || part.UserID == name {
			speaker = "USER"
		} else if part.UserID == "output" || part.UserID == bot || isRTCAgentUser(part.UserID) {
			speaker = "ASSISTANT"
		}
		key := hash(jsonBytes([]any{part.UserID, *part.RoundID, *part.Sequence}))
		if isRTCAgentUser(part.UserID) {
			// Keep the authenticated, session-specific RTC identity for recording lookup.
			// A conflicting UID must never reassign a call to a different room.
			tag, err := tx.Exec(ctx, `UPDATE call_session SET budget_snapshot=jsonb_set(budget_snapshot,'{rtc_agent_user_id}',to_jsonb($2::text)) WHERE id=$1::uuid AND (budget_snapshot->>'rtc_agent_user_id' IS NULL OR budget_snapshot->>'rtc_agent_user_id'=$2)`, sid, part.UserID)
			if err != nil {
				return err
			}
			if tag.RowsAffected() != 1 {
				return fault(409, "RECORDING_IDENTITY_CONFLICT")
			}
		}
		var previous string
		e = tx.QueryRow(ctx, `SELECT text_content FROM call_utterance WHERE session_id=$1::uuid AND utterance_id=$2`, sid, key).Scan(&previous)
		if e == nil {
			if previous != part.Text {
				return fault(409, "SUBTITLE_CONFLICT")
			}
			continue
		}
		if !errors.Is(e, pgx.ErrNoRows) {
			return e
		}
		_, e = tx.Exec(ctx, `INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES($1::uuid,$2,0,$3,$4,false,now())`, sid, key, speaker, part.Text)
		if e != nil {
			return e
		}
		if speaker == "USER" {
			if e = s.applySpokenVolume(ctx, tx, sid, part); e != nil {
				return e
			}
		}
		changed = true
		if storyAllowed {
			story.observe(sid, key, speaker, part.Text, s.now().Sub(created).Milliseconds())
		}
	}
	if changed {
		if storyAllowed {
			if _, e = tx.Exec(ctx, `UPDATE call_session SET budget_snapshot=jsonb_set(budget_snapshot,'{story_context}',$2::jsonb) WHERE id=$1::uuid`, sid, jsonBytes(story)); e != nil {
				return e
			}
		}
		_, e = tx.Exec(ctx, `UPDATE call_session SET last_text_at=now(),transcript_revision=transcript_revision+1,call_state=CASE WHEN call_state='STARTING' AND ended_at IS NULL THEN 'TALKING' ELSE call_state END,transcript_state=CASE WHEN ended_at IS NULL THEN 'COLLECTING' ELSE 'DRAINING' END WHERE id=$1::uuid`, sid)
		if e != nil {
			return e
		}
	}
	if e = s.activateDirect(ctx, tx, sid); e != nil {
		return e
	}
	return tx.Commit(ctx)
}

// The hardware RTC gateway uses bot_<17-digit timestamp> for the room's AI
// participant, not the persistent Aibot ID. Verified against real callbacks.
func isRTCAgentUser(id string) bool {
	return len(id) == 21 && strings.HasPrefix(id, "bot_") && strings.Trim(id[4:], "0123456789") == ""
}

// First successful direct chat replaces the old separate conversation-test
// handshake. Activation still needs an authenticated device stop AND actual
// supplier subtitle evidence; sending a config alone does not activate it.
func (s *Service) activateDirect(ctx context.Context, tx pgx.Tx, sid string) error {
	var device, tenant, job string
	var generation int64
	e := tx.QueryRow(ctx, `SELECT d.id::text,d.tenant_id::text,j.id::text,d.binding_generation FROM call_session c JOIN device d ON d.id=c.device_id JOIN device_activation_job j ON j.device_id=d.id AND j.binding_generation=d.binding_generation JOIN device_binding b ON b.device_id=d.id AND b.generation=d.binding_generation AND b.unbound_at IS NULL WHERE c.id=$1::uuid AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND c.binding_generation=d.binding_generation AND c.call_state='ENDED' AND c.sdk_stopped AND d.activation_state='PROVISIONING' AND j.status='REGISTERED_OBSERVED' AND j.sdk_registered_at IS NOT NULL AND EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id AND u.speaker='ASSISTANT')`, sid).Scan(&device, &tenant, &job, &generation)
	if errors.Is(e, pgx.ErrNoRows) {
		return nil
	}
	if e != nil {
		return e
	}
	if _, e = deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: device, TenantID: tenant, Target: deviceactivation.Active, Reason: "直连聊天收到供应商字幕且设备已结束", Source: "onboarding", ActorType: "system", RequestID: sid}); e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `UPDATE device_activation_job SET status='READY',ready_binding_generation=$2,version=version+1,updated_at=now() WHERE id=$1::uuid`, job, generation)
	return e
}