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.
899 lines
38 KiB
Go
899 lines
38 KiB
Go
package saas
|
|
|
|
import (
|
|
"context"
|
|
"data-terminal-cloud/internal/devicefleet"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/audit"
|
|
"data-terminal-cloud/internal/concurrency"
|
|
"data-terminal-cloud/internal/database"
|
|
"data-terminal-cloud/internal/elderagent"
|
|
"data-terminal-cloud/internal/identity"
|
|
"data-terminal-cloud/internal/outbox"
|
|
"data-terminal-cloud/internal/pagination"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
var (
|
|
ErrAccessDenied = errors.New("resource access denied")
|
|
ErrNotFound = errors.New("resource not found")
|
|
ErrVersionConflict = errors.New("resource version conflict")
|
|
ErrEffectiveConsentExist = errors.New("an effective consent already exists")
|
|
ErrPolicyVersionMismatch = errors.New("consent policy version mismatch")
|
|
ErrDeviceClaimInvalid = errors.New("device claim is invalid")
|
|
ErrDeviceClaimUsed = errors.New("device claim was used for another binding")
|
|
ErrDeviceAlreadyBound = errors.New("device already has an active binding")
|
|
ErrDeviceBindingBlocked = errors.New("device binding change has active blockers")
|
|
ErrDeviceBindingStale = errors.New("device binding changed since impact review")
|
|
ErrSubjectDeleteConflict = errors.New("subject cannot be deleted while a conversation is active")
|
|
ErrDeleteConfirmation = errors.New("subject deletion confirmation does not match")
|
|
)
|
|
|
|
const (
|
|
ConsentTranscriptStorage = "transcript_storage"
|
|
ConsentRawAudioStorage = "raw_audio_storage"
|
|
ConsentStoryExtraction = "story_extraction"
|
|
ConsentPersonalMemory = "personalized_memory"
|
|
ConsentFamilySharing = "family_sharing"
|
|
)
|
|
|
|
var consentTypes = []string{
|
|
ConsentTranscriptStorage,
|
|
ConsentRawAudioStorage,
|
|
ConsentStoryExtraction,
|
|
ConsentPersonalMemory,
|
|
ConsentFamilySharing,
|
|
}
|
|
|
|
type Household struct {
|
|
ID string `json:"id"`
|
|
TenantID string `json:"tenant_id"`
|
|
Name string `json:"name"`
|
|
Status string `json:"status"`
|
|
Role identity.Role `json:"role"`
|
|
Capabilities []identity.Capability `json:"capabilities"`
|
|
Version int64 `json:"version"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
type Subject struct {
|
|
VoiceAgent *elderagent.Info `json:"voice_agent,omitempty"`
|
|
ID string `json:"id"`
|
|
HouseholdID string `json:"household_id"`
|
|
PreferredName string `json:"preferred_name"`
|
|
Language string `json:"language"`
|
|
Timezone string `json:"timezone"`
|
|
ProfileData map[string]any `json:"profile_data"`
|
|
ConversationPreferences map[string]any `json:"conversation_preferences"`
|
|
Status string `json:"status"`
|
|
Version int64 `json:"version"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
type Consent struct {
|
|
ID string `json:"id"`
|
|
SubjectID string `json:"subject_id"`
|
|
Type string `json:"type"`
|
|
Scope map[string]any `json:"scope"`
|
|
PolicyVersion string `json:"policy_version"`
|
|
NoticeDigest string `json:"notice_digest"`
|
|
ActorUserID string `json:"actor_user_id"`
|
|
EvidenceRef string `json:"evidence_ref,omitempty"`
|
|
GrantedAt time.Time `json:"granted_at"`
|
|
ExpiresAt *time.Time `json:"expires_at,omitempty"`
|
|
WithdrawnAt *time.Time `json:"withdrawn_at,omitempty"`
|
|
Effective bool `json:"effective"`
|
|
Version int64 `json:"version"`
|
|
}
|
|
|
|
type CreateSubjectInput struct {
|
|
InitialConsents InitialConsentSelection `json:"initial_consents,omitempty"`
|
|
PreferredName string `json:"preferred_name"`
|
|
Language string `json:"language"`
|
|
Timezone string `json:"timezone"`
|
|
ProfileData map[string]any `json:"profile_data"`
|
|
ConversationPreferences map[string]any `json:"conversation_preferences"`
|
|
}
|
|
|
|
type PatchSubjectInput struct {
|
|
PreferredName *string `json:"preferred_name,omitempty"`
|
|
Language *string `json:"language,omitempty"`
|
|
Timezone *string `json:"timezone,omitempty"`
|
|
ProfileData *map[string]any `json:"profile_data,omitempty"`
|
|
ConversationPreferences *map[string]any `json:"conversation_preferences,omitempty"`
|
|
Status *string `json:"status,omitempty"`
|
|
}
|
|
|
|
type GrantConsentInput struct {
|
|
Type string `json:"type"`
|
|
Scope map[string]any `json:"scope"`
|
|
PolicyVersion string `json:"policy_version"`
|
|
NoticeDigest string `json:"notice_digest"`
|
|
EvidenceRef string `json:"evidence_ref,omitempty"`
|
|
GrantedAt time.Time `json:"granted_at"`
|
|
ExpiresAt *time.Time `json:"expires_at,omitempty"`
|
|
}
|
|
|
|
type Service struct {
|
|
Fleet *devicefleet.Service
|
|
pool *pgxpool.Pool
|
|
audit *audit.Repository
|
|
redis *redis.Client
|
|
cursor *pagination.Codec
|
|
exportSigner exportURLSigner
|
|
exportDownloadTTL time.Duration
|
|
firmwareStore firmwareObjectStore
|
|
firmwareURLTTL time.Duration
|
|
providerMemory bool
|
|
voiceAgents *elderagent.Service
|
|
}
|
|
|
|
func (s *Service) ConfigureVoiceAgents(agents *elderagent.Service) { s.voiceAgents = agents }
|
|
|
|
func (s *Service) EnsureSubjectVoiceAgent(ctx context.Context, principal identity.Principal, publicID string) (elderagent.Info, error) {
|
|
if _, _, err := resolveSubject(ctx, s.pool, principal, publicID, identity.CapabilitySubjectWrite, false); err != nil {
|
|
return elderagent.Info{}, err
|
|
}
|
|
if s.voiceAgents == nil {
|
|
return elderagent.Info{}, errors.New("voice agents not configured")
|
|
}
|
|
return s.voiceAgents.Ensure(ctx, publicID)
|
|
}
|
|
|
|
func (s *Service) ConfigureProviderMemory(enabled bool) {
|
|
s.providerMemory = enabled
|
|
}
|
|
|
|
func NewService(pool *pgxpool.Pool, auditRepository *audit.Repository, redisClient *redis.Client, cursorCodec *pagination.Codec) *Service {
|
|
return &Service{pool: pool, audit: auditRepository, redis: redisClient, cursor: cursorCodec}
|
|
}
|
|
|
|
func IsConsentType(value string) bool {
|
|
for _, consentType := range consentTypes {
|
|
if value == consentType {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (s *Service) ListHouseholds(ctx context.Context, principal identity.Principal, tenantPublicID string) ([]Household, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT h.public_id::text, t.public_id::text, h.name, h.status, hm.role,
|
|
h.version, h.created_at, h.updated_at
|
|
FROM user_account ua
|
|
JOIN tenant t ON t.id=ua.tenant_id AND t.status='active'
|
|
JOIN household_member hm ON hm.user_id=ua.id AND hm.tenant_id=ua.tenant_id AND hm.status='active'
|
|
JOIN household h ON h.id=hm.household_id AND h.tenant_id=hm.tenant_id
|
|
WHERE ua.oidc_issuer=$1 AND ua.oidc_subject=$2 AND ua.status='active'
|
|
AND h.status <> 'deleted'
|
|
AND (NULLIF($3,'') IS NULL OR t.public_id=NULLIF($3,'')::uuid)
|
|
ORDER BY h.updated_at DESC, h.public_id`, principal.Issuer, principal.Subject, tenantPublicID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list households: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]Household, 0)
|
|
for rows.Next() {
|
|
var item Household
|
|
if err := rows.Scan(&item.ID, &item.TenantID, &item.Name, &item.Status, &item.Role,
|
|
&item.Version, &item.CreatedAt, &item.UpdatedAt); err != nil {
|
|
return nil, fmt.Errorf("scan household: %w", err)
|
|
}
|
|
item.Capabilities = identity.CapabilitiesForRole(item.Role)
|
|
result = append(result, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate households: %w", err)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Service) CreateHousehold(ctx context.Context, principal identity.Principal, tenantPublicID, name, requestID string) (Household, error) {
|
|
var result Household
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
var tenantID, userID, userPublicID string
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT t.id::text, ua.id::text, ua.public_id::text
|
|
FROM tenant t JOIN user_account ua ON ua.tenant_id=t.id AND ua.status='active'
|
|
WHERE t.public_id=$1::uuid AND t.status='active'
|
|
AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3
|
|
FOR SHARE OF t, ua`, tenantPublicID, principal.Issuer, principal.Subject,
|
|
).Scan(&tenantID, &userID, &userPublicID); errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrAccessDenied
|
|
} else if err != nil {
|
|
return fmt.Errorf("resolve tenant account: %w", err)
|
|
}
|
|
if err := tx.QueryRow(ctx, `
|
|
INSERT INTO household (tenant_id, name, owner_user_id)
|
|
VALUES ($1::uuid, $2, $3::uuid)
|
|
RETURNING public_id::text, $4::text, name, status, version, created_at, updated_at`,
|
|
tenantID, name, userID, tenantPublicID,
|
|
).Scan(&result.ID, &result.TenantID, &result.Name, &result.Status,
|
|
&result.Version, &result.CreatedAt, &result.UpdatedAt); err != nil {
|
|
return fmt.Errorf("insert household: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO household_member (tenant_id, household_id, user_id, role)
|
|
SELECT $1::uuid, id, $2::uuid, 'owner' FROM household WHERE public_id=$3::uuid`,
|
|
tenantID, userID, result.ID); err != nil {
|
|
return fmt.Errorf("insert household owner membership: %w", err)
|
|
}
|
|
result.Role = identity.RoleOwner
|
|
result.Capabilities = identity.CapabilitiesForRole(result.Role)
|
|
return s.audit.Record(ctx, tx, audit.Event{
|
|
TenantID: strPtr(tenantID), ActorType: "user", ActorPublicID: strPtr(userPublicID),
|
|
Action: "household.create", ResourceType: "household", ResourceID: strPtr(result.ID),
|
|
RequestID: requestID, Result: "success", Detail: map[string]any{"version": result.Version},
|
|
})
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) GetHousehold(ctx context.Context, principal identity.Principal, householdPublicID string) (Household, error) {
|
|
var result Household
|
|
if err := s.pool.QueryRow(ctx, householdSelectSQL+` AND h.public_id=$3::uuid`,
|
|
principal.Issuer, principal.Subject, householdPublicID,
|
|
).Scan(&result.ID, &result.TenantID, &result.Name, &result.Status, &result.Role,
|
|
&result.Version, &result.CreatedAt, &result.UpdatedAt); errors.Is(err, pgx.ErrNoRows) {
|
|
return Household{}, ErrAccessDenied
|
|
} else if err != nil {
|
|
return Household{}, fmt.Errorf("get household: %w", err)
|
|
}
|
|
if !identity.Can(result.Role, identity.CapabilityHouseholdRead) {
|
|
return Household{}, ErrAccessDenied
|
|
}
|
|
result.Capabilities = identity.CapabilitiesForRole(result.Role)
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Service) PatchHousehold(ctx context.Context, principal identity.Principal, householdPublicID, name string, expectedVersion int64, requestID string) (Household, error) {
|
|
var result Household
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, err := resolveHouseholdAccess(ctx, tx, principal, householdPublicID, identity.CapabilityHouseholdManage, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := concurrency.RequireVersion(expectedVersion, access.version); err != nil {
|
|
return ErrVersionConflict
|
|
}
|
|
if err := tx.QueryRow(ctx, `
|
|
UPDATE household SET name=$1, version=version+1, updated_at=now()
|
|
WHERE id=$2::uuid AND tenant_id=$3::uuid
|
|
RETURNING public_id::text, $4::text, name, status, version, created_at, updated_at`,
|
|
name, access.householdID, access.tenantID, access.tenantPublicID,
|
|
).Scan(&result.ID, &result.TenantID, &result.Name, &result.Status,
|
|
&result.Version, &result.CreatedAt, &result.UpdatedAt); err != nil {
|
|
return fmt.Errorf("update household: %w", err)
|
|
}
|
|
result.Role = access.role
|
|
result.Capabilities = identity.CapabilitiesForRole(result.Role)
|
|
return s.audit.Record(ctx, tx, audit.Event{
|
|
TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID),
|
|
Action: "household.update", ResourceType: "household", ResourceID: strPtr(result.ID),
|
|
RequestID: requestID, Result: "success", Detail: map[string]any{"from_version": expectedVersion, "to_version": result.Version},
|
|
})
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) ListSubjects(ctx context.Context, principal identity.Principal, householdPublicID string) ([]Subject, error) {
|
|
access, err := resolveHouseholdAccess(ctx, s.pool, principal, householdPublicID, identity.CapabilitySubjectRead, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT s.public_id::text, $1::text, s.preferred_name, s.language, s.timezone,
|
|
s.profile_data, s.conversation_preferences, s.status, s.profile_version, s.created_at, s.updated_at
|
|
FROM subject_profile s
|
|
WHERE s.tenant_id=$2::uuid AND s.household_id=$3::uuid AND s.status IN ('active','suspended')
|
|
ORDER BY s.updated_at DESC, s.public_id`, householdPublicID, access.tenantID, access.householdID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list subjects: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]Subject, 0)
|
|
for rows.Next() {
|
|
item, err := scanSubject(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate subjects: %w", err)
|
|
}
|
|
rows.Close()
|
|
if s.voiceAgents != nil {
|
|
for n := range result {
|
|
info, e := s.voiceAgents.Get(ctx, result[n].ID)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
result[n].VoiceAgent = &info
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Service) CreateSubject(ctx context.Context, principal identity.Principal, householdPublicID, requestID string, input CreateSubjectInput) (Subject, error) {
|
|
if err := ValidateInitialConsents(input.InitialConsents); err != nil {
|
|
return Subject{}, err
|
|
}
|
|
var result Subject
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, err := resolveHouseholdAccess(ctx, tx, principal, householdPublicID, identity.CapabilitySubjectWrite, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
profileJSON, _ := json.Marshal(input.ProfileData)
|
|
preferencesJSON, _ := json.Marshal(input.ConversationPreferences)
|
|
row := tx.QueryRow(ctx, `
|
|
INSERT INTO subject_profile (
|
|
tenant_id, household_id, preferred_name, language, timezone,
|
|
profile_data, conversation_preferences, created_by, updated_by
|
|
) VALUES ($1::uuid, $2::uuid, $3, $4, $5, $6::jsonb, $7::jsonb, $8::uuid, $8::uuid)
|
|
RETURNING public_id::text, $9::text, preferred_name, language, timezone,
|
|
profile_data, conversation_preferences, status, profile_version, created_at, updated_at`,
|
|
access.tenantID, access.householdID, input.PreferredName, input.Language, input.Timezone,
|
|
profileJSON, preferencesJSON, access.userID, householdPublicID)
|
|
result, err = scanSubject(row)
|
|
if err != nil {
|
|
return fmt.Errorf("insert subject: %w", err)
|
|
}
|
|
if err := InitializeSubjectMemory(ctx, tx, access.tenantID, result.ID, access.userID, requestID, input.InitialConsents); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{
|
|
TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID),
|
|
Action: "subject.create", ResourceType: "subject", ResourceID: strPtr(result.ID),
|
|
RequestID: requestID, Result: "success", Detail: map[string]any{"household_id": householdPublicID, "version": result.Version},
|
|
})
|
|
})
|
|
if err == nil && s.voiceAgents != nil {
|
|
info, provisionErr := s.voiceAgents.Ensure(ctx, result.ID)
|
|
if provisionErr != nil && info.Status == "" {
|
|
info = elderagent.Info{Status: "FAILED", Message: "老人已保存，智能体暂未就绪，请重试"}
|
|
}
|
|
result.VoiceAgent = &info
|
|
}
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) GetSubject(ctx context.Context, principal identity.Principal, subjectPublicID string) (Subject, error) {
|
|
access, result, err := resolveSubject(ctx, s.pool, principal, subjectPublicID, identity.CapabilitySubjectRead, false)
|
|
_ = access
|
|
if err == nil && s.voiceAgents != nil {
|
|
info, e := s.voiceAgents.Get(ctx, subjectPublicID)
|
|
if e != nil {
|
|
return result, e
|
|
}
|
|
result.VoiceAgent = &info
|
|
}
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) PatchSubject(ctx context.Context, principal identity.Principal, subjectPublicID, requestID string, expectedVersion int64, input PatchSubjectInput) (Subject, error) {
|
|
var result Subject
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, current, err := resolveSubject(ctx, tx, principal, subjectPublicID, identity.CapabilitySubjectWrite, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if current.Version != expectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
preferredName, language, timezone, status := current.PreferredName, current.Language, current.Timezone, current.Status
|
|
profile, preferences := current.ProfileData, current.ConversationPreferences
|
|
if input.PreferredName != nil {
|
|
preferredName = *input.PreferredName
|
|
}
|
|
if input.Language != nil {
|
|
language = *input.Language
|
|
}
|
|
if input.Timezone != nil {
|
|
timezone = *input.Timezone
|
|
}
|
|
if input.Status != nil {
|
|
status = *input.Status
|
|
}
|
|
if input.ProfileData != nil {
|
|
profile = *input.ProfileData
|
|
}
|
|
if input.ConversationPreferences != nil {
|
|
preferences = *input.ConversationPreferences
|
|
}
|
|
profileJSON, _ := json.Marshal(profile)
|
|
preferencesJSON, _ := json.Marshal(preferences)
|
|
row := tx.QueryRow(ctx, `
|
|
UPDATE subject_profile SET preferred_name=$1, language=$2, timezone=$3,
|
|
profile_data=$4::jsonb, conversation_preferences=$5::jsonb, status=$6,
|
|
updated_by=$7::uuid, profile_version=profile_version+1, updated_at=now()
|
|
WHERE id=$8::uuid AND tenant_id=$9::uuid
|
|
RETURNING public_id::text, $10::text, preferred_name, language, timezone,
|
|
profile_data, conversation_preferences, status, profile_version, created_at, updated_at`,
|
|
preferredName, language, timezone, profileJSON, preferencesJSON, status,
|
|
access.userID, access.subjectID, access.tenantID, access.householdPublicID)
|
|
result, err = scanSubject(row)
|
|
if err != nil {
|
|
return fmt.Errorf("update subject: %w", err)
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{
|
|
TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID),
|
|
Action: "subject.update", ResourceType: "subject", ResourceID: strPtr(result.ID),
|
|
RequestID: requestID, Result: "success", Detail: map[string]any{"from_version": expectedVersion, "to_version": result.Version},
|
|
})
|
|
})
|
|
if err == nil && s.voiceAgents != nil {
|
|
// The database trigger queues this atomically with the profile change.
|
|
// A provider failure must not undo a successfully saved profile.
|
|
info, syncErr := s.voiceAgents.Sync(ctx, result.ID)
|
|
if syncErr != nil && info.Status == "" {
|
|
info = elderagent.Info{Status: "PENDING", PersonaSyncStatus: "PENDING", Message: "资料已保存，等待同步"}
|
|
}
|
|
result.VoiceAgent = &info
|
|
}
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) ListConsents(ctx context.Context, principal identity.Principal, subjectPublicID string) ([]Consent, error) {
|
|
access, _, err := resolveSubject(ctx, s.pool, principal, subjectPublicID, identity.CapabilitySubjectRead, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT c.public_id::text, $1::text, c.consent_type, c.scope, c.policy_version,
|
|
c.notice_digest, ua.public_id::text, COALESCE(c.evidence_ref,''), c.granted_at,
|
|
c.expires_at, c.withdrawn_at, c.version
|
|
FROM consent_record c JOIN user_account ua ON ua.id=c.actor_user_id AND ua.tenant_id=c.tenant_id
|
|
WHERE c.tenant_id=$2::uuid AND c.subject_id=$3::uuid
|
|
ORDER BY c.granted_at DESC, c.id DESC`, subjectPublicID, access.tenantID, access.subjectID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list consents: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]Consent, 0)
|
|
for rows.Next() {
|
|
item, err := scanConsent(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate consents: %w", err)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Service) GrantConsent(ctx context.Context, principal identity.Principal, subjectPublicID, requestID string, input GrantConsentInput) (Consent, error) {
|
|
var result Consent
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, _, err := resolveSubject(ctx, tx, principal, subjectPublicID, identity.CapabilitySubjectWrite, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if input.PolicyVersion != access.policyVersion {
|
|
return ErrPolicyVersionMismatch
|
|
}
|
|
var existing bool
|
|
if err := tx.QueryRow(ctx, `SELECT EXISTS(
|
|
SELECT 1 FROM consent_record WHERE tenant_id=$1::uuid AND subject_id=$2::uuid
|
|
AND consent_type=$3 AND withdrawn_at IS NULL AND (expires_at IS NULL OR expires_at > now())
|
|
)`, access.tenantID, access.subjectID, input.Type).Scan(&existing); err != nil {
|
|
return fmt.Errorf("check effective consent: %w", err)
|
|
}
|
|
if existing {
|
|
return ErrEffectiveConsentExist
|
|
}
|
|
scopeJSON, _ := json.Marshal(input.Scope)
|
|
row := tx.QueryRow(ctx, `
|
|
INSERT INTO consent_record (
|
|
tenant_id, subject_id, consent_type, scope, policy_version, notice_digest,
|
|
actor_user_id, evidence_ref, granted_at, expires_at
|
|
) VALUES ($1::uuid,$2::uuid,$3,$4::jsonb,$5,$6,$7::uuid,NULLIF($8,''),$9,$10)
|
|
RETURNING public_id::text, $11::text, consent_type, scope, policy_version,
|
|
notice_digest, $12::text, COALESCE(evidence_ref,''), granted_at, expires_at, withdrawn_at, version`,
|
|
access.tenantID, access.subjectID, input.Type, scopeJSON, input.PolicyVersion,
|
|
input.NoticeDigest, access.userID, input.EvidenceRef, input.GrantedAt, input.ExpiresAt,
|
|
subjectPublicID, access.userPublicID)
|
|
result, err = scanConsent(row)
|
|
if err != nil {
|
|
return fmt.Errorf("insert consent: %w", err)
|
|
}
|
|
if err := createConsentSnapshot(ctx, tx, access.tenantID, access.subjectID); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{
|
|
TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID),
|
|
Action: "consent.grant", ResourceType: "consent", ResourceID: strPtr(result.ID),
|
|
RequestID: requestID, Result: "success", Detail: map[string]any{"subject_id": subjectPublicID, "consent_type": result.Type},
|
|
})
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) WithdrawConsent(ctx context.Context, principal identity.Principal, subjectPublicID, consentPublicID, requestID string) (Consent, error) {
|
|
var result Consent
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, _, err := resolveSubject(ctx, tx, principal, subjectPublicID, identity.CapabilitySubjectWrite, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
row := tx.QueryRow(ctx, `
|
|
SELECT c.public_id::text, $1::text, c.consent_type, c.scope, c.policy_version,
|
|
c.notice_digest, ua.public_id::text, COALESCE(c.evidence_ref,''), c.granted_at,
|
|
c.expires_at, c.withdrawn_at, c.version
|
|
FROM consent_record c JOIN user_account ua ON ua.id=c.actor_user_id AND ua.tenant_id=c.tenant_id
|
|
WHERE c.public_id=$2::uuid AND c.tenant_id=$3::uuid AND c.subject_id=$4::uuid
|
|
FOR UPDATE OF c`, subjectPublicID, consentPublicID, access.tenantID, access.subjectID)
|
|
result, err = scanConsent(row)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("get consent for withdrawal: %w", err)
|
|
}
|
|
if result.WithdrawnAt != nil {
|
|
return nil
|
|
}
|
|
row = tx.QueryRow(ctx, `
|
|
UPDATE consent_record SET withdrawn_at=now(), version=version+1, updated_at=now()
|
|
WHERE public_id=$1::uuid
|
|
RETURNING public_id::text, $2::text, consent_type, scope, policy_version,
|
|
notice_digest, $3::text, COALESCE(evidence_ref,''), granted_at, expires_at, withdrawn_at, version`,
|
|
consentPublicID, subjectPublicID, result.ActorUserID)
|
|
result, err = scanConsent(row)
|
|
if err != nil {
|
|
return fmt.Errorf("withdraw consent: %w", err)
|
|
}
|
|
if err := createConsentSnapshot(ctx, tx, access.tenantID, access.subjectID); err != nil {
|
|
return err
|
|
}
|
|
if result.Type == "personalized_memory" {
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE curated_memory SET status='DISABLED',version=version+1,updated_at=now()
|
|
WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status='ACTIVE'`, access.tenantID, access.subjectID); err != nil {
|
|
return fmt.Errorf("disable memories after consent withdrawal: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE memory_index_entry SET embedding=NULL,status='STALE',last_error='personalized memory consent withdrawn',updated_at=now()
|
|
WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status IN ('PENDING','ACTIVE')`, access.tenantID, access.subjectID); err != nil {
|
|
return fmt.Errorf("invalidate memory index after consent withdrawal: %w", err)
|
|
}
|
|
if s.providerMemory {
|
|
rows, err := tx.Query(ctx, `SELECT id::text FROM curated_memory WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status<>'DELETED' ORDER BY id`, access.tenantID, access.subjectID)
|
|
if err != nil {
|
|
return fmt.Errorf("list provider memories after consent withdrawal: %w", err)
|
|
}
|
|
memoryIDs := make([]string, 0)
|
|
for rows.Next() {
|
|
var memoryID string
|
|
if err := rows.Scan(&memoryID); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
memoryIDs = append(memoryIDs, memoryID)
|
|
}
|
|
rows.Close()
|
|
if err := rows.Err(); err != nil {
|
|
return err
|
|
}
|
|
for _, memoryID := range memoryIDs {
|
|
dedupe := "memory-provider-delete-consent:" + memoryID + ":" + result.ID
|
|
payload, _ := json.Marshal(map[string]string{"reason": "personalized memory consent withdrawn", "consent_id": result.ID})
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &access.tenantID, AggregateType: "curated_memory", AggregateID: memoryID, EventType: "memory.provider_delete_requested", DeduplicationKey: &dedupe, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
affectedConversations, err := s.stopConversationsAfterConsentWithdrawal(ctx, tx, access.tenantID, access.subjectID, result.Type)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{
|
|
TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID),
|
|
Action: "consent.withdraw", ResourceType: "consent", ResourceID: strPtr(result.ID),
|
|
RequestID: requestID, Result: "success", Detail: map[string]any{
|
|
"subject_id": subjectPublicID, "consent_type": result.Type,
|
|
"stopped_conversation_ids": affectedConversations,
|
|
},
|
|
})
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func consentWithdrawalStopsConversation(consentType string) bool {
|
|
switch consentType {
|
|
case ConsentTranscriptStorage, ConsentRawAudioStorage, ConsentStoryExtraction, ConsentPersonalMemory:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (s *Service) stopConversationsAfterConsentWithdrawal(
|
|
ctx context.Context,
|
|
tx pgx.Tx,
|
|
tenantID, subjectID, consentType string,
|
|
) ([]string, error) {
|
|
if !consentWithdrawalStopsConversation(consentType) {
|
|
return []string{}, nil
|
|
}
|
|
type candidate struct {
|
|
internalID, publicID, status, runtimeType string
|
|
}
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT c.id::text,c.public_id::text,c.status,c.runtime_type
|
|
FROM conversation c
|
|
JOIN consent_snapshot cs ON cs.id=c.consent_snapshot_id AND cs.tenant_id=c.tenant_id
|
|
LEFT JOIN rtc_task rt ON rt.conversation_id=c.id
|
|
LEFT JOIN hardware_agent_session hs ON hs.conversation_id=c.id
|
|
WHERE c.tenant_id=$1::uuid AND c.subject_id=$2::uuid
|
|
AND c.status IN ('PREPARING','ROOM_ISSUED','STARTING_AGENT','ACTIVE')
|
|
AND cs.effective_consents @> jsonb_build_object($3::text,true)
|
|
ORDER BY c.created_at,c.id
|
|
FOR UPDATE OF c`, tenantID, subjectID, consentType)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("find conversations affected by consent withdrawal: %w", err)
|
|
}
|
|
candidates := make([]candidate, 0)
|
|
for rows.Next() {
|
|
var item candidate
|
|
if err := rows.Scan(&item.internalID, &item.publicID, &item.status, &item.runtimeType); err != nil {
|
|
rows.Close()
|
|
return nil, fmt.Errorf("scan consent withdrawal conversation: %w", err)
|
|
}
|
|
candidates = append(candidates, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
rows.Close()
|
|
return nil, fmt.Errorf("iterate consent withdrawal conversations: %w", err)
|
|
}
|
|
rows.Close()
|
|
|
|
affected := make([]string, 0, len(candidates))
|
|
stopReason := "consent_withdrawn:" + consentType
|
|
for _, item := range candidates {
|
|
toStatus := "EXPIRED"
|
|
if item.status == "STARTING_AGENT" || item.status == "ACTIVE" {
|
|
toStatus = "STOPPING"
|
|
}
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE conversation SET status=$2::varchar,stop_reason=$3,ended_at=CASE WHEN $2::text='EXPIRED' THEN now() ELSE ended_at END,
|
|
lease_expires_at=NULL,version=version+1,updated_at=now()
|
|
WHERE id=$1::uuid AND status=$4::varchar`, item.internalID, toStatus, stopReason, item.status)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("stop conversation after consent withdrawal: %w", err)
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return nil, fmt.Errorf("conversation changed during consent withdrawal")
|
|
}
|
|
if item.runtimeType == "volc_hardware_agent" && toStatus == "EXPIRED" {
|
|
_, err = tx.Exec(ctx, `UPDATE hardware_agent_session SET status='EXPIRED',updated_at=now() WHERE conversation_id=$1::uuid`, item.internalID)
|
|
} else if item.runtimeType == "volc_hardware_agent" {
|
|
_, err = tx.Exec(ctx, `UPDATE hardware_agent_session SET status='STOPPING',updated_at=now() WHERE conversation_id=$1::uuid`, item.internalID)
|
|
} else if toStatus == "EXPIRED" {
|
|
_, err = tx.Exec(ctx, `UPDATE rtc_task SET status='COMPLETED',lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`, item.internalID)
|
|
} else {
|
|
_, err = tx.Exec(ctx, `UPDATE rtc_task SET status=CASE WHEN status='ACTIVE' THEN 'STOPPING' ELSE 'UNKNOWN' END,lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`, item.internalID)
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("update RTC after consent withdrawal: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source)
|
|
VALUES($1::uuid,$2,$3,$4,'platform')`, item.internalID, item.status, toStatus, "consent withdrawn: "+consentType); err != nil {
|
|
return nil, fmt.Errorf("record consent withdrawal stop: %w", err)
|
|
}
|
|
if toStatus == "STOPPING" && item.runtimeType != "volc_hardware_agent" {
|
|
deduplicationKey := "consent-withdrawal-stop:" + item.internalID + ":" + consentType
|
|
payload, _ := json.Marshal(map[string]string{
|
|
"conversation_id": item.publicID,
|
|
"reason": stopReason,
|
|
})
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{
|
|
TenantID: strPtr(tenantID), AggregateType: "conversation", AggregateID: item.internalID,
|
|
EventType: "conversation.agent_stop_requested", DeduplicationKey: &deduplicationKey, Payload: payload,
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("enqueue consent withdrawal stop: %w", err)
|
|
}
|
|
}
|
|
affected = append(affected, item.publicID)
|
|
}
|
|
return affected, nil
|
|
}
|
|
|
|
const householdSelectSQL = `
|
|
SELECT h.public_id::text, t.public_id::text, h.name, h.status, hm.role,
|
|
h.version, h.created_at, h.updated_at
|
|
FROM household h
|
|
JOIN tenant t ON t.id=h.tenant_id AND t.status='active'
|
|
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 ua.oidc_issuer=$1 AND ua.oidc_subject=$2 AND h.status <> 'deleted'`
|
|
|
|
type queryRow interface {
|
|
QueryRow(context.Context, string, ...any) pgx.Row
|
|
}
|
|
|
|
type resourceAccess struct {
|
|
tenantID, tenantPublicID, userID, userPublicID string
|
|
householdID, householdPublicID, subjectID string
|
|
policyVersion string
|
|
role identity.Role
|
|
version int64
|
|
}
|
|
|
|
func resolveHouseholdAccess(ctx context.Context, q queryRow, principal identity.Principal, publicID string, capability identity.Capability, lock bool) (resourceAccess, error) {
|
|
sql := `
|
|
SELECT t.id::text,t.public_id::text,t.policy_version,ua.id::text,ua.public_id::text,
|
|
h.id::text,h.public_id::text,hm.role,h.version
|
|
FROM household h
|
|
JOIN tenant t ON t.id=h.tenant_id AND t.status='active'
|
|
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 h.public_id=$1::uuid AND h.status <> 'deleted'
|
|
AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`
|
|
if lock {
|
|
sql += ` FOR UPDATE OF h`
|
|
}
|
|
var a resourceAccess
|
|
var role string
|
|
err := q.QueryRow(ctx, sql, publicID, principal.Issuer, principal.Subject).Scan(
|
|
&a.tenantID, &a.tenantPublicID, &a.policyVersion, &a.userID, &a.userPublicID,
|
|
&a.householdID, &a.householdPublicID, &role, &a.version)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return resourceAccess{}, ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return resourceAccess{}, fmt.Errorf("resolve household access: %w", err)
|
|
}
|
|
a.role = identity.Role(role)
|
|
if !identity.Can(a.role, capability) {
|
|
return resourceAccess{}, ErrAccessDenied
|
|
}
|
|
return a, nil
|
|
}
|
|
|
|
func resolveSubject(ctx context.Context, q queryRow, principal identity.Principal, publicID string, capability identity.Capability, lock bool) (resourceAccess, Subject, error) {
|
|
sql := `
|
|
SELECT t.id::text,t.public_id::text,t.policy_version,ua.id::text,ua.public_id::text,
|
|
h.id::text,h.public_id::text,s.id::text,hm.role,
|
|
s.public_id::text,h.public_id::text,s.preferred_name,s.language,s.timezone,
|
|
s.profile_data,s.conversation_preferences,s.status,s.profile_version,s.created_at,s.updated_at
|
|
FROM subject_profile s
|
|
JOIN household h ON h.id=s.household_id AND h.tenant_id=s.tenant_id AND h.status <> 'deleted'
|
|
JOIN tenant t ON t.id=s.tenant_id AND t.status='active'
|
|
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 s.public_id=$1::uuid AND s.status IN ('active','suspended')
|
|
AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`
|
|
if lock {
|
|
sql += ` FOR UPDATE OF s`
|
|
}
|
|
var a resourceAccess
|
|
var role string
|
|
var result Subject
|
|
var profileJSON, preferencesJSON []byte
|
|
err := q.QueryRow(ctx, sql, publicID, principal.Issuer, principal.Subject).Scan(
|
|
&a.tenantID, &a.tenantPublicID, &a.policyVersion, &a.userID, &a.userPublicID,
|
|
&a.householdID, &a.householdPublicID, &a.subjectID, &role,
|
|
&result.ID, &result.HouseholdID, &result.PreferredName, &result.Language, &result.Timezone,
|
|
&profileJSON, &preferencesJSON, &result.Status, &result.Version, &result.CreatedAt, &result.UpdatedAt)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return resourceAccess{}, Subject{}, ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return resourceAccess{}, Subject{}, fmt.Errorf("resolve subject: %w", err)
|
|
}
|
|
a.role = identity.Role(role)
|
|
if !identity.Can(a.role, capability) {
|
|
return resourceAccess{}, Subject{}, ErrAccessDenied
|
|
}
|
|
if err := decodeObject(profileJSON, &result.ProfileData); err != nil {
|
|
return resourceAccess{}, Subject{}, err
|
|
}
|
|
if err := decodeObject(preferencesJSON, &result.ConversationPreferences); err != nil {
|
|
return resourceAccess{}, Subject{}, err
|
|
}
|
|
return a, result, nil
|
|
}
|
|
|
|
type rowScanner interface{ Scan(...any) error }
|
|
|
|
func scanSubject(row rowScanner) (Subject, error) {
|
|
var item Subject
|
|
var profileJSON, preferencesJSON []byte
|
|
if err := row.Scan(&item.ID, &item.HouseholdID, &item.PreferredName, &item.Language, &item.Timezone,
|
|
&profileJSON, &preferencesJSON, &item.Status, &item.Version, &item.CreatedAt, &item.UpdatedAt); err != nil {
|
|
return Subject{}, err
|
|
}
|
|
if err := decodeObject(profileJSON, &item.ProfileData); err != nil {
|
|
return Subject{}, err
|
|
}
|
|
if err := decodeObject(preferencesJSON, &item.ConversationPreferences); err != nil {
|
|
return Subject{}, err
|
|
}
|
|
return item, nil
|
|
}
|
|
|
|
func scanConsent(row rowScanner) (Consent, error) {
|
|
var item Consent
|
|
var scopeJSON []byte
|
|
if err := row.Scan(&item.ID, &item.SubjectID, &item.Type, &scopeJSON, &item.PolicyVersion,
|
|
&item.NoticeDigest, &item.ActorUserID, &item.EvidenceRef, &item.GrantedAt,
|
|
&item.ExpiresAt, &item.WithdrawnAt, &item.Version); err != nil {
|
|
return Consent{}, err
|
|
}
|
|
if err := decodeObject(scopeJSON, &item.Scope); err != nil {
|
|
return Consent{}, err
|
|
}
|
|
item.Effective = item.WithdrawnAt == nil && (item.ExpiresAt == nil || item.ExpiresAt.After(time.Now()))
|
|
return item, nil
|
|
}
|
|
|
|
func createConsentSnapshot(ctx context.Context, tx pgx.Tx, tenantID, subjectID string) error {
|
|
effective := map[string]bool{"voice_chat": true}
|
|
for _, consentType := range consentTypes {
|
|
effective[consentType] = false
|
|
}
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT public_id::text, consent_type FROM consent_record
|
|
WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND withdrawn_at IS NULL
|
|
AND (expires_at IS NULL OR expires_at > now())
|
|
ORDER BY granted_at,id`, tenantID, subjectID)
|
|
if err != nil {
|
|
return fmt.Errorf("query effective consents: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
sourceIDs := make([]string, 0)
|
|
for rows.Next() {
|
|
var publicID, consentType string
|
|
if err := rows.Scan(&publicID, &consentType); err != nil {
|
|
return fmt.Errorf("scan effective consent: %w", err)
|
|
}
|
|
effective[consentType] = true
|
|
sourceIDs = append(sourceIDs, publicID)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return fmt.Errorf("iterate effective consents: %w", err)
|
|
}
|
|
effectiveJSON, _ := json.Marshal(effective)
|
|
sourceJSON, _ := json.Marshal(sourceIDs)
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO consent_snapshot (tenant_id,subject_id,snapshot_version,effective_consents,source_consent_ids)
|
|
SELECT $1::uuid,$2::uuid,COALESCE(MAX(snapshot_version),0)+1,$3::jsonb,$4::jsonb
|
|
FROM consent_snapshot WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`,
|
|
tenantID, subjectID, effectiveJSON, sourceJSON); err != nil {
|
|
return fmt.Errorf("insert consent snapshot: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func decodeObject(value []byte, target *map[string]any) error {
|
|
if len(value) == 0 {
|
|
*target = map[string]any{}
|
|
return nil
|
|
}
|
|
if err := json.Unmarshal(value, target); err != nil {
|
|
return fmt.Errorf("decode JSON object: %w", err)
|
|
}
|
|
if *target == nil {
|
|
*target = map[string]any{}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func strPtr(value string) *string { return &value }
|