yovision/Sense/internal/store/postgres.go

package store

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

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

	"yovision/sense/internal/device"
)

// Postgres persists Sense state in the sense schema and consumes only Bell's
// versioned quota and Area views. Migrations are installed out of process.
type Postgres struct {
	db *sql.DB
}

func OpenPostgres(ctx context.Context, dsn string) (*Postgres, error) {
	configuration, err := pgx.ParseConfig(dsn)
	if err != nil {
		return nil, errors.New("invalid postgres DSN")
	}
	if configuration.RuntimeParams == nil {
		configuration.RuntimeParams = make(map[string]string)
	}
	configuration.RuntimeParams["application_name"] = "yovision-sense"
	db := stdlib.OpenDB(*configuration)
	db.SetMaxOpenConns(16)
	db.SetMaxIdleConns(4)
	db.SetConnMaxLifetime(30 * time.Minute)
	if err := db.PingContext(ctx); err != nil {
		db.Close()
		return nil, errors.New("connect postgres database")
	}
	store := &Postgres{db: db}
	if err := store.verifySchemaAndPrivileges(ctx); err != nil {
		db.Close()
		return nil, err
	}
	return store, nil
}

func (s *Postgres) Close() error {
	return s.db.Close()
}

func (s *Postgres) verifySchemaAndPrivileges(ctx context.Context) error {
	var version sql.NullInt64
	if err := s.db.QueryRowContext(ctx,
		`SELECT MAX(version) FROM sense.schema_migrations`).Scan(&version); err != nil || !version.Valid || version.Int64 < 5 {
		return errors.New("postgres sense schema migration v5 is required")
	}
	var canReadQuotaView, canWriteQuotaView, canReadSiteSource, canWriteSiteSource bool
	var canReadAreaView, canWriteAreaView, canReadAreaSource, canWriteAreaSource bool
	if err := s.db.QueryRowContext(ctx, `SELECT
		has_table_privilege(current_user, 'bell.site_quota_v1', 'SELECT'),
		has_table_privilege(current_user, 'bell.site_quota_v1', 'INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'bell.sites', 'SELECT'),
		has_table_privilege(current_user, 'bell.sites', 'INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'bell.area_policy_v1', 'SELECT'),
		has_table_privilege(current_user, 'bell.area_policy_v1', 'INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'bell.areas', 'SELECT'),
		has_table_privilege(current_user, 'bell.areas', 'INSERT,UPDATE,DELETE')`).
		Scan(
			&canReadQuotaView, &canWriteQuotaView, &canReadSiteSource, &canWriteSiteSource,
			&canReadAreaView, &canWriteAreaView, &canReadAreaSource, &canWriteAreaSource,
		); err != nil {
		return errors.New("verify postgres Bell projection privileges")
	}
	if !canReadQuotaView || canWriteQuotaView || canReadSiteSource || canWriteSiteSource ||
		!canReadAreaView || canWriteAreaView || canReadAreaSource || canWriteAreaSource {
		return errors.New("postgres role violates Bell projection privilege boundary")
	}
	var canUseReceipts, canUseOperations, canUseOperationItems bool
	var publicReceipts, publicOperations, publicOperationItems bool
	if err := s.db.QueryRowContext(ctx, `SELECT
		has_table_privilege(current_user, 'sense.control_idempotency_receipts', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'sense.batch_operations', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'sense.batch_operation_items', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege('public', 'sense.control_idempotency_receipts', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege('public', 'sense.batch_operations', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege('public', 'sense.batch_operation_items', 'SELECT,INSERT,UPDATE,DELETE')`).Scan(
		&canUseReceipts, &canUseOperations, &canUseOperationItems,
		&publicReceipts, &publicOperations, &publicOperationItems,
	); err != nil {
		return errors.New("verify postgres Control API state privileges")
	}
	if !canUseReceipts || !canUseOperations || !canUseOperationItems ||
		publicReceipts || publicOperations || publicOperationItems {
		return errors.New("postgres role violates Control API state privilege boundary")
	}
	var canUseOwnership, canUseLeases, canUseScans, canUseFindings, canUseActions bool
	var publicOwnership, publicLeases, publicScans, publicFindings, publicActions bool
	if err := s.db.QueryRowContext(ctx, `SELECT
		has_table_privilege(current_user, 'sense.media_path_ownership', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'sense.operational_leases', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'sense.orphan_scan_runs', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'sense.orphan_scan_findings', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege(current_user, 'sense.orphan_cleanup_actions', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege('public', 'sense.media_path_ownership', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege('public', 'sense.operational_leases', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege('public', 'sense.orphan_scan_runs', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege('public', 'sense.orphan_scan_findings', 'SELECT,INSERT,UPDATE,DELETE'),
		has_table_privilege('public', 'sense.orphan_cleanup_actions', 'SELECT,INSERT,UPDATE,DELETE')`).Scan(
		&canUseOwnership, &canUseLeases, &canUseScans, &canUseFindings, &canUseActions,
		&publicOwnership, &publicLeases, &publicScans, &publicFindings, &publicActions,
	); err != nil {
		return errors.New("verify postgres reconciliation safety privileges")
	}
	if !canUseOwnership || !canUseLeases || !canUseScans || !canUseFindings || !canUseActions ||
		publicOwnership || publicLeases || publicScans || publicFindings || publicActions {
		return errors.New("postgres role violates reconciliation safety privilege boundary")
	}
	return nil
}

