yovision/Sense/internal/store/control_postgres.go

package store

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"crypto/subtle"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"net/url"
	"sort"
	"strings"
	"time"

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

	"yovision/sense/internal/device"
)

const controlReceiptTTL = 24 * time.Hour

const controlDeviceColumns = `d.id, d.tenant_id, d.site_id, d.serial_number, d.name, d.modality,
	d.area_id, d.desired_state, d.actual_state,
	(d.endpoint_ref <> ''), (d.credential_ref <> ''),
	d.generation, d.resource_version,
	r.observed_generation, r.failure_count, r.next_attempt_at, r.last_error_code,
	d.quota_source_version, d.area_policy_source_version,
	GREATEST(
		(SELECT q.synced_at FROM sense.site_quota_projection_state q
		 WHERE q.tenant_id = d.tenant_id AND q.site_id = d.site_id
		   AND q.source_version = d.quota_source_version),
		(SELECT a.synced_at FROM sense.area_policy_projection_state a
		 WHERE a.tenant_id = d.tenant_id AND a.site_id = d.site_id
		   AND a.area_id = d.area_id AND a.source_version = d.area_policy_source_version)
	),
	d.created_at, d.updated_at,
	COALESCE((SELECT jsonb_agg(c.capability ORDER BY c.capability)
	 FROM sense.device_capabilities c WHERE c.device_id = d.id), '[]'::jsonb)::text`

const controlDeviceSelect = `SELECT ` + controlDeviceColumns + `
	FROM sense.devices d JOIN sense.reconcile_state r ON r.device_id = d.id`

type controlScanner interface {
	Scan(...any) error
}

func scanControlDevice(row controlScanner) (ControlDevice, error) {
	var value ControlDevice
	var nextAttempt, syncedAt sql.NullTime
	var lastError sql.NullString
	var quotaVersion, areaVersion sql.NullInt64
	var capabilitiesJSON string
	err := row.Scan(
		&value.ID, &value.TenantID, &value.SiteID, &value.SerialNumber, &value.Name,
		&value.Modality, &value.AreaID, &value.DesiredState, &value.ActualState,
		&value.EndpointConfigured, &value.CredentialConfigured,
		&value.Generation, &value.ResourceVersion, &value.ObservedGeneration,
		&value.FailureCount, &nextAttempt, &lastError, &quotaVersion, &areaVersion,
		&syncedAt, &value.CreatedAt, &value.UpdatedAt, &capabilitiesJSON,
	)
	if err != nil {
		return ControlDevice{}, err
	}
	if err := json.Unmarshal([]byte(capabilitiesJSON), &value.Capabilities); err != nil {
		return ControlDevice{}, errors.New("decode postgres control device capabilities")
	}
	if value.Capabilities == nil {
		value.Capabilities = make([]device.Capability, 0)
	}
	if nextAttempt.Valid {
		point := nextAttempt.Time.UTC()
		value.NextAttemptAt = &point
	}
	if lastError.Valid {
		code := lastError.String
		value.LastErrorCode = &code
	}
	if quotaVersion.Valid {
		version := quotaVersion.Int64
		value.ProjectionVersions.QuotaSourceVersion = &version
	}
	if areaVersion.Valid {
		version := areaVersion.Int64
		value.ProjectionVersions.AreaPolicySourceVersion = &version
	}
	if syncedAt.Valid {
		point := syncedAt.Time.UTC()
		value.ProjectionVersions.SyncedAt = &point
	}
	value.Converged = value.ObservedGeneration >= value.Generation && value.FailureCount == 0
	value.AdapterStatus = controlAdapterStatus(value)
	return value, nil
}

func controlAdapterStatus(value ControlDevice) string {
	if value.LastErrorCode != nil {
		switch *value.LastErrorCode {
		case "authentication_failed":
			return "authentication_failed"
		case "adapter_not_ready":
			return "adapter_not_ready"
		default:
			return "unavailable"
		}
	}
	if value.Converged {
		return "ready"
	}
	if value.ActualState == device.ActualFailed || value.ActualState == device.ActualOffline {
		return "unavailable"
	}
	return "pending"
}

func (s *Postgres) GetControlDevice(
	ctx context.Context, tenantID, siteID, deviceID string,
) (ControlDevice, error) {
	value, err := scanControlDevice(s.db.QueryRowContext(ctx, controlDeviceSelect+`
		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3`, tenantID, siteID, deviceID))
	if errors.Is(err, sql.ErrNoRows) {
		return ControlDevice{}, ErrNotFound
	}
	if err != nil {
		return ControlDevice{}, errors.New("get postgres control device")
	}
	return value, nil
}

