Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
245 lines
13 KiB
Go
245 lines
13 KiB
Go
package dataexport
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/audit"
|
|
"data-terminal-cloud/internal/database"
|
|
"data-terminal-cloud/internal/outbox"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type objectStore interface {
|
|
PutJSON(context.Context, string, []byte, string) (int64, error)
|
|
Remove(context.Context, string) error
|
|
}
|
|
|
|
type Processor struct {
|
|
pool *pgxpool.Pool
|
|
store objectStore
|
|
audit *audit.Repository
|
|
objectTTL time.Duration
|
|
}
|
|
|
|
func NewProcessor(pool *pgxpool.Pool, store objectStore, objectTTL time.Duration) (*Processor, error) {
|
|
if pool == nil || store == nil || objectTTL <= 0 {
|
|
return nil, errors.New("database, object store and positive object TTL are required")
|
|
}
|
|
return &Processor{pool: pool, store: store, audit: audit.NewRepository(), objectTTL: objectTTL}, nil
|
|
}
|
|
|
|
func (p *Processor) Handle(ctx context.Context, event outbox.Event) error {
|
|
if err := p.process(ctx, event.AggregateID); err != nil {
|
|
_, _ = p.pool.Exec(ctx, `UPDATE export_task SET status='FAILED',last_error='export generation failed',updated_at=now() WHERE id=$1::uuid AND status<>'EXPIRED'`, event.AggregateID)
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (p *Processor) process(ctx context.Context, taskID string) error {
|
|
var tenantID, tenantPublicID, subjectID, subjectPublicID, taskPublicID, status, subjectStatus string
|
|
var scopeJSON []byte
|
|
found := false
|
|
err := database.Transact(ctx, p.pool, func(tx pgx.Tx) error {
|
|
if err := tx.QueryRow(ctx, `SELECT et.tenant_id::text,t.public_id::text,et.subject_id::text,sp.public_id::text,et.public_id::text,et.status,sp.status,et.scope FROM export_task et JOIN tenant t ON t.id=et.tenant_id JOIN subject_profile sp ON sp.id=et.subject_id WHERE et.id=$1::uuid FOR UPDATE OF et`, taskID).Scan(&tenantID, &tenantPublicID, &subjectID, &subjectPublicID, &taskPublicID, &status, &subjectStatus, &scopeJSON); errors.Is(err, pgx.ErrNoRows) {
|
|
return nil
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
found = true
|
|
if status == "READY" || status == "EXPIRED" {
|
|
return nil
|
|
}
|
|
if subjectStatus != "active" && subjectStatus != "suspended" {
|
|
_, err := tx.Exec(ctx, `UPDATE export_task SET status='EXPIRED',scope='{}'::jsonb,object_ref=NULL,object_size=NULL,content_digest=NULL,expires_at=NULL,last_error=NULL,updated_at=now() WHERE id=$1::uuid`, taskID)
|
|
status = "EXPIRED"
|
|
return err
|
|
}
|
|
_, err := tx.Exec(ctx, `UPDATE export_task SET status='RUNNING',attempts=attempts+1,last_error=NULL,updated_at=now() WHERE id=$1::uuid`, taskID)
|
|
return err
|
|
})
|
|
if err != nil || !found || status == "READY" || status == "EXPIRED" {
|
|
return err
|
|
}
|
|
|
|
categories, err := categoriesFromScope(scopeJSON)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
data := make(map[string]json.RawMessage, len(categories))
|
|
for _, category := range categories {
|
|
value, err := p.collectCategory(ctx, category, tenantID, subjectID)
|
|
if err != nil {
|
|
return fmt.Errorf("collect export category %s: %w", category, err)
|
|
}
|
|
data[category] = value
|
|
}
|
|
document := struct {
|
|
SchemaVersion string `json:"schema_version"`
|
|
GeneratedAt time.Time `json:"generated_at"`
|
|
SubjectID string `json:"subject_id"`
|
|
Scope []string `json:"scope"`
|
|
Data map[string]json.RawMessage `json:"data"`
|
|
}{"subject-export-v1", time.Now().UTC(), subjectPublicID, categories, data}
|
|
content, err := json.MarshalIndent(document, "", " ")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
digestValue := sha256.Sum256(content)
|
|
digest := hex.EncodeToString(digestValue[:])
|
|
tenantDigest := sha256.Sum256([]byte(tenantID))
|
|
objectRef := fmt.Sprintf("tenants/%s/exports/%s/v1.json", hex.EncodeToString(tenantDigest[:16]), taskPublicID)
|
|
size, err := p.store.PutJSON(ctx, objectRef, content, digest)
|
|
if err != nil {
|
|
return outbox.Retryable(err)
|
|
}
|
|
expiresAt := time.Now().UTC().Add(p.objectTTL)
|
|
finalizeErr := database.Transact(ctx, p.pool, func(tx pgx.Tx) error {
|
|
var currentSubjectStatus string
|
|
if err := tx.QueryRow(ctx, `SELECT status FROM subject_profile WHERE tenant_id=$1::uuid AND id=$2::uuid FOR SHARE`, tenantID, subjectID).Scan(¤tSubjectStatus); err != nil {
|
|
return err
|
|
}
|
|
if currentSubjectStatus != "active" && currentSubjectStatus != "suspended" {
|
|
_, err := tx.Exec(ctx, `UPDATE export_task SET status='EXPIRED',scope='{}'::jsonb,object_ref=NULL,object_size=NULL,content_digest=NULL,expires_at=NULL,last_error=NULL,updated_at=now() WHERE id=$1::uuid`, taskID)
|
|
return err
|
|
}
|
|
tag, err := tx.Exec(ctx, `UPDATE export_task SET status='READY',object_ref=$2,object_size=$3,content_digest=$4,expires_at=$5,completed_at=now(),last_error=NULL,updated_at=now() WHERE id=$1::uuid AND status IN ('RUNNING','FAILED')`, taskID, objectRef, size, digest, expiresAt)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return errors.New("export task state changed before completion")
|
|
}
|
|
return p.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "system", Action: "export.ready", ResourceType: "export_task", ResourceID: &taskPublicID, Result: "success", Detail: map[string]any{"tenant_public_id": tenantPublicID, "object_size": size, "content_digest": digest, "expires_at": expiresAt}})
|
|
})
|
|
if finalizeErr != nil {
|
|
_ = p.store.Remove(ctx, objectRef)
|
|
return finalizeErr
|
|
}
|
|
var finalStatus string
|
|
if err := p.pool.QueryRow(ctx, `SELECT status FROM export_task WHERE id=$1::uuid`, taskID).Scan(&finalStatus); err != nil {
|
|
_ = p.store.Remove(ctx, objectRef)
|
|
return err
|
|
}
|
|
if finalStatus != "READY" {
|
|
return p.store.Remove(ctx, objectRef)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func categoriesFromScope(scopeJSON []byte) ([]string, error) {
|
|
var scope struct {
|
|
Categories []string `json:"categories"`
|
|
}
|
|
if err := json.Unmarshal(scopeJSON, &scope); err != nil {
|
|
return nil, err
|
|
}
|
|
if len(scope.Categories) == 0 {
|
|
return []string{"profile", "consents", "device_bindings", "conversations", "transcripts", "summaries", "stories", "memories"}, nil
|
|
}
|
|
return scope.Categories, nil
|
|
}
|
|
|
|
func (p *Processor) collectCategory(ctx context.Context, category, tenantID, subjectID string) (json.RawMessage, error) {
|
|
queries := map[string]string{
|
|
"profile": `SELECT jsonb_build_object('id',public_id,'preferred_name',preferred_name,'language',language,'timezone',timezone,'profile_data',profile_data,'conversation_preferences',conversation_preferences,'status',status,'version',profile_version,'created_at',created_at,'updated_at',updated_at) FROM subject_profile WHERE tenant_id=$1::uuid AND id=$2::uuid`,
|
|
"consents": `SELECT COALESCE(jsonb_agg(jsonb_build_object('id',public_id,'type',consent_type,'scope',scope,'policy_version',policy_version,'notice_digest',notice_digest,'granted_at',granted_at,'expires_at',expires_at,'withdrawn_at',withdrawn_at,'version',version) ORDER BY granted_at,id),'[]'::jsonb) FROM consent_record WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
|
|
"device_bindings": `SELECT COALESCE(jsonb_agg(jsonb_build_object('id',db.public_id,'device_id',d.public_id,'model',d.model,'bound_at',db.bound_at,'unbound_at',db.unbound_at,'reason',db.reason) ORDER BY db.bound_at,db.id),'[]'::jsonb) FROM device_binding db JOIN device d ON d.id=db.device_id AND d.tenant_id=db.tenant_id WHERE db.tenant_id=$1::uuid AND db.subject_id=$2::uuid`,
|
|
"conversations": `SELECT COALESCE(jsonb_agg(jsonb_build_object('id',public_id,'engine_mode',engine_mode,'retention_mode',retention_mode,'status',status,'started_at',started_at,'ended_at',ended_at,'stop_reason',stop_reason,'failure_code',failure_code,'created_at',created_at) ORDER BY created_at,id),'[]'::jsonb) FROM conversation WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
|
|
"transcripts": `SELECT COALESCE(jsonb_agg(jsonb_build_object('id',public_id,'conversation_id',(SELECT public_id FROM conversation WHERE id=conversation_turn.conversation_id),'sequence_no',sequence_no,'speaker',speaker,'start_at',start_at,'end_at',end_at,'final_text',final_text,'interrupted',interrupted,'source_type',source_type) ORDER BY start_at,id),'[]'::jsonb) FROM conversation_turn WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
|
|
"summaries": `SELECT COALESCE(jsonb_agg(jsonb_build_object('id',public_id,'conversation_id',(SELECT public_id FROM conversation WHERE id=conversation_summary.conversation_id),'type',summary_type,'content',content,'source_turn_ids',source_turn_ids,'generation_version',generation_version,'status',status,'created_at',created_at) ORDER BY created_at,id),'[]'::jsonb) FROM conversation_summary WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
|
|
"stories": `SELECT COALESCE(jsonb_agg(jsonb_build_object('id',sd.public_id,'conversation_id',c.public_id,'call_session_id',sd.call_session_id,'import_id',sd.import_id,'event_time',sd.event_time,'title',sd.title,'status',sd.status,'sensitivity',sd.sensitivity,'version',sd.version,'content',sr.content,'revision_no',sr.revision_no,'sources',(SELECT COALESCE(jsonb_agg(jsonb_build_object('turn_id',COALESCE(ct.public_id::text,u.utterance_id,it.id::text),'call_session_id',ss.call_session_id,'utterance_id',ss.call_utterance_id,'source_revision',ss.call_source_revision,'import_turn_id',it.id,'sequence_no',COALESCE(ct.sequence_no,it.sequence_no),'speaker',ss.source_role,'text',COALESCE(ct.final_text,u.text_content,it.content)) ORDER BY ct.sequence_no,u.source_at,u.utterance_id),'[]'::jsonb) FROM story_source ss LEFT JOIN conversation_turn ct ON ct.id=ss.conversation_turn_id LEFT JOIN call_utterance u ON u.session_id=ss.call_session_id AND u.utterance_id=ss.call_utterance_id AND u.source_revision=ss.call_source_revision LEFT JOIN memoir_import_turn it ON it.id=ss.import_turn_id WHERE ss.story_revision_id=sr.id),'facts',(SELECT COALESCE(jsonb_agg(jsonb_build_object('id',fc.public_id,'type',fc.fact_type,'content',fc.content,'basis',fc.statement_basis,'status',fc.status,'source_turn_ids',fc.source_turn_ids) ORDER BY fc.created_at,fc.id),'[]'::jsonb) FROM fact_candidate fc WHERE fc.story_id=sd.id)) ORDER BY sd.created_at,sd.id),'[]'::jsonb) FROM story_draft sd LEFT JOIN conversation c ON c.id=sd.conversation_id LEFT JOIN story_revision sr ON sr.id=sd.current_revision_id WHERE sd.tenant_id=$1::uuid AND sd.subject_id=$2::uuid AND call_story_available(sd.id)`,
|
|
"memories": `SELECT COALESCE(jsonb_agg(jsonb_build_object('id',cm.public_id,'status',cm.status,'sensitivity',cm.sensitivity,'valid_from',cm.valid_from,'valid_to',cm.valid_to,'version',cm.version,'memory_version',mv.version_no,'content',mv.content,'source_refs',mv.source_refs,'confirmed_at',mv.confirmed_at,'change_reason',mv.change_reason) ORDER BY cm.created_at,cm.id),'[]'::jsonb) FROM curated_memory cm LEFT JOIN memory_version mv ON mv.id=cm.current_version_id WHERE cm.tenant_id=$1::uuid AND cm.subject_id=$2::uuid AND cm.status<>'DELETED'`,
|
|
}
|
|
query, ok := queries[category]
|
|
if !ok {
|
|
return nil, fmt.Errorf("unsupported category %q", category)
|
|
}
|
|
var value []byte
|
|
if err := p.pool.QueryRow(ctx, query, tenantID, subjectID).Scan(&value); err != nil {
|
|
return nil, err
|
|
}
|
|
return json.RawMessage(value), nil
|
|
}
|
|
|
|
type ExpiryScanner struct {
|
|
pool *pgxpool.Pool
|
|
store objectStore
|
|
logger *slog.Logger
|
|
interval time.Duration
|
|
}
|
|
|
|
func NewExpiryScanner(pool *pgxpool.Pool, store objectStore, logger *slog.Logger, interval time.Duration) (*ExpiryScanner, error) {
|
|
if pool == nil || store == nil || logger == nil || interval <= 0 {
|
|
return nil, errors.New("database, object store, logger and interval are required")
|
|
}
|
|
return &ExpiryScanner{pool: pool, store: store, logger: logger, interval: interval}, nil
|
|
}
|
|
|
|
func (s *ExpiryScanner) Run(ctx context.Context) {
|
|
ticker := time.NewTicker(s.interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
s.scan(ctx)
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *ExpiryScanner) scan(ctx context.Context) {
|
|
rows, err := s.pool.Query(ctx, `SELECT id::text,tenant_id::text,public_id::text,object_ref FROM export_task WHERE status='READY' AND expires_at<=now() ORDER BY expires_at LIMIT 50`)
|
|
if err != nil {
|
|
s.logger.Error("scan expired exports failed", "error", err)
|
|
return
|
|
}
|
|
type item struct{ id, tenantID, publicID, objectRef string }
|
|
items := []item{}
|
|
for rows.Next() {
|
|
var value item
|
|
if err := rows.Scan(&value.id, &value.tenantID, &value.publicID, &value.objectRef); err != nil {
|
|
rows.Close()
|
|
s.logger.Error("scan expired export row failed", "error", err)
|
|
return
|
|
}
|
|
items = append(items, value)
|
|
}
|
|
rows.Close()
|
|
for _, value := range items {
|
|
if err := s.store.Remove(ctx, value.objectRef); err != nil {
|
|
s.logger.Warn("remove expired export failed", "export_id", value.publicID, "error", err)
|
|
continue
|
|
}
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
tag, err := tx.Exec(ctx, `UPDATE export_task SET status='EXPIRED',object_ref=NULL,updated_at=now() WHERE id=$1::uuid AND status='READY' AND expires_at<=now()`, value.id)
|
|
if err != nil || tag.RowsAffected() != 1 {
|
|
return err
|
|
}
|
|
return audit.NewRepository().Record(ctx, tx, audit.Event{TenantID: &value.tenantID, ActorType: "system", Action: "export.expire", ResourceType: "export_task", ResourceID: &value.publicID, Result: "success"})
|
|
})
|
|
if err != nil {
|
|
s.logger.Error("mark export expired failed", "export_id", value.publicID, "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func sanitizeError(value string) string {
|
|
value = strings.TrimSpace(value)
|
|
if len(value) > 500 {
|
|
return value[:500]
|
|
}
|
|
return value
|
|
}
|