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.
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
|
|
}
|