func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error {
	if value.Generation == 0 {
		value.Generation = 1
	}
	if value.ActualState == "" {
		value.ActualState = device.ActualPending
	}
	if value.ResourceVersion == 0 {
		value.ResourceVersion = 1
	}
	if err := value.Validate(); err != nil {
		return fmt.Errorf("validate device: %w", err)
	}
	now := time.Now().UTC()
	if value.CreatedAt.IsZero() {
		value.CreatedAt = now
	}
	value.UpdatedAt = now

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin postgres create device")
	}
	defer tx.Rollback()
	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() {
		version, quotaErr := checkPostgresVideoQuota(ctx, tx, value.TenantID, value.SiteID, now)
		if quotaErr != nil {
			return quotaErr
		}
		quotaVersion = version
	}
	_, 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 {
		return errors.New("insert postgres 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 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 reconcile state")
	}
	if err := 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,
		},
	}); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return errors.New("commit postgres create device")
	}
	return nil
}

func checkPostgresAreaPolicy(
	ctx context.Context,
	tx *sql.Tx,
	tenantID, siteID, areaID string,
	imaging bool,
	now time.Time,
) (int64, error) {
	if strings.TrimSpace(areaID) == "" {
		return 0, areaPolicyUnavailable()
	}
	// Area projection observation is serialized before the Site quota lock.
	// No admission path acquires these locks in the opposite order.
	if _, err := tx.ExecContext(ctx,
		`SELECT pg_advisory_xact_lock(hashtext($1), hashtext('area:' || $2))`, tenantID, areaID); err != nil {
		return 0, errors.New("lock postgres Area admission")
	}
	var capturePolicy string
	var sourceVersion int64
	var sourceUpdatedAt time.Time
	err := tx.QueryRowContext(ctx, `SELECT capture_policy, source_version, source_updated_at
		FROM bell.area_policy_v1
		WHERE tenant_id = $1 AND site_id = $2 AND area_id = $3`, tenantID, siteID, areaID).
		Scan(&capturePolicy, &sourceVersion, &sourceUpdatedAt)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, areaPolicyUnavailable()
	}
	if err != nil {
		return 0, areaPolicyUnavailable()
	}
	if (capturePolicy != "video_allowed" && capturePolicy != "non_imaging_only") ||
		sourceVersion < 1 || sourceUpdatedAt.IsZero() {
		return 0, areaPolicyInvalid()
	}
	var previous sql.NullInt64
	err = tx.QueryRowContext(ctx, `SELECT source_version
		FROM sense.area_policy_projection_state
		WHERE tenant_id = $1 AND site_id = $2 AND area_id = $3`, tenantID, siteID, areaID).
		Scan(&previous)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return 0, errors.New("read postgres Area projection state")
	}
	if previous.Valid && sourceVersion < previous.Int64 {
		return 0, areaPolicyInvalid()
	}
	if imaging && capturePolicy == "non_imaging_only" {
		return 0, areaPolicyDenied()
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO sense.area_policy_projection_state(
		tenant_id, site_id, area_id, source_version, synced_at
	) VALUES ($1, $2, $3, $4, $5)
	ON CONFLICT (tenant_id, site_id, area_id) DO UPDATE SET
		source_version = EXCLUDED.source_version,
		synced_at = EXCLUDED.synced_at`, tenantID, siteID, areaID, sourceVersion, now); err != nil {
		return 0, errors.New("record postgres Area projection state")
	}
	return sourceVersion, nil
}

func checkPostgresVideoQuota(
	ctx context.Context,
	tx *sql.Tx,
	tenantID, siteID string,
	now time.Time,
) (int64, error) {
	// A transaction-scoped lock shared by all Sense instances makes count +
	// write atomic per logical site without locking Bell-owned rows.
	if _, err := tx.ExecContext(ctx,
		`SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))`, tenantID, siteID); err != nil {
		return 0, errors.New("lock postgres site quota admission")
	}
	var limit int
	var sourceVersion int64
	var sourceUpdatedAt time.Time
	err := tx.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(&limit, &sourceVersion, &sourceUpdatedAt)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, projectionUnavailable()
	}
	if err != nil {
		return 0, projectionUnavailable()
	}
	if limit < 1 || limit > device.MaximumVideoChannels || sourceVersion < 1 || sourceUpdatedAt.IsZero() {
		return 0, projectionInvalid()
	}
	var previous sql.NullInt64
	err = tx.QueryRowContext(ctx, `SELECT source_version
		FROM sense.site_quota_projection_state WHERE tenant_id = $1 AND site_id = $2`,
		tenantID, siteID).Scan(&previous)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return 0, errors.New("read postgres quota projection state")
	}
	if previous.Valid && sourceVersion < previous.Int64 {
		return 0, projectionInvalid()
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO sense.site_quota_projection_state(
		tenant_id, site_id, source_version, synced_at
	) VALUES ($1, $2, $3, $4)
	ON CONFLICT (tenant_id, site_id) DO UPDATE SET
		source_version = EXCLUDED.source_version,
		synced_at = EXCLUDED.synced_at`, tenantID, siteID, sourceVersion, now); err != nil {
		return 0, errors.New("record postgres quota projection state")
	}
	var current int
	if err := tx.QueryRowContext(ctx, `SELECT COUNT(*)
		FROM sense.devices d
		JOIN sense.device_capabilities c ON c.device_id = d.id
		WHERE d.tenant_id = $1 AND d.site_id = $2
			AND d.desired_state = 'enabled'
			AND c.capability = 'video_capture'`, tenantID, siteID).Scan(&current); err != nil {
		return 0, errors.New("count postgres site video channels")
	}
	if current >= limit {
		return 0, &device.QuotaExceededError{TenantID: tenantID, SiteID: siteID, Limit: limit}
	}
	return sourceVersion, nil
}

