feat(store): add Area admission and audit outbox [T-010]

Sense/internal/store/postgres.go

@@ -16,7 +16,7 @@ import (
 )
 
 // Postgres persists Sense state in the sense schema and consumes only Bell's
-// versioned quota view. Migrations are deliberately installed out of process.
+// versioned quota and Area views. Migrations are installed out of process.
 type Postgres struct {
 	db *sql.DB
 }
@@ -53,20 +53,29 @@ func (s *Postgres) Close() error {
 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")
+		`SELECT MAX(version) FROM sense.schema_migrations`).Scan(&version); err != nil || !version.Valid || version.Int64 < 3 {
+		return errors.New("postgres sense schema migration v3 is required")
 	}
-	var canReadView, canWriteView, canReadSource, canWriteSource bool
+	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')`).
-		Scan(&canReadView, &canWriteView, &canReadSource, &canWriteSource); err != nil {
-		return errors.New("verify postgres quota privileges")
+		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 !canReadView || canWriteView || canReadSource || canWriteSource {
-		return errors.New("postgres role violates Bell quota privilege boundary")
+	if !canReadQuotaView || canWriteQuotaView || canReadSiteSource || canWriteSiteSource ||
+		!canReadAreaView || canWriteAreaView || canReadAreaSource || canWriteAreaSource {
+		return errors.New("postgres role violates Bell projection privilege boundary")
 	}
 	return nil
 }
@@ -92,7 +101,14 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error
 		return errors.New("begin postgres create device")
 	}
 	defer tx.Rollback()
-	var quotaVersion any
+	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 {
@@ -101,14 +117,15 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error
 		quotaVersion = version
 	}
 	_, err = tx.ExecContext(ctx, `INSERT INTO sense.devices(
-		id, tenant_id, site_id, serial_number, name, modality,
+		id, tenant_id, site_id, area_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,
+		path_name, generation, 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)`,
+		value.ID, value.TenantID, value.SiteID, value.AreaID, value.SerialNumber, value.Name,
 		value.Modality, value.DesiredState, value.ActualState, value.EndpointRef,
-		value.CredentialRef, value.PathName, value.Generation, quotaVersion,
-		value.CreatedAt, value.UpdatedAt)
+		value.CredentialRef, value.PathName, value.Generation, nullableVersion(quotaVersion),
+		areaVersion, value.CreatedAt, value.UpdatedAt)
 	if err != nil {
 		return errors.New("insert postgres device")
 	}
@@ -123,12 +140,83 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error
 		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,
@@ -199,10 +287,17 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic
 	}
 	defer tx.Rollback()
 	var tenantID, siteID, endpointRef, pathName string
+	var areaID sql.NullString
 	var current device.DesiredState
-	err = tx.QueryRowContext(ctx, `SELECT tenant_id, site_id, desired_state, endpoint_ref, path_name
+	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, &current, &endpointRef, &pathName)
+		Scan(
+			&tenantID, &siteID, &areaID, &current, &endpointRef, &pathName,
+			&generation, &storedQuotaVersion, &storedAreaVersion,
+		)
 	if errors.Is(err, sql.ErrNoRows) {
 		return ErrNotFound
 	}
@@ -210,9 +305,25 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic
 		return errors.New("read postgres device desired state")
 	}
 	if current == desired {
-		return tx.Commit()
+		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 any
+	var quotaVersion, areaVersion int64
 	if desired == device.DesiredEnabled {
 		var hasVideo bool
 		if err := tx.QueryRowContext(ctx, `SELECT EXISTS(
@@ -225,6 +336,13 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic
 			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
@@ -232,21 +350,41 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic
 			quotaVersion = version
 		}
 	}
-	result, err := tx.ExecContext(ctx, `UPDATE sense.devices SET
+	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,
-		quota_source_version = COALESCE($2, quota_source_version), updated_at = $3
-		WHERE id = $4`, desired, quotaVersion, time.Now().UTC(), id)
+		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 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 {
+		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")
 	}
@@ -288,18 +426,24 @@ func (s *Postgres) ListDueReconcile(ctx context.Context, now time.Time, limit in
 	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,
-			&candidate.Device.SerialNumber, &candidate.Device.Name, &candidate.Device.Modality,
+			&areaID, &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,
+			&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
@@ -483,20 +627,26 @@ func (s *Postgres) ConvergenceSnapshot(ctx context.Context) (ConvergenceSnapshot
 	return snapshot, nil
 }
 
-const postgresDeviceColumns = `d.id, d.tenant_id, d.site_id, d.serial_number, d.name, d.modality,
+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.path_name, d.generation, d.created_at, d.updated_at`
+		d.path_name, d.generation, 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, &value.SerialNumber,
+		&value.ID, &value.TenantID, &value.SiteID, &areaID, &value.SerialNumber,
 		&value.Name, &value.Modality, &value.DesiredState, &value.ActualState,
 		&value.EndpointRef, &value.CredentialRef, &value.PathName,
-		&value.Generation, &value.CreatedAt, &value.UpdatedAt,
+		&value.Generation, &quotaVersion, &areaVersion, &value.CreatedAt, &value.UpdatedAt,
 	)
+	value.AreaID = areaID.String
+	value.QuotaSourceVersion = quotaVersion.Int64
+	value.AreaPolicySourceVersion = areaVersion.Int64
 	return value, err
 }