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 view. Migrations are deliberately 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 < 1 {
		return errors.New("postgres sense schema migration v1 is required")
	}
	var canReadView, canWriteView, canReadSource, canWriteSource 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')`).
		Scan(&canReadView, &canWriteView, &canReadSource, &canWriteSource); err != nil {
		return errors.New("verify postgres quota privileges")
	}
	if !canReadView || canWriteView || canReadSource || canWriteSource {
		return errors.New("postgres role violates Bell quota 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 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()
	var quotaVersion any
	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, serial_number, name, modality,
		desired_state, actual_state, endpoint_ref, credential_ref,
		path_name, generation, quota_source_version, created_at, updated_at
	) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)`,
		value.ID, value.TenantID, value.SiteID, value.SerialNumber, value.Name,
		value.Modality, value.DesiredState, value.ActualState, value.EndpointRef,
		value.CredentialRef, value.PathName, value.Generation, quotaVersion,
		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 := tx.Commit(); err != nil {
		return errors.New("commit postgres create device")
	}
	return 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 current device.DesiredState
	err = tx.QueryRowContext(ctx, `SELECT tenant_id, site_id, desired_state, endpoint_ref, path_name
		FROM sense.devices WHERE id = $1 FOR UPDATE`, id).
		Scan(&tenantID, &siteID, &current, &endpointRef, &pathName)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrNotFound
	}
	if err != nil {
		return errors.New("read postgres device desired state")
	}
	if current == desired {
		return tx.Commit()
	}
	var quotaVersion any
	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, quotaErr := checkPostgresVideoQuota(ctx, tx, tenantID, siteID, time.Now().UTC())
			if quotaErr != nil {
				return quotaErr
			}
			quotaVersion = version
		}
	}
	result, err := tx.ExecContext(ctx, `UPDATE sense.devices SET
		desired_state = $1, actual_state = 'pending', generation = generation + 1,
		quota_source_version = COALESCE($2, quota_source_version), updated_at = $3
		WHERE id = $4`, desired, quotaVersion, time.Now().UTC(), id)
	if err != nil {
		return errors.New("update postgres desired state")
	}
	if affected, _ := result.RowsAffected(); affected != 1 {
		return ErrNotFound
	}
	if _, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
		failure_count = 0, next_attempt_at = NULL, last_error_code = NULL, updated_at = $1
		WHERE device_id = $2`, time.Now().UTC(), id); err != nil {
		return errors.New("reset postgres reconcile state")
	}
	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 d.desired_state = 'enabled'
			AND 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 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 nextAttempt sql.NullTime
		if err := rows.Scan(
			&candidate.Device.ID, &candidate.Device.TenantID, &candidate.Device.SiteID,
			&candidate.Device.SerialNumber, &candidate.Device.Name, &candidate.Device.Modality,
			&candidate.Device.DesiredState, &candidate.Device.ActualState,
			&candidate.Device.EndpointRef, &candidate.Device.CredentialRef,
			&candidate.Device.PathName, &candidate.Device.Generation,
			&candidate.Device.CreatedAt, &candidate.Device.UpdatedAt,
			&candidate.FailureCount, &nextAttempt,
		); err != nil {
			return nil, errors.New("scan postgres due reconcile device")
		}
		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) 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 = 'pending', 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.serial_number, d.name, d.modality,
		d.desired_state, d.actual_state, d.endpoint_ref, d.credential_ref,
		d.path_name, d.generation, 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
	err := row.Scan(
		&value.ID, &value.TenantID, &value.SiteID, &value.SerialNumber,
		&value.Name, &value.Modality, &value.DesiredState, &value.ActualState,
		&value.EndpointRef, &value.CredentialRef, &value.PathName,
		&value.Generation, &value.CreatedAt, &value.UpdatedAt,
	)
	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
}