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.
56 lines
2.1 KiB
Go
56 lines
2.1 KiB
Go
package callmemory
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
|
|
"data-terminal-cloud/internal/outbox"
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
// Story review is independent of automatic Viking memory extraction. Schedule
|
|
// only drained, authorized calls; late subtitles get a new revision and event.
|
|
func (s *Service) enqueueDirectStories(ctx context.Context, tx pgx.Tx) error {
|
|
rows, err := tx.Query(ctx, `SELECT c.id::text,c.tenant_id::text,c.transcript_revision,c.privacy_epoch
|
|
FROM call_session c WHERE c.ended_at IS NOT NULL AND c.transcript_state IN ('COMPLETE','PARTIAL')
|
|
AND c.transcript_revision>0 AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1'
|
|
AND c.budget_snapshot->>'story_extraction_allowed'='true' AND call_story_authorized(c.id)
|
|
AND NOT EXISTS(SELECT 1 FROM call_story_extraction e WHERE e.session_id=c.id AND e.transcript_revision=c.transcript_revision AND e.privacy_epoch=c.privacy_epoch)
|
|
AND NOT EXISTS(SELECT 1 FROM outbox_event o WHERE o.aggregate_id=c.id AND o.event_type='call.story_extraction_requested'
|
|
AND o.deduplication_key='call-story-v1:'||c.id::text||':'||c.transcript_revision::text||':'||c.privacy_epoch::text)
|
|
ORDER BY c.ended_at,c.id LIMIT 20`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
type pending struct {
|
|
session, tenant string
|
|
revision, epoch int64
|
|
}
|
|
items := []pending{}
|
|
for rows.Next() {
|
|
var item pending
|
|
if err = rows.Scan(&item.session, &item.tenant, &item.revision, &item.epoch); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, item := range items {
|
|
key := fmt.Sprintf("call-story-v1:%s:%d:%d", item.session, item.revision, item.epoch)
|
|
payload, _ := json.Marshal(map[string]int64{"generation_version": item.revision, "privacy_epoch": item.epoch})
|
|
if _, err = outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{
|
|
TenantID: &item.tenant, AggregateType: "call_session", AggregateID: item.session,
|
|
EventType: "call.story_extraction_requested", DeduplicationKey: &key, Payload: payload,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|