286 lines
11 KiB
Go
286 lines
11 KiB
Go
|
package callmemory
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"data-terminal-cloud/internal/audit"
|
||
|
|
"data-terminal-cloud/internal/identity"
|
||
|
|
"errors"
|
||
|
|
"net/url"
|
||
|
|
"regexp"
|
||
|
|
"sort"
|
||
|
|
"strconv"
|
||
|
|
"strings"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
"github.com/volcengine/ve-tos-golang-sdk/v2/tos"
|
||
|
|
"github.com/volcengine/ve-tos-golang-sdk/v2/tos/enum"
|
||
|
|
)
|
||
|
|
|
||
|
|
// Only metadata and locally signed URLs pass through this service. There is
|
||
|
|
// deliberately no GetObject operation or audio proxy in this interface.
|
||
|
|
type RecordingStore interface {
|
||
|
|
List(context.Context, string, string) ([]CloudRecording, error)
|
||
|
|
Sign(context.Context, string, string, int64) (string, error)
|
||
|
|
PlaybackReady() bool
|
||
|
|
}
|
||
|
|
type CloudRecording struct {
|
||
|
|
ID string `json:"id"`
|
||
|
|
Speaker string `json:"speaker"`
|
||
|
|
Duration int64 `json:"duration_seconds"`
|
||
|
|
Bytes int64 `json:"bytes"`
|
||
|
|
Key string `json:"-"`
|
||
|
|
}
|
||
|
|
type TOSRecordings struct {
|
||
|
|
client *tos.ClientV2
|
||
|
|
bucket string
|
||
|
|
playbackEndpoint string
|
||
|
|
}
|
||
|
|
|
||
|
|
var recordingDevice = regexp.MustCompile(`^[A-Za-z0-9_-]{1,128}$`)
|
||
|
|
var recordingSuffix = regexp.MustCompile(`^[0-9]{13}_([0-9]{6})_[A-Za-z0-9]+\.(?:mp4|mp3)$`)
|
||
|
|
|
||
|
|
func NewTOSRecordings(bucket, region, ak, sk, token string) (*TOSRecordings, error) {
|
||
|
|
if !regexp.MustCompile(`^[a-z0-9][a-z0-9-]{1,61}[a-z0-9]$`).MatchString(bucket) ||
|
||
|
|
!regexp.MustCompile(`^[a-z]{2}-[a-z]+[0-9]*$`).MatchString(region) || ak == "" || sk == "" {
|
||
|
|
return nil, errors.New("invalid TOS recording configuration")
|
||
|
|
}
|
||
|
|
credentials := tos.NewStaticCredentials(ak, sk)
|
||
|
|
if token != "" {
|
||
|
|
credentials.WithSecurityToken(token)
|
||
|
|
}
|
||
|
|
client, err := tos.NewClientV2("https://tos-"+region+".volces.com", tos.WithRegion(region), tos.WithCredentials(credentials), tos.WithMaxRetryCount(0))
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return &TOSRecordings{client: client, bucket: bucket}, nil
|
||
|
|
}
|
||
|
|
func (s *Service) ConfigureRecordings(store RecordingStore) { s.recordings = store }
|
||
|
|
|
||
|
|
// TOS default domains force attachment responses. Playback requires a bound
|
||
|
|
// HTTPS custom domain; signing still uses the same private bucket credentials.
|
||
|
|
func (s *TOSRecordings) ConfigurePlayback(endpoint string) error {
|
||
|
|
if endpoint == "" {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
u, e := url.Parse(endpoint)
|
||
|
|
if e != nil || u.Scheme != "https" || u.Hostname() == "" || u.User != nil || u.Port() != "" || (u.Path != "" && u.Path != "/") || u.RawQuery != "" || u.Fragment != "" {
|
||
|
|
return errors.New("invalid recording playback domain")
|
||
|
|
}
|
||
|
|
s.playbackEndpoint = "https://" + u.Host
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
func (s *TOSRecordings) PlaybackReady() bool { return s.playbackEndpoint != "" }
|
||
|
|
|
||
|
|
func recordingPrefixes(device, bot string) ([]string, error) {
|
||
|
|
if !recordingDevice.MatchString(device) || !isRTCAgentUser(bot) {
|
||
|
|
return nil, errors.New("unverified recording identity")
|
||
|
|
}
|
||
|
|
// Hardware gateway identity, corroborated by the authenticated subtitle UID.
|
||
|
|
room := "aibotrtc_OPUS_" + device + "_" + strings.TrimPrefix(bot, "bot_")
|
||
|
|
return []string{device + "_" + room + "_" + device + "_C_", bot + "_" + room + "_" + bot + "_C_"}, nil
|
||
|
|
}
|
||
|
|
func parseRecording(key, prefix, speaker string, size int64) (CloudRecording, bool) {
|
||
|
|
if !strings.HasPrefix(key, prefix) {
|
||
|
|
return CloudRecording{}, false
|
||
|
|
}
|
||
|
|
parts := recordingSuffix.FindStringSubmatch(strings.TrimPrefix(key, prefix))
|
||
|
|
if len(parts) != 2 || size <= 0 {
|
||
|
|
return CloudRecording{}, false
|
||
|
|
}
|
||
|
|
duration, _ := strconv.ParseInt(parts[1], 10, 64)
|
||
|
|
return CloudRecording{ID: hash([]byte(key)), Key: key, Speaker: speaker, Duration: duration, Bytes: size}, true
|
||
|
|
}
|
||
|
|
func (s *TOSRecordings) List(ctx context.Context, device, bot string) ([]CloudRecording, error) {
|
||
|
|
prefixes, err := recordingPrefixes(device, bot)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
items := []CloudRecording{}
|
||
|
|
room := "aibotrtc_OPUS_" + device + "_" + strings.TrimPrefix(bot, "bot_")
|
||
|
|
prefixes = append([]string{mixedPrefix(room)}, prefixes...)
|
||
|
|
for i, prefix := range prefixes {
|
||
|
|
cursor := ""
|
||
|
|
for page := 0; ; page++ {
|
||
|
|
if page >= 20 {
|
||
|
|
return nil, errors.New("recording listing limit exceeded")
|
||
|
|
}
|
||
|
|
out, e := s.client.ListObjectsType2(ctx, &tos.ListObjectsType2Input{Bucket: s.bucket, Prefix: prefix, MaxKeys: 100, ContinuationToken: cursor, ListOnlyOnce: true})
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
for _, object := range out.Contents {
|
||
|
|
if item, ok := parseRecording(object.Key, prefix, []string{"DIALOG", "USER", "ASSISTANT"}[i], object.Size); ok {
|
||
|
|
items = append(items, item)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if !out.IsTruncated {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
if out.NextContinuationToken == "" || out.NextContinuationToken == cursor {
|
||
|
|
return nil, errors.New("invalid recording pagination")
|
||
|
|
}
|
||
|
|
cursor = out.NextContinuationToken
|
||
|
|
}
|
||
|
|
if i == 0 && len(items) > 0 {
|
||
|
|
break
|
||
|
|
} // Prefer the complete conversation over separate tracks.
|
||
|
|
}
|
||
|
|
sort.Slice(items, func(i, j int) bool { return items[i].Key < items[j].Key })
|
||
|
|
return items, nil
|
||
|
|
}
|
||
|
|
func (s *TOSRecordings) Sign(ctx context.Context, key, action string, seconds int64) (string, error) {
|
||
|
|
if action != "PLAY" && action != "DOWNLOAD" {
|
||
|
|
return "", errors.New("invalid recording action")
|
||
|
|
}
|
||
|
|
disposition := "inline"
|
||
|
|
if action == "DOWNLOAD" {
|
||
|
|
disposition = "attachment"
|
||
|
|
}
|
||
|
|
mimeType, filename := "audio/mp4", "conversation.mp4"
|
||
|
|
if strings.HasSuffix(key, ".mp3") {
|
||
|
|
mimeType = "audio/mpeg"
|
||
|
|
filename = "conversation.mp3"
|
||
|
|
}
|
||
|
|
in := &tos.PreSignedURLInput{HTTPMethod: enum.HttpMethodGet, Bucket: s.bucket, Key: key, Expires: seconds,
|
||
|
|
Query: map[string]string{"response-content-type": mimeType, "response-content-disposition": disposition + `; filename="` + filename + `"`}}
|
||
|
|
if action == "PLAY" {
|
||
|
|
if !s.PlaybackReady() {
|
||
|
|
return "", errors.New("recording playback domain required")
|
||
|
|
}
|
||
|
|
custom := true
|
||
|
|
in.IsCustomDomain = &custom
|
||
|
|
in.AlternativeEndpoint = s.playbackEndpoint
|
||
|
|
}
|
||
|
|
out, e := s.client.PreSignedURL(in)
|
||
|
|
if e != nil {
|
||
|
|
return "", e
|
||
|
|
}
|
||
|
|
return out.SignedUrl, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
type recordingInput struct {
|
||
|
|
ElderID string `json:"elder_id"`
|
||
|
|
SessionID string `json:"business_session_id"`
|
||
|
|
RecordingID string `json:"recording_id,omitempty"`
|
||
|
|
Action string `json:"action,omitempty"`
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) callRecordings(ctx context.Context, tx pgx.Tx, p identity.Principal, op string, raw []byte) (any, error) {
|
||
|
|
var in recordingInput
|
||
|
|
if e := StrictJSON(raw, &in); e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if !validUUID(in.SessionID) {
|
||
|
|
return nil, fault(400, "INVALID_CALL_ID")
|
||
|
|
}
|
||
|
|
if op == "calls/recording-access" && (len(in.RecordingID) != 64 || !oneOf(in.Action, "PLAY", "DOWNLOAD")) {
|
||
|
|
return nil, fault(400, "INVALID_RECORDING_REQUEST")
|
||
|
|
}
|
||
|
|
a, e := ResolveAccess(ctx, tx, p, in.ElderID, false)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if !a.Sensitive {
|
||
|
|
return nil, fault(403, "RECORDING_ACCESS_DENIED")
|
||
|
|
}
|
||
|
|
var device, bot, publicUser string
|
||
|
|
var consent, play, download, mixed bool
|
||
|
|
var playUntil, downloadUntil *time.Time
|
||
|
|
e = tx.QueryRow(ctx, `SELECT COALESCE(c.budget_snapshot->>'device_name',''),COALESCE(c.budget_snapshot->>'rtc_agent_user_id',''),ua.public_id::text,COALESCE(c.budget_snapshot->>'mixed_recording','false')='true',
|
||
|
|
EXISTS(SELECT 1 FROM consent_record cr WHERE cr.tenant_id=c.tenant_id AND cr.subject_id=c.subject_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()))
|
||
|
|
AND NOT EXISTS(SELECT 1 FROM privacy_request pr WHERE pr.subject_id=c.subject_id AND pr.request_type='DELETE'),
|
||
|
|
(SELECT max(expires_at) FROM recording_access_grant WHERE tenant_id=c.tenant_id AND subject_id=c.subject_id AND grantee_user_id=ua.id AND permission='PLAY' AND revoked_at IS NULL AND expires_at>now()),
|
||
|
|
(SELECT max(expires_at) FROM recording_access_grant WHERE tenant_id=c.tenant_id AND subject_id=c.subject_id AND grantee_user_id=ua.id AND permission='DOWNLOAD' AND revoked_at IS NULL AND expires_at>now())
|
||
|
|
FROM call_session c JOIN user_account ua ON ua.id=$4::uuid AND ua.tenant_id=c.tenant_id
|
||
|
|
WHERE c.id=$1::uuid AND c.subject_id=$2::uuid AND c.tenant_id=$3::uuid`, in.SessionID, a.Subject, a.Tenant, a.User).Scan(&device, &bot, &publicUser, &mixed, &consent, &playUntil, &downloadUntil)
|
||
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
||
|
|
return nil, fault(404, "CALL_NOT_FOUND")
|
||
|
|
}
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
play = consent && playUntil != nil
|
||
|
|
download = consent && downloadUntil != nil
|
||
|
|
playbackReady := s.recordings != nil && s.recordings.PlaybackReady()
|
||
|
|
result := map[string]any{"items": []CloudRecording{}, "can_play": play, "can_download": download, "can_manage": a.Manage, "requester_id": publicUser, "consent": consent, "state": "READY", "playback_ready": playbackReady}
|
||
|
|
result["mixed"] = mixed
|
||
|
|
if op == "calls/recording-access" && in.Action == "PLAY" && !playbackReady {
|
||
|
|
return nil, fault(409, "RECORDING_PLAYBACK_NOT_CONFIGURED")
|
||
|
|
}
|
||
|
|
if op == "calls/recording-access" && ((!play && in.Action == "PLAY") || (!download && in.Action == "DOWNLOAD")) {
|
||
|
|
return nil, fault(403, "RECORDING_ACCESS_DENIED")
|
||
|
|
}
|
||
|
|
if !consent {
|
||
|
|
result["state"] = "CONSENT_REQUIRED"
|
||
|
|
return result, nil
|
||
|
|
}
|
||
|
|
if s.recordings == nil {
|
||
|
|
result["state"] = "NOT_CONFIGURED"
|
||
|
|
if op == "calls/recording-access" {
|
||
|
|
return nil, fault(503, "RECORDING_NOT_CONFIGURED")
|
||
|
|
}
|
||
|
|
return result, nil
|
||
|
|
}
|
||
|
|
if _, e = recordingPrefixes(device, bot); e != nil {
|
||
|
|
result["state"] = "IDENTITY_PENDING"
|
||
|
|
if op == "calls/recording-access" {
|
||
|
|
return nil, fault(404, "RECORDING_NOT_FOUND")
|
||
|
|
}
|
||
|
|
return result, nil
|
||
|
|
}
|
||
|
|
items, e := s.recordings.List(ctx, device, bot)
|
||
|
|
if e != nil {
|
||
|
|
return nil, fault(503, "RECORDING_STORAGE_UNAVAILABLE")
|
||
|
|
}
|
||
|
|
if mixed {
|
||
|
|
complete := []CloudRecording{}
|
||
|
|
for _, item := range items {
|
||
|
|
if item.Speaker == "DIALOG" {
|
||
|
|
complete = append(complete, item)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
items = complete
|
||
|
|
}
|
||
|
|
if op == "calls/recordings" {
|
||
|
|
result["items"] = items
|
||
|
|
return result, nil
|
||
|
|
}
|
||
|
|
for _, item := range items {
|
||
|
|
if item.ID == in.RecordingID {
|
||
|
|
until := playUntil
|
||
|
|
if in.Action == "DOWNLOAD" {
|
||
|
|
until = downloadUntil
|
||
|
|
}
|
||
|
|
seconds := int64(300)
|
||
|
|
if remaining := int64(until.Sub(s.now()).Seconds()); remaining < seconds {
|
||
|
|
seconds = remaining
|
||
|
|
}
|
||
|
|
// Never issue an address that outlives its consent or recording grant.
|
||
|
|
var consentUntil *time.Time
|
||
|
|
e = tx.QueryRow(ctx, `SELECT CASE WHEN bool_or(expires_at IS NULL) THEN NULL ELSE max(expires_at) END FROM consent_record WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND consent_type='raw_audio_storage' AND withdrawn_at IS NULL AND (expires_at IS NULL OR expires_at>now())`, a.Tenant, a.Subject).Scan(&consentUntil)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if consentUntil != nil {
|
||
|
|
if remaining := int64(consentUntil.Sub(s.now()).Seconds()); remaining < seconds {
|
||
|
|
seconds = remaining
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if seconds < 1 {
|
||
|
|
return nil, fault(403, "RECORDING_ACCESS_DENIED")
|
||
|
|
}
|
||
|
|
url, e := s.recordings.Sign(ctx, item.Key, in.Action, seconds)
|
||
|
|
if e != nil {
|
||
|
|
return nil, fault(503, "RECORDING_STORAGE_UNAVAILABLE")
|
||
|
|
}
|
||
|
|
e = audit.NewRepository().Record(ctx, tx, audit.Event{TenantID: &a.Tenant, ActorType: "user", ActorPublicID: &publicUser, Action: "call.recording." + strings.ToLower(in.Action), ResourceType: "call_session", ResourceID: &in.SessionID, Result: "success", Purpose: "customer conversation review", Detail: map[string]any{"recording_id": item.ID, "expires_seconds": seconds}})
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
return map[string]any{"access_url": url, "access_until": s.now().Add(time.Duration(seconds) * time.Second)}, nil
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return nil, fault(404, "RECORDING_NOT_FOUND")
|
||
|
|
}
|