func (s *Postgres) ListControlDevices(
	ctx context.Context, tenantID, siteID string, filter ControlListFilter,
) (ControlDevicePage, error) {
	quota, err := s.controlSiteQuota(ctx, tenantID, siteID)
	if err != nil {
		return ControlDevicePage{}, err
	}
	query := controlDeviceSelect + ` WHERE d.tenant_id = $1 AND d.site_id = $2`
	arguments := []any{tenantID, siteID}
	appendCondition := func(clause string, value any) {
		arguments = append(arguments, value)
		query += fmt.Sprintf(clause, len(arguments))
	}
	if filter.Modality != nil {
		appendCondition(` AND d.modality = $%d`, *filter.Modality)
	}
	if filter.Capability != nil {
		appendCondition(` AND EXISTS (SELECT 1 FROM sense.device_capabilities fc
			WHERE fc.device_id = d.id AND fc.capability = $%d)`, *filter.Capability)
	}
	if filter.DesiredState != nil {
		appendCondition(` AND d.desired_state = $%d`, *filter.DesiredState)
	}
	if filter.ActualState != nil {
		appendCondition(` AND d.actual_state = $%d`, *filter.ActualState)
	}
	if filter.AfterCreated != nil {
		arguments = append(arguments, filter.AfterCreated.UTC(), filter.AfterDeviceID)
		query += fmt.Sprintf(` AND (d.created_at, d.id) > ($%d, $%d)`, len(arguments)-1, len(arguments))
	}
	arguments = append(arguments, filter.Limit+1)
	query += fmt.Sprintf(` ORDER BY d.created_at ASC, d.id ASC LIMIT $%d`, len(arguments))
	rows, err := s.db.QueryContext(ctx, query, arguments...)
	if err != nil {
		return ControlDevicePage{}, errors.New("list postgres control devices")
	}
	defer rows.Close()
	values := make([]ControlDevice, 0, filter.Limit+1)
	for rows.Next() {
		value, scanErr := scanControlDevice(rows)
		if scanErr != nil {
			return ControlDevicePage{}, errors.New("scan postgres control device page")
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return ControlDevicePage{}, errors.New("iterate postgres control device page")
	}
	hasMore := len(values) > filter.Limit
	if hasMore {
		values = values[:filter.Limit]
	}
	return ControlDevicePage{Items: values, HasMore: hasMore, Quota: quota}, nil
}

func (s *Postgres) controlSiteQuota(ctx context.Context, tenantID, siteID string) (ControlSiteQuota, error) {
	var maximum int
	var sourceVersion int64
	var syncedAt time.Time
	err := s.db.QueryRowContext(ctx, `SELECT max_video_channels, source_version, source_updated_at
		FROM bell.site_quota_v1 WHERE tenant_id = $1 AND site_id = $2`, tenantID, siteID).
		Scan(&maximum, &sourceVersion, &syncedAt)
	if errors.Is(err, sql.ErrNoRows) {
		return ControlSiteQuota{}, ErrNotFound
	}
	if err != nil {
		return ControlSiteQuota{}, errors.New("read postgres control site quota")
	}
	var used int
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM sense.devices d
		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.desired_state = 'enabled'
		AND EXISTS (SELECT 1 FROM sense.device_capabilities c
			WHERE c.device_id = d.id AND c.capability = 'video_capture')`, tenantID, siteID).Scan(&used); err != nil {
		return ControlSiteQuota{}, errors.New("count postgres control site video channels")
	}
	status := "current"
	if maximum < 1 || maximum > device.MaximumVideoChannels || sourceVersion < 1 || syncedAt.IsZero() {
		status = "invalid"
		return ControlSiteQuota{Status: status, UsedVideoChannels: used, OverLimit: false}, nil
	}
	available := maximum - used
	if available < 0 {
		available = 0
	}
	point := syncedAt.UTC()
	return ControlSiteQuota{
		Status: status, UsedVideoChannels: used, MaxVideoChannels: &maximum,
		AvailableVideoChannels: &available, OverLimit: used > maximum,
		SourceVersion: &sourceVersion, SyncedAt: &point,
	}, nil
}

type controlReceipt struct {
	Status    int
	Body      []byte
	ETag      string
	Location  string
	TraceID   string
	CreatedAt time.Time
}

func controlScopeHash(scope IdempotencyScope) [sha256.Size]byte {
	encoded, _ := json.Marshal([]string{
		scope.PrincipalID, scope.TenantID, scope.SiteID, scope.Operation, scope.Key,
	})
	return sha256.Sum256(encoded)
}

func readControlReceipt(
	ctx context.Context, tx *sql.Tx, scope IdempotencyScope, now time.Time,
) (controlReceipt, bool, error) {
	var receipt controlReceipt
	scopeHash := controlScopeHash(scope)
	if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext($1))`, hex.EncodeToString(scopeHash[:])); err != nil {
		return receipt, false, errors.New("lock postgres Control API idempotency scope")
	}
	var storedRequestHash []byte
	var etag, location sql.NullString
	var expiresAt time.Time
	err := tx.QueryRowContext(ctx, `SELECT request_hash, response_status, response_body::text,
		response_etag, response_location, trace_id, created_at, expires_at
		FROM sense.control_idempotency_receipts WHERE scope_hash = $1`, scopeHash[:]).
		Scan(&storedRequestHash, &receipt.Status, &receipt.Body, &etag, &location,
			&receipt.TraceID, &receipt.CreatedAt, &expiresAt)
	if errors.Is(err, sql.ErrNoRows) {
		return receipt, false, nil
	}
	if err != nil {
		return receipt, false, errors.New("read postgres Control API idempotency receipt")
	}
	if !expiresAt.After(now) {
		if _, err := tx.ExecContext(ctx, `DELETE FROM sense.control_idempotency_receipts
			WHERE scope_hash = $1`, scopeHash[:]); err != nil {
			return receipt, false, errors.New("expire postgres Control API idempotency receipt")
		}
		return controlReceipt{}, false, nil
	}
	if subtle.ConstantTimeCompare(storedRequestHash, scope.RequestHash[:]) != 1 {
		return receipt, false, ErrIdempotencyConflict
	}
	receipt.ETag = etag.String
	receipt.Location = location.String
	return receipt, true, nil
}