func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired device.DesiredState) error {
	if desired != device.DesiredEnabled && desired != device.DesiredDisabled {
		return fmt.Errorf("invalid desired state %q", desired)
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin postgres desired-state update")
	}
	defer tx.Rollback()
	var tenantID, siteID, endpointRef, pathName string
	var areaID sql.NullString
	var current device.DesiredState
	var generation int64
	var storedQuotaVersion, storedAreaVersion sql.NullInt64
	err = tx.QueryRowContext(ctx, `SELECT tenant_id, site_id, area_id, desired_state,
		endpoint_ref, path_name, generation, quota_source_version, area_policy_source_version
		FROM sense.devices WHERE id = $1 FOR UPDATE`, id).
		Scan(
			&tenantID, &siteID, &areaID, &current, &endpointRef, &pathName,
			&generation, &storedQuotaVersion, &storedAreaVersion,
		)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrNotFound
	}
	if err != nil {
		return errors.New("read postgres device desired state")
	}
	if current == desired {
		if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{
			EventType: "device.desired_state.accepted", TenantID: tenantID, SiteID: siteID,
			DeviceID: id, Generation: generation,
			QuotaSourceVersion:      storedQuotaVersion.Int64,
			AreaPolicySourceVersion: storedAreaVersion.Int64,
			OccurredAt:              time.Now().UTC(),
			Payload: map[string]any{
				"kind": "desired_state_accepted", "previous_desired_state": current,
				"desired_state": desired, "changed": false,
			},
		}); err != nil {
			return err
		}
		if err := tx.Commit(); err != nil {
			return errors.New("commit postgres no-op desired-state audit")
		}
		return nil
	}
	var quotaVersion, areaVersion int64
	if desired == device.DesiredEnabled {
		var hasVideo bool
		if err := tx.QueryRowContext(ctx, `SELECT EXISTS(
			SELECT 1 FROM sense.device_capabilities
			WHERE device_id = $1 AND capability = 'video_capture'
		)`, id).Scan(&hasVideo); err != nil {
			return errors.New("read postgres video capability")
		}
		if hasVideo {
			if strings.TrimSpace(endpointRef) == "" || strings.TrimSpace(pathName) == "" {
				return errors.New("enabled video devices require endpoint ref and path name")
			}
			version, areaErr := checkPostgresAreaPolicy(
				ctx, tx, tenantID, siteID, areaID.String, true, time.Now().UTC(),
			)
			if areaErr != nil {
				return areaErr
			}
			areaVersion = version
			version, quotaErr := checkPostgresVideoQuota(ctx, tx, tenantID, siteID, time.Now().UTC())
			if quotaErr != nil {
				return quotaErr
			}
			quotaVersion = version
		}
	}
	now := time.Now().UTC()
	var updatedQuotaVersion, updatedAreaVersion sql.NullInt64
	err = tx.QueryRowContext(ctx, `UPDATE sense.devices SET
		desired_state = $1, actual_state = 'pending', generation = generation + 1,
		resource_version = resource_version + 1,
		quota_source_version = COALESCE($2, quota_source_version),
		area_policy_source_version = COALESCE($3, area_policy_source_version),
		updated_at = $4
		WHERE id = $5
		RETURNING generation, quota_source_version, area_policy_source_version`,
		desired, nullableVersion(quotaVersion), nullableVersion(areaVersion), now, id).
		Scan(&generation, &updatedQuotaVersion, &updatedAreaVersion)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrNotFound
	}
	if err != nil {
		return errors.New("update postgres 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, id); err != nil {
		return errors.New("reset postgres reconcile state")
	}
	if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{
		EventType: "device.desired_state.accepted", TenantID: tenantID, SiteID: siteID,
		DeviceID: id, Generation: generation,
		QuotaSourceVersion:      updatedQuotaVersion.Int64,
		AreaPolicySourceVersion: updatedAreaVersion.Int64,
		OccurredAt:              now,
		Payload: map[string]any{
			"kind": "desired_state_accepted", "previous_desired_state": current,
			"desired_state": desired, "changed": true,
		},
	}); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return errors.New("commit postgres desired-state update")
	}
	return nil
}

