272 lines
15 KiB
Go
272 lines
15 KiB
Go
package media
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/audit"
|
|
"data-terminal-cloud/internal/database"
|
|
"data-terminal-cloud/internal/identity"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
type RecordingAccessGrant struct {
|
|
ID string `json:"id"`
|
|
SubjectID string `json:"subject_id"`
|
|
GranteeUserID string `json:"grantee_user_id"`
|
|
Permission string `json:"permission"`
|
|
Purpose string `json:"purpose"`
|
|
GrantedBy string `json:"granted_by"`
|
|
GrantedAt time.Time `json:"granted_at"`
|
|
ExpiresAt time.Time `json:"expires_at"`
|
|
RevokedAt *time.Time `json:"revoked_at,omitempty"`
|
|
}
|
|
|
|
type ConversationMedia struct {
|
|
ConversationID string `json:"conversation_id"`
|
|
CanPlay bool `json:"can_play"`
|
|
CanDownload bool `json:"can_download"`
|
|
Segments []Segment `json:"segments"`
|
|
}
|
|
|
|
type MediaAccess struct {
|
|
SegmentID string `json:"segment_id"`
|
|
Action string `json:"action"`
|
|
AccessURL string `json:"access_url"`
|
|
AccessUntil time.Time `json:"access_until"`
|
|
}
|
|
|
|
func (s *Service) ListRecordingAccessGrants(ctx context.Context, principal identity.Principal, subjectPublicID string) ([]RecordingAccessGrant, error) {
|
|
var tenantID, subjectID, role string
|
|
err := s.pool.QueryRow(ctx, `
|
|
SELECT sp.tenant_id::text,sp.id::text,hm.role
|
|
FROM subject_profile sp
|
|
JOIN household_member hm ON hm.household_id=sp.household_id AND hm.tenant_id=sp.tenant_id AND hm.status='active'
|
|
JOIN user_account ua ON ua.id=hm.user_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active'
|
|
WHERE sp.public_id=$1::uuid AND sp.status='active'`, subjectPublicID, principal.Issuer, principal.Subject,
|
|
).Scan(&tenantID, &subjectID, &role)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, ErrRecordingAccessDenied
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if role != "owner" && role != "admin" {
|
|
return nil, ErrRecordingAccessDenied
|
|
}
|
|
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT rag.public_id::text,$3::text,grantee.public_id::text,rag.permission,rag.purpose,
|
|
grantor.public_id::text,rag.granted_at,rag.expires_at,rag.revoked_at
|
|
FROM recording_access_grant rag
|
|
JOIN user_account grantee ON grantee.id=rag.grantee_user_id AND grantee.tenant_id=rag.tenant_id
|
|
JOIN user_account grantor ON grantor.id=rag.granted_by AND grantor.tenant_id=rag.tenant_id
|
|
WHERE rag.tenant_id=$1::uuid AND rag.subject_id=$2::uuid
|
|
ORDER BY rag.granted_at DESC LIMIT 200`, tenantID, subjectID, subjectPublicID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := make([]RecordingAccessGrant, 0)
|
|
for rows.Next() {
|
|
var item RecordingAccessGrant
|
|
if err := rows.Scan(&item.ID, &item.SubjectID, &item.GranteeUserID, &item.Permission, &item.Purpose, &item.GrantedBy, &item.GrantedAt, &item.ExpiresAt, &item.RevokedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) GrantRecordingAccess(ctx context.Context, principal identity.Principal, subjectPublicID, granteePublicID, permission, purpose, requestID string, expiresAt time.Time) (RecordingAccessGrant, error) {
|
|
permission = strings.ToUpper(strings.TrimSpace(permission))
|
|
purpose = strings.TrimSpace(purpose)
|
|
if (permission != "PLAY" && permission != "DOWNLOAD") || purpose == "" || expiresAt.Before(time.Now().UTC().Add(time.Minute)) {
|
|
return RecordingAccessGrant{}, ErrInvalidDeclaration
|
|
}
|
|
var result RecordingAccessGrant
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
var tenantID, subjectID, requesterID, requesterPublicID, role, granteeID string
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT sp.tenant_id::text,sp.id::text,ua.id::text,ua.public_id::text,hm.role,grantee.id::text
|
|
FROM subject_profile sp
|
|
JOIN household_member hm ON hm.household_id=sp.household_id AND hm.tenant_id=sp.tenant_id AND hm.status='active'
|
|
JOIN user_account ua ON ua.id=hm.user_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active'
|
|
JOIN user_account grantee ON grantee.public_id=$4::uuid AND grantee.tenant_id=sp.tenant_id AND grantee.status='active'
|
|
JOIN household_member ghm ON ghm.household_id=sp.household_id AND ghm.tenant_id=sp.tenant_id AND ghm.user_id=grantee.id AND ghm.status='active'
|
|
WHERE sp.public_id=$1::uuid AND sp.status='active' FOR UPDATE OF sp`, subjectPublicID, principal.Issuer, principal.Subject, granteePublicID,
|
|
).Scan(&tenantID, &subjectID, &requesterID, &requesterPublicID, &role, &granteeID)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrRecordingAccessDenied
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if role != "owner" && role != "admin" {
|
|
return ErrRecordingAccessDenied
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE recording_access_grant SET revoked_at=now(),revoked_by=$1::uuid,updated_at=now() WHERE tenant_id=$2::uuid AND subject_id=$3::uuid AND grantee_user_id=$4::uuid AND permission=$5 AND revoked_at IS NULL AND expires_at>now()`, requesterID, tenantID, subjectID, granteeID, permission)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = tx.QueryRow(ctx, `
|
|
INSERT INTO recording_access_grant(tenant_id,subject_id,grantee_user_id,permission,purpose,granted_by,expires_at)
|
|
VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,$6::uuid,$7)
|
|
RETURNING public_id::text,$8::text,$9::text,$4,$5,$10::text,granted_at,expires_at,revoked_at`, tenantID, subjectID, granteeID, permission, purpose, requesterID, expiresAt, subjectPublicID, granteePublicID, requesterPublicID,
|
|
).Scan(&result.ID, &result.SubjectID, &result.GranteeUserID, &result.Permission, &result.Purpose, &result.GrantedBy, &result.GrantedAt, &result.ExpiresAt, &result.RevokedAt)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "user", ActorPublicID: &requesterPublicID, Action: "recording.access.grant", ResourceType: "recording_access_grant", ResourceID: &result.ID, RequestID: requestID, Purpose: purpose, Result: "success", Detail: map[string]any{"subject_id": subjectPublicID, "grantee_user_id": granteePublicID, "permission": permission, "expires_at": expiresAt}})
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) RevokeRecordingAccess(ctx context.Context, principal identity.Principal, grantPublicID, requestID string) (RecordingAccessGrant, error) {
|
|
var result RecordingAccessGrant
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
var tenantID, requesterID, requesterPublicID, role string
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT rag.tenant_id::text,ua.id::text,ua.public_id::text,hm.role,rag.public_id::text,sp.public_id::text,grantee.public_id::text,
|
|
rag.permission,rag.purpose,grantor.public_id::text,rag.granted_at,rag.expires_at,rag.revoked_at
|
|
FROM recording_access_grant rag
|
|
JOIN subject_profile sp ON sp.id=rag.subject_id AND sp.tenant_id=rag.tenant_id
|
|
JOIN household_member hm ON hm.household_id=sp.household_id AND hm.tenant_id=sp.tenant_id AND hm.status='active'
|
|
JOIN user_account ua ON ua.id=hm.user_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active'
|
|
JOIN user_account grantee ON grantee.id=rag.grantee_user_id
|
|
JOIN user_account grantor ON grantor.id=rag.granted_by
|
|
WHERE rag.public_id=$1::uuid FOR UPDATE OF rag`, grantPublicID, principal.Issuer, principal.Subject,
|
|
).Scan(&tenantID, &requesterID, &requesterPublicID, &role, &result.ID, &result.SubjectID, &result.GranteeUserID, &result.Permission, &result.Purpose, &result.GrantedBy, &result.GrantedAt, &result.ExpiresAt, &result.RevokedAt)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if role != "owner" && role != "admin" {
|
|
return ErrRecordingAccessDenied
|
|
}
|
|
if result.RevokedAt == nil {
|
|
if err := tx.QueryRow(ctx, `UPDATE recording_access_grant SET revoked_at=now(),revoked_by=$2::uuid,updated_at=now() WHERE public_id=$1::uuid RETURNING revoked_at`, grantPublicID, requesterID).Scan(&result.RevokedAt); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "user", ActorPublicID: &requesterPublicID, Action: "recording.access.revoke", ResourceType: "recording_access_grant", ResourceID: &result.ID, RequestID: requestID, Result: "success"})
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) GetConversationMedia(ctx context.Context, principal identity.Principal, conversationPublicID string) (ConversationMedia, error) {
|
|
var result ConversationMedia
|
|
var tenantID, subjectID, userID string
|
|
err := s.pool.QueryRow(ctx, `
|
|
SELECT c.tenant_id::text,c.subject_id::text,ua.id::text,c.public_id::text,
|
|
EXISTS(SELECT 1 FROM recording_access_grant rag WHERE rag.tenant_id=c.tenant_id AND rag.subject_id=c.subject_id AND rag.grantee_user_id=ua.id AND rag.permission='PLAY' AND rag.revoked_at IS NULL AND rag.expires_at>now()),
|
|
EXISTS(SELECT 1 FROM recording_access_grant rag WHERE rag.tenant_id=c.tenant_id AND rag.subject_id=c.subject_id AND rag.grantee_user_id=ua.id AND rag.permission='DOWNLOAD' AND rag.revoked_at IS NULL AND rag.expires_at>now())
|
|
FROM conversation c
|
|
JOIN household_member hm ON hm.household_id=c.household_id AND hm.tenant_id=c.tenant_id AND hm.status='active'
|
|
JOIN user_account ua ON ua.id=hm.user_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active'
|
|
WHERE c.public_id=$1::uuid`, conversationPublicID, principal.Issuer, principal.Subject,
|
|
).Scan(&tenantID, &subjectID, &userID, &result.ConversationID, &result.CanPlay, &result.CanDownload)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ConversationMedia{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return ConversationMedia{}, err
|
|
}
|
|
if !result.CanPlay && !result.CanDownload {
|
|
return ConversationMedia{}, ErrRecordingAccessDenied
|
|
}
|
|
var currentConsent bool
|
|
if err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 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()))`, tenantID, subjectID).Scan(¤tConsent); err != nil {
|
|
return ConversationMedia{}, err
|
|
}
|
|
if !currentConsent {
|
|
return ConversationMedia{}, ErrRecordingAccessDenied
|
|
}
|
|
rows, err := s.pool.Query(ctx, `SELECT public_id::text,track,sequence_no,codec,content_type,duration_ms,expected_size,expected_digest,status,COALESCE(last_error_code,''),verified_at,local_delete_after FROM media_segment WHERE tenant_id=$1::uuid AND conversation_id=(SELECT id FROM conversation WHERE public_id=$2::uuid) AND status='VERIFIED' ORDER BY track,sequence_no`, tenantID, conversationPublicID)
|
|
if err != nil {
|
|
return ConversationMedia{}, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var item Segment
|
|
if err := rows.Scan(&item.ID, &item.Track, &item.SequenceNo, &item.Codec, &item.ContentType, &item.DurationMS, &item.ExpectedSize, &item.ExpectedDigest, &item.Status, &item.LastErrorCode, &item.VerifiedAt, &item.LocalDeleteAfter); err != nil {
|
|
return ConversationMedia{}, err
|
|
}
|
|
result.Segments = append(result.Segments, item)
|
|
}
|
|
_ = userID
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *Service) CreateMediaAccess(ctx context.Context, principal identity.Principal, segmentPublicID, action, purpose, requestID string) (MediaAccess, error) {
|
|
action = strings.ToUpper(strings.TrimSpace(action))
|
|
purpose = strings.TrimSpace(purpose)
|
|
if (action != "PLAY" && action != "DOWNLOAD") || purpose == "" {
|
|
return MediaAccess{}, ErrInvalidDeclaration
|
|
}
|
|
var result MediaAccess
|
|
denied := false
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
var tenantID, userID, userPublicID, conversationID, conversationPublicID, segmentID, objectRef, contentType, track string
|
|
var sequenceNo int
|
|
var allowed, currentConsent bool
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT ms.tenant_id::text,ua.id::text,ua.public_id::text,c.id::text,c.public_id::text,ms.id::text,ms.object_ref,ms.content_type,ms.track,ms.sequence_no,
|
|
(hm.id IS NOT NULL AND EXISTS(SELECT 1 FROM recording_access_grant rag WHERE rag.tenant_id=ms.tenant_id AND rag.subject_id=c.subject_id AND rag.grantee_user_id=ua.id AND rag.permission=$4 AND rag.revoked_at IS NULL AND rag.expires_at>now())),
|
|
EXISTS(SELECT 1 FROM consent_record cr WHERE cr.tenant_id=ms.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()))
|
|
FROM media_segment ms JOIN conversation c ON c.id=ms.conversation_id
|
|
JOIN user_account ua ON ua.tenant_id=ms.tenant_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active'
|
|
LEFT JOIN household_member hm ON hm.household_id=c.household_id AND hm.tenant_id=c.tenant_id AND hm.user_id=ua.id AND hm.status='active'
|
|
WHERE ms.public_id=$1::uuid AND ms.status='VERIFIED'`, segmentPublicID, principal.Issuer, principal.Subject, action,
|
|
).Scan(&tenantID, &userID, &userPublicID, &conversationID, &conversationPublicID, &segmentID, &objectRef, &contentType, &track, &sequenceNo, &allowed, ¤tConsent)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !allowed || !currentConsent {
|
|
denied = true
|
|
if _, err := tx.Exec(ctx, `INSERT INTO media_access_log(tenant_id,actor_user_id,conversation_id,media_segment_id,action,purpose,result,reason_code,request_id) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5,$6,'DENIED','ACCESS_GRANT_REQUIRED',$7)`, tenantID, userID, conversationID, segmentID, action, purpose, requestID); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "user", ActorPublicID: &userPublicID, Action: "recording.access", ResourceType: "media_segment", ResourceID: &segmentPublicID, RequestID: requestID, Purpose: purpose, Result: "denied", Detail: map[string]any{"action": action}})
|
|
}
|
|
extension := ".bin"
|
|
if contentType == "audio/wav" {
|
|
extension = ".wav"
|
|
} else if contentType == "audio/opus" {
|
|
extension = ".opus"
|
|
} else if contentType == "audio/ogg" {
|
|
extension = ".ogg"
|
|
}
|
|
filename := fmt.Sprintf("%s-%s-%08d%s", conversationPublicID, track, sequenceNo, extension)
|
|
until := time.Now().UTC().Add(s.accessTTL)
|
|
url, err := s.store.PresignMediaDownload(ctx, objectRef, s.accessTTL, filename, contentType, action == "PLAY")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO media_access_log(tenant_id,actor_user_id,conversation_id,media_segment_id,action,purpose,result,request_id) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5,$6,'SUCCESS',$7)`, tenantID, userID, conversationID, segmentID, action, purpose, requestID); err != nil {
|
|
return err
|
|
}
|
|
if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "user", ActorPublicID: &userPublicID, Action: "recording.access", ResourceType: "media_segment", ResourceID: &segmentPublicID, RequestID: requestID, Purpose: purpose, Result: "success", Detail: map[string]any{"action": action}}); err != nil {
|
|
return err
|
|
}
|
|
result = MediaAccess{SegmentID: segmentPublicID, Action: action, AccessURL: url, AccessUntil: until}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return MediaAccess{}, err
|
|
}
|
|
if denied {
|
|
return MediaAccess{}, ErrRecordingAccessDenied
|
|
}
|
|
return result, nil
|
|
}
|