351 lines
17 KiB
Go
351 lines
17 KiB
Go
package media
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/audit"
|
|
"data-terminal-cloud/internal/database"
|
|
"data-terminal-cloud/internal/deviceauth"
|
|
"data-terminal-cloud/internal/outbox"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
var (
|
|
ErrAccessDenied = errors.New("media upload access denied")
|
|
ErrNotFound = errors.New("media upload batch not found")
|
|
ErrConflict = errors.New("media upload request conflicts with existing data")
|
|
ErrExpired = errors.New("media upload batch expired")
|
|
ErrInvalidDeclaration = errors.New("media segment declaration is invalid")
|
|
ErrRecordingAccessDenied = errors.New("recording access denied")
|
|
ErrLocalDeleteTooEarly = errors.New("media local deletion is not due")
|
|
)
|
|
|
|
type objectStore interface {
|
|
PresignUpload(context.Context, string, time.Duration) (string, error)
|
|
PresignMediaDownload(context.Context, string, time.Duration, string, string, bool) (string, error)
|
|
}
|
|
|
|
type SegmentDeclaration struct {
|
|
Track string `json:"track"`
|
|
SequenceNo int `json:"sequence_no"`
|
|
Codec string `json:"codec"`
|
|
ContentType string `json:"content_type"`
|
|
DurationMS int `json:"duration_ms"`
|
|
ExpectedSize int64 `json:"expected_size"`
|
|
ExpectedDigest string `json:"expected_digest"`
|
|
}
|
|
|
|
type Segment struct {
|
|
ID string `json:"id"`
|
|
Track string `json:"track"`
|
|
SequenceNo int `json:"sequence_no"`
|
|
Codec string `json:"codec"`
|
|
ContentType string `json:"content_type"`
|
|
DurationMS int `json:"duration_ms"`
|
|
ExpectedSize int64 `json:"expected_size"`
|
|
ExpectedDigest string `json:"expected_digest"`
|
|
Status string `json:"status"`
|
|
LastErrorCode string `json:"last_error_code,omitempty"`
|
|
UploadURL string `json:"upload_url,omitempty"`
|
|
UploadUntil *time.Time `json:"upload_until,omitempty"`
|
|
VerifiedAt *time.Time `json:"verified_at,omitempty"`
|
|
LocalDeleteAfter *time.Time `json:"local_delete_after,omitempty"`
|
|
LocalDeleteStatus string `json:"local_delete_status,omitempty"`
|
|
LocalDeleteReportedAt *time.Time `json:"local_delete_reported_at,omitempty"`
|
|
LocalDeleteErrorCode string `json:"local_delete_error_code,omitempty"`
|
|
}
|
|
|
|
type UploadBatch struct {
|
|
ID string `json:"id"`
|
|
ConversationID string `json:"conversation_id"`
|
|
Status string `json:"status"`
|
|
SegmentCount int `json:"segment_count"`
|
|
ExpectedTotalSize int64 `json:"expected_total_size"`
|
|
VerifiedCount int `json:"verified_count"`
|
|
RejectedCount int `json:"rejected_count"`
|
|
VerificationAttempts int `json:"verification_attempts"`
|
|
LastErrorCode string `json:"last_error_code,omitempty"`
|
|
ExpiresAt time.Time `json:"expires_at"`
|
|
Segments []Segment `json:"segments"`
|
|
}
|
|
|
|
type Service struct {
|
|
pool *pgxpool.Pool
|
|
outbox *outbox.Store
|
|
audit *audit.Repository
|
|
store objectStore
|
|
uploadTTL time.Duration
|
|
accessTTL time.Duration
|
|
}
|
|
|
|
func NewService(pool *pgxpool.Pool, outboxStore *outbox.Store, auditRepository *audit.Repository, store objectStore, uploadTTL time.Duration) (*Service, error) {
|
|
if pool == nil || outboxStore == nil || auditRepository == nil || store == nil || uploadTTL < 10*time.Minute || uploadTTL > 20*time.Minute {
|
|
return nil, errors.New("media service requires database, outbox, audit, object store and a 10-20 minute upload TTL")
|
|
}
|
|
return &Service{pool: pool, outbox: outboxStore, audit: auditRepository, store: store, uploadTTL: uploadTTL, accessTTL: 5 * time.Minute}, nil
|
|
}
|
|
|
|
func (s *Service) PrepareUploadBatch(ctx context.Context, identity deviceauth.Identity, conversationPublicID, idempotencyKey, requestID string, declarations []SegmentDeclaration) (UploadBatch, error) {
|
|
manifestDigest, normalized, totalSize, err := normalizeManifest(declarations)
|
|
if err != nil {
|
|
return UploadBatch{}, err
|
|
}
|
|
var batchPublicID string
|
|
err = database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
var conversationID, retentionMode, conversationStatus string
|
|
var currentConsent bool
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT c.id::text,c.retention_mode,c.status,
|
|
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())
|
|
)
|
|
FROM conversation c
|
|
WHERE c.public_id=$1::uuid AND c.tenant_id=$2::uuid AND c.device_id=$3::uuid`,
|
|
conversationPublicID, identity.TenantID, identity.ID,
|
|
).Scan(&conversationID, &retentionMode, &conversationStatus, ¤tConsent)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if retentionMode != "transcript_and_audio" || !currentConsent || (conversationStatus != "COMPLETED" && conversationStatus != "FAILED") {
|
|
return ErrAccessDenied
|
|
}
|
|
|
|
var storedDigest string
|
|
err = tx.QueryRow(ctx, `SELECT public_id::text,manifest_digest FROM upload_batch WHERE device_id=$1::uuid AND conversation_id=$2::uuid AND idempotency_key=$3 FOR UPDATE`, identity.ID, conversationID, idempotencyKey).Scan(&batchPublicID, &storedDigest)
|
|
if err == nil {
|
|
if storedDigest != manifestDigest {
|
|
return ErrConflict
|
|
}
|
|
return nil
|
|
}
|
|
if !errors.Is(err, pgx.ErrNoRows) {
|
|
return err
|
|
}
|
|
|
|
var batchID string
|
|
expiresAt := time.Now().UTC().Add(s.uploadTTL)
|
|
err = tx.QueryRow(ctx, `
|
|
INSERT INTO upload_batch(tenant_id,conversation_id,device_id,idempotency_key,manifest_digest,status,segment_count,expected_total_size,expires_at)
|
|
VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,'PENDING',$6,$7,$8)
|
|
RETURNING id::text,public_id::text`, identity.TenantID, conversationID, identity.ID, idempotencyKey, manifestDigest, len(normalized), totalSize, expiresAt,
|
|
).Scan(&batchID, &batchPublicID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
tenantHash := sha256.Sum256([]byte(identity.TenantID))
|
|
type manifestRow struct {
|
|
Track string `json:"track"`
|
|
SequenceNo int `json:"sequence_no"`
|
|
Codec string `json:"codec"`
|
|
ContentType string `json:"content_type"`
|
|
DurationMS int `json:"duration_ms"`
|
|
ExpectedSize int64 `json:"expected_size"`
|
|
ExpectedDigest string `json:"expected_digest"`
|
|
ObjectRef string `json:"object_ref"`
|
|
}
|
|
manifestRows := make([]manifestRow, 0, len(normalized))
|
|
for _, declaration := range normalized {
|
|
manifestRows = append(manifestRows, manifestRow{
|
|
Track: declaration.Track, SequenceNo: declaration.SequenceNo, Codec: declaration.Codec,
|
|
ContentType: declaration.ContentType, DurationMS: declaration.DurationMS,
|
|
ExpectedSize: declaration.ExpectedSize, ExpectedDigest: declaration.ExpectedDigest,
|
|
ObjectRef: fmt.Sprintf("tenants/%s/media/%s/%s/%08d-%s.bin", hex.EncodeToString(tenantHash[:16]), conversationPublicID, declaration.Track, declaration.SequenceNo, declaration.ExpectedDigest[:16]),
|
|
})
|
|
}
|
|
manifestJSON, err := json.Marshal(manifestRows)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = tx.Exec(ctx, `
|
|
INSERT INTO media_segment(tenant_id,conversation_id,device_id,track,sequence_no,codec,content_type,duration_ms,expected_size,expected_digest,object_ref,status)
|
|
SELECT $1::uuid,$2::uuid,$3::uuid,x.track,x.sequence_no,x.codec,x.content_type,x.duration_ms,x.expected_size,x.expected_digest,x.object_ref,'AUTHORIZED'
|
|
FROM jsonb_to_recordset($4::jsonb) AS x(track text,sequence_no integer,codec text,content_type text,duration_ms integer,expected_size bigint,expected_digest text,object_ref text)
|
|
ON CONFLICT (conversation_id,track,sequence_no) DO NOTHING`, identity.TenantID, conversationID, identity.ID, manifestJSON); err != nil {
|
|
return err
|
|
}
|
|
var foundCount, validCount int
|
|
if err = tx.QueryRow(ctx, `
|
|
SELECT count(*),count(*) FILTER(WHERE ms.codec=x.codec AND ms.content_type=x.content_type AND ms.duration_ms=x.duration_ms
|
|
AND ms.expected_size=x.expected_size AND ms.expected_digest=x.expected_digest AND ms.status<>'DELETED')
|
|
FROM jsonb_to_recordset($2::jsonb) AS x(track text,sequence_no integer,codec text,content_type text,duration_ms integer,expected_size bigint,expected_digest text)
|
|
JOIN media_segment ms ON ms.conversation_id=$1::uuid AND ms.track=x.track AND ms.sequence_no=x.sequence_no`, conversationID, manifestJSON).Scan(&foundCount, &validCount); err != nil {
|
|
return err
|
|
}
|
|
if foundCount != len(normalized) || validCount != len(normalized) {
|
|
return ErrConflict
|
|
}
|
|
if _, err = tx.Exec(ctx, `
|
|
UPDATE media_segment ms SET status='AUTHORIZED',last_error_code=NULL,updated_at=now()
|
|
FROM jsonb_to_recordset($2::jsonb) AS x(track text,sequence_no integer)
|
|
WHERE ms.conversation_id=$1::uuid AND ms.track=x.track AND ms.sequence_no=x.sequence_no AND ms.status IN('REJECTED','DECLARED')`, conversationID, manifestJSON); err != nil {
|
|
return err
|
|
}
|
|
linkTag, err := tx.Exec(ctx, `
|
|
INSERT INTO upload_batch_segment(upload_batch_id,media_segment_id,tenant_id)
|
|
SELECT $1::uuid,ms.id,$2::uuid
|
|
FROM jsonb_to_recordset($4::jsonb) AS x(track text,sequence_no integer)
|
|
JOIN media_segment ms ON ms.conversation_id=$3::uuid AND ms.track=x.track AND ms.sequence_no=x.sequence_no`, batchID, identity.TenantID, conversationID, manifestJSON)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if linkTag.RowsAffected() != int64(len(normalized)) {
|
|
return ErrConflict
|
|
}
|
|
if _, err = tx.Exec(ctx, `UPDATE upload_batch SET status='UPLOADING',updated_at=now() WHERE id=$1::uuid`, batchID); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: &identity.TenantID, ActorType: "device", ActorPublicID: &identity.PublicID, Action: "media.upload_batch.prepare", ResourceType: "upload_batch", ResourceID: &batchPublicID, RequestID: requestID, Result: "success", Detail: map[string]any{"conversation_id": conversationPublicID, "segment_count": len(normalized)}})
|
|
})
|
|
if err != nil {
|
|
return UploadBatch{}, err
|
|
}
|
|
return s.GetUploadBatch(ctx, identity, batchPublicID, true)
|
|
}
|
|
|
|
func (s *Service) CompleteUploadBatch(ctx context.Context, identity deviceauth.Identity, batchPublicID, requestID string) (UploadBatch, error) {
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
var batchID, status string
|
|
var verificationAttempts int
|
|
var expiresAt time.Time
|
|
err := tx.QueryRow(ctx, `SELECT id::text,status,verification_attempts,expires_at FROM upload_batch WHERE public_id=$1::uuid AND tenant_id=$2::uuid AND device_id=$3::uuid FOR UPDATE`, batchPublicID, identity.TenantID, identity.ID).Scan(&batchID, &status, &verificationAttempts, &expiresAt)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if status == "COMPLETED" || status == "VERIFYING" {
|
|
return nil
|
|
}
|
|
if time.Now().UTC().After(expiresAt) || status == "EXPIRED" {
|
|
return ErrExpired
|
|
}
|
|
if status != "UPLOADING" && status != "FAILED" {
|
|
return ErrConflict
|
|
}
|
|
if _, err = tx.Exec(ctx, `UPDATE media_segment ms SET status='VERIFYING',uploaded_at=COALESCE(uploaded_at,now()),last_error_code=NULL,updated_at=now() FROM upload_batch_segment ubs WHERE ubs.upload_batch_id=$1::uuid AND ubs.media_segment_id=ms.id AND ms.status IN ('AUTHORIZED','REJECTED')`, batchID); err != nil {
|
|
return err
|
|
}
|
|
if _, err = tx.Exec(ctx, `UPDATE upload_batch SET status='VERIFYING',last_error_code=NULL,updated_at=now() WHERE id=$1::uuid`, batchID); err != nil {
|
|
return err
|
|
}
|
|
payload, _ := json.Marshal(map[string]string{"batch_id": batchPublicID})
|
|
deduplicationKey := fmt.Sprintf("media-verify:%s:%d", batchPublicID, verificationAttempts+1)
|
|
if _, err = s.outbox.Enqueue(ctx, tx, outbox.NewEvent{TenantID: &identity.TenantID, AggregateType: "upload_batch", AggregateID: batchPublicID, EventType: "media.verify_requested", DeduplicationKey: &deduplicationKey, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: &identity.TenantID, ActorType: "device", ActorPublicID: &identity.PublicID, Action: "media.upload_batch.complete", ResourceType: "upload_batch", ResourceID: &batchPublicID, RequestID: requestID, Result: "success", Detail: map[string]any{"status": "VERIFYING"}})
|
|
})
|
|
if err != nil {
|
|
return UploadBatch{}, err
|
|
}
|
|
return s.GetUploadBatch(ctx, identity, batchPublicID, false)
|
|
}
|
|
|
|
func (s *Service) GetUploadBatch(ctx context.Context, identity deviceauth.Identity, batchPublicID string, includeUploadURLs bool) (UploadBatch, error) {
|
|
var item UploadBatch
|
|
err := s.pool.QueryRow(ctx, `
|
|
SELECT ub.public_id::text,c.public_id::text,ub.status,ub.segment_count,ub.expected_total_size,
|
|
ub.verified_count,ub.rejected_count,ub.verification_attempts,COALESCE(ub.last_error_code,''),ub.expires_at
|
|
FROM upload_batch ub JOIN conversation c ON c.id=ub.conversation_id
|
|
WHERE ub.public_id=$1::uuid AND ub.tenant_id=$2::uuid AND ub.device_id=$3::uuid`, batchPublicID, identity.TenantID, identity.ID,
|
|
).Scan(&item.ID, &item.ConversationID, &item.Status, &item.SegmentCount, &item.ExpectedTotalSize, &item.VerifiedCount, &item.RejectedCount, &item.VerificationAttempts, &item.LastErrorCode, &item.ExpiresAt)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return UploadBatch{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return UploadBatch{}, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT ms.public_id::text,ms.track,ms.sequence_no,ms.codec,ms.content_type,ms.duration_ms,ms.expected_size,
|
|
ms.expected_digest,ms.status,COALESCE(ms.last_error_code,''),ms.verified_at,ms.local_delete_after,
|
|
COALESCE(ms.local_delete_status,''),ms.local_delete_reported_at,COALESCE(ms.local_delete_error_code,''),ms.object_ref
|
|
FROM media_segment ms JOIN upload_batch_segment ubs ON ubs.media_segment_id=ms.id
|
|
JOIN upload_batch ub ON ub.id=ubs.upload_batch_id
|
|
WHERE ub.public_id=$1::uuid AND ub.tenant_id=$2::uuid ORDER BY ms.track,ms.sequence_no`, batchPublicID, identity.TenantID)
|
|
if err != nil {
|
|
return UploadBatch{}, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var segment Segment
|
|
var objectRef string
|
|
if err := rows.Scan(&segment.ID, &segment.Track, &segment.SequenceNo, &segment.Codec, &segment.ContentType, &segment.DurationMS, &segment.ExpectedSize, &segment.ExpectedDigest, &segment.Status, &segment.LastErrorCode, &segment.VerifiedAt, &segment.LocalDeleteAfter, &segment.LocalDeleteStatus, &segment.LocalDeleteReportedAt, &segment.LocalDeleteErrorCode, &objectRef); err != nil {
|
|
return UploadBatch{}, err
|
|
}
|
|
if includeUploadURLs && segment.Status == "AUTHORIZED" && time.Now().UTC().Before(item.ExpiresAt) {
|
|
url, signErr := s.store.PresignUpload(ctx, objectRef, time.Until(item.ExpiresAt))
|
|
if signErr != nil {
|
|
return UploadBatch{}, signErr
|
|
}
|
|
segment.UploadURL = url
|
|
until := item.ExpiresAt
|
|
segment.UploadUntil = &until
|
|
}
|
|
item.Segments = append(item.Segments, segment)
|
|
}
|
|
return item, rows.Err()
|
|
}
|
|
|
|
func normalizeManifest(declarations []SegmentDeclaration) (string, []SegmentDeclaration, int64, error) {
|
|
if len(declarations) == 0 || len(declarations) > 1000 {
|
|
return "", nil, 0, ErrInvalidDeclaration
|
|
}
|
|
normalized := append([]SegmentDeclaration(nil), declarations...)
|
|
sort.Slice(normalized, func(i, j int) bool {
|
|
if normalized[i].Track == normalized[j].Track {
|
|
return normalized[i].SequenceNo < normalized[j].SequenceNo
|
|
}
|
|
return normalized[i].Track < normalized[j].Track
|
|
})
|
|
seen := make(map[string]struct{}, len(normalized))
|
|
var total int64
|
|
for index := range normalized {
|
|
item := &normalized[index]
|
|
item.Track = strings.ToLower(strings.TrimSpace(item.Track))
|
|
item.Codec = strings.ToLower(strings.TrimSpace(item.Codec))
|
|
item.ContentType = strings.ToLower(strings.TrimSpace(item.ContentType))
|
|
item.ExpectedDigest = strings.ToLower(strings.TrimSpace(item.ExpectedDigest))
|
|
key := fmt.Sprintf("%s:%d", item.Track, item.SequenceNo)
|
|
_, duplicate := seen[key]
|
|
if (item.Track != "near" && item.Track != "far") || item.SequenceNo < 0 || duplicate || !validCodec(item.Codec, item.ContentType) || item.DurationMS < 1 || item.DurationMS > 600000 || item.ExpectedSize < 1 || item.ExpectedSize > 64*1024*1024 || len(item.ExpectedDigest) != 64 {
|
|
return "", nil, 0, ErrInvalidDeclaration
|
|
}
|
|
if _, err := hex.DecodeString(item.ExpectedDigest); err != nil {
|
|
return "", nil, 0, ErrInvalidDeclaration
|
|
}
|
|
seen[key] = struct{}{}
|
|
if total > 2*1024*1024*1024-item.ExpectedSize {
|
|
return "", nil, 0, ErrInvalidDeclaration
|
|
}
|
|
total += item.ExpectedSize
|
|
}
|
|
encoded, _ := json.Marshal(normalized)
|
|
digest := sha256.Sum256(encoded)
|
|
return hex.EncodeToString(digest[:]), normalized, total, nil
|
|
}
|
|
|
|
func validCodec(codec, contentType string) bool {
|
|
allowed := map[string]string{
|
|
"wav_pcm_s16le": "audio/wav",
|
|
}
|
|
return allowed[codec] == contentType
|
|
}
|