252 lines
11 KiB
Go
252 lines
11 KiB
Go
|
package saas
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"errors"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"data-terminal-cloud/internal/callmemory"
|
||
|
|
"data-terminal-cloud/internal/identity"
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
)
|
||
|
|
|
||
|
|
type HomeOverviewDay struct {
|
||
|
|
Date string `json:"date"`
|
||
|
|
ConversationCount int `json:"conversation_count"`
|
||
|
|
DurationMS int64 `json:"duration_ms"`
|
||
|
|
}
|
||
|
|
|
||
|
|
type HomeOverviewConversation struct {
|
||
|
|
ID string `json:"id"`
|
||
|
|
Source string `json:"source"`
|
||
|
|
Status string `json:"status"`
|
||
|
|
StartedAt time.Time `json:"started_at"`
|
||
|
|
EndedAt time.Time `json:"ended_at"`
|
||
|
|
DurationMS int64 `json:"duration_ms"`
|
||
|
|
Summary string `json:"summary,omitempty"`
|
||
|
|
SummaryState string `json:"summary_state"`
|
||
|
|
SummaryKind string `json:"summary_kind,omitempty"`
|
||
|
|
}
|
||
|
|
|
||
|
|
type HomeOverview struct {
|
||
|
|
SubjectID string `json:"subject_id"`
|
||
|
|
Timezone string `json:"timezone"`
|
||
|
|
WeekStart string `json:"week_start"`
|
||
|
|
WeekEnd string `json:"week_end"`
|
||
|
|
ConversationCount int `json:"conversation_count"`
|
||
|
|
TotalDurationMS int64 `json:"total_duration_ms"`
|
||
|
|
Daily []HomeOverviewDay `json:"daily"`
|
||
|
|
LatestConversation *HomeOverviewConversation `json:"latest_conversation"`
|
||
|
|
}
|
||
|
|
|
||
|
|
// These are separate session stores, not joins against turns or memories: a
|
||
|
|
// session contributes once regardless of how many utterances or memories it has.
|
||
|
|
// DIRECT has no reliable connected_at. Its existing created_at -> ended_at
|
||
|
|
// duration is retained, but a finalized USER/ASSISTANT utterance must establish
|
||
|
|
// actual interaction so failed preparations and abandoned leases do not count.
|
||
|
|
const homeConversationsSQL = `WITH eligible AS (
|
||
|
|
SELECT c.public_id AS id,c.id AS internal_id,'conversation'::text AS source,c.status,c.started_at,c.ended_at
|
||
|
|
FROM conversation c
|
||
|
|
WHERE c.tenant_id=$1::uuid AND c.subject_id=$2::uuid
|
||
|
|
AND c.status IN ('COMPLETED','FAILED','EXPIRED') AND c.started_at IS NOT NULL
|
||
|
|
AND c.ended_at>c.started_at AND c.ended_at<=$3
|
||
|
|
AND NOT EXISTS(SELECT 1 FROM memory_source_suppression ss WHERE ss.subject_id=c.subject_id AND ss.scope='SOURCE_AND_DERIVED'
|
||
|
|
AND (ss.source_refs @> jsonb_build_array(jsonb_build_object('conversation_id',c.public_id::text))
|
||
|
|
OR ss.source_refs @> jsonb_build_array(jsonb_build_object('conversation_id',c.id::text))
|
||
|
|
OR EXISTS(SELECT 1 FROM story_draft sd WHERE sd.conversation_id=c.id AND sd.subject_id=c.subject_id
|
||
|
|
AND ss.source_refs @> jsonb_build_array(jsonb_build_object('story_id',sd.public_id::text)))))
|
||
|
|
UNION ALL
|
||
|
|
SELECT c.id,c.id,'call_memory',c.call_state,c.created_at,c.ended_at
|
||
|
|
FROM call_session c
|
||
|
|
WHERE c.tenant_id=$1::uuid AND c.subject_id=$2::uuid AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1'
|
||
|
|
AND c.call_state IN ('ENDED','FAILED','CANCELLED') AND c.ended_at>c.created_at AND c.ended_at<=$3
|
||
|
|
AND EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id AND u.speaker IN ('USER','ASSISTANT')
|
||
|
|
AND regexp_replace(u.text_content,'[[:space:]]','','g') NOT IN ('','[deleted]'))
|
||
|
|
AND NOT EXISTS(SELECT 1 FROM memory_source_suppression ss WHERE ss.subject_id=c.subject_id AND ss.scope='SOURCE_AND_DERIVED'
|
||
|
|
AND ss.source_refs @> jsonb_build_array(jsonb_build_object('session_id',c.id::text)))
|
||
|
|
), latest AS (SELECT source,id FROM eligible ORDER BY started_at DESC,id DESC,source DESC LIMIT 1)
|
||
|
|
SELECT e.id::text,e.internal_id::text,e.source,e.status,e.started_at,e.ended_at
|
||
|
|
FROM eligible e WHERE (e.started_at >= $4 AND e.started_at < $5)
|
||
|
|
OR EXISTS(SELECT 1 FROM latest l WHERE l.source=e.source AND l.id=e.id)
|
||
|
|
ORDER BY e.started_at DESC,e.id DESC,e.source DESC`
|
||
|
|
|
||
|
|
// Home calendars use the product timezone even for historical profiles that
|
||
|
|
// store another timezone. This changes presentation boundaries, not the profile.
|
||
|
|
func homeCalendarWeek(now time.Time) (time.Time, time.Time, error) {
|
||
|
|
location, err := time.LoadLocation("Asia/Shanghai")
|
||
|
|
if err != nil {
|
||
|
|
return time.Time{}, time.Time{}, err
|
||
|
|
}
|
||
|
|
local := now.In(location)
|
||
|
|
start := time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, location)
|
||
|
|
start = start.AddDate(0, 0, -(int(start.Weekday())+6)%7)
|
||
|
|
return start, start.AddDate(0, 0, 7), nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) SubjectHomeOverview(ctx context.Context, principal identity.Principal, subjectPublicID string) (HomeOverview, error) {
|
||
|
|
return s.subjectHomeOverviewAt(ctx, principal, subjectPublicID, time.Now())
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) subjectHomeOverviewAt(ctx context.Context, principal identity.Principal, subjectPublicID string, now time.Time) (HomeOverview, error) {
|
||
|
|
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||
|
|
defer cancel()
|
||
|
|
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly})
|
||
|
|
if err != nil {
|
||
|
|
return HomeOverview{}, err
|
||
|
|
}
|
||
|
|
defer tx.Rollback(context.Background())
|
||
|
|
a, subject, err := resolveSubject(ctx, tx, principal, subjectPublicID, identity.CapabilityConversationRead, false)
|
||
|
|
if err != nil {
|
||
|
|
return HomeOverview{}, err
|
||
|
|
}
|
||
|
|
start, end, err := homeCalendarWeek(now)
|
||
|
|
if err != nil {
|
||
|
|
return HomeOverview{}, err
|
||
|
|
}
|
||
|
|
location := start.Location()
|
||
|
|
result := HomeOverview{SubjectID: subject.ID, Timezone: location.String(), WeekStart: start.Format("2006-01-02"), WeekEnd: end.Format("2006-01-02"), Daily: make([]HomeOverviewDay, 7)}
|
||
|
|
dayIndexes := map[string]int{}
|
||
|
|
for i := range result.Daily {
|
||
|
|
day := start.AddDate(0, 0, i).Format("2006-01-02")
|
||
|
|
result.Daily[i].Date = day
|
||
|
|
dayIndexes[day] = i
|
||
|
|
}
|
||
|
|
rows, err := tx.Query(ctx, homeConversationsSQL, a.tenantID, a.subjectID, now, start, end)
|
||
|
|
if err != nil {
|
||
|
|
return HomeOverview{}, err
|
||
|
|
}
|
||
|
|
var latestInternalID string
|
||
|
|
for rows.Next() {
|
||
|
|
var item HomeOverviewConversation
|
||
|
|
var internalID string
|
||
|
|
if err = rows.Scan(&item.ID, &internalID, &item.Source, &item.Status, &item.StartedAt, &item.EndedAt); err != nil {
|
||
|
|
rows.Close()
|
||
|
|
return HomeOverview{}, err
|
||
|
|
}
|
||
|
|
item.DurationMS = item.EndedAt.Sub(item.StartedAt).Milliseconds()
|
||
|
|
item.SummaryState = "UNAVAILABLE"
|
||
|
|
if result.LatestConversation == nil {
|
||
|
|
result.LatestConversation = &item
|
||
|
|
latestInternalID = internalID
|
||
|
|
}
|
||
|
|
// Entire duration belongs to the local start date, including midnight
|
||
|
|
// crossings. AddDate above preserves calendar weeks through DST.
|
||
|
|
if i, ok := dayIndexes[item.StartedAt.In(location).Format("2006-01-02")]; ok {
|
||
|
|
result.Daily[i].ConversationCount++
|
||
|
|
result.Daily[i].DurationMS += item.DurationMS
|
||
|
|
result.ConversationCount++
|
||
|
|
result.TotalDurationMS += item.DurationMS
|
||
|
|
}
|
||
|
|
}
|
||
|
|
err = rows.Err()
|
||
|
|
rows.Close()
|
||
|
|
if err != nil {
|
||
|
|
return HomeOverview{}, err
|
||
|
|
}
|
||
|
|
if result.LatestConversation != nil {
|
||
|
|
grant, grantErr := callmemory.ResolveAccess(ctx, tx, principal, subjectPublicID, false)
|
||
|
|
if grantErr == nil {
|
||
|
|
err = s.homeConversationSummary(ctx, tx, a.subjectID, latestInternalID, grant, result.LatestConversation)
|
||
|
|
if err != nil {
|
||
|
|
return HomeOverview{}, err
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
var denied *callmemory.Error
|
||
|
|
if !errors.As(grantErr, &denied) || denied.Status != 403 {
|
||
|
|
return HomeOverview{}, grantErr
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return result, tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) homeConversationSummary(ctx context.Context, tx pgx.Tx, subject, id string, grant callmemory.Access, item *HomeOverviewConversation) error {
|
||
|
|
if item.Source == "call_memory" {
|
||
|
|
var allowed, hasUser, hasAssistant bool
|
||
|
|
var transcript, state, lastError string
|
||
|
|
err := tx.QueryRow(ctx, `SELECT c.memory_mode<>'OFF' AND managed_call_memory_authorized(c.subject_id)
|
||
|
|
AND c.privacy_epoch=COALESCE(mus.privacy_epoch,0),c.transcript_state,COALESCE(j.state,''),COALESCE(j.last_error,''),
|
||
|
|
EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id AND u.speaker='USER' AND regexp_replace(u.text_content,'[[:space:]]','','g') NOT IN ('','[deleted]')),
|
||
|
|
EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id AND u.speaker='ASSISTANT' AND regexp_replace(u.text_content,'[[:space:]]','','g') NOT IN ('','[deleted]'))
|
||
|
|
FROM call_session c LEFT JOIN memory_user_state mus ON mus.subject_id=c.subject_id
|
||
|
|
LEFT JOIN call_memory_sync j ON j.session_id=c.id AND j.transcript_revision=c.transcript_revision
|
||
|
|
WHERE c.id=$1::uuid AND c.subject_id=$2::uuid`, id, subject).Scan(&allowed, &transcript, &state, &lastError, &hasUser, &hasAssistant)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if !allowed {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
// A current, active, normal memory from this exact session is a short
|
||
|
|
// memory recap, not a synthesized whole-conversation summary.
|
||
|
|
err = tx.QueryRow(ctx, `SELECT mv.content FROM managed_memory_origin o
|
||
|
|
JOIN curated_memory cm ON cm.id=o.memory_id AND cm.subject_id=o.subject_id
|
||
|
|
JOIN memory_version mv ON mv.id=cm.current_version_id AND mv.memory_id=cm.id
|
||
|
|
WHERE o.subject_id=$1::uuid AND o.session_id=$2::uuid AND cm.status='ACTIVE' AND cm.sensitivity='normal'
|
||
|
|
AND NOT EXISTS(SELECT 1 FROM memory_source_suppression ss WHERE ss.subject_id=cm.subject_id AND ss.source_refs @> mv.source_refs)
|
||
|
|
ORDER BY cm.created_at DESC,cm.public_id DESC LIMIT 1`, subject, id).Scan(&item.Summary)
|
||
|
|
if err == nil {
|
||
|
|
item.SummaryState = "READY"
|
||
|
|
item.SummaryKind = "memory"
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if !errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
switch {
|
||
|
|
case transcript == "DRAINING" || transcript == "COLLECTING":
|
||
|
|
item.SummaryState = "PROCESSING"
|
||
|
|
case !hasUser || !hasAssistant:
|
||
|
|
item.SummaryState = "INSUFFICIENT_TRANSCRIPT"
|
||
|
|
case lastError != "":
|
||
|
|
item.SummaryState = "SYNC_FAILED"
|
||
|
|
case state == "SYNCED" || state == "SKIPPED":
|
||
|
|
item.SummaryState = "NO_SUMMARY"
|
||
|
|
default:
|
||
|
|
item.SummaryState = "PROCESSING"
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
// Legacy summaries have no sensitivity classification; only a reader with
|
||
|
|
// sensitive access may see them. Metadata permission alone never exposes text.
|
||
|
|
if !grant.Sensitive {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
var allowed, hasUser bool
|
||
|
|
var jobState string
|
||
|
|
err := tx.QueryRow(ctx, `SELECT c.retention_mode<>'none'
|
||
|
|
AND EXISTS(SELECT 1 FROM consent_record cr WHERE cr.subject_id=c.subject_id AND cr.consent_type='transcript_storage' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now()))
|
||
|
|
AND EXISTS(SELECT 1 FROM consent_record cr WHERE cr.subject_id=c.subject_id AND cr.consent_type='story_extraction' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now())),
|
||
|
|
EXISTS(SELECT 1 FROM conversation_turn ct WHERE ct.conversation_id=c.id AND ct.speaker='subject' AND regexp_replace(ct.final_text,'[[:space:]]','','g') NOT IN ('','[deleted]')),
|
||
|
|
COALESCE((SELECT o.status FROM outbox_event o WHERE o.aggregate_id=c.id AND o.event_type='conversation.content_extraction_requested' ORDER BY o.created_at DESC,o.id DESC LIMIT 1),'')
|
||
|
|
FROM conversation c WHERE c.id=$1::uuid AND c.subject_id=$2::uuid`, id, subject).Scan(&allowed, &hasUser, &jobState)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if !allowed {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
var state string
|
||
|
|
err = tx.QueryRow(ctx, `SELECT content,status FROM conversation_summary WHERE subject_id=$1::uuid AND conversation_id=$2::uuid AND summary_type='conversation_summary' AND status<>'SUPERSEDED' ORDER BY generation_version DESC LIMIT 1`, subject, id).Scan(&item.Summary, &state)
|
||
|
|
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if err == nil && state == "READY" {
|
||
|
|
item.SummaryState = "READY"
|
||
|
|
item.SummaryKind = "conversation_summary"
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
item.Summary = ""
|
||
|
|
switch {
|
||
|
|
case !hasUser:
|
||
|
|
item.SummaryState = "INSUFFICIENT_TRANSCRIPT"
|
||
|
|
case state == "FAILED" || jobState == "dead":
|
||
|
|
item.SummaryState = "SYNC_FAILED"
|
||
|
|
case state == "PROCESSING" || jobState == "pending" || jobState == "processing":
|
||
|
|
item.SummaryState = "PROCESSING"
|
||
|
|
default:
|
||
|
|
item.SummaryState = "NO_SUMMARY"
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|