643 lines
36 KiB
Go
643 lines
36 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,
|
|
"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' 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": "rt.provider_task_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, rtcStatus string
|
|
err := tx.QueryRow(operationContext, `SELECT c.id::text,d.public_id::text,c.status,rt.status FROM conversation c JOIN device d ON d.id=c.device_id JOIN rtc_task rt ON rt.conversation_id=c.id WHERE c.tenant_id=$1::uuid AND c.public_id=$2::uuid FOR UPDATE OF c,rt`, a.tenantID, conversationPublicID).Scan(&internalID, &deviceID, &status, &rtcStatus)
|
|
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 {
|
|
_, 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", true
|
|
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 {
|
|
_, 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 {
|
|
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{"from_rtc_status": rtcStatus, "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, rtcStatus string
|
|
queued := false
|
|
conversationQueryErr := tx.QueryRow(op, `SELECT c.id::text,c.public_id::text,c.status,rt.status FROM conversation c JOIN rtc_task rt ON rt.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,rt`, a.tenantID, deviceID).Scan(&conversationInternalID, &conversationPublicID, &conversationStatus, &rtcStatus)
|
|
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
|
|
}
|
|
if _, updateErr := tx.Exec(op, `UPDATE rtc_task SET status='COMPLETED',lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`, conversationInternalID); updateErr != nil {
|
|
return idempotency.Result{}, updateErr
|
|
}
|
|
case "STARTING_AGENT", "ACTIVE":
|
|
conversationStatus, queued = "STOPPING", true
|
|
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
|
|
}
|
|
if _, updateErr := tx.Exec(op, `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`, conversationInternalID); updateErr != nil {
|
|
return idempotency.Result{}, updateErr
|
|
}
|
|
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["from_rtc_status"] = rtcStatus
|
|
}
|
|
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 string
|
|
if queryErr := tx.QueryRow(operationContext, `SELECT c.id::text,d.public_id::text,c.status,rt.provider_task_ref FROM conversation c JOIN device d ON d.id=c.device_id JOIN rtc_task rt ON rt.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); 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 != ""
|
|
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, "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,''),COALESCE(rt.provider_task_ref,''),COALESCE(rt.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 LATERAL (
|
|
SELECT count(*) OVER() event_count,event_type,COALESCE(sanitized_payload->>'status','') event_status,received_at
|
|
FROM provider_event WHERE conversation_id=c.id 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)
|
|
}
|