func writeControlReceipt(
	ctx context.Context, tx *sql.Tx, scope IdempotencyScope, receipt controlReceipt,
) error {
	scopeHash := controlScopeHash(scope)
	_, err := tx.ExecContext(ctx, `INSERT INTO sense.control_idempotency_receipts(
		scope_hash, request_hash, operation_name, principal_id, tenant_id, site_id,
		response_status, response_body, response_etag, response_location, trace_id,
		created_at, expires_at
	) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, NULLIF($9, ''), NULLIF($10, ''), $11, $12, $13)`,
		scopeHash[:], scope.RequestHash[:], scope.Operation, scope.PrincipalID,
		scope.TenantID, scope.SiteID, receipt.Status, string(receipt.Body),
		receipt.ETag, receipt.Location, receipt.TraceID, receipt.CreatedAt,
		receipt.CreatedAt.Add(controlReceiptTTL))
	if err != nil {
		return errors.New("write postgres Control API idempotency receipt")
	}
	// Bound opportunistic cleanup; never scans or deletes unexpired receipts.
	_, _ = tx.ExecContext(ctx, `DELETE FROM sense.control_idempotency_receipts
		WHERE scope_hash IN (SELECT scope_hash FROM sense.control_idempotency_receipts
			WHERE expires_at <= $1 ORDER BY expires_at LIMIT 32)`, receipt.CreatedAt)
	return nil
}

func (s *Postgres) CreateControlDevice(
	ctx context.Context, request ControlCreateRequest,
) (ControlCreateResult, error) {
	now := time.Now().UTC()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return ControlCreateResult{}, errors.New("begin postgres Control API device create")
	}
	defer tx.Rollback()
	receipt, found, err := readControlReceipt(ctx, tx, request.Scope, now)
	if err != nil {
		return ControlCreateResult{}, err
	}
	if found {
		var value ControlDevice
		if err := json.Unmarshal(receipt.Body, &value); err != nil {
			return ControlCreateResult{}, errors.New("decode postgres device creation receipt")
		}
		if err := tx.Commit(); err != nil {
			return ControlCreateResult{}, errors.New("commit postgres device creation replay")
		}
		return ControlCreateResult{
			Device: value, AcceptedAt: receipt.CreatedAt, TraceID: receipt.TraceID,
			ETag: receipt.ETag, Location: receipt.Location, Replay: true,
		}, nil
	}
	value := request.Device
	if value.Generation == 0 {
		value.Generation = 1
	}
	if value.ResourceVersion == 0 {
		value.ResourceVersion = 1
	}
	if value.ActualState == "" {
		value.ActualState = device.ActualPending
	}
	value.CreatedAt = now
	value.UpdatedAt = now
	if err := createControlDeviceTx(ctx, tx, value, now); err != nil {
		return ControlCreateResult{}, err
	}
	created, err := scanControlDevice(tx.QueryRowContext(ctx, controlDeviceSelect+`
		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3`, value.TenantID, value.SiteID, value.ID))
	if err != nil {
		return ControlCreateResult{}, errors.New("read created postgres control device")
	}
	responseBody, err := json.Marshal(created)
	if err != nil {
		return ControlCreateResult{}, errors.New("encode created postgres control device")
	}
	etag := DeviceETag(created.ID, created.ResourceVersion)
	location := "/api/v1/sites/" + url.PathEscape(created.SiteID) + "/devices/" + url.PathEscape(created.ID)
	receipt = controlReceipt{
		Status: 201, Body: responseBody, ETag: etag, Location: location,
		TraceID: request.Scope.TraceID, CreatedAt: now,
	}
	if err := writeControlReceipt(ctx, tx, request.Scope, receipt); err != nil {
		return ControlCreateResult{}, err
	}
	if err := tx.Commit(); err != nil {
		return ControlCreateResult{}, errors.New("commit postgres Control API device create")
	}
	return ControlCreateResult{
		Device: created, AcceptedAt: now, TraceID: receipt.TraceID,
		ETag: etag, Location: location,
	}, nil
}

