Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
231 lines
12 KiB
Go
231 lines
12 KiB
Go
package callmemory
|
|
|
|
import (
|
|
"context"
|
|
"data-terminal-cloud/internal/deviceauth"
|
|
"encoding/json"
|
|
"errors"
|
|
"github.com/jackc/pgx/v5"
|
|
"time"
|
|
)
|
|
|
|
func (s *Service) heartbeat(ctx context.Context, tx pgx.Tx, d deviceauth.Identity, in HeartbeatInput, cached []byte) (any, error) {
|
|
if in.SessionID == nil || !validUUID(in.ManifestID) || !validUUID(in.AttemptID) || in.Uptime < 0 || in.LastSeq < 0 || in.Epoch < 0 || !oneOf(in.Runtime, "WARMING", "PREPARING", "STARTING", "LISTENING", "THINKING", "ANSWERING", "STOPPING", "ON_HOOK", "ERROR") {
|
|
return nil, fault(400, "INVALID_HEARTBEAT")
|
|
}
|
|
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 || c.ManifestID != in.ManifestID || (in.HookID != "" && c.Hook != in.HookID) {
|
|
return nil, fault(409, "CALL_IDENTITY_MISMATCH")
|
|
}
|
|
var epoch int64
|
|
var consent, validBinding bool
|
|
var attempt string
|
|
e = tx.QueryRow(ctx, `SELECT us.privacy_epoch,(cs.memory_mode='OFF' OR EXISTS(SELECT 1 FROM consent_record cr WHERE cr.subject_id=us.subject_id AND cr.consent_type='personalized_memory' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now()))),EXISTS(SELECT 1 FROM device_binding b JOIN device d ON d.id=b.device_id WHERE b.device_id=$2::uuid AND b.subject_id=us.subject_id AND b.unbound_at IS NULL AND b.generation=$3 AND d.binding_generation=$3 AND d.status='registered' AND d.activation_state NOT IN ('REVOKED','LOST')),COALESCE(cs.media_attempt_id,'') FROM memory_user_state us JOIN call_session cs ON cs.subject_id=us.subject_id WHERE cs.id=$1::uuid`, c.ID, d.ID, in.Binding).Scan(&epoch, &consent, &validBinding, &attempt)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if attempt != "" && attempt != in.AttemptID {
|
|
return nil, fault(409, "MEDIA_ATTEMPT_CONFLICT")
|
|
}
|
|
now := s.now()
|
|
var voiceAllowed 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')`, c.Subject).Scan(&voiceAllowed)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
consent = consent && voiceAllowed
|
|
allowed := validBinding && consent && c.ManifestStatus == "ISSUED" && c.Epoch == epoch && in.Epoch == epoch && c.Lease != nil && c.Lease.After(now) && !oneOf(c.State, "ENDED", "CANCELLED", "FAILED", "STOPPING")
|
|
if allowed {
|
|
// Replayed replies retain the original absolute expiry; cannot create a new lease.
|
|
if len(cached) > 0 {
|
|
var old map[string]any
|
|
if e = json.Unmarshal(cached, &old); e != nil {
|
|
return nil, e
|
|
}
|
|
if deadline, ok := old["lease_expires_at_ms"].(float64); ok && int64(deadline) > now.UnixMilli() {
|
|
return old, nil
|
|
}
|
|
allowed = false
|
|
}
|
|
}
|
|
var expiry int64
|
|
if allowed {
|
|
expiry = now.UnixMilli() + LeaseMS
|
|
var issued time.Time
|
|
var manifest []byte
|
|
e = tx.QueryRow(ctx, `SELECT created_at,decode(translate(manifest_b64,'-_','+/') || repeat('=',(4-length(manifest_b64)%4)%4),'base64') FROM call_session WHERE id=$1::uuid`, c.ID).Scan(&issued, &manifest)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
var m Manifest
|
|
if e = json.Unmarshal(manifest, &m); e != nil {
|
|
return nil, e
|
|
}
|
|
max := issued.Add(time.Duration(m.MaxCallMS) * time.Millisecond).UnixMilli()
|
|
if expiry > max {
|
|
expiry = max
|
|
}
|
|
if expiry <= now.UnixMilli() {
|
|
allowed = false
|
|
}
|
|
}
|
|
if allowed {
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET media_attempt_id=COALESCE(media_attempt_id,$2),lease_expires_at=$3,authorization_renewed_at=now() WHERE id=$1::uuid`, c.ID, in.AttemptID, time.UnixMilli(expiry))
|
|
} else {
|
|
expiry = now.UnixMilli()
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET manifest_status=CASE WHEN manifest_status='ISSUED' THEN 'REVOKED' ELSE manifest_status END,call_state=CASE WHEN call_state IN ('PREPARING','STARTING','TALKING') THEN 'STOPPING' ELSE call_state END,lease_expires_at=LEAST(lease_expires_at,now()) WHERE id=$1::uuid`, c.ID)
|
|
if e == nil {
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_command(session_id,privacy_epoch,type,reason,expires_at) VALUES($1::uuid,$2,'STOP_CALL','AUTHORIZATION_EXPIRED',now()+interval '7 days') ON CONFLICT DO NOTHING`, c.ID, epoch)
|
|
}
|
|
}
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
commands, e := s.commands(ctx, tx, c.ID)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
lease := expiry - now.UnixMilli()
|
|
if !allowed {
|
|
lease = 0
|
|
}
|
|
transcriptAllowed, e := transcriptPermission(ctx, tx, c.Subject)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
return map[string]any{"allowed": allowed, "lease_ms": lease, "lease_expires_at_ms": expiry, "current_privacy_epoch": epoch, "transcript_upload_allowed": allowed && transcriptAllowed, "commands": commands}, nil
|
|
}
|
|
func (s *Service) end(ctx context.Context, tx pgx.Tx, d deviceauth.Identity, in EndInput) (any, error) {
|
|
if !validUUID(in.BootID) || !short(in.HookID) || in.Generation < 0 || in.Binding < 0 || in.Elapsed < 0 || in.LastSeq < 0 || !oneOf(in.SDKStatus, "NOT_STARTED", "STOP_PENDING", "STOPPED_OBSERVED", "UNKNOWN") || !oneOf(in.Reason, "ON_HOOK", "CANCEL_BEFORE_START", "START_FAILED", "NO_FIRST_AUDIO", "NETWORK_TIMEOUT", "LEASE_EXPIRED", "POLICY_REVOKED", "INACTIVITY", "MAX_DURATION", "DEVICE_REBOOT", "LOCAL_ERROR") {
|
|
return nil, fault(400, "INVALID_END")
|
|
}
|
|
c, e := readSession(ctx, tx, d.ID, in.CallKey)
|
|
if e != nil {
|
|
var f *Error
|
|
if !errors.As(e, &f) || f.Status != 404 {
|
|
return nil, e
|
|
}
|
|
if in.SessionID != nil {
|
|
return nil, e
|
|
}
|
|
// A pre-prepare cancellation contains no personal data or provider state.
|
|
if in.Binding != d.BindingGeneration {
|
|
return nil, fault(409, "BINDING_GENERATION_CONFLICT")
|
|
}
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_session(device_id,tenant_id,boot_id,hook_event_id,call_generation,binding_generation,call_state,remote_state,transcript_state,local_audio_off,sdk_stopped,ended_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,$6,'CANCELLED','NOT_STARTED','EMPTY',true,true,now()) ON CONFLICT(device_id,boot_id,hook_event_id) DO NOTHING`, d.ID, d.TenantID, in.BootID, in.HookID, in.Generation, in.Binding)
|
|
return map[string]any{"accepted": true, "business_session_id": nil, "call_state": "CANCELLED", "remote_state": "NOT_STARTED"}, e
|
|
}
|
|
if c.Boot != in.BootID || c.Hook != in.HookID || c.Generation != in.Generation || c.Binding != in.Binding {
|
|
return nil, fault(409, "CALL_IDENTITY_MISMATCH")
|
|
}
|
|
stopped := in.SDKStatus == "STOPPED_OBSERVED"
|
|
remote := c.Remote
|
|
if stopped && remote != "STOPPED_OBSERVED" {
|
|
remote = "UNKNOWN" // Local SDK completion is not a provider-side observation.
|
|
}
|
|
if in.SDKStatus == "NOT_STARTED" {
|
|
var started bool
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_event WHERE session_id=$1::uuid AND event_type IN ('START_INTENT','MEDIA_CONNECTED','FIRST_AUDIO_RECEIVED','FIRST_PLAYBACK_STARTED'))`, c.ID).Scan(&started)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if started {
|
|
return nil, fault(409, "STOP_EVIDENCE_CONFLICT")
|
|
}
|
|
remote = "NOT_STARTED"
|
|
stopped = true
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET call_state='ENDED',manifest_status=CASE WHEN $2='NOT_STARTED' THEN 'DISCARDED' ELSE 'REVOKED' END,remote_state=$2,local_audio_off=local_audio_off OR $3,sdk_stopped=sdk_stopped OR $4,lease_expires_at=LEAST(lease_expires_at,now()),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`, c.ID, remote, in.AudioStopped, stopped)
|
|
return map[string]any{"accepted": true, "business_session_id": c.ID, "call_state": "ENDED", "remote_state": remote}, e
|
|
}
|
|
func (s *Service) commands(ctx context.Context, tx pgx.Tx, sid string) ([]map[string]any, error) {
|
|
rows, e := tx.Query(ctx, `SELECT cc.id::text,cc.type,cc.reason,cc.expires_at,cc.privacy_epoch,cs.call_generation,COALESCE(cs.manifest_id::text,'') FROM call_command cc JOIN call_session cs ON cs.id=cc.session_id WHERE cc.session_id=$1::uuid AND NOT(cs.local_audio_off AND cs.sdk_stopped) ORDER BY cc.created_at`, sid)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
defer rows.Close()
|
|
out := []map[string]any{}
|
|
for rows.Next() {
|
|
var id, kind, reason, manifest string
|
|
var expires time.Time
|
|
var epoch, generation int64
|
|
if e = rows.Scan(&id, &kind, &reason, &expires, &epoch, &generation, &manifest); e != nil {
|
|
return nil, e
|
|
}
|
|
out = append(out, map[string]any{"command_id": id, "type": kind, "business_session_id": sid, "call_generation": generation, "reason": reason, "expires_at_ms": expires.UnixMilli(), "privacy_epoch": epoch, "required_privacy_epoch": epoch, "manifest_id": manifest})
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
func (s *Service) status(ctx context.Context, tx pgx.Tx, c session) (any, error) {
|
|
commands, e := s.commands(ctx, tx, c.ID)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
epoch := c.Epoch
|
|
if c.Subject != "" {
|
|
if e = tx.QueryRow(ctx, `SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid`, c.Subject).Scan(&epoch); e != nil {
|
|
return nil, e
|
|
}
|
|
}
|
|
return map[string]any{"business_session_id": c.ID, "call_state": c.State, "remote_state": c.Remote, "transcript_state": c.Transcript, "last_accepted_seq": c.LastSeq, "manifest_status": c.ManifestStatus, "current_privacy_epoch": epoch, "commands": commands}, nil
|
|
}
|
|
func (s *Service) ack(ctx context.Context, tx pgx.Tx, d deviceauth.Identity, in AckInput) (any, error) {
|
|
if !validUUID(in.SessionID) || !validUUID(in.CommandID) || !oneOf(in.Phase, "LOCAL_AUDIO_OFF", "SDK_STOPPED", "MANIFEST_DISCARDED", "NOTICE_PLAYED", "NOTICE_SKIPPED", "CANNOT_COMPLETE") {
|
|
return nil, fault(400, "INVALID_COMMAND_ACK")
|
|
}
|
|
c, e := readSession(ctx, tx, d.ID, CallKey{SessionID: &in.SessionID})
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if in.Generation != c.Generation {
|
|
return nil, fault(409, "CALL_IDENTITY_MISMATCH")
|
|
}
|
|
var kind string
|
|
e = tx.QueryRow(ctx, `SELECT type FROM call_command WHERE id=$1::uuid AND session_id=$2::uuid FOR UPDATE`, in.CommandID, c.ID).Scan(&kind)
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
|
return nil, fault(404, "COMMAND_NOT_FOUND")
|
|
}
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if oneOf(in.Phase, "NOTICE_PLAYED", "NOTICE_SKIPPED") {
|
|
if kind != "PLAY_LOCAL_NOTICE" {
|
|
return nil, fault(409, "COMMAND_PHASE_CONFLICT")
|
|
}
|
|
} else if kind != "STOP_CALL" {
|
|
return nil, fault(409, "COMMAND_PHASE_CONFLICT")
|
|
}
|
|
switch in.Phase {
|
|
case "MANIFEST_DISCARDED":
|
|
var started bool
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_event WHERE session_id=$1::uuid AND event_type IN ('START_INTENT','MEDIA_CONNECTED','FIRST_AUDIO_RECEIVED','FIRST_PLAYBACK_STARTED'))`, c.ID).Scan(&started)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if started || c.Remote == "ACTIVE" {
|
|
return nil, fault(409, "STOP_EVIDENCE_CONFLICT")
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET manifest_status='DISCARDED',remote_state='NOT_STARTED',local_audio_off=true,sdk_stopped=true,call_state='ENDED',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`, c.ID)
|
|
case "LOCAL_AUDIO_OFF":
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET local_audio_off=true WHERE id=$1::uuid`, c.ID)
|
|
case "SDK_STOPPED":
|
|
_, 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)
|
|
}
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE call_command SET phase=$2,evidence=evidence || jsonb_build_object($2::text,true) WHERE id=$1::uuid`, in.CommandID, in.Phase)
|
|
return map[string]any{"accepted": true}, e
|
|
}
|
|
func oneOf(s string, values ...string) bool {
|
|
for _, v := range values {
|
|
if s == v {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|