story/internal/callmemory/records.go

package callmemory

import (
	"context"
	"data-terminal-cloud/internal/identity"
	"encoding/base64"
	"time"

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

type CallRecordsInput struct {
	ElderID   string `json:"elder_id"`
	SessionID string `json:"business_session_id,omitempty"`
	Cursor    string `json:"cursor,omitempty"`
	Limit     int    `json:"limit,omitempty"`
}
type recordCursor struct {
	Time time.Time `json:"time"`
	ID   string    `json:"id"`
}

func (s *Service) callRecords(ctx context.Context, tx pgx.Tx, p identity.Principal, op string, raw []byte) (any, error) {
	var in CallRecordsInput
	if e := StrictJSON(raw, &in); e != nil {
		return nil, e
	}
	if in.Limit == 0 {
		in.Limit = 20
	}
	if in.Limit < 1 || in.Limit > 100 || len(in.Cursor) > 1024 {
		return nil, fault(400, "INVALID_PAGINATION")
	}
	a, e := ResolveAccess(ctx, tx, p, in.ElderID, false)
	if e != nil {
		return nil, e
	}
	// Transcripts have not been classified per sentence. A general read grant
	// alone must not expose potentially sensitive conversation text.
	if !a.Sensitive {
		return nil, fault(403, "TRANSCRIPT_ACCESS_DENIED")
	}
	c := recordCursor{}
	if in.Cursor != "" {
		b, err := base64.RawURLEncoding.DecodeString(in.Cursor)
		if err != nil || StrictJSON(b, &c) != nil || c.Time.IsZero() || c.ID == "" {
			return nil, fault(400, "INVALID_CURSOR")
		}
	}
	var rows pgx.Rows
	if op == "calls/query" {
		if in.SessionID != "" || (c.ID != "" && !validUUID(c.ID)) {
			return nil, fault(400, "INVALID_CALL_ID")
		}
		rows, e = tx.Query(ctx, `SELECT id::text,created_at,ended_at,call_state,transcript_state,
   GREATEST(0,COALESCE(EXTRACT(EPOCH FROM (ended_at-created_at))*1000,0))::bigint
   FROM call_session WHERE subject_id=$1::uuid AND tenant_id=$2::uuid
   AND ($3='' OR (created_at,id)<($4,NULLIF($3,'')::uuid))
   ORDER BY created_at DESC,id DESC LIMIT $5`, a.Subject, a.Tenant, c.ID, c.Time, in.Limit+1)
	} else {
		if !validUUID(in.SessionID) {
			return nil, fault(400, "INVALID_CALL_ID")
		}
		var exists bool
		e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM call_session WHERE id=$1::uuid AND subject_id=$2::uuid AND tenant_id=$3::uuid)`, in.SessionID, a.Subject, a.Tenant).Scan(&exists)
		if e != nil {
			return nil, e
		}
		if !exists {
			return nil, fault(404, "CALL_NOT_FOUND")
		}
		rows, e = tx.Query(ctx, `SELECT u.utterance_id,u.source_at,u.speaker,u.text_content,u.interrupted,u.playback_complete
   FROM call_utterance u JOIN call_session c ON c.id=u.session_id
   WHERE c.id=$1::uuid AND c.subject_id=$2::uuid AND c.tenant_id=$3::uuid
   AND NOT EXISTS(SELECT 1 FROM privacy_request pr WHERE pr.subject_id=c.subject_id AND pr.request_type='DELETE')
   AND ($4='' OR (u.source_at,u.utterance_id)>($5,$4))
   ORDER BY u.source_at,u.utterance_id LIMIT $6`, in.SessionID, a.Subject, a.Tenant, c.ID, c.Time, in.Limit+1)
	}
	if e != nil {
		return nil, e
	}
	defer rows.Close()
	items := []map[string]any{}
	next := ""
	last := recordCursor{}
	for rows.Next() {
		if len(items) == in.Limit {
			next = base64.RawURLEncoding.EncodeToString(jsonBytes(last))
			break
		}
		var id string
		var at time.Time
		item := map[string]any{}
		if op == "calls/query" {
			var ended *time.Time
			var state, transcript string
			var duration int64
			e = rows.Scan(&id, &at, &ended, &state, &transcript, &duration)
			item = map[string]any{"id": id, "created_at": at, "ended_at": ended, "status": state, "transcript_state": transcript, "duration_ms": duration, "source": "call_memory"}
		} else {
			var speaker, content string
			var interrupted, complete bool
			e = rows.Scan(&id, &at, &speaker, &content, &interrupted, &complete)
			item = map[string]any{"id": id, "created_at": at, "speaker": speaker, "final_text": content, "interrupted": interrupted, "playback_complete": complete}
		}
		if e != nil {
			return nil, e
		}
		last = recordCursor{at, id}
		items = append(items, item)
	}
	if e = rows.Err(); e != nil {
		return nil, e
	}
	return map[string]any{"items": items, "next_cursor": next}, nil
}