func createControlDeviceTx(ctx context.Context, tx *sql.Tx, value device.Device, now time.Time) error {
	if err := value.Validate(); err != nil {
		return fmt.Errorf("validate Control API device: %w", err)
	}
	areaVersion, err := checkPostgresAreaPolicy(
		ctx, tx, value.TenantID, value.SiteID, value.AreaID,
		value.HasCapability(device.CapabilityVideoCapture), now,
	)
	if err != nil {
		return err
	}
	var quotaVersion int64
	if value.ConsumesVideoChannel() {
		quotaVersion, err = checkPostgresVideoQuota(ctx, tx, value.TenantID, value.SiteID, now)
		if err != nil {
			return err
		}
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO sense.devices(
		id, tenant_id, site_id, area_id, serial_number, name, modality,
		desired_state, actual_state, endpoint_ref, credential_ref, profile_token,
		path_name, generation, resource_version, quota_source_version,
		area_policy_source_version, created_at, updated_at
	) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19)`,
		value.ID, value.TenantID, value.SiteID, value.AreaID, value.SerialNumber, value.Name,
		value.Modality, value.DesiredState, value.ActualState, value.EndpointRef,
		value.CredentialRef, value.ProfileToken, value.PathName, value.Generation,
		value.ResourceVersion, nullableVersion(quotaVersion), areaVersion,
		value.CreatedAt, value.UpdatedAt)
	if err != nil {
		var postgresError *pgconn.PgError
		if errors.As(err, &postgresError) && postgresError.Code == "23505" &&
			strings.Contains(postgresError.ConstraintName, "serial_number") {
			return ErrDuplicateSerialNumber
		}
		return errors.New("insert postgres Control API device")
	}
	for _, capability := range sortedCapabilities(value.Capabilities) {
		if _, err := tx.ExecContext(ctx, `INSERT INTO sense.device_capabilities(device_id, capability)
			VALUES ($1, $2)`, value.ID, capability); err != nil {
			return errors.New("insert postgres Control API device capability")
		}
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO sense.reconcile_state(device_id, updated_at)
		VALUES ($1, $2)`, value.ID, now); err != nil {
		return errors.New("insert postgres Control API reconcile state")
	}
	return insertPostgresAudit(ctx, tx, postgresAuditEvent{
		EventType: "device.created", TenantID: value.TenantID, SiteID: value.SiteID,
		DeviceID: value.ID, Generation: value.Generation,
		QuotaSourceVersion: quotaVersion, AreaPolicySourceVersion: areaVersion,
		OccurredAt: now,
		Payload: map[string]any{
			"kind": "device_created", "area_id": value.AreaID, "modality": value.Modality,
			"capabilities": sortedCapabilities(value.Capabilities), "desired_state": value.DesiredState,
		},
	})
}

var _ ControlRepository = (*Postgres)(nil)

