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.
1010 lines
52 KiB
Go
1010 lines
52 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/callmemory"
|
|
"data-terminal-cloud/internal/database"
|
|
"data-terminal-cloud/internal/identity"
|
|
"data-terminal-cloud/internal/outbox"
|
|
"data-terminal-cloud/internal/pagination"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
var (
|
|
ErrStorySourceRequired = errors.New("story source is required")
|
|
ErrPersonalizedMemoryConsentNeeded = errors.New("personalized memory consent is required")
|
|
ErrReviewStateConflict = errors.New("content review state does not allow this operation")
|
|
ErrFactReviewStateConflict = errors.New("fact candidate review state does not allow this operation")
|
|
ErrConversationSummaryNotReady = errors.New("conversation is not ready for summary generation")
|
|
)
|
|
|
|
type ConversationTurn 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"`
|
|
SourceType string `json:"source_type"`
|
|
}
|
|
|
|
type ConversationTurnPage struct {
|
|
Items []ConversationTurn `json:"items"`
|
|
NextCursor string `json:"next_cursor,omitempty"`
|
|
}
|
|
|
|
type ConversationContentSummary struct {
|
|
ID string `json:"id"`
|
|
Type string `json:"type"`
|
|
Content string `json:"content"`
|
|
SourceTurnIDs []string `json:"source_turn_ids"`
|
|
ModelRef string `json:"model_ref"`
|
|
PromptVersion string `json:"prompt_version"`
|
|
GenerationVersion int64 `json:"generation_version"`
|
|
Status string `json:"status"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
}
|
|
|
|
type StorySource struct {
|
|
TurnID string `json:"turn_id"`
|
|
CallSessionID string `json:"call_session_id,omitempty"`
|
|
ImportID string `json:"import_id,omitempty"`
|
|
PrivateSources bool `json:"-"`
|
|
UtteranceID string `json:"utterance_id,omitempty"`
|
|
SourceRevision *int64 `json:"source_revision,omitempty"`
|
|
SequenceNo int64 `json:"sequence_no"`
|
|
Speaker string `json:"speaker"`
|
|
Text string `json:"text"`
|
|
StartOffset *int `json:"start_offset,omitempty"`
|
|
EndOffset *int `json:"end_offset,omitempty"`
|
|
}
|
|
|
|
type FactCandidate struct {
|
|
ID string `json:"id"`
|
|
FactType string `json:"fact_type"`
|
|
Content string `json:"content"`
|
|
StatementBasis string `json:"statement_basis"`
|
|
Confidence *float64 `json:"confidence,omitempty"`
|
|
Sensitivity string `json:"sensitivity"`
|
|
SourceTurnIDs []string `json:"source_turn_ids"`
|
|
Status string `json:"status"`
|
|
Version int64 `json:"version"`
|
|
}
|
|
|
|
type UpdateFactCandidateInput struct {
|
|
FactType *string
|
|
Content *string
|
|
Sensitivity *string
|
|
Status *string
|
|
ErrorType *string
|
|
Reason string
|
|
}
|
|
|
|
type Story struct {
|
|
ID string `json:"id"`
|
|
SubjectID string `json:"subject_id"`
|
|
ConversationID string `json:"conversation_id,omitempty"`
|
|
CallSessionID string `json:"call_session_id,omitempty"`
|
|
ImportID string `json:"import_id,omitempty"`
|
|
PrivateSources bool `json:"-"`
|
|
Title string `json:"title"`
|
|
Content string `json:"content"`
|
|
Status string `json:"status"`
|
|
Confidence *float64 `json:"confidence,omitempty"`
|
|
Sensitivity string `json:"sensitivity"`
|
|
GenerationVersion int64 `json:"generation_version"`
|
|
RevisionNo int64 `json:"revision_no"`
|
|
AuthorType string `json:"author_type"`
|
|
ReviewReason string `json:"review_reason,omitempty"`
|
|
Version int64 `json:"version"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
Sources []StorySource `json:"sources"`
|
|
Facts []FactCandidate `json:"facts"`
|
|
}
|
|
|
|
type CuratedMemory struct {
|
|
ID string `json:"id"`
|
|
SubjectID string `json:"subject_id"`
|
|
Content string `json:"content"`
|
|
SourceRefs []map[string]any `json:"source_refs"`
|
|
Status string `json:"status"`
|
|
Sensitivity string `json:"sensitivity"`
|
|
VersionNo int64 `json:"version_no"`
|
|
Version int64 `json:"version"`
|
|
ConfirmedBy string `json:"confirmed_by"`
|
|
ConfirmedAt time.Time `json:"confirmed_at"`
|
|
ChangeReason string `json:"change_reason,omitempty"`
|
|
IndexStatus string `json:"index_status"`
|
|
LastIndexError string `json:"last_index_error,omitempty"`
|
|
ProviderIndexStatus string `json:"provider_index_status"`
|
|
ProviderIndexError string `json:"provider_index_error,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
type UpdateMemoryInput struct {
|
|
Content *string
|
|
Status *string
|
|
ErrorType *string
|
|
Reason string
|
|
}
|
|
|
|
var contentErrorTypes = map[string]bool{
|
|
"PERSON_ERROR": true, "TIME_ERROR": true, "PLACE_ERROR": true, "RELATION_ERROR": true,
|
|
"OVER_INFERENCE": true, "KEY_POINT_OMISSION": true, "DUPLICATE_CONTENT": true,
|
|
"UNNATURAL_EXPRESSION": true, "SENSITIVE_INFORMATION": true,
|
|
}
|
|
|
|
func IsContentErrorType(value string) bool { return contentErrorTypes[value] }
|
|
|
|
func (s *Service) ListConversationTurns(ctx context.Context, principal identity.Principal, conversationPublicID string, limit int, cursor string) (ConversationTurnPage, error) {
|
|
access, conversationID, err := s.resolveConversationContent(ctx, s.pool, principal, conversationPublicID, identity.CapabilityConversationRead, false)
|
|
if err != nil {
|
|
return ConversationTurnPage{}, err
|
|
}
|
|
if s.cursor == nil {
|
|
return ConversationTurnPage{}, errors.New("pagination cursor codec is unavailable")
|
|
}
|
|
var afterSequence int64
|
|
var afterID string
|
|
if cursor != "" {
|
|
decoded, err := s.cursor.Decode(cursor)
|
|
if err != nil {
|
|
return ConversationTurnPage{}, pagination.ErrInvalidCursor
|
|
}
|
|
if _, err := fmt.Sscan(decoded.SortKey, &afterSequence); err != nil {
|
|
return ConversationTurnPage{}, pagination.ErrInvalidCursor
|
|
}
|
|
afterID = decoded.ID
|
|
}
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT public_id::text,sequence_no,speaker,start_at,end_at,final_text,interrupted,source_type
|
|
FROM conversation_turn WHERE tenant_id=$1::uuid AND conversation_id=$2::uuid
|
|
AND ($3::bigint=0 OR (sequence_no,public_id)>($3,$4::uuid))
|
|
ORDER BY sequence_no,public_id LIMIT $5`, access.tenantID, conversationID, afterSequence, nullUUID(afterID), limit+1)
|
|
if err != nil {
|
|
return ConversationTurnPage{}, fmt.Errorf("list conversation turns: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
items := make([]ConversationTurn, 0, limit+1)
|
|
for rows.Next() {
|
|
var item ConversationTurn
|
|
if err := rows.Scan(&item.ID, &item.SequenceNo, &item.Speaker, &item.StartAt, &item.EndAt, &item.FinalText, &item.Interrupted, &item.SourceType); err != nil {
|
|
return ConversationTurnPage{}, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return ConversationTurnPage{}, err
|
|
}
|
|
page := ConversationTurnPage{Items: items}
|
|
if len(items) > limit {
|
|
page.Items = items[:limit]
|
|
last := page.Items[len(page.Items)-1]
|
|
page.NextCursor, err = s.cursor.Encode(fmt.Sprint(last.SequenceNo), last.ID)
|
|
if err != nil {
|
|
return ConversationTurnPage{}, err
|
|
}
|
|
}
|
|
return page, nil
|
|
}
|
|
|
|
func (s *Service) ListConversationContentSummaries(ctx context.Context, principal identity.Principal, conversationPublicID string) ([]ConversationContentSummary, error) {
|
|
access, conversationID, err := s.resolveConversationContent(ctx, s.pool, principal, conversationPublicID, identity.CapabilityConversationRead, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `SELECT public_id::text,summary_type,content,source_turn_ids,model_ref,prompt_version,generation_version,status,created_at FROM conversation_summary WHERE tenant_id=$1::uuid AND conversation_id=$2::uuid ORDER BY generation_version DESC,summary_type`, access.tenantID, conversationID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list conversation summaries: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
items := make([]ConversationContentSummary, 0)
|
|
for rows.Next() {
|
|
var item ConversationContentSummary
|
|
var sourceJSON []byte
|
|
if err := rows.Scan(&item.ID, &item.Type, &item.Content, &sourceJSON, &item.ModelRef, &item.PromptVersion, &item.GenerationVersion, &item.Status, &item.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := json.Unmarshal(sourceJSON, &item.SourceTurnIDs); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) RegenerateConversationSummary(ctx context.Context, principal identity.Principal, conversationPublicID, requestID string) (int64, error) {
|
|
var generationVersion int64
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, conversationID, err := s.resolveConversationContent(ctx, tx, principal, conversationPublicID, identity.CapabilityStoryReview, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var status string
|
|
var maxSequence int64
|
|
if err := tx.QueryRow(ctx, `SELECT c.status,(SELECT COALESCE(MAX(sequence_no),0) FROM conversation_turn ct WHERE ct.conversation_id=c.id AND ct.tenant_id=c.tenant_id) FROM conversation c WHERE c.id=$1::uuid`, conversationID).Scan(&status, &maxSequence); err != nil {
|
|
return err
|
|
}
|
|
if (status != "COMPLETED" && status != "FAILED") || maxSequence == 0 {
|
|
return ErrConversationSummaryNotReady
|
|
}
|
|
if err := tx.QueryRow(ctx, `SELECT GREATEST($2::bigint,COALESCE(MAX(generation_version),0))+1 FROM conversation_summary WHERE tenant_id=$1::uuid AND conversation_id=$3::uuid`, access.tenantID, maxSequence, conversationID).Scan(&generationVersion); err != nil {
|
|
return err
|
|
}
|
|
payload, _ := json.Marshal(map[string]any{"generation_version": generationVersion, "reason": "manual_regeneration"})
|
|
dedupe := fmt.Sprintf("content-regenerate:%s:%d", conversationID, generationVersion)
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &access.tenantID, AggregateType: "conversation", AggregateID: conversationID, EventType: "conversation.content_extraction_requested", DeduplicationKey: &dedupe, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
resourceID := conversationPublicID
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: &access.tenantID, ActorType: "user", ActorPublicID: &access.userPublicID, Action: "conversation.summary_regenerate", ResourceType: "conversation", ResourceID: &resourceID, RequestID: requestID, Purpose: "使用当前激活的大模型重新总结聊天", Result: "queued", Detail: map[string]any{"generation_version": generationVersion}})
|
|
})
|
|
return generationVersion, err
|
|
}
|
|
|
|
func (s *Service) ListSubjectStories(ctx context.Context, principal identity.Principal, subjectPublicID, status string) ([]Story, error) {
|
|
access, _, err := resolveSubject(ctx, s.pool, principal, subjectPublicID, identity.CapabilitySubjectRead, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
callAccess, callAccessErr := callmemory.ResolveAccess(ctx, s.pool, principal, subjectPublicID, false)
|
|
canReadCallSources := callAccessErr == nil && callAccess.Sensitive
|
|
rows, err := s.pool.Query(ctx, storySelectSQL+` WHERE sd.tenant_id=$1::uuid AND sd.subject_id=$2::uuid AND call_story_available(sd.id) AND (NULLIF($3,'') IS NULL OR sd.status=$3) ORDER BY sd.created_at DESC,sd.public_id`, access.tenantID, access.subjectID, status)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list subject stories: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
items := make([]Story, 0)
|
|
for rows.Next() {
|
|
item, err := scanStory(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if item.PrivateSources && !canReadCallSources {
|
|
continue
|
|
}
|
|
item.Sources = []StorySource{}
|
|
item.Facts = []FactCandidate{}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) GetStory(ctx context.Context, principal identity.Principal, storyPublicID string) (Story, error) {
|
|
_, story, err := s.resolveStory(ctx, s.pool, principal, storyPublicID, identity.CapabilitySubjectRead, false)
|
|
if err != nil {
|
|
return Story{}, err
|
|
}
|
|
if err := s.loadStoryChildren(ctx, &story); err != nil {
|
|
return Story{}, err
|
|
}
|
|
return story, nil
|
|
}
|
|
|
|
func (s *Service) UpdateFactCandidate(ctx context.Context, principal identity.Principal, storyPublicID, factPublicID, requestID string, expectedVersion int64, input UpdateFactCandidateInput) (FactCandidate, error) {
|
|
var result FactCandidate
|
|
if input.ErrorType != nil && !IsContentErrorType(*input.ErrorType) {
|
|
return result, ErrFactReviewStateConflict
|
|
}
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, story, err := s.resolveStory(ctx, tx, principal, storyPublicID, identity.CapabilityStoryReview, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if story.Status == "CONFIRMED" || story.Status == "REJECTED" || story.Status == "SUPERSEDED" {
|
|
return ErrFactReviewStateConflict
|
|
}
|
|
var internalID, factType, content, statementBasis, sensitivity, status string
|
|
var confidence *float64
|
|
var sourceJSON []byte
|
|
var version int64
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT fc.id::text,fc.fact_type,fc.content,fc.statement_basis,fc.confidence,fc.sensitivity,fc.source_turn_ids,fc.status,fc.version
|
|
FROM fact_candidate fc JOIN story_draft sd ON sd.id=fc.story_id AND sd.tenant_id=fc.tenant_id
|
|
WHERE fc.public_id=$1::uuid AND sd.public_id=$2::uuid FOR UPDATE OF fc`, factPublicID, storyPublicID).Scan(
|
|
&internalID, &factType, &content, &statementBasis, &confidence, &sensitivity, &sourceJSON, &status, &version,
|
|
); errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNotFound
|
|
} else if err != nil {
|
|
return fmt.Errorf("lock fact candidate: %w", err)
|
|
}
|
|
if version != expectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
if status == "REJECTED" || status == "SUPERSEDED" || (status == "CONFIRMED" && (input.Status == nil || *input.Status != "SUPERSEDED")) {
|
|
return ErrFactReviewStateConflict
|
|
}
|
|
previousFactType, previousContent, previousSensitivity, previousStatus := factType, content, sensitivity, status
|
|
if input.FactType != nil {
|
|
factType = strings.TrimSpace(*input.FactType)
|
|
}
|
|
if input.Content != nil {
|
|
content = strings.TrimSpace(*input.Content)
|
|
}
|
|
if input.Sensitivity != nil {
|
|
sensitivity = *input.Sensitivity
|
|
}
|
|
if input.Status != nil {
|
|
status = *input.Status
|
|
}
|
|
if status != "CANDIDATE" && status != "CONFIRMED" && status != "REJECTED" && status != "SUPERSEDED" {
|
|
return ErrFactReviewStateConflict
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO fact_candidate_revision(fact_candidate_id,version_no,fact_type,content,sensitivity,status,edited_by,edit_reason)
|
|
VALUES($1::uuid,$2,$3,$4,$5,$6,$7::uuid,$8) ON CONFLICT(fact_candidate_id,version_no) DO NOTHING`, internalID, version, previousFactType, previousContent, previousSensitivity, previousStatus, access.userID, input.Reason); err != nil {
|
|
return err
|
|
}
|
|
row := tx.QueryRow(ctx, `
|
|
UPDATE fact_candidate SET fact_type=$1,content=$2,sensitivity=$3,status=$4,
|
|
reviewed_by=$5::uuid,reviewed_at=now(),review_reason=$6,version=version+1,updated_at=now()
|
|
WHERE id=$7::uuid
|
|
RETURNING public_id::text,fact_type,content,statement_basis,confidence,sensitivity,source_turn_ids,status,version`,
|
|
factType, content, sensitivity, status, access.userID, input.Reason, internalID)
|
|
if err := scanFactCandidate(row, &result); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO fact_candidate_revision(fact_candidate_id,version_no,fact_type,content,sensitivity,status,edited_by,edit_reason,error_type)
|
|
VALUES($1::uuid,$2,$3,$4,$5,$6,$7::uuid,$8,$9)`, internalID, result.Version, result.FactType, result.Content, result.Sensitivity, result.Status, access.userID, input.Reason, input.ErrorType); err != nil {
|
|
return err
|
|
}
|
|
action := reviewAction(result.Status, input.Content != nil || input.FactType != nil || input.Sensitivity != nil)
|
|
if _, err := tx.Exec(ctx, `INSERT INTO content_review_signal(tenant_id,subject_id,resource_type,resource_public_id,action,error_type,reason,model_ref,prompt_version,generation_version,actor_id)
|
|
SELECT sd.tenant_id,sd.subject_id,'fact',$1::uuid,$2,$3,$4,sr.model_ref,sr.prompt_version,sd.generation_version,$5::uuid
|
|
FROM story_draft sd JOIN story_revision sr ON sr.story_id=sd.id AND sr.revision_no=1 WHERE sd.public_id=$6::uuid`, factPublicID, action, input.ErrorType, input.Reason, access.userID, storyPublicID); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{
|
|
TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID),
|
|
Action: "fact_candidate.update", ResourceType: "fact_candidate", ResourceID: strPtr(factPublicID),
|
|
RequestID: requestID, Result: "success", Detail: map[string]any{"story_id": storyPublicID, "status": result.Status, "reason": input.Reason},
|
|
})
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) ReviseStory(ctx context.Context, principal identity.Principal, storyPublicID, content, reason, requestID string, expectedVersion int64, errorType string) (Story, error) {
|
|
var result Story
|
|
if errorType != "" && !IsContentErrorType(errorType) {
|
|
return result, ErrReviewStateConflict
|
|
}
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, current, err := s.resolveStory(ctx, tx, principal, storyPublicID, identity.CapabilityStoryReview, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if current.Version != expectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
if current.Status == "CONFIRMED" || current.Status == "REJECTED" || current.Status == "SUPERSEDED" {
|
|
return ErrReviewStateConflict
|
|
}
|
|
var currentRevisionID, newRevisionID string
|
|
if err := tx.QueryRow(ctx, `SELECT current_revision_id::text FROM story_draft WHERE public_id=$1::uuid`, storyPublicID).Scan(¤tRevisionID); err != nil {
|
|
return err
|
|
}
|
|
if err := tx.QueryRow(ctx, `INSERT INTO story_revision(story_id,revision_no,content,author_type,author_id,edit_reason,base_revision_id,error_type) SELECT id,$2,$3,'user',$4::uuid,$5,$6::uuid,NULLIF($7,'') FROM story_draft WHERE public_id=$1::uuid RETURNING id::text`, storyPublicID, current.RevisionNo+1, content, access.userID, reason, currentRevisionID, errorType).Scan(&newRevisionID); err != nil {
|
|
return fmt.Errorf("insert story revision: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO story_source(story_revision_id,conversation_turn_id,call_session_id,call_utterance_id,call_source_revision,import_turn_id,source_start_offset,source_end_offset,source_role) SELECT $1::uuid,conversation_turn_id,call_session_id,call_utterance_id,call_source_revision,import_turn_id,source_start_offset,source_end_offset,source_role FROM story_source WHERE story_revision_id=$2::uuid`, newRevisionID, currentRevisionID); err != nil {
|
|
return err
|
|
}
|
|
row := tx.QueryRow(ctx, storyUpdateReturningSQL, `IN_REVIEW`, newRevisionID, access.userID, reason, storyPublicID)
|
|
result, err = scanStory(row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := insertStoryReviewSignal(ctx, tx, access, storyPublicID, "EDITED", errorType, reason); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID), Action: "story.revise", ResourceType: "story", ResourceID: strPtr(storyPublicID), RequestID: requestID, Result: "success", Detail: map[string]any{"revision_no": result.RevisionNo, "reason": reason}})
|
|
})
|
|
if err != nil {
|
|
return Story{}, err
|
|
}
|
|
if err := s.loadStoryChildren(ctx, &result); err != nil {
|
|
return Story{}, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Service) RejectStory(ctx context.Context, principal identity.Principal, storyPublicID, reason, requestID string, expectedVersion int64, errorType string) (Story, error) {
|
|
var result Story
|
|
if errorType != "" && !IsContentErrorType(errorType) {
|
|
return result, ErrReviewStateConflict
|
|
}
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, current, err := s.resolveStory(ctx, tx, principal, storyPublicID, identity.CapabilityStoryReview, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if current.Version != expectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
if current.Status == "CONFIRMED" || current.Status == "REJECTED" || current.Status == "SUPERSEDED" {
|
|
return ErrReviewStateConflict
|
|
}
|
|
row := tx.QueryRow(ctx, storyStatusReturningSQL, "REJECTED", access.userID, reason, storyPublicID)
|
|
result, err = scanStory(row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE fact_candidate SET status='REJECTED',reviewed_by=$1::uuid,reviewed_at=now(),review_reason=$2,version=version+1,updated_at=now() WHERE story_id=(SELECT id FROM story_draft WHERE public_id=$3::uuid) AND status='CANDIDATE'`, access.userID, reason, storyPublicID); err != nil {
|
|
return err
|
|
}
|
|
if err := insertStoryReviewSignal(ctx, tx, access, storyPublicID, "REJECTED", errorType, reason); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID), Action: "story.reject", ResourceType: "story", ResourceID: strPtr(storyPublicID), RequestID: requestID, Result: "success", Detail: map[string]any{"reason": reason}})
|
|
})
|
|
if err != nil {
|
|
return Story{}, err
|
|
}
|
|
if err := s.loadStoryChildren(ctx, &result); err != nil {
|
|
return Story{}, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Service) ConfirmStory(ctx context.Context, principal identity.Principal, storyPublicID, reason, requestID string, expectedVersion int64) (Story, CuratedMemory, error) {
|
|
var story Story
|
|
var memory CuratedMemory
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
access, current, err := s.resolveStory(ctx, tx, principal, storyPublicID, identity.CapabilityStoryReview, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
grant, e := callmemory.ResolveAccess(ctx, tx, principal, current.SubjectID, true)
|
|
if e != nil || (!grant.Sensitive && current.Sensitivity != "normal") {
|
|
return ErrAccessDenied
|
|
}
|
|
if current.Version != expectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
if current.Status == "CONFIRMED" || current.Status == "REJECTED" || current.Status == "SUPERSEDED" {
|
|
return ErrReviewStateConflict
|
|
}
|
|
var sourceCount int
|
|
if err := tx.QueryRow(ctx, `SELECT count(*) FROM story_source WHERE story_revision_id=(SELECT current_revision_id FROM story_draft WHERE public_id=$1::uuid)`, storyPublicID).Scan(&sourceCount); err != nil {
|
|
return err
|
|
}
|
|
if sourceCount == 0 {
|
|
return ErrStorySourceRequired
|
|
}
|
|
var consent 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='personalized_memory' AND withdrawn_at IS NULL AND (expires_at IS NULL OR expires_at>now()))`, access.tenantID, access.subjectID).Scan(&consent); err != nil {
|
|
return err
|
|
}
|
|
if !consent {
|
|
return ErrPersonalizedMemoryConsentNeeded
|
|
}
|
|
storyRow := tx.QueryRow(ctx, storyStatusReturningSQL, "CONFIRMED", access.userID, reason, storyPublicID)
|
|
story, err = scanStory(storyRow)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sourceRefs := []map[string]any{{"type": "story", "story_id": storyPublicID, "revision_no": story.RevisionNo}}
|
|
if story.PrivateSources {
|
|
rows, err := tx.Query(ctx, `SELECT ss.call_session_id::text,ss.call_utterance_id,ss.call_source_revision FROM story_source ss JOIN story_draft sd ON sd.current_revision_id=ss.story_revision_id WHERE sd.public_id=$1::uuid AND ss.call_session_id IS NOT NULL ORDER BY ss.call_utterance_id`, storyPublicID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for rows.Next() {
|
|
var sessionID, utteranceID string
|
|
var sourceRevision int64
|
|
if err := rows.Scan(&sessionID, &utteranceID, &sourceRevision); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
sourceRefs = append(sourceRefs, map[string]any{"type": "call_utterance", "session_id": sessionID, "utterance_id": utteranceID, "source_revision": sourceRevision})
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
importRows, err := tx.Query(ctx, `SELECT it.import_id::text,it.id::text FROM story_source ss JOIN story_draft sd ON sd.current_revision_id=ss.story_revision_id JOIN memoir_import_turn it ON it.id=ss.import_turn_id WHERE sd.public_id=$1::uuid ORDER BY it.sequence_no`, storyPublicID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for importRows.Next() {
|
|
var importID, turnID string
|
|
if err := importRows.Scan(&importID, &turnID); err != nil {
|
|
importRows.Close()
|
|
return err
|
|
}
|
|
sourceRefs = append(sourceRefs, map[string]any{"type": "import_turn", "import_id": importID, "turn_id": turnID})
|
|
}
|
|
err = importRows.Err()
|
|
importRows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sourceJSON, _ := json.Marshal(sourceRefs)
|
|
digest := sha256.Sum256([]byte(story.Content))
|
|
var memoryID, memoryVersionID string
|
|
if err := tx.QueryRow(ctx, `INSERT INTO curated_memory(tenant_id,subject_id,status,sensitivity,valid_from) VALUES($1::uuid,$2::uuid,'ACTIVE',$3,now()) RETURNING id::text,public_id::text`, access.tenantID, access.subjectID, story.Sensitivity).Scan(&memoryID, &memory.ID); err != nil {
|
|
return fmt.Errorf("create curated memory: %w", err)
|
|
}
|
|
if err := tx.QueryRow(ctx, `INSERT INTO memory_version(memory_id,version_no,content,source_refs,confirmed_by,change_reason,content_digest) VALUES($1::uuid,1,$2,$3::jsonb,$4::uuid,$5,$6) RETURNING id::text`, memoryID, story.Content, sourceJSON, access.userID, reason, hex.EncodeToString(digest[:])).Scan(&memoryVersionID); err != nil {
|
|
return fmt.Errorf("create memory version: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE curated_memory SET current_version_id=$1::uuid WHERE id=$2::uuid`, memoryVersionID, memoryID); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO memory_index_entry(tenant_id,subject_id,memory_id,memory_version_id,index_type,status) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,'pgvector','PENDING')`, access.tenantID, access.subjectID, memoryID, memoryVersionID); err != nil {
|
|
return err
|
|
}
|
|
if s.providerMemory {
|
|
if _, err := tx.Exec(ctx, `INSERT INTO memory_index_entry(tenant_id,subject_id,memory_id,memory_version_id,index_type,status) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,'provider','PENDING')`, access.tenantID, access.subjectID, memoryID, memoryVersionID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE fact_candidate SET status='CONFIRMED',reviewed_by=$1::uuid,reviewed_at=now(),review_reason=$2,version=version+1,updated_at=now() WHERE story_id=(SELECT id FROM story_draft WHERE public_id=$3::uuid) AND status='CANDIDATE'`, access.userID, reason, storyPublicID); err != nil {
|
|
return err
|
|
}
|
|
if err := insertStoryReviewSignal(ctx, tx, access, storyPublicID, "CONFIRMED", "", reason); err != nil {
|
|
return err
|
|
}
|
|
dedupe := "memory-index-v1:" + memoryVersionID
|
|
payload, _ := json.Marshal(map[string]any{"memory_version_id": memoryVersionID})
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &access.tenantID, AggregateType: "curated_memory", AggregateID: memoryID, EventType: "memory.index_requested", DeduplicationKey: &dedupe, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
if s.providerMemory {
|
|
providerDedupe := "memory-provider-sync-v1:" + memoryVersionID
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &access.tenantID, AggregateType: "curated_memory", AggregateID: memoryID, EventType: "memory.provider_sync_requested", DeduplicationKey: &providerDedupe, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID), Action: "story.confirm", ResourceType: "story", ResourceID: strPtr(storyPublicID), RequestID: requestID, Result: "success", Detail: map[string]any{"memory_id": memory.ID, "revision_no": story.RevisionNo}}); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return Story{}, CuratedMemory{}, err
|
|
}
|
|
memory, err = s.GetMemory(ctx, principal, memory.ID)
|
|
if err != nil {
|
|
return Story{}, CuratedMemory{}, err
|
|
}
|
|
if err := s.loadStoryChildren(ctx, &story); err != nil {
|
|
return Story{}, CuratedMemory{}, err
|
|
}
|
|
return story, memory, nil
|
|
}
|
|
|
|
func (s *Service) ListSubjectMemories(ctx context.Context, principal identity.Principal, subjectPublicID, query string) ([]CuratedMemory, error) {
|
|
grant, grantErr := callmemory.ResolveAccess(ctx, s.pool, principal, subjectPublicID, false)
|
|
if grantErr != nil {
|
|
return nil, ErrAccessDenied
|
|
}
|
|
access, _, err := resolveSubject(ctx, s.pool, principal, subjectPublicID, identity.CapabilitySubjectRead, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, memorySelectSQL+` WHERE cm.tenant_id=$1::uuid AND cm.subject_id=$2::uuid AND cm.status<>'DELETED' AND (NULLIF($3,'') IS NULL OR strpos(lower(mv.content),lower($3))>0) ORDER BY cm.updated_at DESC,cm.public_id`, access.tenantID, access.subjectID, strings.TrimSpace(query))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := make([]CuratedMemory, 0)
|
|
for rows.Next() {
|
|
item, err := scanMemory(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
if !grant.Sensitive && item.Sensitivity != "normal" {
|
|
items = items[:len(items)-1]
|
|
}
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) GetMemory(ctx context.Context, principal identity.Principal, memoryPublicID string) (CuratedMemory, error) {
|
|
if err := s.checkMemoryGrant(ctx, s.pool, principal, memoryPublicID, false); err != nil {
|
|
return CuratedMemory{}, err
|
|
}
|
|
result, err := scanMemory(s.pool.QueryRow(ctx, memorySelectSQL+` JOIN household h ON h.id=sp.household_id AND h.tenant_id=sp.tenant_id JOIN household_member hm ON hm.household_id=h.id AND hm.status='active' JOIN user_account member_account ON member_account.id=hm.user_id AND member_account.tenant_id=h.tenant_id AND member_account.status='active' WHERE cm.public_id=$1::uuid AND sp.status IN ('active','suspended') AND member_account.oidc_issuer=$2 AND member_account.oidc_subject=$3`, memoryPublicID, principal.Issuer, principal.Subject))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return CuratedMemory{}, ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return CuratedMemory{}, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Service) UpdateMemory(ctx context.Context, principal identity.Principal, memoryPublicID, requestID string, expectedVersion int64, input UpdateMemoryInput) (CuratedMemory, error) {
|
|
if input.ErrorType != nil && !IsContentErrorType(*input.ErrorType) {
|
|
return CuratedMemory{}, ErrReviewStateConflict
|
|
}
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
if err := s.checkMemoryGrant(ctx, tx, principal, memoryPublicID, true); err != nil {
|
|
return err
|
|
}
|
|
var internalID, tenantID, subjectID, userID, userPublicID, role string
|
|
var currentVersionID, currentContent, currentStatus string
|
|
var currentVersionNo, resourceVersion int64
|
|
var sourceRefs []byte
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT cm.id::text,cm.tenant_id::text,cm.subject_id::text,member_account.id::text,
|
|
member_account.public_id::text,hm.role,cm.current_version_id::text,mv.version_no,
|
|
mv.content,mv.source_refs,cm.status,cm.version
|
|
FROM curated_memory cm
|
|
JOIN memory_version mv ON mv.id=cm.current_version_id
|
|
JOIN subject_profile sp ON sp.id=cm.subject_id AND sp.tenant_id=cm.tenant_id
|
|
JOIN household h ON h.id=sp.household_id AND h.tenant_id=sp.tenant_id
|
|
JOIN household_member hm ON hm.household_id=h.id AND hm.status='active'
|
|
JOIN user_account member_account ON member_account.id=hm.user_id AND member_account.tenant_id=h.tenant_id AND member_account.status='active'
|
|
WHERE cm.public_id=$1::uuid AND cm.status<>'DELETED' AND sp.status IN ('active','suspended')
|
|
AND member_account.oidc_issuer=$2 AND member_account.oidc_subject=$3
|
|
FOR UPDATE OF cm`, memoryPublicID, principal.Issuer, principal.Subject).Scan(
|
|
&internalID, &tenantID, &subjectID, &userID, &userPublicID, &role, ¤tVersionID,
|
|
¤tVersionNo, ¤tContent, &sourceRefs, ¤tStatus, &resourceVersion)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("resolve memory update: %w", err)
|
|
}
|
|
if !identity.Can(identity.Role(role), identity.CapabilityStoryReview) {
|
|
return ErrAccessDenied
|
|
}
|
|
if resourceVersion != expectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
if currentStatus == "PENDING_DELETE" {
|
|
return ErrReviewStateConflict
|
|
}
|
|
|
|
targetContent := currentContent
|
|
if input.Content != nil {
|
|
targetContent = strings.TrimSpace(*input.Content)
|
|
}
|
|
targetStatus := currentStatus
|
|
if input.Status != nil {
|
|
targetStatus = *input.Status
|
|
}
|
|
if targetContent == currentContent && targetStatus == currentStatus {
|
|
return ErrReviewStateConflict
|
|
}
|
|
if targetStatus == "ACTIVE" {
|
|
var consent 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='personalized_memory' AND withdrawn_at IS NULL AND (expires_at IS NULL OR expires_at>now()))`, tenantID, subjectID).Scan(&consent); err != nil {
|
|
return err
|
|
}
|
|
if !consent {
|
|
return ErrPersonalizedMemoryConsentNeeded
|
|
}
|
|
}
|
|
|
|
newVersionID := currentVersionID
|
|
if targetContent != currentContent {
|
|
digest := sha256.Sum256([]byte(targetContent))
|
|
if err := tx.QueryRow(ctx, `
|
|
INSERT INTO memory_version(memory_id,version_no,content,source_refs,confirmed_by,change_reason,content_digest,error_type)
|
|
VALUES($1::uuid,$2,$3,$4::jsonb,$5::uuid,$6,$7,$8) RETURNING id::text`,
|
|
internalID, currentVersionNo+1, targetContent, sourceRefs, userID, input.Reason, hex.EncodeToString(digest[:]), input.ErrorType).Scan(&newVersionID); err != nil {
|
|
return fmt.Errorf("create memory revision: %w", err)
|
|
}
|
|
}
|
|
|
|
if _, err := tx.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,status='STALE',last_error='memory version or status changed',updated_at=now() WHERE memory_id=$1::uuid AND status IN ('PENDING','ACTIVE')`, internalID); err != nil {
|
|
return err
|
|
}
|
|
if targetStatus == "ACTIVE" {
|
|
if newVersionID == currentVersionID {
|
|
if _, err := tx.Exec(ctx, `UPDATE memory_index_entry SET status='PENDING',last_error=NULL,updated_at=now() WHERE memory_version_id=$1::uuid AND index_type='pgvector'`, newVersionID); err != nil {
|
|
return err
|
|
}
|
|
} else if _, err := tx.Exec(ctx, `INSERT INTO memory_index_entry(tenant_id,subject_id,memory_id,memory_version_id,index_type,status) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,'pgvector','PENDING')`, tenantID, subjectID, internalID, newVersionID); err != nil {
|
|
return err
|
|
}
|
|
if s.providerMemory {
|
|
if _, err := tx.Exec(ctx, `INSERT INTO memory_index_entry(tenant_id,subject_id,memory_id,memory_version_id,index_type,status,last_error,updated_at)
|
|
VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,'provider','PENDING',NULL,now())
|
|
ON CONFLICT(memory_version_id,index_type) DO UPDATE SET status='PENDING',last_error=NULL,updated_at=now()`, tenantID, subjectID, internalID, newVersionID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE curated_memory SET current_version_id=$1::uuid,status=$2,version=version+1,updated_at=now() WHERE id=$3::uuid`, newVersionID, targetStatus, internalID); err != nil {
|
|
return err
|
|
}
|
|
action := "EDITED"
|
|
if targetStatus == "DISABLED" {
|
|
action = "DISABLED"
|
|
} else if targetContent == currentContent {
|
|
action = "CONFIRMED"
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO content_review_signal(tenant_id,subject_id,resource_type,resource_public_id,action,error_type,reason,model_ref,prompt_version,generation_version,actor_id)
|
|
SELECT $1::uuid,$2::uuid,'memory',$3::uuid,$4,$5,$6,sr.model_ref,sr.prompt_version,sd.generation_version,$7::uuid
|
|
FROM memory_version mv
|
|
LEFT JOIN story_draft sd ON sd.public_id::text=mv.source_refs->0->>'story_id'
|
|
LEFT JOIN story_revision sr ON sr.id=sd.current_revision_id
|
|
WHERE mv.id=$8::uuid`, tenantID, subjectID, memoryPublicID, action, input.ErrorType, input.Reason, userID, newVersionID); err != nil {
|
|
return err
|
|
}
|
|
if targetStatus == "ACTIVE" {
|
|
dedupe := fmt.Sprintf("memory-index-update-v%d:%s", resourceVersion+1, newVersionID)
|
|
payload, _ := json.Marshal(map[string]any{"memory_version_id": newVersionID})
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &tenantID, AggregateType: "curated_memory", AggregateID: internalID, EventType: "memory.index_requested", DeduplicationKey: &dedupe, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
if s.providerMemory {
|
|
providerDedupe := fmt.Sprintf("memory-provider-sync-update-v%d:%s", resourceVersion+1, newVersionID)
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &tenantID, AggregateType: "curated_memory", AggregateID: internalID, EventType: "memory.provider_sync_requested", DeduplicationKey: &providerDedupe, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
} else if s.providerMemory {
|
|
providerDedupe := fmt.Sprintf("memory-provider-delete-disabled-v%d:%s", resourceVersion+1, internalID)
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &tenantID, AggregateType: "curated_memory", AggregateID: internalID, EventType: "memory.provider_delete_requested", DeduplicationKey: &providerDedupe, Payload: []byte(`{"reason":"memory disabled"}`)}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(tenantID), ActorType: "user", ActorPublicID: strPtr(userPublicID), Action: "memory.update", ResourceType: "memory", ResourceID: strPtr(memoryPublicID), RequestID: requestID, Result: "success", Detail: map[string]any{"from_status": currentStatus, "to_status": targetStatus, "content_revised": targetContent != currentContent, "reason": input.Reason}})
|
|
})
|
|
if err != nil {
|
|
return CuratedMemory{}, err
|
|
}
|
|
return s.GetMemory(ctx, principal, memoryPublicID)
|
|
}
|
|
|
|
func (s *Service) RequestMemoryDeletion(ctx context.Context, principal identity.Principal, memoryPublicID, reason, requestID string, expectedVersion int64) (CuratedMemory, error) {
|
|
err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
|
if err := s.checkMemoryGrant(ctx, tx, principal, memoryPublicID, true); err != nil {
|
|
return err
|
|
}
|
|
var internalID, tenantID, userPublicID, role, status string
|
|
var version int64
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT cm.id::text,cm.tenant_id::text,member_account.public_id::text,hm.role,cm.status,cm.version
|
|
FROM curated_memory cm
|
|
JOIN subject_profile sp ON sp.id=cm.subject_id AND sp.tenant_id=cm.tenant_id
|
|
JOIN household h ON h.id=sp.household_id AND h.tenant_id=sp.tenant_id
|
|
JOIN household_member hm ON hm.household_id=h.id AND hm.status='active'
|
|
JOIN user_account member_account ON member_account.id=hm.user_id AND member_account.tenant_id=h.tenant_id AND member_account.status='active'
|
|
WHERE cm.public_id=$1::uuid AND sp.status IN ('active','suspended') AND member_account.oidc_issuer=$2 AND member_account.oidc_subject=$3
|
|
FOR UPDATE OF cm`, memoryPublicID, principal.Issuer, principal.Subject).Scan(&internalID, &tenantID, &userPublicID, &role, &status, &version)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("resolve memory deletion: %w", err)
|
|
}
|
|
if !identity.Can(identity.Role(role), identity.CapabilityStoryReview) {
|
|
return ErrAccessDenied
|
|
}
|
|
if version != expectedVersion {
|
|
return ErrVersionConflict
|
|
}
|
|
if status == "DELETED" || status == "PENDING_DELETE" {
|
|
return nil
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE curated_memory SET status='PENDING_DELETE',valid_to=COALESCE(valid_to,now()),version=version+1,updated_at=now() WHERE id=$1::uuid`, internalID); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,status='STALE',last_error='memory deletion requested',updated_at=now() WHERE memory_id=$1::uuid AND status<>'DELETED'`, internalID); err != nil {
|
|
return err
|
|
}
|
|
dedupe := "memory-delete:" + internalID
|
|
payload, _ := json.Marshal(map[string]any{"reason": reason})
|
|
if s.providerMemory {
|
|
providerDedupe := "memory-provider-delete:" + internalID
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &tenantID, AggregateType: "curated_memory", AggregateID: internalID, EventType: "memory.provider_delete_requested", DeduplicationKey: &providerDedupe, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &tenantID, AggregateType: "curated_memory", AggregateID: internalID, EventType: "memory.delete_requested", DeduplicationKey: &dedupe, Payload: payload}); err != nil {
|
|
return err
|
|
}
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(tenantID), ActorType: "user", ActorPublicID: strPtr(userPublicID), Action: "memory.delete.request", ResourceType: "memory", ResourceID: strPtr(memoryPublicID), RequestID: requestID, Result: "success", Detail: map[string]any{"reason": reason}})
|
|
})
|
|
if err != nil {
|
|
return CuratedMemory{}, err
|
|
}
|
|
return s.GetMemory(ctx, principal, memoryPublicID)
|
|
}
|
|
|
|
type contentAccess struct {
|
|
tenantID, subjectID, userID, userPublicID string
|
|
role identity.Role
|
|
}
|
|
|
|
func reviewAction(status string, edited bool) string {
|
|
if edited && status == "CANDIDATE" {
|
|
return "EDITED"
|
|
}
|
|
switch status {
|
|
case "CONFIRMED":
|
|
return "CONFIRMED"
|
|
case "REJECTED", "SUPERSEDED":
|
|
return "REJECTED"
|
|
default:
|
|
return "EDITED"
|
|
}
|
|
}
|
|
|
|
func insertStoryReviewSignal(ctx context.Context, tx pgx.Tx, access contentAccess, storyPublicID, action, errorType, reason string) error {
|
|
_, err := tx.Exec(ctx, `INSERT INTO content_review_signal(tenant_id,subject_id,resource_type,resource_public_id,action,error_type,reason,model_ref,prompt_version,generation_version,actor_id)
|
|
SELECT sd.tenant_id,sd.subject_id,'story',sd.public_id,$2,NULLIF($3,''),$4,sr.model_ref,sr.prompt_version,sd.generation_version,$5::uuid
|
|
FROM story_draft sd JOIN story_revision sr ON sr.story_id=sd.id AND sr.revision_no=1 WHERE sd.public_id=$1::uuid`, storyPublicID, action, errorType, reason, access.userID)
|
|
return err
|
|
}
|
|
|
|
func (s *Service) resolveConversationContent(ctx context.Context, q queryRow, p identity.Principal, publicID string, cap identity.Capability, lock bool) (contentAccess, string, error) {
|
|
sql := `SELECT c.tenant_id::text,c.subject_id::text,ua.id::text,ua.public_id::text,hm.role,c.id::text FROM conversation c JOIN subject_profile sp ON sp.id=c.subject_id AND sp.tenant_id=c.tenant_id AND sp.status IN ('active','suspended') JOIN household h ON h.id=c.household_id AND h.tenant_id=c.tenant_id JOIN household_member hm ON hm.household_id=h.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 c.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`
|
|
if lock {
|
|
sql += ` FOR UPDATE OF c`
|
|
}
|
|
var a contentAccess
|
|
var role, conversationID string
|
|
err := q.QueryRow(ctx, sql, publicID, p.Issuer, p.Subject).Scan(&a.tenantID, &a.subjectID, &a.userID, &a.userPublicID, &role, &conversationID)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return contentAccess{}, "", ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return contentAccess{}, "", err
|
|
}
|
|
a.role = identity.Role(role)
|
|
if !identity.Can(a.role, cap) {
|
|
return contentAccess{}, "", ErrAccessDenied
|
|
}
|
|
return a, conversationID, nil
|
|
}
|
|
func (s *Service) resolveStory(ctx context.Context, q queryRow, p identity.Principal, publicID string, cap identity.Capability, lock bool) (contentAccess, Story, error) {
|
|
if lock {
|
|
access, current, err := s.resolveStory(ctx, q, p, publicID, cap, false)
|
|
if err != nil {
|
|
return contentAccess{}, Story{}, err
|
|
}
|
|
if current.PrivateSources {
|
|
// Serialize all draft edits with extraction and source deletion, using
|
|
// the same subject-before-story lock order as memory mutations.
|
|
var epoch int64
|
|
if err := q.QueryRow(ctx, `SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid AND tenant_id=$2::uuid FOR UPDATE`, access.subjectID, access.tenantID).Scan(&epoch); err != nil {
|
|
return contentAccess{}, Story{}, err
|
|
}
|
|
}
|
|
}
|
|
sql := storySelectSQL + ` JOIN household h ON h.id=sp.household_id AND h.tenant_id=sp.tenant_id JOIN household_member hm ON hm.household_id=h.id AND hm.status='active' JOIN user_account member_account ON member_account.id=hm.user_id AND member_account.tenant_id=h.tenant_id AND member_account.status='active' WHERE sd.public_id=$1::uuid AND call_story_available(sd.id) AND sp.status IN ('active','suspended') AND member_account.oidc_issuer=$2 AND member_account.oidc_subject=$3`
|
|
if lock {
|
|
sql += ` FOR UPDATE OF sd`
|
|
}
|
|
var a contentAccess
|
|
var story Story
|
|
var role, tenantID, subjectInternal string
|
|
targets := storyScanTargets(&story, &tenantID, &subjectInternal)
|
|
targets = append(targets, &a.userID, &a.userPublicID, &role)
|
|
row := q.QueryRow(ctx, strings.Replace(sql, `sd.subject_id::text AS subject_internal_id`, `sd.subject_id::text AS subject_internal_id,member_account.id::text,member_account.public_id::text,hm.role`, 1), publicID, p.Issuer, p.Subject)
|
|
err := row.Scan(targets...)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return contentAccess{}, Story{}, ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return contentAccess{}, Story{}, err
|
|
}
|
|
a.tenantID = tenantID
|
|
a.subjectID = subjectInternal
|
|
a.role = identity.Role(role)
|
|
if !identity.Can(a.role, cap) {
|
|
return contentAccess{}, Story{}, ErrAccessDenied
|
|
}
|
|
if story.PrivateSources {
|
|
if lock {
|
|
var available bool
|
|
if err := q.QueryRow(ctx, `SELECT call_story_available(id) FROM story_draft WHERE public_id=$1::uuid`, publicID).Scan(&available); err != nil {
|
|
return contentAccess{}, Story{}, err
|
|
}
|
|
if !available {
|
|
return contentAccess{}, Story{}, ErrAccessDenied
|
|
}
|
|
}
|
|
grant, err := callmemory.ResolveAccess(ctx, q, p, story.SubjectID, cap == identity.CapabilityStoryReview)
|
|
if err != nil || !grant.Sensitive {
|
|
return contentAccess{}, Story{}, ErrAccessDenied
|
|
}
|
|
}
|
|
return a, story, nil
|
|
}
|
|
|
|
const storySelectSQL = `SELECT sd.public_id::text AS story_public_id,sp.public_id::text AS subject_public_id,COALESCE(c.public_id::text,'') AS conversation_public_id,COALESCE(sd.call_session_id::text,''),COALESCE(sd.import_id::text,''),EXISTS(SELECT 1 FROM story_source ps WHERE ps.story_revision_id=sd.current_revision_id AND (ps.call_session_id IS NOT NULL OR ps.import_turn_id IS NOT NULL)),sd.title,sr.content,sd.status,sd.confidence,sd.sensitivity,sd.generation_version,sr.revision_no,sr.author_type,COALESCE(sd.review_reason,''),sd.version,sd.created_at,sd.updated_at,sd.tenant_id::text AS tenant_internal_id,sd.subject_id::text AS subject_internal_id FROM story_draft sd JOIN subject_profile sp ON sp.id=sd.subject_id AND sp.tenant_id=sd.tenant_id LEFT JOIN conversation c ON c.id=sd.conversation_id AND c.tenant_id=sd.tenant_id JOIN story_revision sr ON sr.id=sd.current_revision_id`
|
|
const storyReturningColumns = `sd.public_id::text,sp.public_id::text,COALESCE((SELECT c.public_id::text FROM conversation c WHERE c.id=sd.conversation_id),''),COALESCE(sd.call_session_id::text,''),COALESCE(sd.import_id::text,''),EXISTS(SELECT 1 FROM story_source ps WHERE ps.story_revision_id=sd.current_revision_id AND (ps.call_session_id IS NOT NULL OR ps.import_turn_id IS NOT NULL)),sd.title,sr.content,sd.status,sd.confidence,sd.sensitivity,sd.generation_version,sr.revision_no,sr.author_type,COALESCE(sd.review_reason,''),sd.version,sd.created_at,sd.updated_at,sd.tenant_id::text,sd.subject_id::text`
|
|
const storyStatusReturningSQL = `UPDATE story_draft sd SET status=$1,reviewed_by=$2::uuid,reviewed_at=now(),review_reason=$3,version=sd.version+1,updated_at=now() FROM subject_profile sp,story_revision sr WHERE sd.public_id=$4::uuid AND sp.id=sd.subject_id AND sr.id=sd.current_revision_id AND call_story_available(sd.id) RETURNING ` + storyReturningColumns
|
|
const storyUpdateReturningSQL = `UPDATE story_draft sd SET status=$1,current_revision_id=$2::uuid,reviewed_by=$3::uuid,reviewed_at=now(),review_reason=$4,version=sd.version+1,updated_at=now() FROM subject_profile sp,story_revision sr WHERE sd.public_id=$5::uuid AND sp.id=sd.subject_id AND sr.id=$2::uuid AND call_story_available(sd.id) RETURNING ` + storyReturningColumns
|
|
|
|
func storyScanTargets(s *Story, tenantID, subjectInternal *string) []any {
|
|
return []any{&s.ID, &s.SubjectID, &s.ConversationID, &s.CallSessionID, &s.ImportID, &s.PrivateSources, &s.Title, &s.Content, &s.Status, &s.Confidence, &s.Sensitivity, &s.GenerationVersion, &s.RevisionNo, &s.AuthorType, &s.ReviewReason, &s.Version, &s.CreatedAt, &s.UpdatedAt, tenantID, subjectInternal}
|
|
}
|
|
func scanStory(row rowScanner) (Story, error) {
|
|
var s Story
|
|
var tenant, subject string
|
|
err := row.Scan(storyScanTargets(&s, &tenant, &subject)...)
|
|
return s, err
|
|
}
|
|
func (s *Service) loadStoryChildren(ctx context.Context, story *Story) error {
|
|
rows, err := s.pool.Query(ctx, `SELECT COALESCE(ct.public_id::text,u.utterance_id,it.id::text),COALESCE(ss.call_session_id::text,''),COALESCE(ss.call_utterance_id,''),ss.call_source_revision,
|
|
COALESCE(ct.sequence_no,it.sequence_no,row_number() OVER(ORDER BY u.source_at,u.utterance_id)),ss.source_role,COALESCE(ct.final_text,u.text_content,it.content),ss.source_start_offset,ss.source_end_offset
|
|
FROM story_draft sd JOIN story_source ss ON ss.story_revision_id=sd.current_revision_id
|
|
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 sd.public_id=$1::uuid AND call_story_available(sd.id) ORDER BY ct.sequence_no,u.source_at,u.utterance_id`, story.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
story.Sources = []StorySource{}
|
|
for rows.Next() {
|
|
var x StorySource
|
|
if err := rows.Scan(&x.TurnID, &x.CallSessionID, &x.UtteranceID, &x.SourceRevision, &x.SequenceNo, &x.Speaker, &x.Text, &x.StartOffset, &x.EndOffset); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
story.Sources = append(story.Sources, x)
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rows, err = s.pool.Query(ctx, `SELECT fc.public_id::text,fc.fact_type,fc.content,fc.statement_basis,fc.confidence,fc.sensitivity,fc.source_turn_ids,fc.status,fc.version FROM fact_candidate fc JOIN story_draft sd ON sd.id=fc.story_id WHERE sd.public_id=$1::uuid AND call_story_available(sd.id) ORDER BY fc.created_at,fc.public_id`, story.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
story.Facts = []FactCandidate{}
|
|
for rows.Next() {
|
|
var x FactCandidate
|
|
if err := scanFactCandidate(rows, &x); err != nil {
|
|
return err
|
|
}
|
|
story.Facts = append(story.Facts, x)
|
|
}
|
|
return rows.Err()
|
|
}
|
|
|
|
func scanFactCandidate(row rowScanner, result *FactCandidate) error {
|
|
var source []byte
|
|
if err := row.Scan(&result.ID, &result.FactType, &result.Content, &result.StatementBasis, &result.Confidence, &result.Sensitivity, &source, &result.Status, &result.Version); err != nil {
|
|
return err
|
|
}
|
|
return json.Unmarshal(source, &result.SourceTurnIDs)
|
|
}
|
|
|
|
const memorySelectSQL = `SELECT cm.public_id::text,sp.public_id::text,mv.content,mv.source_refs,cm.status,cm.sensitivity,mv.version_no,cm.version,COALESCE(ua.public_id::text,''),mv.confirmed_at,COALESCE(mv.change_reason,''),COALESCE(mi.status,'PENDING'),COALESCE(mi.last_error,''),COALESCE(provider_mi.status,'DISABLED'),COALESCE(provider_mi.last_error,''),cm.created_at,cm.updated_at FROM curated_memory cm JOIN subject_profile sp ON sp.id=cm.subject_id AND sp.tenant_id=cm.tenant_id JOIN memory_version mv ON mv.id=cm.current_version_id LEFT JOIN user_account ua ON ua.id=mv.confirmed_by LEFT JOIN memory_index_entry mi ON mi.memory_version_id=mv.id AND mi.index_type='pgvector' LEFT JOIN memory_index_entry provider_mi ON provider_mi.memory_version_id=mv.id AND provider_mi.index_type='provider'`
|
|
|
|
func scanMemory(row rowScanner) (CuratedMemory, error) {
|
|
var m CuratedMemory
|
|
var sourceJSON []byte
|
|
if err := row.Scan(&m.ID, &m.SubjectID, &m.Content, &sourceJSON, &m.Status, &m.Sensitivity, &m.VersionNo, &m.Version, &m.ConfirmedBy, &m.ConfirmedAt, &m.ChangeReason, &m.IndexStatus, &m.LastIndexError, &m.ProviderIndexStatus, &m.ProviderIndexError, &m.CreatedAt, &m.UpdatedAt); err != nil {
|
|
return CuratedMemory{}, err
|
|
}
|
|
if err := json.Unmarshal(sourceJSON, &m.SourceRefs); err != nil {
|
|
return CuratedMemory{}, err
|
|
}
|
|
return m, nil
|
|
}
|
|
func nullUUID(value string) any {
|
|
if value == "" {
|
|
return nil
|
|
}
|
|
return value
|
|
}
|