func (s *Postgres) GetDevice(ctx context.Context, id string) (device.Device, error) {
	value, err := scanPostgresDevice(s.db.QueryRowContext(ctx, postgresDeviceSelect+` WHERE d.id = $1`, id))
	if errors.Is(err, sql.ErrNoRows) {
		return device.Device{}, ErrNotFound
	}
	if err != nil {
		return device.Device{}, errors.New("get postgres device")
	}
	value.Capabilities, err = s.capabilities(ctx, value.ID)
	if err != nil {
		return device.Device{}, err
	}
	return value, nil
}

func (s *Postgres) ListDueReconcile(ctx context.Context, now time.Time, limit int) ([]ReconcileCandidate, error) {
	if limit <= 0 {
		return nil, nil
	}
	rows, err := s.db.QueryContext(ctx, `SELECT `+postgresDeviceColumns+`, r.failure_count, r.next_attempt_at
		FROM sense.devices d
		JOIN sense.reconcile_state r ON r.device_id = d.id
		WHERE EXISTS (SELECT 1 FROM sense.device_capabilities c
				WHERE c.device_id = d.id AND c.capability = 'video_capture')
			AND (r.observed_generation < d.generation
				OR (d.desired_state = 'enabled' AND r.failure_count > 0))
			AND (r.next_attempt_at IS NULL OR r.next_attempt_at <= $1)
		ORDER BY d.updated_at, d.id LIMIT $2`, now, limit)
	if err != nil {
		return nil, errors.New("list postgres due reconcile devices")
	}
	defer rows.Close()
	values := make([]ReconcileCandidate, 0)
	for rows.Next() {
		var candidate ReconcileCandidate
		var areaID sql.NullString
		var quotaVersion, areaVersion sql.NullInt64
		var nextAttempt sql.NullTime
		if err := rows.Scan(
			&candidate.Device.ID, &candidate.Device.TenantID, &candidate.Device.SiteID,
			&areaID, &candidate.Device.SerialNumber, &candidate.Device.Name, &candidate.Device.Modality,
			&candidate.Device.DesiredState, &candidate.Device.ActualState,
			&candidate.Device.EndpointRef, &candidate.Device.CredentialRef,
			&candidate.Device.ProfileToken, &candidate.Device.PathName,
			&candidate.Device.Generation, &candidate.Device.ResourceVersion,
			&quotaVersion, &areaVersion,
			&candidate.Device.CreatedAt, &candidate.Device.UpdatedAt,
			&candidate.FailureCount, &nextAttempt,
		); err != nil {
			return nil, errors.New("scan postgres due reconcile device")
		}
		candidate.Device.AreaID = areaID.String
		candidate.Device.QuotaSourceVersion = quotaVersion.Int64
		candidate.Device.AreaPolicySourceVersion = areaVersion.Int64
		if nextAttempt.Valid {
			value := nextAttempt.Time
			candidate.NextAttempt = &value
		}
		values = append(values, candidate)
	}
	if err := rows.Err(); err != nil {
		return nil, errors.New("iterate postgres due reconcile devices")
	}
	for index := range values {
		values[index].Device.Capabilities, err = s.capabilities(ctx, values[index].Device.ID)
		if err != nil {
			return nil, err
		}
	}
	return values, nil
}

