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.
297 lines
14 KiB
Go
297 lines
14 KiB
Go
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
|
|
}
|