func (s *Postgres) PatchControlDevice(
	ctx context.Context, tenantID, siteID, deviceID, expectedETag string, patch ControlPatch,
) (ControlMutationResult, error) {
	now := time.Now().UTC()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return ControlMutationResult{}, errors.New("begin postgres Control API device patch")
	}
	defer tx.Rollback()
	var name, areaID, endpointRef, credentialRef, profileToken string
	var generation, resourceVersion int64
	var desired device.DesiredState
	var quotaVersion, areaVersion sql.NullInt64
	var hasVideo bool
	err = tx.QueryRowContext(ctx, `SELECT d.name, d.area_id, d.endpoint_ref,
		d.credential_ref, d.profile_token, d.generation, d.resource_version,
		d.desired_state, d.quota_source_version, d.area_policy_source_version,
		EXISTS (SELECT 1 FROM sense.device_capabilities c
			WHERE c.device_id = d.id AND c.capability = 'video_capture')
		FROM sense.devices d
		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3 FOR UPDATE`,
		tenantID, siteID, deviceID).Scan(
		&name, &areaID, &endpointRef, &credentialRef, &profileToken,
		&generation, &resourceVersion, &desired, &quotaVersion, &areaVersion, &hasVideo,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return ControlMutationResult{}, ErrNotFound
	}
	if err != nil {
		return ControlMutationResult{}, errors.New("read postgres Control API device patch state")
	}
	if DeviceETag(deviceID, resourceVersion) != expectedETag {
		return ControlMutationResult{}, ErrETagMismatch
	}
	changedFields := make([]string, 0, 5)
	reconcileChanged := false
	if patch.Name != nil && *patch.Name != name {
		name = *patch.Name
		changedFields = append(changedFields, "name")
	}
	if patch.AreaID != nil && *patch.AreaID != areaID {
		version, policyErr := checkPostgresAreaPolicy(
			ctx, tx, tenantID, siteID, *patch.AreaID, hasVideo, now,
		)
		if policyErr != nil {
			return ControlMutationResult{}, policyErr
		}
		areaID = *patch.AreaID
		areaVersion = sql.NullInt64{Int64: version, Valid: true}
		changedFields = append(changedFields, "area_id")
	}
	if patch.EndpointRef != nil && *patch.EndpointRef != endpointRef {
		endpointRef = *patch.EndpointRef
		changedFields = append(changedFields, "endpoint_ref")
		reconcileChanged = true
	}
	if patch.CredentialRef != nil && *patch.CredentialRef != credentialRef {
		credentialRef = *patch.CredentialRef
		changedFields = append(changedFields, "credential_ref")
		reconcileChanged = true
	}
	if patch.ProfileToken != nil && *patch.ProfileToken != profileToken {
		profileToken = *patch.ProfileToken
		changedFields = append(changedFields, "profile_token")
		reconcileChanged = true
	}
	if len(changedFields) > 0 {
		resourceVersion++
		if reconcileChanged {
			generation++
		}
		_, err = tx.ExecContext(ctx, `UPDATE sense.devices SET
			name = $1, area_id = $2, endpoint_ref = $3, credential_ref = $4,
			profile_token = $5, generation = $6, resource_version = $7,
			area_policy_source_version = $8,
			actual_state = CASE WHEN $9 THEN 'pending' ELSE actual_state END,
			updated_at = $10
			WHERE tenant_id = $11 AND site_id = $12 AND id = $13`,
			name, areaID, endpointRef, credentialRef, profileToken, generation,
			resourceVersion, nullableVersion(areaVersion.Int64), reconcileChanged, now,
			tenantID, siteID, deviceID)
		if err != nil {
			return ControlMutationResult{}, errors.New("update postgres Control API device configuration")
		}
		if reconcileChanged {
			if _, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
				failure_count = 0, next_attempt_at = NULL, last_error_code = NULL,
				lease_owner = NULL, lease_token = NULL, lease_until = NULL,
				updated_at = $1 WHERE device_id = $2`, now, deviceID); err != nil {
				return ControlMutationResult{}, errors.New("reset postgres Control API reconcile state")
			}
		}
	}
	if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{
		EventType: "device.configuration.accepted", TenantID: tenantID, SiteID: siteID,
		DeviceID: deviceID, Generation: generation,
		QuotaSourceVersion: quotaVersion.Int64, AreaPolicySourceVersion: areaVersion.Int64,
		OccurredAt: now,
		Payload: map[string]any{
			"kind": "configuration_accepted", "changed": len(changedFields) > 0,
			"changed_fields": changedFields, "area_id": areaID,
		},
	}); err != nil {
		return ControlMutationResult{}, err
	}
	updated, err := scanControlDevice(tx.QueryRowContext(ctx, controlDeviceSelect+`
		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3`, tenantID, siteID, deviceID))
	if err != nil {
		return ControlMutationResult{}, errors.New("read patched postgres control device")
	}
	if err := tx.Commit(); err != nil {
		return ControlMutationResult{}, errors.New("commit postgres Control API device patch")
	}
	return ControlMutationResult{
		Device: updated, AcceptedAt: now, TraceID: auditFromContext(ctx).TraceID,
		ETag: DeviceETag(updated.ID, updated.ResourceVersion),
	}, nil
}

func (s *Postgres) SetControlDesiredState(
	ctx context.Context, tenantID, siteID, deviceID, expectedETag string, desired device.DesiredState,
) (ControlMutationResult, error) {
	now := time.Now().UTC()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return ControlMutationResult{}, errors.New("begin postgres Control API desired-state update")
	}
	defer tx.Rollback()
	updated, err := setControlDesiredStateTx(
		ctx, tx, tenantID, siteID, deviceID, expectedETag, desired, now,
	)
	if err != nil {
		return ControlMutationResult{}, err
	}
	if err := tx.Commit(); err != nil {
		return ControlMutationResult{}, errors.New("commit postgres Control API desired-state update")
	}
	return ControlMutationResult{
		Device: updated, AcceptedAt: now, TraceID: auditFromContext(ctx).TraceID,
		ETag: DeviceETag(updated.ID, updated.ResourceVersion),
	}, nil
}

func setControlDesiredStateTx(
	ctx context.Context, tx *sql.Tx, tenantID, siteID, deviceID, expectedETag string,
	desired device.DesiredState, now time.Time,
) (ControlDevice, error) {
	var areaID, endpointRef, pathName string
	var current device.DesiredState
	var generation, resourceVersion int64
	var quotaVersion, areaVersion sql.NullInt64
	var hasVideo bool
	err := tx.QueryRowContext(ctx, `SELECT d.area_id, d.desired_state, d.endpoint_ref,
		d.path_name, d.generation, d.resource_version, d.quota_source_version,
		d.area_policy_source_version,
		EXISTS (SELECT 1 FROM sense.device_capabilities c
			WHERE c.device_id = d.id AND c.capability = 'video_capture')
		FROM sense.devices d
		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3 FOR UPDATE`,
		tenantID, siteID, deviceID).Scan(
		&areaID, &current, &endpointRef, &pathName, &generation, &resourceVersion,
		&quotaVersion, &areaVersion, &hasVideo,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return ControlDevice{}, ErrNotFound
	}
	if err != nil {
		return ControlDevice{}, errors.New("read postgres Control API desired state")
	}
	if DeviceETag(deviceID, resourceVersion) != expectedETag {
		return ControlDevice{}, ErrETagMismatch
	}
	if current != desired {
		var admittedQuota, admittedArea int64
		if desired == device.DesiredEnabled && hasVideo {
			if strings.TrimSpace(endpointRef) == "" || strings.TrimSpace(pathName) == "" {
				return ControlDevice{}, errors.New("video adapter configuration is incomplete")
			}
			admittedArea, err = checkPostgresAreaPolicy(ctx, tx, tenantID, siteID, areaID, true, now)
			if err != nil {
				return ControlDevice{}, err
			}
			admittedQuota, err = checkPostgresVideoQuota(ctx, tx, tenantID, siteID, now)
			if err != nil {
				return ControlDevice{}, err
			}
		}
		generation++
		resourceVersion++
		err = tx.QueryRowContext(ctx, `UPDATE sense.devices SET
			desired_state = $1, actual_state = 'pending', generation = $2,
			resource_version = $3,
			quota_source_version = COALESCE($4, quota_source_version),
			area_policy_source_version = COALESCE($5, area_policy_source_version),
			updated_at = $6
			WHERE tenant_id = $7 AND site_id = $8 AND id = $9
			RETURNING quota_source_version, area_policy_source_version`,
			desired, generation, resourceVersion, nullableVersion(admittedQuota),
			nullableVersion(admittedArea), now, tenantID, siteID, deviceID).
			Scan(&quotaVersion, &areaVersion)
		if err != nil {
			return ControlDevice{}, errors.New("update postgres Control API desired state")
		}
		if _, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
			failure_count = 0, next_attempt_at = NULL, last_error_code = NULL,
			lease_owner = NULL, lease_token = NULL, lease_until = NULL,
			updated_at = $1 WHERE device_id = $2`, now, deviceID); err != nil {
			return ControlDevice{}, errors.New("reset postgres Control API desired-state reconciliation")
		}
	}
	if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{
		EventType: "device.desired_state.accepted", TenantID: tenantID, SiteID: siteID,
		DeviceID: deviceID, Generation: generation,
		QuotaSourceVersion: quotaVersion.Int64, AreaPolicySourceVersion: areaVersion.Int64,
		OccurredAt: now,
		Payload: map[string]any{
			"kind": "desired_state_accepted", "previous_desired_state": current,
			"desired_state": desired, "changed": current != desired,
		},
	}); err != nil {
		return ControlDevice{}, err
	}
	value, err := scanControlDevice(tx.QueryRowContext(ctx, controlDeviceSelect+`
		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3`, tenantID, siteID, deviceID))
	if err != nil {
		return ControlDevice{}, errors.New("read updated postgres control desired state")
	}
	return value, nil
}

