package opsaccess import ( "context" "encoding/json" "errors" "fmt" "sort" "strings" "time" "data-terminal-cloud/internal/audit" "data-terminal-cloud/internal/database" "data-terminal-cloud/internal/idempotency" "data-terminal-cloud/internal/identity" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" ) var ( ErrInvalid = errors.New("temporary access request is invalid") ErrAccessDenied = errors.New("temporary access is denied") ErrNotFound = errors.New("temporary access request was not found") ErrConflict = errors.New("temporary access request conflicts with current state") ) const ( CapabilityTranscript = "TRANSCRIPT_READ" CapabilityRecording = "RECORDING_PLAY" ) type objectStore interface { PresignMediaDownload(context.Context, string, time.Duration, string, string, bool) (string, error) } type Request struct { ID string `json:"id"` ConversationID string `json:"conversation_id"` RequesterUserID string `json:"requester_user_id"` Capabilities []string `json:"capabilities"` Reason string `json:"reason"` Status string `json:"status"` RequestedExpiresAt time.Time `json:"requested_expires_at"` ReviewedBy string `json:"reviewed_by,omitempty"` ReviewReason string `json:"review_reason,omitempty"` ReviewedAt *time.Time `json:"reviewed_at,omitempty"` RevokedAt *time.Time `json:"revoked_at,omitempty"` Version int64 `json:"version"` CreatedAt time.Time `json:"created_at"` } type Turn struct { ID string `json:"id"` SequenceNo int64 `json:"sequence_no"` Speaker string `json:"speaker"` StartAt time.Time `json:"start_at"` EndAt time.Time `json:"end_at"` FinalText string `json:"final_text"` Interrupted bool `json:"interrupted"` } type Segment struct { ID string `json:"id"` Track string `json:"track"` SequenceNo int `json:"sequence_no"` ContentType string `json:"content_type"` DurationMS int `json:"duration_ms"` } type SensitiveConversation struct { ConversationID string `json:"conversation_id"` GrantID string `json:"grant_id"` AccessUntil time.Time `json:"access_until"` Turns []Turn `json:"turns"` Segments []Segment `json:"segments"` } type RecordingAccess struct { SegmentID string `json:"segment_id"` AccessURL string `json:"access_url"` AccessUntil time.Time `json:"access_until"` } type Service struct { pool *pgxpool.Pool idem *idempotency.Executor audit *audit.Repository store objectStore accessTTL time.Duration now func() time.Time } func NewService(pool *pgxpool.Pool, idem *idempotency.Executor, auditRepository *audit.Repository, store objectStore, accessTTL time.Duration) (*Service, error) { if pool == nil || idem == nil || auditRepository == nil || store == nil || accessTTL <= 0 || accessTTL > 15*time.Minute { return nil, errors.New("temporary access dependencies and a short positive access TTL are required") } return &Service{pool: pool, idem: idem, audit: auditRepository, store: store, accessTTL: accessTTL, now: time.Now}, nil } type actor struct{ tenantID, userID, publicID, role string } func (s *Service) resolveActor(ctx context.Context, principal identity.Principal, tenantPublicID string, roles ...string) (actor, error) { if _, err := uuid.Parse(tenantPublicID); err != nil { return actor{}, ErrInvalid } var a actor err := s.pool.QueryRow(ctx, `SELECT t.id::text,ua.id::text,ua.public_id::text,ua.platform_role FROM tenant t JOIN user_account ua ON ua.tenant_id=t.id AND ua.status='active' AND ua.account_kind='operator' WHERE t.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, tenantPublicID, principal.Issuer, principal.Subject).Scan(&a.tenantID, &a.userID, &a.publicID, &a.role) if errors.Is(err, pgx.ErrNoRows) { return actor{}, ErrAccessDenied } if err != nil { return actor{}, err } for _, role := range roles { if a.role == role { return a, nil } } return actor{}, ErrAccessDenied } func normalizeCapabilities(values []string) ([]string, error) { seen := map[string]bool{} for _, value := range values { value = strings.ToUpper(strings.TrimSpace(value)) if value != CapabilityTranscript && value != CapabilityRecording { return nil, ErrInvalid } seen[value] = true } result := make([]string, 0, len(seen)) for value := range seen { result = append(result, value) } sort.Strings(result) if len(result) == 0 { return nil, ErrInvalid } return result, nil } func (s *Service) CreateRequest(ctx context.Context, principal identity.Principal, tenantPublicID, conversationPublicID string, capabilities []string, reason string, expiresAt time.Time, key, requestID string) (Request, bool, error) { reason = strings.TrimSpace(reason) capabilities, err := normalizeCapabilities(capabilities) now := s.now().UTC() if err != nil || reason == "" || len([]rune(reason)) > 500 || expiresAt.Before(now.Add(5*time.Minute)) || expiresAt.After(now.Add(4*time.Hour)) { return Request{}, false, ErrInvalid } if _, err = uuid.Parse(conversationPublicID); err != nil { return Request{}, false, ErrInvalid } a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin") if err != nil { return Request{}, false, err } canonical, _ := json.Marshal(map[string]any{"conversation_id": conversationPublicID, "capabilities": capabilities, "reason": reason, "expires_at": expiresAt.UTC()}) result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.publicID, Scope: "POST /api/v1/ops/sensitive-access-requests", Key: key, RequestHash: idempotency.Hash(canonical), ExpiresAfter: 24 * time.Hour}, func(op context.Context, tx pgx.Tx) (idempotency.Result, error) { var conversationID string if queryErr := tx.QueryRow(op, `SELECT id::text FROM conversation WHERE tenant_id=$1::uuid AND public_id=$2::uuid`, a.tenantID, conversationPublicID).Scan(&conversationID); errors.Is(queryErr, pgx.ErrNoRows) { return idempotency.Result{}, ErrNotFound } else if queryErr != nil { return idempotency.Result{}, queryErr } var item Request insertErr := tx.QueryRow(op, `INSERT INTO ops_sensitive_access_request(tenant_id,conversation_id,requester_user_id,capabilities,reason,requested_expires_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,$6) RETURNING public_id::text,$7::text,$8::text,capabilities,reason,status,requested_expires_at,version,created_at`, a.tenantID, conversationID, a.userID, capabilities, reason, expiresAt.UTC(), conversationPublicID, a.publicID).Scan(&item.ID, &item.ConversationID, &item.RequesterUserID, &item.Capabilities, &item.Reason, &item.Status, &item.RequestedExpiresAt, &item.Version, &item.CreatedAt) var pgErr *pgconn.PgError if errors.As(insertErr, &pgErr) && pgErr.Code == "23505" { return idempotency.Result{}, ErrConflict } if insertErr != nil { return idempotency.Result{}, insertErr } if auditErr := s.audit.Record(op, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.publicID, Action: "ops.sensitive_access.request", ResourceType: "ops_sensitive_access_request", ResourceID: &item.ID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"conversation_id": conversationPublicID, "capabilities": capabilities, "expires_at": expiresAt.UTC()}}); auditErr != nil { return idempotency.Result{}, auditErr } encoded, _ := json.Marshal(item) return idempotency.Result{StatusCode: 201, Body: encoded, ResourceID: &item.ID}, nil }) if err != nil { return Request{}, false, err } var item Request if err = json.Unmarshal(result.Body, &item); err != nil { return Request{}, false, err } return item, !result.Replayed, nil } func (s *Service) ListRequests(ctx context.Context, principal identity.Principal, tenantPublicID string) ([]Request, error) { a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin") if err != nil { return nil, err } if _, err = s.pool.Exec(ctx, `UPDATE ops_sensitive_access_request SET status='EXPIRED',version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND status IN('PENDING','APPROVED') AND requested_expires_at<=now()`, a.tenantID); err != nil { return nil, err } filter := " AND r.requester_user_id=$2::uuid" args := []any{a.tenantID, a.userID} if a.role == "ops_approver" || a.role == "ops_admin" { filter, args = "", []any{a.tenantID} } rows, err := s.pool.Query(ctx, `SELECT r.public_id::text,c.public_id::text,requester.public_id::text,r.capabilities,r.reason,r.status,r.requested_expires_at,COALESCE(reviewer.public_id::text,''),COALESCE(r.review_reason,''),r.reviewed_at,r.revoked_at,r.version,r.created_at FROM ops_sensitive_access_request r JOIN conversation c ON c.id=r.conversation_id JOIN user_account requester ON requester.id=r.requester_user_id LEFT JOIN user_account reviewer ON reviewer.id=r.reviewed_by WHERE r.tenant_id=$1::uuid`+filter+` ORDER BY r.created_at DESC LIMIT 200`, args...) if err != nil { return nil, err } defer rows.Close() items := []Request{} for rows.Next() { var item Request if err := rows.Scan(&item.ID, &item.ConversationID, &item.RequesterUserID, &item.Capabilities, &item.Reason, &item.Status, &item.RequestedExpiresAt, &item.ReviewedBy, &item.ReviewReason, &item.ReviewedAt, &item.RevokedAt, &item.Version, &item.CreatedAt); err != nil { return nil, err } items = append(items, item) } return items, rows.Err() } func (s *Service) ReviewRequest(ctx context.Context, principal identity.Principal, tenantPublicID, accessPublicID, decision, reason, key, requestID string) (Request, bool, error) { decision, reason = strings.ToUpper(strings.TrimSpace(decision)), strings.TrimSpace(reason) if _, err := uuid.Parse(accessPublicID); err != nil || (decision != "APPROVE" && decision != "REJECT") || reason == "" || len([]rune(reason)) > 500 { return Request{}, false, ErrInvalid } a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_approver", "ops_admin") if err != nil { return Request{}, false, err } body, _ := json.Marshal(map[string]string{"id": accessPublicID, "decision": decision, "reason": reason}) result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.publicID, Scope: "POST /api/v1/ops/sensitive-access-requests/review", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(op context.Context, tx pgx.Tx) (idempotency.Result, error) { var internalID, requesterID, conversationPublicID string var item Request if queryErr := tx.QueryRow(op, `SELECT r.id::text,r.requester_user_id::text,c.public_id::text,r.status,r.requested_expires_at FROM ops_sensitive_access_request r JOIN conversation c ON c.id=r.conversation_id WHERE r.tenant_id=$1::uuid AND r.public_id=$2::uuid FOR UPDATE OF r`, a.tenantID, accessPublicID).Scan(&internalID, &requesterID, &conversationPublicID, &item.Status, &item.RequestedExpiresAt); errors.Is(queryErr, pgx.ErrNoRows) { return idempotency.Result{}, ErrNotFound } else if queryErr != nil { return idempotency.Result{}, queryErr } if requesterID == a.userID || item.Status != "PENDING" || !item.RequestedExpiresAt.After(s.now().UTC()) { return idempotency.Result{}, ErrConflict } status := map[string]string{"APPROVE": "APPROVED", "REJECT": "REJECTED"}[decision] if updateErr := tx.QueryRow(op, `UPDATE ops_sensitive_access_request SET status=$2,reviewed_by=$3::uuid,review_reason=$4,reviewed_at=now(),version=version+1,updated_at=now() WHERE id=$1::uuid RETURNING public_id::text,$5::text,(SELECT public_id::text FROM user_account WHERE id=requester_user_id),capabilities,reason,status,requested_expires_at,$6::text,review_reason,reviewed_at,revoked_at,version,created_at`, internalID, status, a.userID, reason, conversationPublicID, a.publicID).Scan(&item.ID, &item.ConversationID, &item.RequesterUserID, &item.Capabilities, &item.Reason, &item.Status, &item.RequestedExpiresAt, &item.ReviewedBy, &item.ReviewReason, &item.ReviewedAt, &item.RevokedAt, &item.Version, &item.CreatedAt); updateErr != nil { return idempotency.Result{}, updateErr } if auditErr := s.audit.Record(op, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.publicID, Action: "ops.sensitive_access.review", ResourceType: "ops_sensitive_access_request", ResourceID: &accessPublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"decision": decision, "requester_user_id": item.RequesterUserID, "conversation_id": conversationPublicID, "expires_at": item.RequestedExpiresAt}}); auditErr != nil { return idempotency.Result{}, auditErr } encoded, _ := json.Marshal(item) return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &internalID}, nil }) if err != nil { return Request{}, false, err } var item Request if err = json.Unmarshal(result.Body, &item); err != nil { return Request{}, false, err } return item, !result.Replayed, nil } type grant struct { requestID, tenantID, actorID, actorPublicID, requesterID, conversationID, conversationPublicID, status, reason string capabilities []string expiresAt time.Time } func (s *Service) loadGrant(ctx context.Context, tx pgx.Tx, principal identity.Principal, tenantPublicID, grantPublicID, conversationPublicID string) (grant, error) { var g grant err := tx.QueryRow(ctx, `SELECT r.id::text,r.tenant_id::text,ua.id::text,ua.public_id::text,r.requester_user_id::text,r.conversation_id::text,c.public_id::text,r.status,r.reason,r.capabilities,r.requested_expires_at FROM ops_sensitive_access_request r JOIN tenant t ON t.id=r.tenant_id JOIN conversation c ON c.id=r.conversation_id JOIN user_account ua ON ua.tenant_id=r.tenant_id AND ua.status='active' AND ua.account_kind='operator' AND ua.oidc_issuer=$3 AND ua.oidc_subject=$4 WHERE t.public_id=$1::uuid AND r.public_id=$2::uuid AND c.public_id=$5::uuid FOR UPDATE OF r`, tenantPublicID, grantPublicID, principal.Issuer, principal.Subject, conversationPublicID).Scan(&g.requestID, &g.tenantID, &g.actorID, &g.actorPublicID, &g.requesterID, &g.conversationID, &g.conversationPublicID, &g.status, &g.reason, &g.capabilities, &g.expiresAt) if errors.Is(err, pgx.ErrNoRows) { return grant{}, ErrAccessDenied } return g, err } func hasCapability(values []string, target string) bool { for _, value := range values { if value == target { return true } } return false } func (s *Service) authorize(ctx context.Context, tx pgx.Tx, g grant, capability, requestID string, segmentID *string, segmentPublicID string, consent bool) (bool, error) { reasonCode := "" if g.actorID != g.requesterID { reasonCode = "REQUESTER_MISMATCH" } else if g.status != "APPROVED" { reasonCode = "GRANT_NOT_APPROVED" } else if !g.expiresAt.After(s.now().UTC()) { reasonCode = "GRANT_EXPIRED" if _, err := tx.Exec(ctx, `UPDATE ops_sensitive_access_request SET status='EXPIRED',version=version+1,updated_at=now() WHERE id=$1::uuid AND status='APPROVED'`, g.requestID); err != nil { return false, err } } else if !hasCapability(g.capabilities, capability) { reasonCode = "CAPABILITY_NOT_GRANTED" } else if !consent { reasonCode = "SOURCE_CONSENT_REQUIRED" } result := "SUCCESS" if reasonCode != "" { result = "DENIED" } var mediaID any if segmentID != nil { mediaID = *segmentID } if _, err := tx.Exec(ctx, `INSERT INTO ops_sensitive_access_log(tenant_id,access_request_id,actor_user_id,conversation_id,media_segment_id,capability,result,reason_code,request_id) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5::uuid,$6,$7,NULLIF($8,''),$9)`, g.tenantID, g.requestID, g.actorID, g.conversationID, mediaID, capability, result, reasonCode, requestID); err != nil { return false, err } action := "ops.sensitive_access.transcript" resourceType, resourceID := "conversation", g.conversationPublicID if capability == CapabilityRecording { action, resourceType = "ops.sensitive_access.recording", "media_segment" if segmentPublicID != "" { resourceID = segmentPublicID } } if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &g.tenantID, ActorType: "user", ActorPublicID: &g.actorPublicID, Action: action, ResourceType: resourceType, ResourceID: &resourceID, RequestID: requestID, Purpose: g.reason, Result: strings.ToLower(result), Detail: map[string]any{"access_request_id": g.requestID, "capability": capability, "reason_code": reasonCode}}); err != nil { return false, err } return reasonCode == "", nil } func (s *Service) GetSensitiveConversation(ctx context.Context, principal identity.Principal, tenantPublicID, conversationPublicID, grantPublicID, requestID string) (SensitiveConversation, error) { if _, err := uuid.Parse(conversationPublicID); err != nil { return SensitiveConversation{}, ErrInvalid } if _, err := uuid.Parse(grantPublicID); err != nil { return SensitiveConversation{}, ErrInvalid } if _, err := uuid.Parse(tenantPublicID); err != nil { return SensitiveConversation{}, ErrInvalid } denied := false var result SensitiveConversation err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error { g, err := s.loadGrant(ctx, tx, principal, tenantPublicID, grantPublicID, conversationPublicID) if err != nil { return err } allowed, err := s.authorize(ctx, tx, g, CapabilityTranscript, requestID, nil, "", true) if err != nil { return err } if !allowed { denied = true return nil } result = SensitiveConversation{ConversationID: conversationPublicID, GrantID: grantPublicID, AccessUntil: g.expiresAt, Turns: []Turn{}, Segments: []Segment{}} rows, err := tx.Query(ctx, `SELECT public_id::text,sequence_no,speaker,start_at,end_at,final_text,interrupted FROM conversation_turn WHERE tenant_id=$1::uuid AND conversation_id=$2::uuid ORDER BY sequence_no,id`, g.tenantID, g.conversationID) if err != nil { return err } for rows.Next() { var item Turn if err := rows.Scan(&item.ID, &item.SequenceNo, &item.Speaker, &item.StartAt, &item.EndAt, &item.FinalText, &item.Interrupted); err != nil { rows.Close() return err } result.Turns = append(result.Turns, item) } if err := rows.Err(); err != nil { rows.Close() return err } rows.Close() if !hasCapability(g.capabilities, CapabilityRecording) { return nil } segmentRows, err := tx.Query(ctx, `SELECT public_id::text,track,sequence_no,content_type,duration_ms FROM media_segment WHERE tenant_id=$1::uuid AND conversation_id=$2::uuid AND status='VERIFIED' ORDER BY track,sequence_no`, g.tenantID, g.conversationID) if err != nil { return err } defer segmentRows.Close() for segmentRows.Next() { var item Segment if err := segmentRows.Scan(&item.ID, &item.Track, &item.SequenceNo, &item.ContentType, &item.DurationMS); err != nil { return err } result.Segments = append(result.Segments, item) } return segmentRows.Err() }) if err != nil { return SensitiveConversation{}, err } if denied { return SensitiveConversation{}, ErrAccessDenied } return result, nil } func (s *Service) CreateRecordingAccess(ctx context.Context, principal identity.Principal, tenantPublicID, segmentPublicID, grantPublicID, requestID string) (RecordingAccess, error) { if _, err := uuid.Parse(segmentPublicID); err != nil { return RecordingAccess{}, ErrInvalid } if _, err := uuid.Parse(grantPublicID); err != nil { return RecordingAccess{}, ErrInvalid } if _, err := uuid.Parse(tenantPublicID); err != nil { return RecordingAccess{}, ErrInvalid } denied := false var result RecordingAccess err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error { var conversationPublicID string if err := tx.QueryRow(ctx, `SELECT c.public_id::text FROM media_segment ms JOIN conversation c ON c.id=ms.conversation_id JOIN tenant t ON t.id=ms.tenant_id WHERE ms.public_id=$1::uuid AND t.public_id=$2::uuid`, segmentPublicID, tenantPublicID).Scan(&conversationPublicID); errors.Is(err, pgx.ErrNoRows) { return ErrNotFound } else if err != nil { return err } g, err := s.loadGrant(ctx, tx, principal, tenantPublicID, grantPublicID, conversationPublicID) if err != nil { return err } var segmentID, objectRef, contentType, track string var sequenceNo int var currentConsent bool if err := tx.QueryRow(ctx, `SELECT ms.id::text,ms.object_ref,ms.content_type,ms.track,ms.sequence_no,EXISTS(SELECT 1 FROM consent_record cr JOIN conversation c ON c.subject_id=cr.subject_id AND c.tenant_id=cr.tenant_id WHERE c.id=ms.conversation_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 WHERE ms.public_id=$1::uuid AND ms.tenant_id=$2::uuid AND ms.conversation_id=$3::uuid AND ms.status='VERIFIED'`, segmentPublicID, g.tenantID, g.conversationID).Scan(&segmentID, &objectRef, &contentType, &track, &sequenceNo, ¤tConsent); errors.Is(err, pgx.ErrNoRows) { return ErrNotFound } else if err != nil { return err } allowed, err := s.authorize(ctx, tx, g, CapabilityRecording, requestID, &segmentID, segmentPublicID, currentConsent) if err != nil { return err } if !allowed { denied = true return nil } 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("ops-%s-%s-%08d%s", conversationPublicID, track, sequenceNo, extension) url, err := s.store.PresignMediaDownload(ctx, objectRef, s.accessTTL, filename, contentType, true) if err != nil { return err } result = RecordingAccess{SegmentID: segmentPublicID, AccessURL: url, AccessUntil: s.now().UTC().Add(s.accessTTL)} return nil }) if err != nil { return RecordingAccess{}, err } if denied { return RecordingAccess{}, ErrAccessDenied } return result, nil }