func (s *Postgres) ClaimDueReconcile(
	ctx context.Context,
	claim ReconcileClaim,
) ([]ReconcileCandidate, error) {
	if claim.Limit <= 0 {
		return nil, nil
	}
	if strings.TrimSpace(claim.Owner) == "" || strings.TrimSpace(claim.Token) == "" ||
		claim.LeaseDuration <= 0 {
		return nil, errors.New("invalid postgres reconcile claim")
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, errors.New("begin postgres reconcile claim")
	}
	defer tx.Rollback()
	rows, err := tx.QueryContext(ctx, `SELECT `+postgresDeviceColumns+`, r.failure_count, r.next_attempt_at
		FROM sense.devices d
		JOIN sense.reconcile_state r ON r.device_id = d.id
		WHERE EXISTS (SELECT 1 FROM sense.device_capabilities c
				WHERE c.device_id = d.id AND c.capability = 'video_capture')
			AND (r.observed_generation < d.generation
				OR (d.desired_state = 'enabled' AND r.failure_count > 0))
			AND (r.next_attempt_at IS NULL OR r.next_attempt_at <= clock_timestamp())
			AND (r.lease_until IS NULL OR r.lease_until <= clock_timestamp())
		ORDER BY d.updated_at, d.id
		FOR UPDATE OF r SKIP LOCKED
		LIMIT $1`, claim.Limit)
	if err != nil {
		return nil, errors.New("select postgres reconcile claims")
	}
	values := make([]ReconcileCandidate, 0)
	for rows.Next() {
		var candidate ReconcileCandidate
		var areaID sql.NullString
		var quotaVersion, areaVersion sql.NullInt64
		var nextAttempt sql.NullTime
		if err := rows.Scan(
			&candidate.Device.ID, &candidate.Device.TenantID, &candidate.Device.SiteID,
			&areaID, &candidate.Device.SerialNumber, &candidate.Device.Name, &candidate.Device.Modality,
			&candidate.Device.DesiredState, &candidate.Device.ActualState,
			&candidate.Device.EndpointRef, &candidate.Device.CredentialRef,
			&candidate.Device.ProfileToken, &candidate.Device.PathName,
			&candidate.Device.Generation, &candidate.Device.ResourceVersion,
			&quotaVersion, &areaVersion,
			&candidate.Device.CreatedAt, &candidate.Device.UpdatedAt,
			&candidate.FailureCount, &nextAttempt,
		); err != nil {
			rows.Close()
			return nil, errors.New("scan postgres reconcile claim")
		}
		candidate.Device.AreaID = areaID.String
		candidate.Device.QuotaSourceVersion = quotaVersion.Int64
		candidate.Device.AreaPolicySourceVersion = areaVersion.Int64
		if nextAttempt.Valid {
			value := nextAttempt.Time
			candidate.NextAttempt = &value
		}
		values = append(values, candidate)
	}
	if err := rows.Close(); err != nil {
		return nil, errors.New("close postgres reconcile claims")
	}
	if err := rows.Err(); err != nil {
		return nil, errors.New("iterate postgres reconcile claims")
	}
	for _, candidate := range values {
		if _, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
			lease_owner = $1, lease_token = $2,
			lease_until = clock_timestamp() + ($3 * interval '1 second'),
			updated_at = clock_timestamp()
			WHERE device_id = $4`,
			claim.Owner, claim.Token, claim.LeaseDuration.Seconds(), candidate.Device.ID,
		); err != nil {
			return nil, errors.New("persist postgres reconcile claim")
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, errors.New("commit postgres reconcile claim")
	}
	for index := range values {
		values[index].Device.Capabilities, err = s.capabilities(ctx, values[index].Device.ID)
		if err != nil {
			return nil, err
		}
	}
	return values, nil
}

func (s *Postgres) RenewReconcileLease(
	ctx context.Context,
	id, owner, token string,
	now time.Time,
	duration time.Duration,
) (bool, error) {
	if strings.TrimSpace(owner) == "" || strings.TrimSpace(token) == "" || duration <= 0 {
		return false, errors.New("invalid postgres reconcile lease renewal")
	}
	result, err := s.db.ExecContext(ctx, `UPDATE sense.reconcile_state SET
		lease_until = clock_timestamp() + ($1 * interval '1 second'),
		updated_at = clock_timestamp()
		WHERE device_id = $2 AND lease_owner = $3 AND lease_token = $4
			AND lease_until > clock_timestamp()`, duration.Seconds(), id, owner, token)
	if err != nil {
		return false, errors.New("renew postgres reconcile lease")
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return false, errors.New("read postgres reconcile lease renewal")
	}
	return affected == 1, nil
}

func (s *Postgres) CompleteReconcile(
	ctx context.Context,
	id string,
	generation int64,
	owner, token string,
	now time.Time,
) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin postgres fenced reconciliation completion")
	}
	defer tx.Rollback()
	result, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
		failure_count = 0, next_attempt_at = NULL, last_error_code = NULL,
		observed_generation = $1, lease_owner = NULL, lease_token = NULL,
		lease_until = NULL, updated_at = $2
		WHERE device_id = $3 AND lease_owner = $4 AND lease_token = $5
			AND lease_until > clock_timestamp()`, generation, now, id, owner, token)
	if err != nil {
		return errors.New("complete postgres fenced reconciliation")
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return errors.New("read postgres fenced reconciliation completion")
	}
	if affected != 1 {
		return ErrReconcileLeaseLost
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO sense.media_path_ownership(
		path_name, device_id, tenant_id, site_id, first_claimed_at, last_confirmed_at
	)
	SELECT path_name, id, tenant_id, site_id, $1, $1
	FROM sense.devices
	WHERE id = $2 AND desired_state = 'enabled' AND btrim(path_name) <> ''
	ON CONFLICT (path_name) DO UPDATE SET
		device_id = EXCLUDED.device_id,
		tenant_id = EXCLUDED.tenant_id,
		site_id = EXCLUDED.site_id,
		last_confirmed_at = EXCLUDED.last_confirmed_at`, now, id); err != nil {
		return errors.New("record postgres MediaMTX path ownership")
	}
	result, err = tx.ExecContext(ctx, `UPDATE sense.devices
		SET actual_state = CASE WHEN desired_state = 'disabled' THEN 'offline' ELSE 'pending' END,
		updated_at = $1 WHERE id = $2`, now, id)
	if err != nil {
		return errors.New("mark postgres fenced device state")
	}
	if affected, _ := result.RowsAffected(); affected != 1 {
		return ErrNotFound
	}
	if err := tx.Commit(); err != nil {
		return errors.New("commit postgres fenced reconciliation completion")
	}
	return nil
}

func (s *Postgres) FailReconcile(
	ctx context.Context,
	id string,
	failureCount int,
	nextAttempt time.Time,
	errorCode, owner, token string,
	now time.Time,
) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin postgres fenced reconciliation failure")
	}
	defer tx.Rollback()
	result, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
		failure_count = $1, next_attempt_at = $2, last_error_code = $3,
		lease_owner = NULL, lease_token = NULL, lease_until = NULL, updated_at = $4
		WHERE device_id = $5 AND lease_owner = $6 AND lease_token = $7
			AND lease_until > clock_timestamp()`, failureCount, nextAttempt, errorCode, now, id, owner, token)
	if err != nil {
		return errors.New("persist postgres fenced reconciliation failure")
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return errors.New("read postgres fenced reconciliation failure")
	}
	if affected != 1 {
		return ErrReconcileLeaseLost
	}
	result, err = tx.ExecContext(ctx, `UPDATE sense.devices
		SET actual_state = 'failed', updated_at = $1 WHERE id = $2`, now, id)
	if err != nil {
		return errors.New("mark postgres fenced failed device")
	}
	if affected, _ := result.RowsAffected(); affected != 1 {
		return ErrNotFound
	}
	if err := tx.Commit(); err != nil {
		return errors.New("commit postgres fenced reconciliation failure")
	}
	return nil
}