func (s *Postgres) BatchSetControlDesiredState(
	ctx context.Context, request ControlBatchRequest,
) (ControlBatchOperation, error) {
	now := time.Now().UTC()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return ControlBatchOperation{}, errors.New("begin postgres Control API batch")
	}
	defer tx.Rollback()
	receipt, found, err := readControlReceipt(ctx, tx, request.Scope, now)
	if err != nil {
		return ControlBatchOperation{}, err
	}
	if found {
		var operation ControlBatchOperation
		if err := json.Unmarshal(receipt.Body, &operation); err != nil {
			return ControlBatchOperation{}, errors.New("decode postgres Control API batch receipt")
		}
		operation.TenantID = request.Scope.TenantID
		operation.SiteID = request.Scope.SiteID
		operation.Replay = true
		if err := tx.Commit(); err != nil {
			return ControlBatchOperation{}, errors.New("commit postgres Control API batch replay")
		}
		return operation, nil
	}
	operationID, err := newControlOperationID(now)
	if err != nil {
		return ControlBatchOperation{}, err
	}
	counts := make(map[string]int, len(request.Items))
	for _, item := range request.Items {
		counts[item.DeviceID]++
	}
	if err := lockControlBatchDevices(
		ctx, tx, request.Scope.TenantID, request.Scope.SiteID, counts,
	); err != nil {
		return ControlBatchOperation{}, err
	}
	results := make([]ControlBatchItemResult, 0, len(request.Items))
	succeeded := 0
	for index, item := range request.Items {
		if counts[item.DeviceID] > 1 {
			code, message := "invalid_request", "device_id is duplicated in this request"
			results = append(results, ControlBatchItemResult{
				DeviceID: item.DeviceID, Status: "rejected", ErrorCode: &code, Message: &message,
			})
			continue
		}
		savepoint := fmt.Sprintf("control_batch_%d", index)
		if _, err := tx.ExecContext(ctx, "SAVEPOINT "+savepoint); err != nil {
			return ControlBatchOperation{}, errors.New("create postgres Control API batch savepoint")
		}
		updated, itemErr := setControlDesiredStateTx(
			ctx, tx, request.Scope.TenantID, request.Scope.SiteID,
			item.DeviceID, item.ETag, item.DesiredState, now,
		)
		if itemErr != nil {
			if _, rollbackErr := tx.ExecContext(ctx, "ROLLBACK TO SAVEPOINT "+savepoint); rollbackErr != nil {
				return ControlBatchOperation{}, errors.New("rollback postgres Control API batch item")
			}
			code, status, message := controlBatchError(itemErr)
			results = append(results, ControlBatchItemResult{
				DeviceID: item.DeviceID, Status: status, ErrorCode: &code, Message: &message,
			})
		} else {
			generation := updated.Generation
			results = append(results, ControlBatchItemResult{
				DeviceID: item.DeviceID, Status: "succeeded", Generation: &generation,
			})
			succeeded++
		}
		if _, err := tx.ExecContext(ctx, "RELEASE SAVEPOINT "+savepoint); err != nil {
			return ControlBatchOperation{}, errors.New("release postgres Control API batch savepoint")
		}
	}
	status := "partially_succeeded"
	if succeeded == len(results) {
		status = "succeeded"
	} else if succeeded == 0 {
		status = "failed"
	}
	completedAt := now
	operation := ControlBatchOperation{
		ID: operationID, TenantID: request.Scope.TenantID, SiteID: request.Scope.SiteID,
		Status: status, SubmittedAt: now, CompletedAt: &completedAt,
		Results: results, TraceID: request.Scope.TraceID,
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO sense.batch_operations(
		id, tenant_id, site_id, principal_id, status, trace_id, submitted_at, completed_at
	) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)`, operation.ID, operation.TenantID,
		operation.SiteID, request.Scope.PrincipalID, operation.Status, operation.TraceID,
		operation.SubmittedAt, operation.CompletedAt); err != nil {
		return ControlBatchOperation{}, errors.New("insert postgres Control API batch operation")
	}
	for index, result := range results {
		if _, err := tx.ExecContext(ctx, `INSERT INTO sense.batch_operation_items(
			operation_id, ordinal, device_id, status, error_code, message, generation
		) VALUES ($1,$2,$3,$4,$5,$6,$7)`, operation.ID, index, result.DeviceID,
			result.Status, result.ErrorCode, result.Message, result.Generation); err != nil {
			return ControlBatchOperation{}, errors.New("insert postgres Control API batch item")
		}
	}
	body, err := json.Marshal(operation)
	if err != nil {
		return ControlBatchOperation{}, errors.New("encode postgres Control API batch operation")
	}
	receipt = controlReceipt{
		Status: 202, Body: body,
		Location: "/api/v1/operations/" + url.PathEscape(operation.ID),
		TraceID:  operation.TraceID, CreatedAt: now,
	}
	if err := writeControlReceipt(ctx, tx, request.Scope, receipt); err != nil {
		return ControlBatchOperation{}, err
	}
	if err := tx.Commit(); err != nil {
		return ControlBatchOperation{}, errors.New("commit postgres Control API batch")
	}
	return operation, nil
}

func lockControlBatchDevices(
	ctx context.Context, tx *sql.Tx, tenantID, siteID string, deviceIDs map[string]int,
) error {
	ordered := make([]string, 0, len(deviceIDs))
	for deviceID := range deviceIDs {
		ordered = append(ordered, deviceID)
	}
	sort.Strings(ordered)
	for _, deviceID := range ordered {
		var locked string
		err := tx.QueryRowContext(ctx, `SELECT id FROM sense.devices
			WHERE tenant_id = $1 AND site_id = $2 AND id = $3 FOR UPDATE`,
			tenantID, siteID, deviceID).Scan(&locked)
		if err != nil && !errors.Is(err, sql.ErrNoRows) {
			return errors.New("lock postgres Control API batch devices")
		}
	}
	return nil
}

func controlBatchError(err error) (code, status, message string) {
	switch {
	case errors.Is(err, ErrNotFound):
		return "not_found", "rejected", "device was not found"
	case errors.Is(err, ErrETagMismatch):
		return "etag_mismatch", "rejected", "device ETag does not match"
	case errors.Is(err, ErrAreaPolicyDenied):
		return "area_policy_denied", "rejected", "Area policy denies this change"
	case errors.Is(err, ErrAreaPolicyUnavailable), errors.Is(err, ErrAreaPolicyInvalid):
		return "area_policy_unavailable", "failed", "Area policy is unavailable"
	case errors.Is(err, ErrQuotaProjectionUnavailable):
		return "quota_projection_unavailable", "failed", "Site quota is unavailable"
	case errors.Is(err, ErrQuotaProjectionInvalid):
		return "quota_projection_invalid", "failed", "Site quota is invalid"
	}
	var quotaError *device.QuotaExceededError
	if errors.As(err, &quotaError) {
		return "quota_exceeded", "rejected", "Site video channel quota is exceeded"
	}
	return "service_unavailable", "failed", "device change could not be accepted"
}

func (s *Postgres) GetControlOperation(
	ctx context.Context, tenantID, operationID string,
) (ControlBatchOperation, error) {
	var value ControlBatchOperation
	var completedAt sql.NullTime
	err := s.db.QueryRowContext(ctx, `SELECT id, tenant_id, site_id, status,
		submitted_at, completed_at, trace_id
		FROM sense.batch_operations WHERE tenant_id = $1 AND id = $2`, tenantID, operationID).
		Scan(&value.ID, &value.TenantID, &value.SiteID, &value.Status,
			&value.SubmittedAt, &completedAt, &value.TraceID)
	if errors.Is(err, sql.ErrNoRows) {
		return ControlBatchOperation{}, ErrNotFound
	}
	if err != nil {
		return ControlBatchOperation{}, errors.New("read postgres Control API batch operation")
	}
	if completedAt.Valid {
		point := completedAt.Time.UTC()
		value.CompletedAt = &point
	}
	rows, err := s.db.QueryContext(ctx, `SELECT device_id, status, error_code, message, generation
		FROM sense.batch_operation_items WHERE operation_id = $1 ORDER BY ordinal`, operationID)
	if err != nil {
		return ControlBatchOperation{}, errors.New("list postgres Control API batch results")
	}
	defer rows.Close()
	value.Results = make([]ControlBatchItemResult, 0)
	for rows.Next() {
		var item ControlBatchItemResult
		var code, message sql.NullString
		var generation sql.NullInt64
		if err := rows.Scan(&item.DeviceID, &item.Status, &code, &message, &generation); err != nil {
			return ControlBatchOperation{}, errors.New("scan postgres Control API batch result")
		}
		if code.Valid {
			item.ErrorCode = &code.String
		}
		if message.Valid {
			item.Message = &message.String
		}
		if generation.Valid {
			item.Generation = &generation.Int64
		}
		value.Results = append(value.Results, item)
	}
	if err := rows.Err(); err != nil {
		return ControlBatchOperation{}, errors.New("iterate postgres Control API batch results")
	}
	return value, nil
}

func newControlOperationID(now time.Time) (string, error) {
	value := make([]byte, 16)
	milliseconds := uint64(now.UTC().UnixMilli())
	value[0], value[1], value[2] = byte(milliseconds>>40), byte(milliseconds>>32), byte(milliseconds>>24)
	value[3], value[4], value[5] = byte(milliseconds>>16), byte(milliseconds>>8), byte(milliseconds)
	if _, err := rand.Read(value[6:]); err != nil {
		return "", errors.New("generate Control API operation ID")
	}
	number := new(big.Int).SetBytes(value)
	base, remainder := big.NewInt(32), new(big.Int)
	const alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
	encoded := make([]byte, 26)
	for index := len(encoded) - 1; index >= 0; index-- {
		number.QuoRem(number, base, remainder)
		encoded[index] = alphabet[remainder.Int64()]
	}
	return "op_" + string(encoded), nil
}