story/internal/saas/device_command_service.go

package saas

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/database"
	"data-terminal-cloud/internal/identity"

	"github.com/jackc/pgx/v5"
)

type DeviceCommand struct {
	ID               string         `json:"id"`
	DeviceID         string         `json:"device_id"`
	Type             string         `json:"type"`
	Payload          map[string]any `json:"payload"`
	Status           string         `json:"status"`
	ResultSummary    map[string]any `json:"result_summary"`
	ExpiresAt        time.Time      `json:"expires_at"`
	DeliveryAttempts int            `json:"delivery_attempts"`
	Version          int64          `json:"version"`
	CreatedAt        time.Time      `json:"created_at"`
	UpdatedAt        time.Time      `json:"updated_at"`
}

func (s *Service) CreateDeviceCommand(ctx context.Context, principal identity.Principal, devicePublicID, idempotencyKey, requestID, commandType string, payload map[string]any, expiresInSec int) (DeviceCommand, bool, error) {
	requestJSON, err := json.Marshal(map[string]any{"type": commandType, "payload": payload, "expires_in_sec": expiresInSec})
	if err != nil {
		return DeviceCommand{}, false, err
	}
	digestValue := sha256.Sum256(requestJSON)
	digest := hex.EncodeToString(digestValue[:])
	var publicID string
	created := false
	err = database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
		expiresAt := time.Now().UTC().Add(time.Duration(expiresInSec) * time.Second)
		access, _, err := resolveDevice(ctx, tx, principal, devicePublicID, identity.CapabilityDeviceManage, true)
		if err != nil {
			return err
		}
		var internalID string
		err = tx.QueryRow(ctx, `INSERT INTO device_command(tenant_id,device_id,command_type,payload,requested_by,idempotency_key,request_digest,expires_at) VALUES($1::uuid,$2::uuid,$3,$4::jsonb,$5::uuid,$6,$7,$8) ON CONFLICT (tenant_id,requested_by,idempotency_key) DO NOTHING RETURNING id::text,public_id::text`, access.tenantID, access.deviceID, commandType, requestJSONForPayload(payload), access.userID, idempotencyKey, digest, expiresAt).Scan(&internalID, &publicID)
		if errors.Is(err, pgx.ErrNoRows) {
			var stored string
			if err := tx.QueryRow(ctx, `SELECT public_id::text,request_digest FROM device_command WHERE tenant_id=$1::uuid AND requested_by=$2::uuid AND idempotency_key=$3`, access.tenantID, access.userID, idempotencyKey).Scan(&publicID, &stored); err != nil {
				return err
			}
			if stored != digest {
				return ErrIdempotencyConflict
			}
			return nil
		}
		if err != nil {
			return fmt.Errorf("create device command: %w", err)
		}
		created = true
		return s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID), Action: "device.command.create", ResourceType: "device_command", ResourceID: strPtr(publicID), RequestID: requestID, Result: "success", Detail: map[string]any{"device_id": devicePublicID, "type": commandType, "expires_at": expiresAt}})
	})
	if err != nil {
		return DeviceCommand{}, false, err
	}
	item, err := s.GetDeviceCommand(ctx, principal, publicID)
	return item, created, err
}

func requestJSONForPayload(payload map[string]any) []byte {
	value, _ := json.Marshal(payload)
	return value
}

func (s *Service) ListDeviceCommands(ctx context.Context, principal identity.Principal, devicePublicID string) ([]DeviceCommand, error) {
	access, _, err := resolveDevice(ctx, s.pool, principal, devicePublicID, identity.CapabilityDeviceManage, false)
	if err != nil {
		return nil, err
	}
	_, _ = s.pool.Exec(ctx, `UPDATE device_command SET status='EXPIRED',completed_at=now(),version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND device_id=$2::uuid AND status IN ('QUEUED','DELIVERED','ACKNOWLEDGED') AND expires_at<=now()`, access.tenantID, access.deviceID)
	rows, err := s.pool.Query(ctx, deviceCommandSelectSQL+` WHERE dc.tenant_id=$1::uuid AND dc.device_id=$2::uuid ORDER BY dc.created_at DESC,dc.id DESC LIMIT 100`, access.tenantID, access.deviceID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []DeviceCommand{}
	for rows.Next() {
		item, err := scanDeviceCommand(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Service) GetDeviceCommand(ctx context.Context, principal identity.Principal, publicID string) (DeviceCommand, error) {
	item, err := scanDeviceCommand(s.pool.QueryRow(ctx, deviceCommandSelectSQL+` 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_member hm ON hm.household_id=b.household_id AND hm.tenant_id=b.tenant_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=b.tenant_id AND ua.status='active' WHERE dc.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, publicID, principal.Issuer, principal.Subject))
	if errors.Is(err, pgx.ErrNoRows) {
		return DeviceCommand{}, ErrAccessDenied
	}
	if err != nil {
		return DeviceCommand{}, err
	}
	var role string
	if err := s.pool.QueryRow(ctx, `SELECT hm.role FROM device_command dc JOIN device_binding b ON b.device_id=dc.device_id AND b.tenant_id=dc.tenant_id AND b.unbound_at IS NULL JOIN household_member hm ON hm.household_id=b.household_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=dc.tenant_id WHERE dc.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, publicID, principal.Issuer, principal.Subject).Scan(&role); err != nil || !identity.Can(identity.Role(role), identity.CapabilityDeviceManage) {
		return DeviceCommand{}, ErrAccessDenied
	}
	return item, nil
}

const deviceCommandSelectSQL = `SELECT dc.public_id::text,d.public_id::text,dc.command_type,dc.payload,dc.status,dc.result_summary,dc.expires_at,dc.delivery_attempts,dc.version,dc.created_at,dc.updated_at FROM device_command dc JOIN device d ON d.id=dc.device_id AND d.tenant_id=dc.tenant_id`

func scanDeviceCommand(row rowScanner) (DeviceCommand, error) {
	var item DeviceCommand
	var payload, result []byte
	if err := row.Scan(&item.ID, &item.DeviceID, &item.Type, &payload, &item.Status, &result, &item.ExpiresAt, &item.DeliveryAttempts, &item.Version, &item.CreatedAt, &item.UpdatedAt); err != nil {
		return DeviceCommand{}, err
	}
	if err := decodeObject(payload, &item.Payload); err != nil {
		return DeviceCommand{}, err
	}
	if err := decodeObject(result, &item.ResultSummary); err != nil {
		return DeviceCommand{}, err
	}
	return item, nil
}