func (s *Postgres) ListEnabledVideoDevices(ctx context.Context, limit int) ([]device.Device, error) {
	if limit <= 0 {
		return nil, nil
	}
	rows, err := s.db.QueryContext(ctx, postgresDeviceSelect+`
		WHERE d.desired_state = 'enabled'
			AND EXISTS (SELECT 1 FROM sense.device_capabilities c
				WHERE c.device_id = d.id AND c.capability = 'video_capture')
		ORDER BY d.id LIMIT $1`, limit)
	if err != nil {
		return nil, errors.New("list postgres enabled video devices")
	}
	defer rows.Close()
	values := make([]device.Device, 0)
	for rows.Next() {
		value, scanErr := scanPostgresDevice(rows)
		if scanErr != nil {
			return nil, errors.New("scan postgres enabled video device")
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, errors.New("iterate postgres enabled video devices")
	}
	for index := range values {
		values[index].Capabilities, err = s.capabilities(ctx, values[index].ID)
		if err != nil {
			return nil, err
		}
	}
	return values, nil
}

func (s *Postgres) MarkReconciled(ctx context.Context, id string, generation int64, now time.Time) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin postgres reconciled update")
	}
	defer tx.Rollback()
	result, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
		failure_count = 0, next_attempt_at = NULL, last_error_code = NULL,
		observed_generation = $1, updated_at = $2 WHERE device_id = $3`, generation, now, id)
	if err != nil {
		return errors.New("mark postgres device reconciled")
	}
	if affected, _ := result.RowsAffected(); affected != 1 {
		return ErrNotFound
	}
	if _, err := tx.ExecContext(ctx, `UPDATE sense.devices
		SET actual_state = CASE WHEN desired_state = 'disabled' THEN 'offline' ELSE 'pending' END,
		updated_at = $1 WHERE id = $2`, now, id); err != nil {
		return errors.New("mark postgres reconciled device pending")
	}
	if err := tx.Commit(); err != nil {
		return errors.New("commit postgres reconciled update")
	}
	return nil
}

func (s *Postgres) MarkReconcileFailure(
	ctx context.Context,
	id string,
	failureCount int,
	nextAttempt time.Time,
	errorCode string,
	now time.Time,
) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin postgres reconcile failure update")
	}
	defer tx.Rollback()
	result, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
		failure_count = $1, next_attempt_at = $2, last_error_code = $3, updated_at = $4
		WHERE device_id = $5`, failureCount, nextAttempt, errorCode, now, id)
	if err != nil {
		return errors.New("mark postgres reconcile failure")
	}
	if affected, _ := result.RowsAffected(); affected != 1 {
		return ErrNotFound
	}
	if _, err := tx.ExecContext(ctx, `UPDATE sense.devices
		SET actual_state = 'failed', updated_at = $1 WHERE id = $2`, now, id); err != nil {
		return errors.New("mark postgres failed device state")
	}
	if err := tx.Commit(); err != nil {
		return errors.New("commit postgres reconcile failure")
	}
	return nil
}

