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.
664 lines
38 KiB
Go
664 lines
38 KiB
Go
package diagnostics
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"regexp"
|
|
"strings"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/audit"
|
|
"data-terminal-cloud/internal/idempotency"
|
|
"data-terminal-cloud/internal/identity"
|
|
"data-terminal-cloud/internal/outbox"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
var (
|
|
ErrInvalid = errors.New("diagnostic request is invalid")
|
|
ErrAccessDenied = errors.New("diagnostic access is denied")
|
|
ErrNotFound = errors.New("diagnostic resource was not found")
|
|
ErrStateConflict = errors.New("diagnostic action conflicts with current state")
|
|
requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.:/-]{2,127}$`)
|
|
searchFieldAllowed = map[string]bool{"device_id": true, "conversation_id": true, "room_id": true, "provider_task_ref": true, "request_id": true}
|
|
retryableJobTypeAllowed = map[string]bool{
|
|
"platform.noop": true,
|
|
"conversation.room_issued": true,
|
|
"conversation.agent_start_requested": true,
|
|
"conversation.agent_start_reconcile_requested": true,
|
|
"conversation.agent_stop_requested": true,
|
|
"conversation.agent_stop_reconcile_requested": true,
|
|
"conversation.content_extraction_requested": true,
|
|
"call.story_extraction_requested": true,
|
|
"memory.index_requested": true,
|
|
"memory.delete_requested": true,
|
|
"privacy.access_requested": true,
|
|
"privacy.delete_requested": true,
|
|
"export.generate_requested": true,
|
|
"media.verify_requested": true,
|
|
"alert.webhook_test_requested": true,
|
|
}
|
|
)
|
|
|
|
type Record struct {
|
|
DeviceID string `json:"device_id"`
|
|
DeviceStatus string `json:"device_status"`
|
|
DeviceOnline bool `json:"device_online"`
|
|
FirmwareVersion string `json:"firmware_version,omitempty"`
|
|
DesiredConfigRevision int64 `json:"desired_config_revision"`
|
|
ReportedConfigRevision *int64 `json:"reported_config_revision,omitempty"`
|
|
LastDeviceErrorCode string `json:"last_device_error_code,omitempty"`
|
|
LastDeviceErrorAt *time.Time `json:"last_device_error_at,omitempty"`
|
|
ConversationID string `json:"conversation_id,omitempty"`
|
|
ConversationStatus string `json:"conversation_status,omitempty"`
|
|
ConversationFailureCode string `json:"conversation_failure_code,omitempty"`
|
|
AllocationExpiresAt *time.Time `json:"allocation_expires_at,omitempty"`
|
|
CredentialIssuedAt *time.Time `json:"credential_issued_at,omitempty"`
|
|
CredentialExpiresAt *time.Time `json:"credential_expires_at,omitempty"`
|
|
RoomID string `json:"room_id,omitempty"`
|
|
Provider string `json:"provider,omitempty"`
|
|
ProviderTaskRef string `json:"provider_task_ref,omitempty"`
|
|
ProviderTaskStatus string `json:"provider_task_status,omitempty"`
|
|
ProviderStartAttempts int `json:"provider_start_attempts"`
|
|
ProviderStopAttempts int `json:"provider_stop_attempts"`
|
|
ProviderNotificationCount int64 `json:"provider_notification_count"`
|
|
LastProviderEvent string `json:"last_provider_event,omitempty"`
|
|
LastProviderEventStatus string `json:"last_provider_event_status,omitempty"`
|
|
LastProviderEventAt *time.Time `json:"last_provider_event_at,omitempty"`
|
|
LastRequestID string `json:"last_request_id,omitempty"`
|
|
Timings *Timings `json:"timings,omitempty"`
|
|
}
|
|
|
|
type Timings struct {
|
|
ConversationCreatedAt time.Time `json:"conversation_created_at"`
|
|
RoomIssuedAt *time.Time `json:"room_issued_at,omitempty"`
|
|
DeviceActivatedAt *time.Time `json:"device_activated_at,omitempty"`
|
|
AgentActiveAt *time.Time `json:"agent_active_at,omitempty"`
|
|
StopRequestedAt *time.Time `json:"stop_requested_at,omitempty"`
|
|
ConversationEndedAt *time.Time `json:"conversation_ended_at,omitempty"`
|
|
RoomIssueMS *int64 `json:"room_issue_ms,omitempty"`
|
|
DeviceJoinMS *int64 `json:"device_join_ms,omitempty"`
|
|
AgentStartMS *int64 `json:"agent_start_ms,omitempty"`
|
|
AgentStopMS *int64 `json:"agent_stop_ms,omitempty"`
|
|
TotalMS *int64 `json:"total_ms,omitempty"`
|
|
}
|
|
|
|
type ActionResult struct {
|
|
Action string `json:"action"`
|
|
DeviceID string `json:"device_id"`
|
|
ConversationID string `json:"conversation_id,omitempty"`
|
|
ConversationStatus string `json:"conversation_status,omitempty"`
|
|
CommandID string `json:"command_id,omitempty"`
|
|
CommandStatus string `json:"command_status,omitempty"`
|
|
JobID string `json:"job_id,omitempty"`
|
|
JobType string `json:"job_type,omitempty"`
|
|
ManualRetryCount int `json:"manual_retry_count,omitempty"`
|
|
Queued bool `json:"queued"`
|
|
}
|
|
|
|
type Job struct {
|
|
ID string `json:"id"`
|
|
EventType string `json:"event_type"`
|
|
AggregateType string `json:"aggregate_type"`
|
|
Status string `json:"status"`
|
|
AttemptCount int `json:"attempt_count"`
|
|
ManualRetryCount int `json:"manual_retry_count"`
|
|
LastErrorCode string `json:"last_error_code,omitempty"`
|
|
LastManualRetryAt *time.Time `json:"last_manual_retry_at,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
}
|
|
|
|
type Bundle struct {
|
|
ID string `json:"id"`
|
|
GeneratedAt time.Time `json:"generated_at"`
|
|
SearchField string `json:"search_field"`
|
|
SearchValue string `json:"search_value"`
|
|
RedactionPolicy string `json:"redaction_policy"`
|
|
Items []Record `json:"items"`
|
|
}
|
|
|
|
type ExternalDependency struct {
|
|
ServiceName string `json:"service_name"`
|
|
Dependency string `json:"dependency"`
|
|
Status string `json:"status"`
|
|
CheckedAt time.Time `json:"checked_at"`
|
|
LastSuccessAt *time.Time `json:"last_success_at,omitempty"`
|
|
LastFailureAt *time.Time `json:"last_failure_at,omitempty"`
|
|
LastErrorCode string `json:"last_error_code,omitempty"`
|
|
LatencyMS int64 `json:"latency_ms"`
|
|
ProbeIntervalMS int64 `json:"probe_interval_ms"`
|
|
ConsecutiveFailures int `json:"consecutive_failures"`
|
|
}
|
|
|
|
type Service struct {
|
|
pool *pgxpool.Pool
|
|
idem *idempotency.Executor
|
|
audit *audit.Repository
|
|
outbox *outbox.Store
|
|
}
|
|
|
|
func NewService(pool *pgxpool.Pool, idem *idempotency.Executor, auditRepository *audit.Repository) (*Service, error) {
|
|
if pool == nil || idem == nil || auditRepository == nil {
|
|
return nil, errors.New("diagnostic service dependencies are required")
|
|
}
|
|
return &Service{pool: pool, idem: idem, audit: auditRepository, outbox: outbox.NewStore(pool)}, nil
|
|
}
|
|
|
|
type actor struct{ tenantID, userID, userPublicID string }
|
|
|
|
func (s *Service) resolveActor(ctx context.Context, principal identity.Principal, tenantPublicID string, roles ...string) (actor, error) {
|
|
if _, err := uuid.Parse(tenantPublicID); err != nil {
|
|
return actor{}, ErrInvalid
|
|
}
|
|
var result actor
|
|
var role string
|
|
err := s.pool.QueryRow(ctx, `SELECT t.id::text,ua.id::text,ua.public_id::text,ua.platform_role FROM tenant t JOIN user_account ua ON ua.tenant_id=t.id AND ua.status='active' AND ua.account_kind='operator' WHERE t.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, tenantPublicID, principal.Issuer, principal.Subject).Scan(&result.tenantID, &result.userID, &result.userPublicID, &role)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return actor{}, ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return actor{}, err
|
|
}
|
|
for _, allowed := range roles {
|
|
if role == allowed {
|
|
return result, nil
|
|
}
|
|
}
|
|
return actor{}, ErrAccessDenied
|
|
}
|
|
|
|
func (s *Service) Search(ctx context.Context, principal identity.Principal, tenantPublicID, field, value string) ([]Record, error) {
|
|
field, value = strings.TrimSpace(field), strings.TrimSpace(value)
|
|
if !searchFieldAllowed[field] || value == "" || len(value) > 200 {
|
|
return nil, ErrInvalid
|
|
}
|
|
if field == "device_id" || field == "conversation_id" {
|
|
if _, err := uuid.Parse(value); err != nil {
|
|
return nil, ErrInvalid
|
|
}
|
|
} else if !requestIDPattern.MatchString(value) {
|
|
return nil, ErrInvalid
|
|
}
|
|
currentActor, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
condition := map[string]string{
|
|
"device_id": "d.public_id=$2::uuid",
|
|
"conversation_id": "c.public_id=$2::uuid",
|
|
"room_id": "rt.room_id=$2",
|
|
"provider_task_ref": "COALESCE(rt.provider_task_ref,hs.provider_session_ref)=$2",
|
|
"request_id": "EXISTS(SELECT 1 FROM audit_event ae WHERE ae.tenant_id=c.tenant_id AND ae.request_id=$2 AND ae.resource_id IN(c.id,c.public_id,d.id,d.public_id))",
|
|
}[field]
|
|
query := diagnosticSelect + " WHERE d.tenant_id=$1::uuid AND " + condition + " ORDER BY c.created_at DESC NULLS LAST LIMIT 50"
|
|
rows, err := s.pool.Query(ctx, query, currentActor.tenantID, value)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []Record{}
|
|
for rows.Next() {
|
|
item, scanErr := scanRecord(rows)
|
|
if scanErr != nil {
|
|
return nil, scanErr
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) ListExternalDependencies(ctx context.Context, principal identity.Principal, tenantPublicID string) ([]ExternalDependency, error) {
|
|
if _, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin"); err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `SELECT service_name,dependency,status,checked_at,last_success_at,last_failure_at,COALESCE(last_error_code,''),latency_ms,probe_interval_ms,consecutive_failures FROM external_dependency_probe_state ORDER BY service_name,dependency`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ExternalDependency{}
|
|
for rows.Next() {
|
|
var item ExternalDependency
|
|
if err = rows.Scan(&item.ServiceName, &item.Dependency, &item.Status, &item.CheckedAt, &item.LastSuccessAt, &item.LastFailureAt, &item.LastErrorCode, &item.LatencyMS, &item.ProbeIntervalMS, &item.ConsecutiveFailures); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) StopConversation(ctx context.Context, principal identity.Principal, tenantPublicID, conversationPublicID, key, requestID, reason string) (ActionResult, bool, error) {
|
|
reason = strings.TrimSpace(reason)
|
|
if _, err := uuid.Parse(conversationPublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
|
|
return ActionResult{}, false, ErrInvalid
|
|
}
|
|
a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
body, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "reason": reason})
|
|
result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/conversations/stop", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
|
|
var internalID, deviceID, status, runtimeStatus, runtimeType string
|
|
err := tx.QueryRow(operationContext, `SELECT c.id::text,d.public_id::text,c.status,c.runtime_type,COALESCE(rt.status,hs.status,'') FROM conversation c JOIN device d ON d.id=c.device_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.public_id=$2::uuid FOR UPDATE OF c`, a.tenantID, conversationPublicID).Scan(&internalID, &deviceID, &status, &runtimeType, &runtimeStatus)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return idempotency.Result{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
queued := false
|
|
fromStatus := status
|
|
switch status {
|
|
case "COMPLETED", "FAILED", "EXPIRED", "STOPPING":
|
|
case "PREPARING", "ROOM_ISSUED":
|
|
status = "EXPIRED"
|
|
if _, err = tx.Exec(operationContext, `UPDATE conversation SET status='EXPIRED',stop_reason='ops_stop_before_start',ended_at=now(),lease_expires_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, internalID); err == nil && runtimeType == "volc_hardware_agent" {
|
|
_, err = tx.Exec(operationContext, `UPDATE hardware_agent_session SET status='EXPIRED',updated_at=now() WHERE conversation_id=$1::uuid`, internalID)
|
|
} else if err == nil {
|
|
_, err = tx.Exec(operationContext, `UPDATE rtc_task SET status='COMPLETED',lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`, internalID)
|
|
}
|
|
case "STARTING_AGENT", "ACTIVE":
|
|
status, queued = "STOPPING", runtimeType != "volc_hardware_agent"
|
|
if _, err = tx.Exec(operationContext, `UPDATE conversation SET status='STOPPING',stop_reason='ops_stop',lease_expires_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, internalID); err == nil && runtimeType == "volc_hardware_agent" {
|
|
_, err = tx.Exec(operationContext, `UPDATE hardware_agent_session SET status='STOPPING',updated_at=now() WHERE conversation_id=$1::uuid`, internalID)
|
|
} else if err == nil {
|
|
_, err = tx.Exec(operationContext, `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`, internalID)
|
|
}
|
|
if err == nil && runtimeType != "volc_hardware_agent" {
|
|
dedupe := "ops-stop:" + internalID
|
|
payload, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "reason": "ops_stop"})
|
|
_, err = s.outbox.Enqueue(operationContext, tx, outbox.NewEvent{TenantID: &a.tenantID, AggregateType: "conversation", AggregateID: internalID, EventType: "conversation.agent_stop_requested", DeduplicationKey: &dedupe, Payload: payload})
|
|
}
|
|
default:
|
|
return idempotency.Result{}, ErrStateConflict
|
|
}
|
|
if err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
if status != fromStatus {
|
|
if _, err = tx.Exec(operationContext, `INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source) VALUES($1::uuid,$2,$3,$4,'operator')`, internalID, fromStatus, status, "ops diagnostics: "+reason); err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
}
|
|
if err = s.audit.Record(operationContext, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.conversation.stop", ResourceType: "conversation", ResourceID: &conversationPublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"runtime_type": runtimeType, "from_runtime_status": runtimeStatus, "to_status": status}}); err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
output := ActionResult{Action: "stop_conversation", DeviceID: deviceID, ConversationID: conversationPublicID, ConversationStatus: status, Queued: queued}
|
|
encoded, _ := json.Marshal(output)
|
|
return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &internalID}, nil
|
|
})
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
var output ActionResult
|
|
if err = json.Unmarshal(result.Body, &output); err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
return output, !result.Replayed, nil
|
|
}
|
|
|
|
func (s *Service) ListDeadJobs(ctx context.Context, principal identity.Principal, tenantPublicID string) ([]Job, error) {
|
|
a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `SELECT public_id::text,event_type,aggregate_type,status,attempt_count,manual_retry_count,COALESCE(last_error_code,''),last_manual_retry_at,created_at FROM outbox_event WHERE tenant_id=$1::uuid AND status='dead' ORDER BY created_at DESC LIMIT 100`, a.tenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []Job{}
|
|
for rows.Next() {
|
|
var item Job
|
|
if err := rows.Scan(&item.ID, &item.EventType, &item.AggregateType, &item.Status, &item.AttemptCount, &item.ManualRetryCount, &item.LastErrorCode, &item.LastManualRetryAt, &item.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) RetryDeadJob(ctx context.Context, principal identity.Principal, tenantPublicID, jobPublicID, key, requestID, reason string) (ActionResult, bool, error) {
|
|
reason = strings.TrimSpace(reason)
|
|
if _, err := uuid.Parse(jobPublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
|
|
return ActionResult{}, false, ErrInvalid
|
|
}
|
|
a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
body, _ := json.Marshal(map[string]string{"job_id": jobPublicID, "reason": reason})
|
|
result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/jobs/retry", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(op context.Context, tx pgx.Tx) (idempotency.Result, error) {
|
|
var internalID, eventType, status string
|
|
var manualRetries int
|
|
if queryErr := tx.QueryRow(op, `SELECT id::text,event_type,status,manual_retry_count FROM outbox_event WHERE tenant_id=$1::uuid AND public_id=$2::uuid FOR UPDATE`, a.tenantID, jobPublicID).Scan(&internalID, &eventType, &status, &manualRetries); errors.Is(queryErr, pgx.ErrNoRows) {
|
|
return idempotency.Result{}, ErrNotFound
|
|
} else if queryErr != nil {
|
|
return idempotency.Result{}, queryErr
|
|
}
|
|
if status != "dead" {
|
|
return idempotency.Result{}, ErrStateConflict
|
|
}
|
|
if !retryableJobTypeAllowed[eventType] {
|
|
return idempotency.Result{}, ErrStateConflict
|
|
}
|
|
manualRetries++
|
|
if _, updateErr := tx.Exec(op, `UPDATE outbox_event SET status='pending',attempt_count=0,available_at=now(),locked_at=NULL,locked_by=NULL,dead_at=NULL,last_error_code=NULL,manual_retry_count=$2,last_manual_retry_at=now(),last_manual_retry_by=$3::uuid,updated_at=now() WHERE id=$1::uuid`, internalID, manualRetries, a.userID); updateErr != nil {
|
|
return idempotency.Result{}, updateErr
|
|
}
|
|
if auditErr := s.audit.Record(op, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.job.retry", ResourceType: "outbox_job", ResourceID: &jobPublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"event_type": eventType, "from_status": status, "to_status": "pending", "manual_retry_count": manualRetries}}); auditErr != nil {
|
|
return idempotency.Result{}, auditErr
|
|
}
|
|
output := ActionResult{Action: "retry_job", JobID: jobPublicID, JobType: eventType, ManualRetryCount: manualRetries, Queued: true}
|
|
encoded, _ := json.Marshal(output)
|
|
return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &internalID}, nil
|
|
})
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
var output ActionResult
|
|
if err = json.Unmarshal(result.Body, &output); err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
return output, !result.Replayed, nil
|
|
}
|
|
|
|
func (s *Service) RevokeDevice(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, key, requestID, reason string) (ActionResult, bool, error) {
|
|
reason = strings.TrimSpace(reason)
|
|
if _, err := uuid.Parse(devicePublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
|
|
return ActionResult{}, false, ErrInvalid
|
|
}
|
|
a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_admin")
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
body, _ := json.Marshal(map[string]string{"device_id": devicePublicID, "reason": reason})
|
|
result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/devices/revoke", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(op context.Context, tx pgx.Tx) (idempotency.Result, error) {
|
|
var deviceID, oldStatus string
|
|
if queryErr := tx.QueryRow(op, `SELECT id::text,status FROM device WHERE tenant_id=$1::uuid AND public_id=$2::uuid FOR UPDATE`, a.tenantID, devicePublicID).Scan(&deviceID, &oldStatus); errors.Is(queryErr, pgx.ErrNoRows) {
|
|
return idempotency.Result{}, ErrNotFound
|
|
} else if queryErr != nil {
|
|
return idempotency.Result{}, queryErr
|
|
}
|
|
if oldStatus != "revoked" {
|
|
if _, updateErr := tx.Exec(op, `UPDATE device SET status='revoked',revoked_at=COALESCE(revoked_at,now()),version=version+1,updated_at=now() WHERE id=$1::uuid`, deviceID); updateErr != nil {
|
|
return idempotency.Result{}, updateErr
|
|
}
|
|
}
|
|
|
|
var conversationInternalID, conversationPublicID, conversationStatus, runtimeStatus, runtimeType string
|
|
queued := false
|
|
conversationQueryErr := tx.QueryRow(op, `SELECT c.id::text,c.public_id::text,c.status,c.runtime_type,COALESCE(rt.status,hs.status,'') FROM conversation c 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.device_id=$2::uuid AND c.status NOT IN('COMPLETED','FAILED','EXPIRED') ORDER BY c.created_at DESC LIMIT 1 FOR UPDATE OF c`, a.tenantID, deviceID).Scan(&conversationInternalID, &conversationPublicID, &conversationStatus, &runtimeType, &runtimeStatus)
|
|
if conversationQueryErr != nil && !errors.Is(conversationQueryErr, pgx.ErrNoRows) {
|
|
return idempotency.Result{}, conversationQueryErr
|
|
}
|
|
if conversationQueryErr == nil {
|
|
fromStatus := conversationStatus
|
|
switch conversationStatus {
|
|
case "PREPARING", "ROOM_ISSUED":
|
|
conversationStatus = "EXPIRED"
|
|
if _, updateErr := tx.Exec(op, `UPDATE conversation SET status='EXPIRED',stop_reason='device_revoked_before_start',ended_at=now(),lease_expires_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, conversationInternalID); updateErr != nil {
|
|
return idempotency.Result{}, updateErr
|
|
}
|
|
query := `UPDATE rtc_task SET status='COMPLETED',lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`
|
|
if runtimeType == "volc_hardware_agent" {
|
|
query = `UPDATE hardware_agent_session SET status='EXPIRED',updated_at=now() WHERE conversation_id=$1::uuid`
|
|
}
|
|
if _, updateErr := tx.Exec(op, query, conversationInternalID); updateErr != nil {
|
|
return idempotency.Result{}, updateErr
|
|
}
|
|
case "STARTING_AGENT", "ACTIVE":
|
|
conversationStatus, queued = "STOPPING", runtimeType != "volc_hardware_agent"
|
|
if _, updateErr := tx.Exec(op, `UPDATE conversation SET status='STOPPING',stop_reason='device_revoked',lease_expires_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, conversationInternalID); updateErr != nil {
|
|
return idempotency.Result{}, updateErr
|
|
}
|
|
query := `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`
|
|
if runtimeType == "volc_hardware_agent" {
|
|
query = `UPDATE hardware_agent_session SET status='STOPPING',updated_at=now() WHERE conversation_id=$1::uuid`
|
|
}
|
|
if _, updateErr := tx.Exec(op, query, conversationInternalID); updateErr != nil {
|
|
return idempotency.Result{}, updateErr
|
|
}
|
|
if runtimeType != "volc_hardware_agent" {
|
|
dedupe := "ops-stop:" + conversationInternalID
|
|
payload, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "reason": "device_revoked"})
|
|
if _, enqueueErr := s.outbox.Enqueue(op, tx, outbox.NewEvent{TenantID: &a.tenantID, AggregateType: "conversation", AggregateID: conversationInternalID, EventType: "conversation.agent_stop_requested", DeduplicationKey: &dedupe, Payload: payload}); enqueueErr != nil {
|
|
return idempotency.Result{}, enqueueErr
|
|
}
|
|
}
|
|
case "STOPPING":
|
|
default:
|
|
return idempotency.Result{}, ErrStateConflict
|
|
}
|
|
if conversationStatus != fromStatus {
|
|
if _, historyErr := tx.Exec(op, `INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source) VALUES($1::uuid,$2,$3,$4,'operator')`, conversationInternalID, fromStatus, conversationStatus, "device revoked: "+reason); historyErr != nil {
|
|
return idempotency.Result{}, historyErr
|
|
}
|
|
}
|
|
}
|
|
|
|
detail := map[string]any{"from_status": oldStatus, "to_status": "revoked", "conversation_queued": queued}
|
|
if conversationPublicID != "" {
|
|
detail["conversation_id"] = conversationPublicID
|
|
detail["conversation_status"] = conversationStatus
|
|
detail["runtime_type"] = runtimeType
|
|
detail["from_runtime_status"] = runtimeStatus
|
|
}
|
|
if auditErr := s.audit.Record(op, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.device.revoke", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: detail}); auditErr != nil {
|
|
return idempotency.Result{}, auditErr
|
|
}
|
|
output := ActionResult{Action: "revoke_device", DeviceID: devicePublicID, ConversationID: conversationPublicID, ConversationStatus: conversationStatus, Queued: queued}
|
|
encoded, _ := json.Marshal(output)
|
|
return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &deviceID}, nil
|
|
})
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
var output ActionResult
|
|
if err = json.Unmarshal(result.Body, &output); err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
return output, !result.Replayed, nil
|
|
}
|
|
|
|
func (s *Service) ExportBundle(ctx context.Context, principal identity.Principal, tenantPublicID, field, value, key, requestID, reason string) (Bundle, bool, error) {
|
|
reason = strings.TrimSpace(reason)
|
|
if reason == "" || len([]rune(reason)) > 500 {
|
|
return Bundle{}, false, ErrInvalid
|
|
}
|
|
a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
|
|
if err != nil {
|
|
return Bundle{}, false, err
|
|
}
|
|
items, err := s.Search(ctx, principal, tenantPublicID, field, value)
|
|
if err != nil {
|
|
return Bundle{}, false, err
|
|
}
|
|
body, _ := json.Marshal(map[string]string{"field": field, "value": value, "reason": reason})
|
|
result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/bundles", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(op context.Context, tx pgx.Tx) (idempotency.Result, error) {
|
|
bundle := Bundle{ID: uuid.NewString(), GeneratedAt: time.Now().UTC(), SearchField: field, SearchValue: value, RedactionPolicy: "ops-diagnostic-minimal-v1", Items: items}
|
|
if auditErr := s.audit.Record(op, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.bundle.export", ResourceType: "diagnostic_bundle", ResourceID: &bundle.ID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"search_field": field, "item_count": len(items), "redaction_policy": bundle.RedactionPolicy}}); auditErr != nil {
|
|
return idempotency.Result{}, auditErr
|
|
}
|
|
encoded, _ := json.Marshal(bundle)
|
|
return idempotency.Result{StatusCode: 201, Body: encoded, ResourceID: &bundle.ID}, nil
|
|
})
|
|
if err != nil {
|
|
return Bundle{}, false, err
|
|
}
|
|
var bundle Bundle
|
|
if err = json.Unmarshal(result.Body, &bundle); err != nil {
|
|
return Bundle{}, false, err
|
|
}
|
|
return bundle, !result.Replayed, nil
|
|
}
|
|
|
|
func (s *Service) ReconcileConversation(ctx context.Context, principal identity.Principal, tenantPublicID, conversationPublicID, key, requestID, reason string) (ActionResult, bool, error) {
|
|
reason = strings.TrimSpace(reason)
|
|
if _, err := uuid.Parse(conversationPublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
|
|
return ActionResult{}, false, ErrInvalid
|
|
}
|
|
a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
body, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "reason": reason})
|
|
result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/conversations/reconcile", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
|
|
var internalID, deviceID, status, taskRef, runtimeType string
|
|
if queryErr := tx.QueryRow(operationContext, `SELECT c.id::text,d.public_id::text,c.status,COALESCE(rt.provider_task_ref,hs.provider_session_ref),c.runtime_type FROM conversation c JOIN device d ON d.id=c.device_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.public_id=$2::uuid FOR UPDATE OF c`, a.tenantID, conversationPublicID).Scan(&internalID, &deviceID, &status, &taskRef, &runtimeType); errors.Is(queryErr, pgx.ErrNoRows) {
|
|
return idempotency.Result{}, ErrNotFound
|
|
} else if queryErr != nil {
|
|
return idempotency.Result{}, queryErr
|
|
}
|
|
eventType := ""
|
|
if status == "STARTING_AGENT" {
|
|
eventType = "conversation.agent_start_reconcile_requested"
|
|
} else if status == "STOPPING" {
|
|
eventType = "conversation.agent_stop_reconcile_requested"
|
|
}
|
|
queued := eventType != "" && runtimeType != "volc_hardware_agent"
|
|
if queued {
|
|
dedupe := "ops-reconcile:" + taskRef + ":" + key
|
|
payload, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "provider_task_ref": taskRef, "reason": "ops_reconcile"})
|
|
if _, enqueueErr := s.outbox.Enqueue(operationContext, tx, outbox.NewEvent{TenantID: &a.tenantID, AggregateType: "conversation", AggregateID: internalID, EventType: eventType, DeduplicationKey: &dedupe, Payload: payload}); enqueueErr != nil {
|
|
return idempotency.Result{}, enqueueErr
|
|
}
|
|
}
|
|
if auditErr := s.audit.Record(operationContext, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.conversation.reconcile", ResourceType: "conversation", ResourceID: &conversationPublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"status": status, "runtime_type": runtimeType, "queued": queued}}); auditErr != nil {
|
|
return idempotency.Result{}, auditErr
|
|
}
|
|
output := ActionResult{Action: "reconcile_conversation", DeviceID: deviceID, ConversationID: conversationPublicID, ConversationStatus: status, Queued: queued}
|
|
encoded, _ := json.Marshal(output)
|
|
return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &internalID}, nil
|
|
})
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
var output ActionResult
|
|
if err = json.Unmarshal(result.Body, &output); err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
return output, !result.Replayed, nil
|
|
}
|
|
|
|
func (s *Service) RefreshDeviceConfig(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, key, requestID, reason string) (ActionResult, bool, error) {
|
|
reason = strings.TrimSpace(reason)
|
|
if _, err := uuid.Parse(devicePublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
|
|
return ActionResult{}, false, ErrInvalid
|
|
}
|
|
a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
body, _ := json.Marshal(map[string]string{"device_id": devicePublicID, "reason": reason})
|
|
result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/devices/config-refresh", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
|
|
var deviceID string
|
|
var revision int64
|
|
if queryErr := tx.QueryRow(operationContext, `SELECT d.id::text,dt.desired_revision FROM device d JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id WHERE d.tenant_id=$1::uuid AND d.public_id=$2::uuid AND d.status='registered' FOR SHARE OF d`, a.tenantID, devicePublicID).Scan(&deviceID, &revision); errors.Is(queryErr, pgx.ErrNoRows) {
|
|
return idempotency.Result{}, ErrNotFound
|
|
} else if queryErr != nil {
|
|
return idempotency.Result{}, queryErr
|
|
}
|
|
commandPublicID := uuid.NewString()
|
|
commandInternalID := uuid.NewString()
|
|
payload, _ := json.Marshal(map[string]any{"desired_revision": revision, "reason_code": "ops_config_refresh"})
|
|
digest := idempotency.Hash(payload)
|
|
_, insertErr := tx.Exec(operationContext, `INSERT INTO device_command(id,public_id,tenant_id,device_id,command_type,payload,status,requested_by,idempotency_key,request_digest,expires_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,'apply_config',$5::jsonb,'QUEUED',$6::uuid,$7,$8,now()+interval '1 hour')`, commandInternalID, commandPublicID, a.tenantID, deviceID, payload, a.userID, "ops-refresh:"+key, digest)
|
|
if insertErr != nil {
|
|
return idempotency.Result{}, insertErr
|
|
}
|
|
if auditErr := s.audit.Record(operationContext, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.device.config_refresh", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"desired_revision": revision, "command_id": commandPublicID}}); auditErr != nil {
|
|
return idempotency.Result{}, auditErr
|
|
}
|
|
output := ActionResult{Action: "refresh_config", DeviceID: devicePublicID, CommandID: commandPublicID, CommandStatus: "QUEUED", Queued: true}
|
|
encoded, _ := json.Marshal(output)
|
|
return idempotency.Result{StatusCode: 201, Body: encoded, ResourceID: &commandInternalID}, nil
|
|
})
|
|
if err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
var output ActionResult
|
|
if err = json.Unmarshal(result.Body, &output); err != nil {
|
|
return ActionResult{}, false, err
|
|
}
|
|
return output, !result.Replayed, nil
|
|
}
|
|
|
|
const diagnosticSelect = `SELECT
|
|
d.public_id::text,d.status,COALESCE(dp.online,false),COALESCE(dt.firmware_version,''),
|
|
COALESCE(dt.desired_revision,1),dt.reported_revision,COALESCE(dt.last_error_code,''),dt.last_error_at,
|
|
COALESCE(c.public_id::text,''),COALESCE(c.status,''),COALESCE(c.failure_code,''),c.allocation_expires_at,
|
|
rt.credential_issued_at,rt.credential_expires_at,
|
|
COALESCE(rt.room_id,''),COALESCE(rt.provider,hs.provider,''),COALESCE(rt.provider_task_ref,hs.provider_session_ref,''),COALESCE(rt.status,hs.status,''),
|
|
COALESCE(rt.start_attempts,0),COALESCE(rt.stop_attempts,0),COALESCE(pe.event_count,0),
|
|
COALESCE(pe.event_type,''),COALESCE(pe.event_status,''),pe.received_at,COALESCE(ae.request_id,''),
|
|
c.created_at,ch.room_issued_at,ch.device_activated_at,ch.agent_active_at,ch.stop_requested_at,c.ended_at
|
|
FROM device d
|
|
LEFT JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id
|
|
LEFT JOIN device_presence_state dp ON dp.device_id=d.id AND dp.tenant_id=d.tenant_id
|
|
LEFT JOIN conversation c ON c.device_id=d.id AND c.tenant_id=d.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
|
|
LEFT JOIN LATERAL (
|
|
SELECT count(*) OVER() event_count,event_type,event_status,received_at FROM (
|
|
SELECT event_type,COALESCE(sanitized_payload->>'status','') event_status,received_at
|
|
FROM provider_event WHERE conversation_id=c.id
|
|
UNION ALL
|
|
SELECT event_type,event_type,occurred_at FROM hardware_agent_event WHERE conversation_id=c.id
|
|
) runtime_event ORDER BY received_at DESC LIMIT 1
|
|
) pe ON true
|
|
LEFT JOIN LATERAL (
|
|
SELECT
|
|
MIN(occurred_at) FILTER (WHERE to_status='ROOM_ISSUED') room_issued_at,
|
|
MIN(occurred_at) FILTER (WHERE to_status='STARTING_AGENT') device_activated_at,
|
|
MIN(occurred_at) FILTER (WHERE to_status='ACTIVE') agent_active_at,
|
|
MIN(occurred_at) FILTER (WHERE to_status='STOPPING') stop_requested_at
|
|
FROM conversation_state_history WHERE conversation_id=c.id
|
|
) ch ON c.id IS NOT NULL
|
|
LEFT JOIN LATERAL (
|
|
SELECT request_id FROM audit_event
|
|
WHERE tenant_id=d.tenant_id AND resource_id IN(d.id,d.public_id,c.id,c.public_id) AND request_id IS NOT NULL
|
|
ORDER BY occurred_at DESC LIMIT 1
|
|
) ae ON true`
|
|
|
|
type rowScanner interface{ Scan(...any) error }
|
|
|
|
func scanRecord(row rowScanner) (Record, error) {
|
|
var item Record
|
|
var createdAt, roomIssuedAt, deviceActivatedAt, agentActiveAt, stopRequestedAt, endedAt *time.Time
|
|
err := row.Scan(&item.DeviceID, &item.DeviceStatus, &item.DeviceOnline, &item.FirmwareVersion, &item.DesiredConfigRevision, &item.ReportedConfigRevision, &item.LastDeviceErrorCode, &item.LastDeviceErrorAt, &item.ConversationID, &item.ConversationStatus, &item.ConversationFailureCode, &item.AllocationExpiresAt, &item.CredentialIssuedAt, &item.CredentialExpiresAt, &item.RoomID, &item.Provider, &item.ProviderTaskRef, &item.ProviderTaskStatus, &item.ProviderStartAttempts, &item.ProviderStopAttempts, &item.ProviderNotificationCount, &item.LastProviderEvent, &item.LastProviderEventStatus, &item.LastProviderEventAt, &item.LastRequestID, &createdAt, &roomIssuedAt, &deviceActivatedAt, &agentActiveAt, &stopRequestedAt, &endedAt)
|
|
if err == nil && createdAt != nil {
|
|
item.Timings = &Timings{
|
|
ConversationCreatedAt: *createdAt,
|
|
RoomIssuedAt: roomIssuedAt,
|
|
DeviceActivatedAt: deviceActivatedAt,
|
|
AgentActiveAt: agentActiveAt,
|
|
StopRequestedAt: stopRequestedAt,
|
|
ConversationEndedAt: endedAt,
|
|
RoomIssueMS: elapsedMilliseconds(createdAt, roomIssuedAt),
|
|
DeviceJoinMS: elapsedMilliseconds(roomIssuedAt, deviceActivatedAt),
|
|
AgentStartMS: elapsedMilliseconds(deviceActivatedAt, agentActiveAt),
|
|
AgentStopMS: elapsedMilliseconds(stopRequestedAt, endedAt),
|
|
TotalMS: elapsedMilliseconds(createdAt, endedAt),
|
|
}
|
|
}
|
|
return item, err
|
|
}
|
|
|
|
func elapsedMilliseconds(start, end *time.Time) *int64 {
|
|
if start == nil || end == nil || end.Before(*start) {
|
|
return nil
|
|
}
|
|
value := end.Sub(*start).Milliseconds()
|
|
return &value
|
|
}
|
|
|
|
func (r Record) String() string {
|
|
return fmt.Sprintf("device=%s conversation=%s", r.DeviceID, r.ConversationID)
|
|
}
|