255 lines
11 KiB
Go
255 lines
11 KiB
Go
|
package callmemory
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"data-terminal-cloud/internal/deviceauth"
|
||
|
|
"errors"
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
"strings"
|
||
|
|
"time"
|
||
|
|
"unicode/utf8"
|
||
|
|
)
|
||
|
|
|
||
|
|
type Event struct {
|
||
|
|
ID string `json:"event_id"`
|
||
|
|
Seq int64 `json:"event_seq"`
|
||
|
|
Type string `json:"type"`
|
||
|
|
Monotonic int64 `json:"monotonic_ms"`
|
||
|
|
Occurred int64 `json:"occurred_at_ms"`
|
||
|
|
AttemptID string `json:"media_attempt_id"`
|
||
|
|
Payload map[string]any `json:"payload"`
|
||
|
|
}
|
||
|
|
type EventsInput struct {
|
||
|
|
CallKey
|
||
|
|
Events []Event `json:"events"`
|
||
|
|
}
|
||
|
|
type Transcript struct {
|
||
|
|
Utterance string `json:"utterance_id"`
|
||
|
|
Revision int64 `json:"source_revision"`
|
||
|
|
Speaker string `json:"speaker"`
|
||
|
|
Final bool `json:"source_final"`
|
||
|
|
Text string `json:"text"`
|
||
|
|
Part int `json:"text_part_index,omitempty"`
|
||
|
|
Parts int `json:"text_part_count,omitempty"`
|
||
|
|
Hash string `json:"full_text_sha256,omitempty"`
|
||
|
|
ProviderID *string `json:"provider_event_id,omitempty"`
|
||
|
|
Delivery string `json:"delivery_state,omitempty"`
|
||
|
|
}
|
||
|
|
|
||
|
|
func transcriptPermission(ctx context.Context, tx pgx.Tx, subject string) (bool, error) {
|
||
|
|
var allowed bool
|
||
|
|
e := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM subject_profile sp WHERE sp.id=$1::uuid AND sp.status='active' AND NOT EXISTS(SELECT 1 FROM privacy_request pr WHERE pr.subject_id=sp.id AND pr.request_type='DELETE') AND EXISTS(SELECT 1 FROM consent_record cr WHERE cr.subject_id=sp.id AND cr.consent_type='transcript_storage' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now())))`, subject).Scan(&allowed)
|
||
|
|
return allowed, e
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) events(ctx context.Context, tx pgx.Tx, d deviceauth.Identity, in EventsInput) (any, error) {
|
||
|
|
if in.SessionID == nil || len(in.Events) < 1 || len(in.Events) > 16 {
|
||
|
|
return nil, fault(400, "INVALID_EVENT_BATCH")
|
||
|
|
}
|
||
|
|
c, e := readSession(ctx, tx, d.ID, in.CallKey)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if c.Boot != in.BootID || c.Generation != in.Generation || c.Binding != in.Binding || (in.HookID != "" && c.Hook != in.HookID) {
|
||
|
|
return nil, fault(409, "CALL_IDENTITY_MISMATCH")
|
||
|
|
}
|
||
|
|
canStore, e := transcriptPermission(ctx, tx, c.Subject)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
accepted := []string{}
|
||
|
|
rejected := []map[string]any{}
|
||
|
|
for _, event := range in.Events {
|
||
|
|
nested, beginErr := tx.Begin(ctx)
|
||
|
|
if beginErr != nil {
|
||
|
|
return nil, beginErr
|
||
|
|
}
|
||
|
|
err := s.event(ctx, nested, c, event, canStore)
|
||
|
|
if err != nil {
|
||
|
|
_ = nested.Rollback(ctx)
|
||
|
|
var f *Error
|
||
|
|
if !errors.As(err, &f) {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
rejected = append(rejected, map[string]any{"event_id": event.ID, "error_code": f.Code, "retryable": false})
|
||
|
|
} else {
|
||
|
|
if err = nested.Commit(ctx); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
accepted = append(accepted, event.ID)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
next := c.LastSeq
|
||
|
|
for {
|
||
|
|
var exists bool
|
||
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_event WHERE session_id=$1::uuid AND seq=$2)`, c.ID, next+1).Scan(&exists)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if !exists {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
next++
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET highest_contiguous_seq=GREATEST(highest_contiguous_seq,$2) WHERE id=$1::uuid`, c.ID, next)
|
||
|
|
return map[string]any{"accepted_event_ids": accepted, "rejected_events": rejected, "highest_contiguous_seq": next}, e
|
||
|
|
}
|
||
|
|
func (s *Service) event(ctx context.Context, tx pgx.Tx, c session, v Event, canStore bool) error {
|
||
|
|
if !validUUID(v.ID) || v.Seq < 1 || v.Seq > 2147483647 || v.Monotonic < 0 || v.Occurred < 0 || !validUUID(v.AttemptID) || v.Payload == nil {
|
||
|
|
return fault(400, "INVALID_EVENT")
|
||
|
|
}
|
||
|
|
if !oneOf(v.Type, "START_INTENT", "MEDIA_CONNECTED", "MEDIA_DISCONNECTED", "FIRST_AUDIO_RECEIVED", "FIRST_PLAYBACK_STARTED", "CONVERSATION_STATUS", "TRANSCRIPT_REVISION", "INTERRUPTED", "GAP_DECLARED", "SDK_STOP_RESULT", "CALL_ERROR", "LOCAL_NOTICE_RESULT") {
|
||
|
|
return fault(400, "UNKNOWN_EVENT_TYPE")
|
||
|
|
}
|
||
|
|
digest := hash(jsonBytes(v))
|
||
|
|
var oldID, oldHash string
|
||
|
|
e := tx.QueryRow(ctx, `SELECT event_id,digest FROM call_event WHERE session_id=$1::uuid AND (event_id=$2 OR seq=$3) LIMIT 1`, c.ID, v.ID, v.Seq).Scan(&oldID, &oldHash)
|
||
|
|
if e == nil {
|
||
|
|
if oldID != v.ID || oldHash != digest {
|
||
|
|
return fault(409, "EVENT_CONFLICT")
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if !errors.Is(e, pgx.ErrNoRows) {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
var attempt string
|
||
|
|
e = tx.QueryRow(ctx, `SELECT COALESCE(media_attempt_id,'') FROM call_session WHERE id=$1::uuid`, c.ID).Scan(&attempt)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if attempt != "" && attempt != v.AttemptID {
|
||
|
|
return fault(409, "MEDIA_ATTEMPT_CONFLICT")
|
||
|
|
}
|
||
|
|
if v.Type == "TRANSCRIPT_REVISION" && !canStore {
|
||
|
|
return fault(403, "TRANSCRIPT_STORAGE_REVOKED")
|
||
|
|
}
|
||
|
|
if v.Type == "START_INTENT" {
|
||
|
|
manifest, _ := v.Payload["manifest_id"].(string)
|
||
|
|
if manifest != c.ManifestID || c.ManifestStatus != "ISSUED" || c.Expires == nil || !c.Expires.After(s.now()) {
|
||
|
|
return fault(409, "MANIFEST_REVOKED")
|
||
|
|
}
|
||
|
|
var renewed bool
|
||
|
|
e = tx.QueryRow(ctx, `SELECT authorization_renewed_at IS NOT NULL AND lease_expires_at>now() FROM call_session WHERE id=$1::uuid`, c.ID).Scan(&renewed)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if !renewed {
|
||
|
|
return fault(409, "FIRST_HEARTBEAT_REQUIRED")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_event(session_id,event_id,seq,event_type,digest,payload) VALUES($1::uuid,$2,$3,$4,$5,$6::jsonb)`, c.ID, v.ID, v.Seq, v.Type, digest, jsonBytes(v))
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
// Event persistence never renews authorization or revives a terminal call.
|
||
|
|
switch v.Type {
|
||
|
|
case "START_INTENT":
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET media_attempt_id=COALESCE(media_attempt_id,$2),call_state=CASE WHEN call_state='PREPARING' THEN 'STARTING' ELSE call_state END WHERE id=$1::uuid`, c.ID, v.AttemptID)
|
||
|
|
case "MEDIA_CONNECTED":
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET media_attempt_id=COALESCE(media_attempt_id,$2),remote_state=CASE WHEN sdk_stopped THEN remote_state ELSE 'ACTIVE' END,call_state=CASE WHEN manifest_status='ISSUED' AND call_state IN ('PREPARING','STARTING') THEN 'TALKING' ELSE call_state END WHERE id=$1::uuid`, c.ID, v.AttemptID)
|
||
|
|
case "MEDIA_DISCONNECTED":
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET remote_state=CASE WHEN sdk_stopped THEN remote_state ELSE 'UNKNOWN' END WHERE id=$1::uuid`, c.ID)
|
||
|
|
case "GAP_DECLARED":
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET text_gap=true,transcript_state=CASE WHEN transcript_state IN ('COMPLETE','EMPTY') THEN 'PARTIAL' ELSE transcript_state END WHERE id=$1::uuid`, c.ID)
|
||
|
|
case "SDK_STOP_RESULT":
|
||
|
|
if v.Payload["outcome"] == "STOPPED_OBSERVED" {
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET sdk_stopped=true,remote_state=CASE WHEN remote_state IN ('NOT_STARTED','STOPPED_OBSERVED') THEN remote_state ELSE 'UNKNOWN' END WHERE id=$1::uuid`, c.ID)
|
||
|
|
}
|
||
|
|
case "TRANSCRIPT_REVISION":
|
||
|
|
var t Transcript
|
||
|
|
if e = StrictJSON(jsonBytes(v.Payload), &t); e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
e = s.transcript(ctx, tx, c, v, t)
|
||
|
|
}
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
func (s *Service) transcript(ctx context.Context, tx pgx.Tx, c session, v Event, t Transcript) error {
|
||
|
|
var suppressed bool
|
||
|
|
if e := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM memory_source_suppression s CROSS JOIN LATERAL jsonb_array_elements(s.source_refs) r WHERE s.subject_id=$1::uuid AND s.scope='SOURCE_AND_DERIVED' AND r->>'session_id'=$2 AND r->>'utterance_id'=$3)`, c.Subject, c.ID, t.Utterance).Scan(&suppressed); e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if suppressed {
|
||
|
|
return fault(410, "SOURCE_REDACTED")
|
||
|
|
}
|
||
|
|
if t.Parts == 0 {
|
||
|
|
t.Parts = 1
|
||
|
|
}
|
||
|
|
if t.Hash == "" && t.Parts == 1 {
|
||
|
|
t.Hash = hash([]byte(t.Text))
|
||
|
|
}
|
||
|
|
if !short(t.Utterance) || t.Revision < 0 || !oneOf(t.Speaker, "USER", "ASSISTANT", "UNKNOWN") || !utf8.ValidString(t.Text) || len(t.Text) > 4096 || len(t.Text) == 0 || t.Parts < 1 || t.Parts > 16 || t.Part < 0 || t.Part >= t.Parts || len(t.Hash) != 64 {
|
||
|
|
return fault(400, "INVALID_TRANSCRIPT_FRAGMENT")
|
||
|
|
}
|
||
|
|
var oldText, oldHash, oldSpeaker string
|
||
|
|
var oldParts int
|
||
|
|
var oldFinal bool
|
||
|
|
e := tx.QueryRow(ctx, `SELECT text_content,content_sha256,speaker,part_count,is_final FROM call_text_fragment WHERE session_id=$1::uuid AND utterance_id=$2 AND source_revision=$3 AND part_index=$4`, c.ID, t.Utterance, t.Revision, t.Part).Scan(&oldText, &oldHash, &oldSpeaker, &oldParts, &oldFinal)
|
||
|
|
if e == nil {
|
||
|
|
if oldText != t.Text || oldHash != t.Hash || oldSpeaker != t.Speaker || oldParts != t.Parts || oldFinal != t.Final {
|
||
|
|
return fault(409, "FRAGMENT_CONFLICT")
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
if !errors.Is(e, pgx.ErrNoRows) {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
var mismatch bool
|
||
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_text_fragment WHERE session_id=$1::uuid AND utterance_id=$2 AND source_revision=$3 AND (content_sha256<>$4 OR part_count<>$5 OR speaker<>$6 OR is_final<>$7))`, c.ID, t.Utterance, t.Revision, t.Hash, t.Parts, t.Speaker, t.Final).Scan(&mismatch)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if mismatch {
|
||
|
|
return fault(409, "FRAGMENT_CONFLICT")
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_text_fragment(session_id,utterance_id,source_revision,part_index,part_count,speaker,text_content,content_sha256,is_final,interrupted,source_at) VALUES($1::uuid,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, c.ID, t.Utterance, t.Revision, t.Part, t.Parts, t.Speaker, t.Text, t.Hash, t.Final, t.Delivery == "PARTIAL", time.UnixMilli(v.Occurred))
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
}
|
||
|
|
var count int
|
||
|
|
var full string
|
||
|
|
e = tx.QueryRow(ctx, `SELECT count(*),COALESCE(string_agg(text_content,'' ORDER BY part_index),'') FROM call_text_fragment WHERE session_id=$1::uuid AND utterance_id=$2 AND source_revision=$3`, c.ID, t.Utterance, t.Revision).Scan(&count, &full)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if count != t.Parts {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if hash([]byte(full)) != strings.ToLower(t.Hash) {
|
||
|
|
return fault(409, "TEXT_HASH_MISMATCH")
|
||
|
|
}
|
||
|
|
if t.Final && t.Speaker == "UNKNOWN" {
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET text_gap=true WHERE id=$1::uuid`, c.ID)
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if !t.Final {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
var priorRevision int64
|
||
|
|
var priorSpeaker string
|
||
|
|
e = tx.QueryRow(ctx, `SELECT source_revision,speaker FROM call_utterance WHERE session_id=$1::uuid AND utterance_id=$2`, c.ID, t.Utterance).Scan(&priorRevision, &priorSpeaker)
|
||
|
|
if e == nil {
|
||
|
|
if priorSpeaker != t.Speaker {
|
||
|
|
return fault(409, "SPEAKER_CONFLICT")
|
||
|
|
}
|
||
|
|
if priorRevision >= t.Revision {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
} else 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,playback_complete,source_at) VALUES($1::uuid,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(session_id,utterance_id) DO UPDATE SET source_revision=excluded.source_revision,text_content=excluded.text_content,interrupted=excluded.interrupted,playback_complete=excluded.playback_complete,source_at=excluded.source_at WHERE call_utterance.source_revision<excluded.source_revision`, c.ID, t.Utterance, t.Revision, t.Speaker, full, t.Delivery == "PARTIAL", t.Delivery == "COMPLETED_OBSERVED", time.UnixMilli(v.Occurred))
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET transcript_revision=transcript_revision+1,last_text_at=now(),transcript_state=CASE WHEN transcript_state IN ('COMPLETE','PARTIAL','EMPTY') THEN 'DRAINING' ELSE transcript_state END,drain_started_at=CASE WHEN transcript_state IN ('COMPLETE','PARTIAL','EMPTY') THEN now() ELSE drain_started_at END WHERE id=$1::uuid`, c.ID)
|
||
|
|
if e != nil {
|
||
|
|
return e
|
||
|
|
}
|
||
|
|
if t.Speaker == "USER" {
|
||
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_postprocess_job(session_id,transcript_revision,extractor_version) SELECT id,transcript_revision,'intent-1' FROM call_session WHERE id=$1::uuid ON CONFLICT DO NOTHING`, c.ID)
|
||
|
|
}
|
||
|
|
return e
|
||
|
|
}
|