package saas import ( "context" "encoding/json" "errors" "fmt" "strconv" "time" "data-terminal-cloud/internal/audit" "data-terminal-cloud/internal/concurrency" "data-terminal-cloud/internal/database" "data-terminal-cloud/internal/identity" "github.com/jackc/pgx/v5" ) type ManagedDevice struct { AccountBindingID string `json:"account_binding_id,omitempty"` ID string `json:"id"` BindingID string `json:"binding_id"` HouseholdID string `json:"household_id"` SubjectID string `json:"subject_id"` SubjectName string `json:"subject_name"` Model string `json:"model"` HardwareRevision *string `json:"hardware_revision,omitempty"` Status string `json:"status"` MinimumSecurityVersion *string `json:"minimum_security_version,omitempty"` MinimumFirmwareVersion *string `json:"minimum_firmware_version,omitempty"` FirmwareVersion *string `json:"firmware_version,omitempty"` DesiredRevision int64 `json:"desired_revision"` ReportedRevision *int64 `json:"reported_revision,omitempty"` LastReportedAt *time.Time `json:"last_reported_at,omitempty"` BoundAt time.Time `json:"bound_at"` RegisteredAt time.Time `json:"registered_at"` Version int64 `json:"version"` } type DeviceTwin struct { DeviceID string `json:"device_id"` DesiredRevision int64 `json:"desired_revision"` DesiredConfig map[string]any `json:"desired_config"` ReportedRevision *int64 `json:"reported_revision,omitempty"` ReportedConfig map[string]any `json:"reported_config"` FirmwareVersion *string `json:"firmware_version,omitempty"` BoardRevision *string `json:"board_revision,omitempty"` RTCSDKVersion *string `json:"rtc_sdk_version,omitempty"` Capabilities map[string]any `json:"capabilities"` LastReportedAt *time.Time `json:"last_reported_at,omitempty"` LastErrorCode *string `json:"last_error_code,omitempty"` LastErrorSeverity *string `json:"last_error_severity,omitempty"` LastErrorAt *time.Time `json:"last_error_at,omitempty"` SyncState string `json:"sync_state"` UpdatedAt time.Time `json:"updated_at"` } type DevicePresence struct { DeviceID string `json:"device_id"` Online bool `json:"online"` Status string `json:"status"` ConversationID *string `json:"conversation_id,omitempty"` SequenceNo *int64 `json:"sequence_no,omitempty"` LastSeenAt *time.Time `json:"last_seen_at,omitempty"` Firmware *string `json:"firmware_version,omitempty"` Source string `json:"source"` } var ErrDevicePowerUnsupported = errors.New("device has not reported power control support") func (s *Service) ListDevices(ctx context.Context, principal identity.Principal) ([]ManagedDevice, error) { rows, err := s.pool.Query(ctx, ` SELECT d.public_id::text,b.public_id::text,h.public_id::text,sp.public_id::text,sp.preferred_name, d.model,d.hardware_revision,d.status,d.minimum_security_version,d.minimum_firmware_version,dt.firmware_version, dt.desired_revision,dt.reported_revision,dt.last_reported_at,b.bound_at,d.registered_at,d.version,hm.role FROM user_account ua JOIN household_member hm ON hm.user_id=ua.id AND hm.tenant_id=ua.tenant_id AND hm.status='active' JOIN household h ON h.id=hm.household_id AND h.tenant_id=hm.tenant_id AND h.status <> 'deleted' JOIN device_binding b ON b.household_id=h.id AND b.tenant_id=h.tenant_id AND b.unbound_at IS NULL JOIN device d ON d.id=b.device_id AND d.tenant_id=b.tenant_id JOIN subject_profile sp ON sp.id=b.subject_id AND sp.tenant_id=b.tenant_id AND sp.status IN ('active','suspended') JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id WHERE ua.oidc_issuer=$1 AND ua.oidc_subject=$2 AND ua.status='active' ORDER BY dt.last_reported_at DESC NULLS LAST,d.public_id`, principal.Issuer, principal.Subject) if err != nil { return nil, fmt.Errorf("list managed devices: %w", err) } defer rows.Close() result := make([]ManagedDevice, 0) for rows.Next() { var item ManagedDevice var role identity.Role if err := rows.Scan(&item.ID, &item.BindingID, &item.HouseholdID, &item.SubjectID, &item.SubjectName, &item.Model, &item.HardwareRevision, &item.Status, &item.MinimumSecurityVersion, &item.MinimumFirmwareVersion, &item.FirmwareVersion, &item.DesiredRevision, &item.ReportedRevision, &item.LastReportedAt, &item.BoundAt, &item.RegisteredAt, &item.Version, &role); err != nil { return nil, fmt.Errorf("scan managed device: %w", err) } if identity.Can(role, identity.CapabilityHouseholdRead) { result = append(result, item) } } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate managed devices: %w", err) } rows.Close() owned, err := s.listAccountDevices(ctx, principal) return append(result, owned...), err } func (s *Service) GetDevice(ctx context.Context, principal identity.Principal, devicePublicID string) (ManagedDevice, error) { access, device, err := resolveDevice(ctx, s.pool, principal, devicePublicID, identity.CapabilityHouseholdRead, false) _ = access return device, err } func (s *Service) GetDeviceTwin(ctx context.Context, principal identity.Principal, devicePublicID string) (DeviceTwin, error) { access, _, err := resolveDevice(ctx, s.pool, principal, devicePublicID, identity.CapabilityHouseholdRead, false) if err != nil { return DeviceTwin{}, err } return readDeviceTwin(ctx, s.pool, access) } func (s *Service) PatchDeviceTwin(ctx context.Context, principal identity.Principal, devicePublicID, requestID string, expectedRevision int64, desired map[string]any) (DeviceTwin, error) { var result DeviceTwin err := database.Transact(ctx, s.pool, func(tx pgx.Tx) error { access, _, err := resolveDevice(ctx, tx, principal, devicePublicID, identity.CapabilityDeviceManage, true) if err != nil { return err } if err := concurrency.RequireVersion(expectedRevision, access.desiredRevision); err != nil { return ErrVersionConflict } if _, changing := desired["power"]; changing { twin, err := readDeviceTwin(ctx, tx, access) if err != nil { return err } power, ok := twin.ReportedConfig["power"].(map[string]any) if !ok || (power["mode"] != "active" && power["mode"] != "standby" && power["mode"] != "transition") { return ErrDevicePowerUnsupported } } desiredJSON, _ := json.Marshal(desired) if _, err := tx.Exec(ctx, `UPDATE device_twin SET desired_config=desired_config || $1::jsonb,desired_revision=desired_revision+1,updated_at=now() WHERE device_id=$2::uuid AND tenant_id=$3::uuid`, desiredJSON, access.deviceID, access.tenantID); err != nil { return fmt.Errorf("update desired device configuration: %w", err) } result, err = readDeviceTwin(ctx, tx, access) if err != nil { return err } return s.audit.Record(ctx, tx, audit.Event{ TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID), Action: "device.twin.update", ResourceType: "device", ResourceID: strPtr(devicePublicID), RequestID: requestID, Result: "success", Detail: map[string]any{"from_revision": expectedRevision, "to_revision": result.DesiredRevision}, }) }) return result, err } func (s *Service) GetDevicePresence(ctx context.Context, principal identity.Principal, devicePublicID string) (DevicePresence, error) { access, device, err := resolveDevice(ctx, s.pool, principal, devicePublicID, identity.CapabilityHouseholdRead, false) if err != nil { return DevicePresence{}, err } result := DevicePresence{DeviceID: devicePublicID, Online: false, Status: "offline", LastSeenAt: device.LastReportedAt, Firmware: device.FirmwareVersion, Source: "database"} // Factory-authenticated hardware reports presence through MQTT, not the // legacy conversation telemetry Redis key. Use the same source as mini OTA. var fleetSeen *time.Time var fleetOnline bool err = s.pool.QueryRow(ctx, `SELECT COALESCE(connected AND last_seen_at>now()-interval '90 seconds',false),last_seen_at FROM fleet_presence WHERE device_id=$1`, access.deviceID).Scan(&fleetOnline, &fleetSeen) if err == nil { result.Online = fleetOnline if fleetOnline { result.Status = "online" } result.LastSeenAt = fleetSeen result.Source = "mqtt" return result, nil } if !errors.Is(err, pgx.ErrNoRows) { return DevicePresence{}, err } if s.redis == nil { return result, nil } fields, err := s.redis.HGetAll(ctx, "voice-platform:device:presence:"+devicePublicID).Result() if err != nil || len(fields) == 0 { return result, nil } result.Source = "redis" result.Online = fields["online"] == "true" if value := fields["status"]; value != "" { result.Status = value } else if result.Online { result.Status = "online" } if value := fields["conversation_id"]; value != "" { result.ConversationID = &value } if value := fields["sequence_no"]; value != "" { if sequence, parseErr := strconv.ParseInt(value, 10, 64); parseErr == nil { result.SequenceNo = &sequence } } if value := fields["last_seen_at"]; value != "" { if observed, parseErr := time.Parse(time.RFC3339Nano, value); parseErr == nil { result.LastSeenAt = &observed } } _ = access return result, nil } type deviceAccess struct { tenantID, tenantPublicID, userID, userPublicID string householdID, householdPublicID, deviceID string role identity.Role desiredRevision int64 } func resolveDevice(ctx context.Context, q queryRow, principal identity.Principal, publicID string, capability identity.Capability, lock bool) (deviceAccess, ManagedDevice, error) { sql := ` SELECT t.id::text,t.public_id::text,ua.id::text,ua.public_id::text,h.id::text,h.public_id::text,d.id::text,hm.role, d.public_id::text,b.public_id::text,h.public_id::text,sp.public_id::text,sp.preferred_name,d.model,d.hardware_revision, d.status,d.minimum_security_version,d.minimum_firmware_version,dt.firmware_version,dt.desired_revision,dt.reported_revision, dt.last_reported_at,b.bound_at,d.registered_at,d.version FROM device d JOIN tenant t ON t.id=d.tenant_id AND t.status='active' JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id JOIN device_binding b ON b.device_id=d.id AND b.tenant_id=d.tenant_id AND b.unbound_at IS NULL JOIN household h ON h.id=b.household_id AND h.tenant_id=b.tenant_id AND h.status <> 'deleted' JOIN subject_profile sp ON sp.id=b.subject_id AND sp.tenant_id=b.tenant_id AND sp.status IN ('active','suspended') JOIN household_member hm ON hm.household_id=h.id AND hm.tenant_id=h.tenant_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=h.tenant_id AND ua.status='active' WHERE d.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3` if lock { sql += ` FOR UPDATE OF dt` } var access deviceAccess var device ManagedDevice var role string err := q.QueryRow(ctx, sql, publicID, principal.Issuer, principal.Subject).Scan( &access.tenantID, &access.tenantPublicID, &access.userID, &access.userPublicID, &access.householdID, &access.householdPublicID, &access.deviceID, &role, &device.ID, &device.BindingID, &device.HouseholdID, &device.SubjectID, &device.SubjectName, &device.Model, &device.HardwareRevision, &device.Status, &device.MinimumSecurityVersion, &device.MinimumFirmwareVersion, &device.FirmwareVersion, &device.DesiredRevision, &device.ReportedRevision, &device.LastReportedAt, &device.BoundAt, &device.RegisteredAt, &device.Version) if errors.Is(err, pgx.ErrNoRows) { return resolveAccountDevice(ctx, q, principal, publicID, capability, lock) } if err != nil { return deviceAccess{}, ManagedDevice{}, fmt.Errorf("resolve device access: %w", err) } access.role = identity.Role(role) access.desiredRevision = device.DesiredRevision if !identity.Can(access.role, capability) { return deviceAccess{}, ManagedDevice{}, ErrAccessDenied } return access, device, nil } func readDeviceTwin(ctx context.Context, q queryRow, access deviceAccess) (DeviceTwin, error) { var result DeviceTwin var desiredJSON, reportedJSON, capabilitiesJSON []byte err := q.QueryRow(ctx, ` SELECT d.public_id::text,dt.desired_revision,dt.desired_config,dt.reported_revision,dt.reported_config, dt.firmware_version,dt.board_revision,dt.rtc_sdk_version,dt.capabilities,dt.last_reported_at, dt.last_error_code,dt.last_error_severity,dt.last_error_at,dt.updated_at FROM device_twin dt JOIN device d ON d.id=dt.device_id AND d.tenant_id=dt.tenant_id WHERE dt.device_id=$1::uuid AND dt.tenant_id=$2::uuid`, access.deviceID, access.tenantID).Scan( &result.DeviceID, &result.DesiredRevision, &desiredJSON, &result.ReportedRevision, &reportedJSON, &result.FirmwareVersion, &result.BoardRevision, &result.RTCSDKVersion, &capabilitiesJSON, &result.LastReportedAt, &result.LastErrorCode, &result.LastErrorSeverity, &result.LastErrorAt, &result.UpdatedAt) if err != nil { return DeviceTwin{}, fmt.Errorf("read device twin: %w", err) } if err := decodeObject(desiredJSON, &result.DesiredConfig); err != nil { return DeviceTwin{}, err } if err := decodeObject(reportedJSON, &result.ReportedConfig); err != nil { return DeviceTwin{}, err } if err := decodeObject(capabilitiesJSON, &result.Capabilities); err != nil { return DeviceTwin{}, err } switch { case result.ReportedRevision == nil: result.SyncState = "never_reported" case *result.ReportedRevision >= result.DesiredRevision: result.SyncState = "applied" default: result.SyncState = "pending" } return result, nil }