186 lines
8.0 KiB
Go
186 lines
8.0 KiB
Go
|
package saas
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"crypto/sha256"
|
||
|
|
"encoding/hex"
|
||
|
|
"encoding/json"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"strings"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"data-terminal-cloud/internal/audit"
|
||
|
|
"data-terminal-cloud/internal/database"
|
||
|
|
"data-terminal-cloud/internal/identity"
|
||
|
|
"data-terminal-cloud/internal/outbox"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
)
|
||
|
|
|
||
|
|
var ErrExportUnavailable = errors.New("export download is not available")
|
||
|
|
|
||
|
|
type exportURLSigner interface {
|
||
|
|
PresignDownload(context.Context, string, time.Duration, string) (string, error)
|
||
|
|
}
|
||
|
|
|
||
|
|
type ExportTask struct {
|
||
|
|
ID string `json:"id"`
|
||
|
|
SubjectID string `json:"subject_id"`
|
||
|
|
Scope map[string]any `json:"scope"`
|
||
|
|
Status string `json:"status"`
|
||
|
|
ObjectSize *int64 `json:"object_size,omitempty"`
|
||
|
|
ContentDigest string `json:"content_digest,omitempty"`
|
||
|
|
ExpiresAt *time.Time `json:"expires_at,omitempty"`
|
||
|
|
Attempts int `json:"attempts"`
|
||
|
|
LastError string `json:"last_error,omitempty"`
|
||
|
|
CreatedAt time.Time `json:"created_at"`
|
||
|
|
UpdatedAt time.Time `json:"updated_at"`
|
||
|
|
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
||
|
|
DownloadURL string `json:"download_url,omitempty"`
|
||
|
|
DownloadUntil *time.Time `json:"download_until,omitempty"`
|
||
|
|
objectRef string
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) ConfigureExportSigning(signer exportURLSigner, downloadTTL time.Duration) {
|
||
|
|
s.exportSigner = signer
|
||
|
|
s.exportDownloadTTL = downloadTTL
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) CreateExportTask(ctx context.Context, principal identity.Principal, subjectPublicID, idempotencyKey, requestID string, scope map[string]any) (ExportTask, bool, error) {
|
||
|
|
if scope == nil {
|
||
|
|
scope = map[string]any{}
|
||
|
|
}
|
||
|
|
requestBytes, err := json.Marshal(scope)
|
||
|
|
if err != nil {
|
||
|
|
return ExportTask{}, false, err
|
||
|
|
}
|
||
|
|
digestValue := sha256.Sum256(requestBytes)
|
||
|
|
digest := hex.EncodeToString(digestValue[:])
|
||
|
|
created := false
|
||
|
|
var publicID string
|
||
|
|
err = database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
||
|
|
access, _, err := resolveSubject(ctx, tx, principal, subjectPublicID, identity.CapabilityPrivacyManage, true)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
scopeJSON, _ := json.Marshal(scope)
|
||
|
|
var internalID string
|
||
|
|
err = tx.QueryRow(ctx, `INSERT INTO export_task(tenant_id,subject_id,requested_by,scope,idempotency_key,request_digest) VALUES($1::uuid,$2::uuid,$3::uuid,$4::jsonb,$5,$6) ON CONFLICT (tenant_id,requested_by,idempotency_key) DO NOTHING RETURNING id::text,public_id::text`, access.tenantID, access.subjectID, access.userID, scopeJSON, idempotencyKey, digest).Scan(&internalID, &publicID)
|
||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
var storedDigest string
|
||
|
|
if err := tx.QueryRow(ctx, `SELECT public_id::text,request_digest FROM export_task WHERE tenant_id=$1::uuid AND requested_by=$2::uuid AND idempotency_key=$3 FOR UPDATE`, access.tenantID, access.userID, idempotencyKey).Scan(&publicID, &storedDigest); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if storedDigest != digest {
|
||
|
|
return ErrIdempotencyConflict
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("create export task: %w", err)
|
||
|
|
}
|
||
|
|
created = true
|
||
|
|
dedupe := "export-generate:" + internalID
|
||
|
|
payload, _ := json.Marshal(map[string]any{"export_task_id": internalID})
|
||
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &access.tenantID, AggregateType: "export_task", AggregateID: internalID, EventType: "export.generate_requested", DeduplicationKey: &dedupe, Payload: payload}); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID), Action: "export.request", ResourceType: "export_task", ResourceID: strPtr(publicID), RequestID: requestID, Result: "success", Detail: map[string]any{"subject_id": subjectPublicID, "scope": scope}})
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
return ExportTask{}, false, err
|
||
|
|
}
|
||
|
|
result, err := s.GetExportTask(ctx, principal, publicID)
|
||
|
|
return result, created, err
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) ListExportTasks(ctx context.Context, principal identity.Principal, subjectPublicID string) ([]ExportTask, error) {
|
||
|
|
access, _, err := resolveSubject(ctx, s.pool, principal, subjectPublicID, identity.CapabilityPrivacyRead, false)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
rows, err := s.pool.Query(ctx, exportTaskSelectSQL+` WHERE et.tenant_id=$1::uuid AND et.subject_id=$2::uuid ORDER BY et.created_at DESC,et.id DESC`, access.tenantID, access.subjectID)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer rows.Close()
|
||
|
|
items := make([]ExportTask, 0)
|
||
|
|
for rows.Next() {
|
||
|
|
item, err := scanExportTask(rows)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
items = append(items, item)
|
||
|
|
}
|
||
|
|
return items, rows.Err()
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) GetExportTask(ctx context.Context, principal identity.Principal, publicID string) (ExportTask, error) {
|
||
|
|
row := s.pool.QueryRow(ctx, exportTaskSelectSQL+` JOIN household h ON h.id=sp.household_id AND h.tenant_id=sp.tenant_id AND h.status<>'deleted' JOIN household_member hm ON hm.household_id=h.id AND hm.tenant_id=h.tenant_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=h.tenant_id AND ua.status='active' WHERE et.public_id=$1::uuid AND sp.status IN ('active','suspended') AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, publicID, principal.Issuer, principal.Subject)
|
||
|
|
item, err := scanExportTask(row)
|
||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
return ExportTask{}, ErrAccessDenied
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return ExportTask{}, err
|
||
|
|
}
|
||
|
|
var role string
|
||
|
|
if err := s.pool.QueryRow(ctx, `SELECT hm.role FROM export_task et JOIN subject_profile sp ON sp.id=et.subject_id AND sp.status IN ('active','suspended') JOIN household_member hm ON hm.household_id=sp.household_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=et.tenant_id AND ua.status='active' WHERE et.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, publicID, principal.Issuer, principal.Subject).Scan(&role); err != nil || !identity.Can(identity.Role(role), identity.CapabilityPrivacyRead) {
|
||
|
|
return ExportTask{}, ErrAccessDenied
|
||
|
|
}
|
||
|
|
if item.Status == "READY" && item.ExpiresAt != nil {
|
||
|
|
if s.exportSigner == nil || s.exportDownloadTTL <= 0 {
|
||
|
|
return ExportTask{}, ErrExportUnavailable
|
||
|
|
}
|
||
|
|
ttl := min(s.exportDownloadTTL, time.Until(*item.ExpiresAt))
|
||
|
|
if ttl <= 0 {
|
||
|
|
item.Status = "EXPIRED"
|
||
|
|
return item, nil
|
||
|
|
}
|
||
|
|
url, err := s.exportSigner.PresignDownload(ctx, item.objectRef, ttl, "subject-data-"+item.ID+".json")
|
||
|
|
if err != nil {
|
||
|
|
return ExportTask{}, fmt.Errorf("presign export download: %w", err)
|
||
|
|
}
|
||
|
|
until := time.Now().Add(ttl)
|
||
|
|
item.DownloadURL = url
|
||
|
|
item.DownloadUntil = &until
|
||
|
|
}
|
||
|
|
return item, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
const exportTaskSelectSQL = `SELECT et.public_id::text,sp.public_id::text,et.scope,et.status,et.object_ref,et.object_size,COALESCE(et.content_digest,''),et.expires_at,et.attempts,COALESCE(et.last_error,''),et.created_at,et.updated_at,et.completed_at FROM export_task et JOIN subject_profile sp ON sp.id=et.subject_id AND sp.tenant_id=et.tenant_id`
|
||
|
|
|
||
|
|
func scanExportTask(row rowScanner) (ExportTask, error) {
|
||
|
|
var item ExportTask
|
||
|
|
var scopeJSON []byte
|
||
|
|
var objectRef *string
|
||
|
|
if err := row.Scan(&item.ID, &item.SubjectID, &scopeJSON, &item.Status, &objectRef, &item.ObjectSize, &item.ContentDigest, &item.ExpiresAt, &item.Attempts, &item.LastError, &item.CreatedAt, &item.UpdatedAt, &item.CompletedAt); err != nil {
|
||
|
|
return ExportTask{}, err
|
||
|
|
}
|
||
|
|
if objectRef != nil {
|
||
|
|
item.objectRef = *objectRef
|
||
|
|
}
|
||
|
|
if err := decodeObject(scopeJSON, &item.Scope); err != nil {
|
||
|
|
return ExportTask{}, err
|
||
|
|
}
|
||
|
|
return item, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func NormalizeExportCategories(values []string) ([]string, error) {
|
||
|
|
allowed := map[string]bool{"profile": true, "consents": true, "device_bindings": true, "conversations": true, "transcripts": true, "summaries": true, "stories": true, "memories": true}
|
||
|
|
seen := map[string]bool{}
|
||
|
|
result := make([]string, 0, len(values))
|
||
|
|
for _, raw := range values {
|
||
|
|
value := strings.TrimSpace(raw)
|
||
|
|
if !allowed[value] {
|
||
|
|
return nil, fmt.Errorf("unsupported export category %q", value)
|
||
|
|
}
|
||
|
|
if !seen[value] {
|
||
|
|
seen[value] = true
|
||
|
|
result = append(result, value)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return result, nil
|
||
|
|
}
|