func (s *Postgres) UpdateActualState(ctx context.Context, id string, state device.ActualState, now time.Time) error {
	if state != device.ActualPending && state != device.ActualOnline &&
		state != device.ActualOffline && state != device.ActualFailed {
		return fmt.Errorf("invalid actual state %q", state)
	}
	result, err := s.db.ExecContext(ctx, `UPDATE sense.devices
		SET actual_state = $1, updated_at = $2 WHERE id = $3`, state, now, id)
	if err != nil {
		return errors.New("update postgres actual state")
	}
	if affected, _ := result.RowsAffected(); affected != 1 {
		return ErrNotFound
	}
	return nil
}

func (s *Postgres) RequestReconcile(ctx context.Context, id string, now time.Time) error {
	result, err := s.db.ExecContext(ctx, `UPDATE sense.reconcile_state
		SET observed_generation = 0, updated_at = $1 WHERE device_id = $2`, now, id)
	if err != nil {
		return errors.New("request postgres device reconciliation")
	}
	if affected, _ := result.RowsAffected(); affected != 1 {
		return ErrNotFound
	}
	return nil
}

func (s *Postgres) ConvergenceSnapshot(ctx context.Context) (ConvergenceSnapshot, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT
		d.id, d.path_name, d.desired_state, d.actual_state, d.generation,
		r.observed_generation, r.failure_count, r.next_attempt_at, r.last_error_code
		FROM sense.devices d
		JOIN sense.reconcile_state r ON r.device_id = d.id
		WHERE d.desired_state = 'enabled'
			AND EXISTS (SELECT 1 FROM sense.device_capabilities c
				WHERE c.device_id = d.id AND c.capability = 'video_capture')
		ORDER BY d.id`)
	if err != nil {
		return ConvergenceSnapshot{}, errors.New("query postgres convergence snapshot")
	}
	defer rows.Close()
	snapshot := ConvergenceSnapshot{Devices: make([]DeviceConvergence, 0)}
	for rows.Next() {
		var value DeviceConvergence
		var nextAttempt sql.NullTime
		var lastError sql.NullString
		if err := rows.Scan(
			&value.ID, &value.PathName, &value.DesiredState, &value.ActualState,
			&value.Generation, &value.ObservedGeneration, &value.FailureCount,
			&nextAttempt, &lastError,
		); err != nil {
			return ConvergenceSnapshot{}, errors.New("scan postgres convergence snapshot")
		}
		if nextAttempt.Valid {
			point := nextAttempt.Time
			value.NextAttemptAt = &point
		}
		if lastError.Valid {
			value.LastErrorCode = lastError.String
		}
		value.Converged = value.ObservedGeneration == value.Generation &&
			value.FailureCount == 0 && value.ActualState == device.ActualOnline
		if !value.Converged {
			snapshot.Unconverged++
		}
		snapshot.Devices = append(snapshot.Devices, value)
	}
	if err := rows.Err(); err != nil {
		return ConvergenceSnapshot{}, errors.New("iterate postgres convergence snapshot")
	}
	snapshot.Total = len(snapshot.Devices)
	return snapshot, nil
}

const postgresDeviceColumns = `d.id, d.tenant_id, d.site_id, d.area_id, d.serial_number, d.name, d.modality,
		d.desired_state, d.actual_state, d.endpoint_ref, d.credential_ref, d.profile_token,
		d.path_name, d.generation, d.resource_version, d.quota_source_version, d.area_policy_source_version,
		d.created_at, d.updated_at`

const postgresDeviceSelect = `SELECT ` + postgresDeviceColumns + ` FROM sense.devices d`

func scanPostgresDevice(row scanner) (device.Device, error) {
	var value device.Device
	var areaID sql.NullString
	var quotaVersion, areaVersion sql.NullInt64
	err := row.Scan(
		&value.ID, &value.TenantID, &value.SiteID, &areaID, &value.SerialNumber,
		&value.Name, &value.Modality, &value.DesiredState, &value.ActualState,
		&value.EndpointRef, &value.CredentialRef, &value.ProfileToken, &value.PathName,
		&value.Generation, &value.ResourceVersion, &quotaVersion, &areaVersion,
		&value.CreatedAt, &value.UpdatedAt,
	)
	value.AreaID = areaID.String
	value.QuotaSourceVersion = quotaVersion.Int64
	value.AreaPolicySourceVersion = areaVersion.Int64
	return value, err
}

func (s *Postgres) capabilities(ctx context.Context, id string) ([]device.Capability, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT capability
		FROM sense.device_capabilities WHERE device_id = $1 ORDER BY capability`, id)
	if err != nil {
		return nil, errors.New("list postgres device capabilities")
	}
	defer rows.Close()
	values := make([]device.Capability, 0)
	for rows.Next() {
		var value device.Capability
		if err := rows.Scan(&value); err != nil {
			return nil, errors.New("scan postgres device capability")
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, errors.New("iterate postgres device capabilities")
	}
	sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
	return values, nil
}