From 2ff7ff5618da1441f43c2d17e0c33ffa2b6c4a32 Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Fri, 7 Aug 2026 17:52:50 +0800 Subject: [PATCH] feat(store): add PostgreSQL foundation [T-009] --- Sense/README.md | 31 +- Sense/cmd/sense-api/main.go | 2 +- Sense/go.mod | 4 + Sense/go.sum | 9 + Sense/internal/config/config.go | 13 + Sense/internal/config/config_test.go | 23 ++ Sense/internal/store/postgres.go | 523 ++++++++++++++++++++++++++ Sense/internal/store/postgres_test.go | 353 +++++++++++++++++ Sense/internal/store/repository.go | 63 ++++ deploy/postgres/001_roles.sql | 24 ++ deploy/postgres/002_bell.sql | 73 ++++ deploy/postgres/003_sense.sql | 86 +++++ deploy/postgres/004_privileges.sql | 21 ++ deploy/postgres/README.md | 25 ++ deploy/postgres/tests/assertions.sql | 65 ++++ docs/03-tech-stack.md | 12 +- docs/04-architecture.md | 9 +- docs/06-tasks.md | 3 +- docs/api.md | 10 +- docs/current-state.md | 21 +- docs/tasks/T-009.md | 12 +- scripts/test_postgres.ps1 | 133 +++++++ tests/test_postgres_contract.py | 102 +++++ 23 files changed, 1593 insertions(+), 24 deletions(-) create mode 100644 Sense/internal/store/postgres.go create mode 100644 Sense/internal/store/postgres_test.go create mode 100644 Sense/internal/store/repository.go create mode 100644 deploy/postgres/001_roles.sql create mode 100644 deploy/postgres/002_bell.sql create mode 100644 deploy/postgres/003_sense.sql create mode 100644 deploy/postgres/004_privileges.sql create mode 100644 deploy/postgres/README.md create mode 100644 deploy/postgres/tests/assertions.sql create mode 100644 scripts/test_postgres.ps1 create mode 100644 tests/test_postgres_contract.py diff --git a/Sense/README.md b/Sense/README.md index 15c1f69..30fec3c 100644 --- a/Sense/README.md +++ b/Sense/README.md @@ -1,6 +1,6 @@ -# Sense M1 骨架 +# Sense M1/M2 接入骨架 -本目录是 YoVision Sense 的 M1 接入骨架。数据库保存期望态，ONVIF 和 MediaMTX 通过端口隔离；默认关闭真实 ONVIF，显式设置 `SENSE_ONVIF_MODE=standard` 后才启用标准 SOAP/WS-Security 适配器。T-006 的真实样机结论仅覆盖已批准的精确海康基线，不能据此宣称多品牌兼容。 +本目录是 YoVision Sense 的 M1/M2 接入骨架。数据库保存期望态，ONVIF 和 MediaMTX 通过端口隔离；M1 默认使用 SQLite，T-009 增加 PostgreSQL 双 schema 生产路径。默认关闭真实 ONVIF，显式设置 `SENSE_ONVIF_MODE=standard` 后才启用标准 SOAP/WS-Security 适配器。T-006 的真实样机结论仅覆盖已批准的精确海康基线，不能据此宣称多品牌兼容。 ## 常用命令 @@ -14,7 +14,7 @@ go build -o bin/sense-api.exe ./cmd/sense-api go run ./cmd/sense-api ``` -Unix 将构建产物改为 `bin/sense-api`。服务默认监听 `127.0.0.1:8080`，SQLite 默认写入 `Sense/data/sense.db`，MediaMTX 控制 API 默认是 `http://127.0.0.1:9997`。当前只有 `/healthz` 与 `/readyz`，设备管理公共 API 尚未冻结。 +Unix 将构建产物改为 `bin/sense-api`。服务默认监听 `127.0.0.1:8080`，SQLite 默认写入 `Sense/data/sense.db`，MediaMTX 控制 API 默认是 `http://127.0.0.1:9997`。当前运行代码只有 `/healthz` 与 `/readyz`；Sense Control API v1 虽已冻结，但 HTTP handler 和认证尚未实现。 常用环境变量： @@ -22,7 +22,8 @@ Unix 将构建产物改为 `bin/sense-api`。服务默认监听 `127.0.0.1:8080` | --- | --- | --- | | `SENSE_HTTP_ADDR` | `127.0.0.1:8080` | HTTP 监听地址 | | `SENSE_ALLOW_NON_LOOPBACK` | `false` | 显式允许监听非回环地址；只应在可信网络及外部认证/防火墙就绪后开启 | -| `SENSE_DB_DSN` | `file:data/sense.db` | SQLite DSN；凭据不得放入该值 | +| `SENSE_DB_DRIVER` | `sqlite` | `sqlite` 或 `postgres`；生产切换必须显式选择 `postgres` | +| `SENSE_DB_DSN` | `file:data/sense.db` | 所选 driver 的私有 DSN；不得写入日志、文档或仓库，PostgreSQL 密码优先由 `PGPASSFILE`/环境密钥提供 | | `SENSE_MEDIAMTX_URL` | `http://127.0.0.1:9997` | MediaMTX 控制 API；不得包含 userinfo | | `SENSE_RECONCILE_INTERVAL` | `5s` | 对账周期 | | `SENSE_PROBE_INTERVAL` | `10s` | path 探活周期 | @@ -44,6 +45,28 @@ SENSE_CREDENTIAL__RTSP_PASSWORD MediaMTX `v1.19.3` 应作为独立二进制启动并只在可信网络开放 API。获取与 SHA-256 校验值见 `docs/03-tech-stack.md`。生成客户端使用固定版本工具和 vendored 官方 OpenAPI；`internal/mtx/generated/client.gen.go` 不可手改。 +## T-009 PostgreSQL 17.10 + +初始化 SQL 位于 `deploy/postgres/`，由高权限部署步骤按文件名前缀执行；Sense 进程不会自动创建角色、schema 或 Bell 对象。`bell_app` 拥有 Site/配额和 `bell.site_quota_v1`，`sense_app` 只能读取该视图，不能读取或写入 Bell 源表。应用登录角色和密码由部署环境创建，不进入仓库。 + +Windows 本机集成测试从仓库根目录执行： + +```powershell +./scripts/test_postgres.ps1 -PgRoot D:\pgsql17 +``` + +脚本要求冻结的 PostgreSQL `17.10`，使用 `initdb` 创建临时 trust 集群并只监听随机回环端口，重放 migration、执行权限断言和 PostgreSQL repository 测试后停止并清理。它会核对现有 `5432` listener 前后未变化，不读取或修改 `D:\pgsql17\data`。生产安装和恢复边界见 `deploy/postgres/README.md`。 + +选择 PostgreSQL 运行前，管理员必须已经安装 migration，并私下设置无密码回显的连接环境： + +```powershell +$env:SENSE_DB_DRIVER = 'postgres' +$env:SENSE_DB_DSN = '由部署环境私下设置' +go run ./cmd/sense-api +``` + +PostgreSQL 启动会检查 Sense migration 版本及当前角色的跨 schema 权限；权限过宽、配额视图不可读或 schema 未安装时 readiness 初始化失败。默认 SQLite 路径和 `cmd/sense-lab` 保持不变，T-009 不搬迁现有 SQLite 数据。 + Windows 本地准备 MediaMTX（从仓库根目录执行）： ```powershell diff --git a/Sense/cmd/sense-api/main.go b/Sense/cmd/sense-api/main.go index 3d30d0f..7196feb 100644 --- a/Sense/cmd/sense-api/main.go +++ b/Sense/cmd/sense-api/main.go @@ -38,7 +38,7 @@ func run(logger *slog.Logger) error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - repository, err := store.OpenSQLite(ctx, cfg.DatabaseDSN) + repository, err := store.OpenRepository(ctx, cfg.DatabaseDriver, cfg.DatabaseDSN) if err != nil { return err } diff --git a/Sense/go.mod b/Sense/go.mod index 9af5bda..04e5cc5 100644 --- a/Sense/go.mod +++ b/Sense/go.mod @@ -5,6 +5,7 @@ go 1.26.0 toolchain go1.26.5 require ( + github.com/jackc/pgx/v5 v5.10.0 github.com/oapi-codegen/runtime v1.6.0 modernc.org/sqlite v1.54.0 ) @@ -17,6 +18,9 @@ require ( github.com/go-openapi/jsonpointer v0.23.1 // indirect github.com/go-openapi/swag/jsonname v0.26.0 // indirect github.com/google/uuid v1.6.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/oapi-codegen/oapi-codegen/v2 v2.8.0 // indirect diff --git a/Sense/go.sum b/Sense/go.sum index cf11ee6..5fbbba3 100644 --- a/Sense/go.sum +++ b/Sense/go.sum @@ -52,6 +52,14 @@ github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -109,6 +117,7 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/vmware-labs/yaml-jsonpath v0.3.2 h1:/5QKeCBGdsInyDCyVNLbXyilb61MXGi9NP674f9Hobk= diff --git a/Sense/internal/config/config.go b/Sense/internal/config/config.go index 4f7558f..0ae54ac 100644 --- a/Sense/internal/config/config.go +++ b/Sense/internal/config/config.go @@ -13,6 +13,7 @@ import ( const ( defaultHTTPAddress = "127.0.0.1:8080" + defaultDatabaseDriver = "sqlite" defaultDatabaseDSN = "file:data/sense.db" defaultMediaMTXURL = "http://127.0.0.1:9997" defaultReconcilePeriod = 5 * time.Second @@ -23,6 +24,7 @@ const ( type Config struct { HTTPAddress string AllowNonLoopback bool + DatabaseDriver string DatabaseDSN string MediaMTXURL string ReconcileInterval time.Duration @@ -58,6 +60,7 @@ func Load() (Config, error) { cfg := Config{ HTTPAddress: stringEnv("SENSE_HTTP_ADDR", defaultHTTPAddress), AllowNonLoopback: allow, + DatabaseDriver: stringEnv("SENSE_DB_DRIVER", defaultDatabaseDriver), DatabaseDSN: stringEnv("SENSE_DB_DSN", defaultDatabaseDSN), MediaMTXURL: stringEnv("SENSE_MEDIAMTX_URL", defaultMediaMTXURL), ReconcileInterval: reconcilePeriod, @@ -86,6 +89,16 @@ func (c Config) Validate() error { if c.DatabaseDSN == "" { return fmt.Errorf("SENSE_DB_DSN must not be empty") } + databaseDriver := strings.ToLower(strings.TrimSpace(c.DatabaseDriver)) + if databaseDriver == "" { + databaseDriver = defaultDatabaseDriver + } + if databaseDriver != "sqlite" && databaseDriver != "postgres" { + return fmt.Errorf("SENSE_DB_DRIVER must be sqlite or postgres") + } + if databaseDriver == "postgres" && strings.HasPrefix(c.DatabaseDSN, "file:") { + return fmt.Errorf("postgres SENSE_DB_DRIVER requires an explicit PostgreSQL SENSE_DB_DSN") + } mediaURL, err := url.Parse(c.MediaMTXURL) if err != nil || mediaURL.Scheme == "" || mediaURL.Host == "" { return fmt.Errorf("invalid SENSE_MEDIAMTX_URL") diff --git a/Sense/internal/config/config_test.go b/Sense/internal/config/config_test.go index 58a6d1c..1c60ade 100644 --- a/Sense/internal/config/config_test.go +++ b/Sense/internal/config/config_test.go @@ -70,3 +70,26 @@ func TestLoadRTSPStripQueryOptIn(t *testing.T) { t.Fatal("explicit RTSP query stripping was not loaded") } } + +func TestValidateDatabaseDriver(t *testing.T) { + t.Parallel() + cfg := Config{ + HTTPAddress: "127.0.0.1:8080", + DatabaseDriver: "postgres", + DatabaseDSN: "file:test.db", + MediaMTXURL: "http://127.0.0.1:9997", + ReconcileInterval: 1, + ProbeInterval: 1, + } + if err := cfg.Validate(); err == nil { + t.Fatal("postgres driver must reject the SQLite default DSN") + } + cfg.DatabaseDSN = "postgres://sense-runtime@127.0.0.1/yovision?sslmode=disable" + if err := cfg.Validate(); err != nil { + t.Fatalf("valid PostgreSQL selection failed: %v", err) + } + cfg.DatabaseDriver = "mysql" + if err := cfg.Validate(); err == nil { + t.Fatal("unknown database driver must be rejected") + } +} diff --git a/Sense/internal/store/postgres.go b/Sense/internal/store/postgres.go new file mode 100644 index 0000000..ab41a50 --- /dev/null +++ b/Sense/internal/store/postgres.go @@ -0,0 +1,523 @@ +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(¤t); 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, ¤t, &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 +} diff --git a/Sense/internal/store/postgres_test.go b/Sense/internal/store/postgres_test.go new file mode 100644 index 0000000..110df1d --- /dev/null +++ b/Sense/internal/store/postgres_test.go @@ -0,0 +1,353 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "os" + "strings" + "sync" + "testing" + "time" + + "yovision/sense/internal/device" +) + +const ( + postgresTestDSNEnv = "YOVISION_TEST_POSTGRES_DSN" + postgresTestAdminDSNEnv = "YOVISION_TEST_POSTGRES_ADMIN_DSN" +) + +func TestPostgresInvalidDSNDoesNotLeakInput(t *testing.T) { + secret := "do-not-echo-this-value" + _, err := OpenPostgres(context.Background(), "postgres://sense:"+secret+"@%zz") + if err == nil { + t.Fatal("invalid PostgreSQL DSN was accepted") + } + if strings.Contains(err.Error(), secret) { + t.Fatal("PostgreSQL configuration error leaked DSN input") + } +} + +func TestOpenRepositoryRejectsUnknownDriver(t *testing.T) { + _, err := OpenRepository(context.Background(), "mysql", "unused") + if err == nil { + t.Fatal("unknown repository driver was accepted") + } +} + +func TestPostgresDefaultAndMaximumQuota(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant-default", "site-default", 0) + for index := 1; index <= device.DefaultVideoChannels; index++ { + if err := store.CreateDevice(ctx, videoDevice(index, "tenant-default", "site-default")); err != nil { + t.Fatalf("create default channel %d: %v", index, err) + } + } + radar := device.Device{ + ID: "radar-default", TenantID: "tenant-default", SiteID: "site-default", + SerialNumber: "radar-default", Name: "Radar", Modality: device.ModalityRadar, + Capabilities: []device.Capability{device.CapabilityTelemetry}, + DesiredState: device.DesiredEnabled, ActualState: device.ActualPending, + } + if err := store.CreateDevice(ctx, radar); err != nil { + t.Fatalf("non-video device must not consume video quota: %v", err) + } + err := store.CreateDevice(ctx, videoDevice(17, "tenant-default", "site-default")) + var quotaError *device.QuotaExceededError + if !errors.As(err, "aError) || quotaError.Limit != 16 { + t.Fatalf("expected default quota error, got %v", err) + } + + insertBellSite(t, admin, "tenant-max", "site-max", 128) + for index := 1; index <= device.MaximumVideoChannels; index++ { + if err := store.CreateDevice(ctx, videoDevice(index+1000, "tenant-max", "site-max")); err != nil { + t.Fatalf("create maximum channel %d: %v", index, err) + } + } + err = store.CreateDevice(ctx, videoDevice(1129, "tenant-max", "site-max")) + if !errors.As(err, "aError) || quotaError.Limit != 128 { + t.Fatalf("expected maximum quota error, got %v", err) + } +} + +func TestPostgresConcurrentAdmissionCannotExceedQuota(t *testing.T) { + store, admin := openPostgresTestStore(t) + insertBellSite(t, admin, "tenant", "site", 1) + ctx := context.Background() + start := make(chan struct{}) + errorsFound := make(chan error, 2) + var wait sync.WaitGroup + for index := 1; index <= 2; index++ { + wait.Add(1) + go func(index int) { + defer wait.Done() + <-start + errorsFound <- store.CreateDevice(ctx, videoDevice(index, "tenant", "site")) + }(index) + } + close(start) + wait.Wait() + close(errorsFound) + var succeeded, rejected int + for err := range errorsFound { + if err == nil { + succeeded++ + continue + } + var quotaError *device.QuotaExceededError + if errors.As(err, "aError) { + rejected++ + continue + } + t.Fatalf("unexpected concurrent admission error: %v", err) + } + if succeeded != 1 || rejected != 1 { + t.Fatalf("expected one success and one quota rejection, got success=%d rejected=%d", succeeded, rejected) + } +} + +func TestPostgresProjectionFailureAndRollbackFailClosed(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + err := store.CreateDevice(ctx, videoDevice(1, "missing-tenant", "missing-site")) + if !errors.Is(err, ErrQuotaProjectionUnavailable) { + t.Fatalf("missing projection must fail closed, got %v", err) + } + + insertBellSite(t, admin, "tenant", "site", 2) + if _, err := store.db.ExecContext(ctx, `INSERT INTO sense.site_quota_projection_state( + tenant_id, site_id, source_version, synced_at + ) VALUES ('tenant', 'site', 99, clock_timestamp())`); err != nil { + t.Fatal(err) + } + err = store.CreateDevice(ctx, videoDevice(2, "tenant", "site")) + if !errors.Is(err, ErrQuotaProjectionInvalid) { + t.Fatalf("projection version rollback must fail closed, got %v", err) + } + var devices int + if err := store.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM sense.devices`).Scan(&devices); err != nil { + t.Fatal(err) + } + if devices != 0 { + t.Fatalf("failed admissions must not persist devices, got %d", devices) + } +} + +func TestPostgresLowerQuotaDoesNotDisableAndEnableStillChecks(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant", "site", 2) + for index := 1; index <= 2; index++ { + if err := store.CreateDevice(ctx, videoDevice(index, "tenant", "site")); err != nil { + t.Fatal(err) + } + } + disabled := videoDevice(3, "tenant", "site") + disabled.DesiredState = device.DesiredDisabled + if err := store.CreateDevice(ctx, disabled); err != nil { + t.Fatal(err) + } + if _, err := admin.ExecContext(ctx, `UPDATE bell.sites SET max_video_channels = 1 + WHERE tenant_id = 'tenant' AND id = 'site'`); err != nil { + t.Fatal(err) + } + for index := 1; index <= 2; index++ { + value, err := store.GetDevice(ctx, fmt.Sprintf("camera-%03d", index)) + if err != nil { + t.Fatal(err) + } + if value.DesiredState != device.DesiredEnabled { + t.Fatalf("quota decrease disabled existing device %d", index) + } + } + err := store.SetDesiredState(ctx, "camera-003", device.DesiredEnabled) + var quotaError *device.QuotaExceededError + if !errors.As(err, "aError) || quotaError.Limit != 1 { + t.Fatalf("new enable after quota decrease must be rejected, got %v", err) + } +} + +func TestPostgresTenantIsolationAndConvergenceParity(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant-a", "site", 1) + insertBellSite(t, admin, "tenant-b", "site", 1) + left := videoDevice(1, "tenant-a", "site") + right := videoDevice(2, "tenant-b", "site") + right.SerialNumber = left.SerialNumber + if err := store.CreateDevice(ctx, left); err != nil { + t.Fatal(err) + } + if err := store.CreateDevice(ctx, right); err != nil { + t.Fatal(err) + } + now := time.Date(2026, 8, 7, 0, 0, 0, 0, time.UTC) + if err := store.MarkReconciled(ctx, left.ID, 1, now); err != nil { + t.Fatal(err) + } + if err := store.UpdateActualState(ctx, left.ID, device.ActualOnline, now); err != nil { + t.Fatal(err) + } + snapshot, err := store.ConvergenceSnapshot(ctx) + if err != nil { + t.Fatal(err) + } + if snapshot.Total != 2 || snapshot.Unconverged != 1 { + t.Fatalf("unexpected convergence snapshot: %+v", snapshot) + } + if err := store.RequestReconcile(ctx, left.ID, now.Add(time.Second)); err != nil { + t.Fatal(err) + } + snapshot, err = store.ConvergenceSnapshot(ctx) + if err != nil { + t.Fatal(err) + } + if snapshot.Unconverged != 2 { + t.Fatalf("runtime reconcile request was not persisted: %+v", snapshot) + } +} + +func TestPostgresRecordsQuotaSourceVersion(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant", "site", 2) + if err := store.CreateDevice(ctx, videoDevice(1, "tenant", "site")); err != nil { + t.Fatal(err) + } + var firstVersion int64 + if err := store.db.QueryRowContext(ctx, `SELECT quota_source_version + FROM sense.devices WHERE id = 'camera-001'`).Scan(&firstVersion); err != nil { + t.Fatal(err) + } + if _, err := admin.ExecContext(ctx, `UPDATE bell.sites SET name = name + WHERE tenant_id = 'tenant' AND id = 'site'`); err != nil { + t.Fatal(err) + } + second := videoDevice(2, "tenant", "site") + if err := store.CreateDevice(ctx, second); err != nil { + t.Fatal(err) + } + var secondVersion int64 + if err := store.db.QueryRowContext(ctx, `SELECT quota_source_version + FROM sense.devices WHERE id = 'camera-002'`).Scan(&secondVersion); err != nil { + t.Fatal(err) + } + if firstVersion != 1 || secondVersion != 2 { + t.Fatalf("expected recorded projection versions 1 and 2, got %d and %d", firstVersion, secondVersion) + } +} + +func TestPostgresReconcilePortParity(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant", "site", 1) + if err := store.CreateDevice(ctx, videoDevice(1, "tenant", "site")); err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + due, err := store.ListDueReconcile(ctx, now.Add(time.Second), 10) + if err != nil || len(due) != 1 || len(due[0].Device.Capabilities) != 2 { + t.Fatalf("unexpected due reconcile list: values=%+v error=%v", due, err) + } + enabled, err := store.ListEnabledVideoDevices(ctx, 10) + if err != nil || len(enabled) != 1 || enabled[0].ID != "camera-001" { + t.Fatalf("unexpected enabled video list: values=%+v error=%v", enabled, err) + } + if err := store.SetDesiredState(ctx, "camera-001", device.DesiredEnabled); err != nil { + t.Fatal(err) + } + value, err := store.GetDevice(ctx, "camera-001") + if err != nil || value.Generation != 1 { + t.Fatalf("idempotent desired state changed generation: value=%+v error=%v", value, err) + } + nextAttempt := now.Add(time.Minute) + if err := store.MarkReconcileFailure(ctx, "camera-001", 1, nextAttempt, "test_failure", now); err != nil { + t.Fatal(err) + } + due, err = store.ListDueReconcile(ctx, now.Add(30*time.Second), 10) + if err != nil || len(due) != 0 { + t.Fatalf("backoff device became due early: values=%+v error=%v", due, err) + } + due, err = store.ListDueReconcile(ctx, now.Add(2*time.Minute), 10) + if err != nil || len(due) != 1 || due[0].FailureCount != 1 { + t.Fatalf("backoff device did not become due: values=%+v error=%v", due, err) + } + value, err = store.GetDevice(ctx, "camera-001") + if err != nil || value.ActualState != device.ActualFailed { + t.Fatalf("reconcile failure did not persist actual state: value=%+v error=%v", value, err) + } +} + +func TestPostgresOpenRejectsOverprivilegedRuntimeRole(t *testing.T) { + _, admin := openPostgresTestStore(t) + ctx := context.Background() + if _, err := admin.ExecContext(ctx, + `GRANT UPDATE ON bell.site_quota_v1 TO yovision_t009_sense`); err != nil { + t.Fatal(err) + } + defer func() { + _, _ = admin.ExecContext(context.Background(), + `REVOKE UPDATE ON bell.site_quota_v1 FROM yovision_t009_sense`) + }() + value, err := OpenPostgres(ctx, os.Getenv(postgresTestDSNEnv)) + if value != nil { + _ = value.Close() + t.Fatal("overprivileged runtime role was accepted") + } + if err == nil || !strings.Contains(err.Error(), "privilege boundary") { + t.Fatalf("expected privilege-boundary error, got %v", err) + } +} + +func openPostgresTestStore(t *testing.T) (*Postgres, *sql.DB) { + t.Helper() + dsn := os.Getenv(postgresTestDSNEnv) + adminDSN := os.Getenv(postgresTestAdminDSNEnv) + if dsn == "" || adminDSN == "" { + t.Skip("PostgreSQL integration DSNs are not configured") + } + admin, err := sql.Open("pgx", adminDSN) + if err != nil { + t.Fatal(err) + } + if err := admin.PingContext(context.Background()); err != nil { + admin.Close() + t.Fatal("connect PostgreSQL test administrator") + } + if _, err := admin.ExecContext(context.Background(), `TRUNCATE + sense.device_capabilities, + sense.reconcile_state, + sense.devices, + sense.site_quota_projection_state, + bell.sites CASCADE`); err != nil { + admin.Close() + t.Fatal(err) + } + store, err := OpenPostgres(context.Background(), dsn) + if err != nil { + admin.Close() + t.Fatal(err) + } + t.Cleanup(func() { + _ = store.Close() + _ = admin.Close() + }) + return store, admin +} + +func insertBellSite(t *testing.T, admin *sql.DB, tenantID, siteID string, quota int) { + t.Helper() + query := `INSERT INTO bell.sites(tenant_id, id, name, max_video_channels) + VALUES ($1, $2, $3, $4)` + arguments := []any{tenantID, siteID, "Test Site", quota} + if quota == 0 { + query = `INSERT INTO bell.sites(tenant_id, id, name) VALUES ($1, $2, $3)` + arguments = arguments[:3] + } + if _, err := admin.ExecContext(context.Background(), query, arguments...); err != nil { + t.Fatal(err) + } +} diff --git a/Sense/internal/store/repository.go b/Sense/internal/store/repository.go new file mode 100644 index 0000000..3d3347e --- /dev/null +++ b/Sense/internal/store/repository.go @@ -0,0 +1,63 @@ +package store + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "yovision/sense/internal/device" +) + +const ( + DriverSQLite = "sqlite" + DriverPostgres = "postgres" +) + +var ( + ErrQuotaProjectionUnavailable = errors.New("quota projection unavailable") + ErrQuotaProjectionInvalid = errors.New("quota projection invalid") +) + +// Repository is the storage boundary used by the Sense process. SQLite stays +// available for M1 development; PostgreSQL implements the M2 production path. +type Repository interface { + Close() error + CreateDevice(context.Context, device.Device) error + SetDesiredState(context.Context, string, device.DesiredState) error + GetDevice(context.Context, string) (device.Device, error) + ListDueReconcile(context.Context, time.Time, int) ([]ReconcileCandidate, error) + ListEnabledVideoDevices(context.Context, int) ([]device.Device, error) + MarkReconciled(context.Context, string, int64, time.Time) error + MarkReconcileFailure(context.Context, string, int, time.Time, string, time.Time) error + UpdateActualState(context.Context, string, device.ActualState, time.Time) error + RequestReconcile(context.Context, string, time.Time) error + ConvergenceSnapshot(context.Context) (ConvergenceSnapshot, error) +} + +func OpenRepository(ctx context.Context, driver, dsn string) (Repository, error) { + switch strings.ToLower(strings.TrimSpace(driver)) { + case "", DriverSQLite: + return OpenSQLite(ctx, dsn) + case DriverPostgres: + return OpenPostgres(ctx, dsn) + default: + return nil, fmt.Errorf("unsupported database driver %q", driver) + } +} + +type quotaProjectionError struct { + kind error +} + +func (e *quotaProjectionError) Error() string { return e.kind.Error() } +func (e *quotaProjectionError) Unwrap() error { return e.kind } + +func projectionUnavailable() error { + return "aProjectionError{kind: ErrQuotaProjectionUnavailable} +} + +func projectionInvalid() error { + return "aProjectionError{kind: ErrQuotaProjectionInvalid} +} diff --git a/deploy/postgres/001_roles.sql b/deploy/postgres/001_roles.sql new file mode 100644 index 0000000..64dcf42 --- /dev/null +++ b/deploy/postgres/001_roles.sql @@ -0,0 +1,24 @@ +-- T-009 cluster roles. Runtime login roles are deployment-specific members of +-- these NOLOGIN roles; passwords and login principals never belong in Git. + +DO $roles$ +DECLARE + role_record pg_roles%ROWTYPE; +BEGIN + SELECT * INTO role_record FROM pg_roles WHERE rolname = 'bell_app'; + IF NOT FOUND THEN + CREATE ROLE bell_app NOLOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION; + ELSIF role_record.rolsuper OR role_record.rolcreatedb OR role_record.rolcreaterole + OR role_record.rolreplication OR role_record.rolcanlogin THEN + RAISE EXCEPTION 'existing bell_app role has unsafe attributes'; + END IF; + + SELECT * INTO role_record FROM pg_roles WHERE rolname = 'sense_app'; + IF NOT FOUND THEN + CREATE ROLE sense_app NOLOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION; + ELSIF role_record.rolsuper OR role_record.rolcreatedb OR role_record.rolcreaterole + OR role_record.rolreplication OR role_record.rolcanlogin THEN + RAISE EXCEPTION 'existing sense_app role has unsafe attributes'; + END IF; +END +$roles$; diff --git a/deploy/postgres/002_bell.sql b/deploy/postgres/002_bell.sql new file mode 100644 index 0000000..c22bc94 --- /dev/null +++ b/deploy/postgres/002_bell.sql @@ -0,0 +1,73 @@ +-- Bell owns Tenant/Site/quota truth. Run as the cluster administrator or a +-- migration role that can SET ROLE to bell_app. + +CREATE SCHEMA IF NOT EXISTS bell AUTHORIZATION bell_app; +ALTER SCHEMA bell OWNER TO bell_app; + +CREATE TABLE IF NOT EXISTS bell.schema_migrations ( + version bigint PRIMARY KEY, + applied_at timestamptz NOT NULL DEFAULT clock_timestamp() +); +ALTER TABLE bell.schema_migrations OWNER TO bell_app; + +CREATE TABLE IF NOT EXISTS bell.sites ( + tenant_id text NOT NULL, + id text NOT NULL, + name text NOT NULL, + max_video_channels smallint NOT NULL DEFAULT 16, + version bigint NOT NULL DEFAULT 1, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + updated_at timestamptz NOT NULL DEFAULT clock_timestamp(), + deleted_at timestamptz, + PRIMARY KEY (tenant_id, id), + CONSTRAINT bell_sites_video_quota_range + CHECK (max_video_channels BETWEEN 1 AND 128), + CONSTRAINT bell_sites_version_positive CHECK (version >= 1), + CONSTRAINT bell_sites_identity_not_blank + CHECK (btrim(tenant_id) <> '' AND btrim(id) <> '' AND btrim(name) <> '') +); +ALTER TABLE bell.sites OWNER TO bell_app; + +CREATE OR REPLACE FUNCTION bell.bump_site_version() +RETURNS trigger +LANGUAGE plpgsql +SECURITY INVOKER +SET search_path = pg_catalog, bell +AS $function$ +BEGIN + IF NEW.tenant_id IS DISTINCT FROM OLD.tenant_id OR NEW.id IS DISTINCT FROM OLD.id THEN + RAISE EXCEPTION 'Bell site identity is immutable'; + END IF; + NEW.version := OLD.version + 1; + NEW.updated_at := clock_timestamp(); + RETURN NEW; +END +$function$; +ALTER FUNCTION bell.bump_site_version() OWNER TO bell_app; + +DROP TRIGGER IF EXISTS bell_sites_bump_version ON bell.sites; +CREATE TRIGGER bell_sites_bump_version +BEFORE UPDATE ON bell.sites +FOR EACH ROW EXECUTE FUNCTION bell.bump_site_version(); + +CREATE OR REPLACE VIEW bell.site_quota_v1 ( + tenant_id, + site_id, + max_video_channels, + source_version, + source_updated_at +) AS +SELECT + site.tenant_id, + site.id, + site.max_video_channels, + site.version, + site.updated_at +FROM bell.sites AS site +WHERE site.deleted_at IS NULL; +ALTER VIEW bell.site_quota_v1 OWNER TO bell_app; +COMMENT ON VIEW bell.site_quota_v1 IS + 'v1 read-only site video quota projection owned by Bell and consumed by Sense'; + +INSERT INTO bell.schema_migrations(version) VALUES (1) +ON CONFLICT (version) DO NOTHING; diff --git a/deploy/postgres/003_sense.sql b/deploy/postgres/003_sense.sql new file mode 100644 index 0000000..48fb257 --- /dev/null +++ b/deploy/postgres/003_sense.sql @@ -0,0 +1,86 @@ +-- Sense owns device desired state and reconciliation progress. It deliberately +-- has no writable Site or quota truth table. + +CREATE SCHEMA IF NOT EXISTS sense AUTHORIZATION sense_app; +ALTER SCHEMA sense OWNER TO sense_app; + +CREATE TABLE IF NOT EXISTS sense.schema_migrations ( + version bigint PRIMARY KEY, + applied_at timestamptz NOT NULL DEFAULT clock_timestamp() +); +ALTER TABLE sense.schema_migrations OWNER TO sense_app; + +CREATE TABLE IF NOT EXISTS sense.devices ( + id text PRIMARY KEY, + tenant_id text NOT NULL, + site_id text NOT NULL, + serial_number text NOT NULL, + name text NOT NULL, + modality text NOT NULL, + desired_state text NOT NULL, + actual_state text NOT NULL, + endpoint_ref text NOT NULL DEFAULT '', + credential_ref text NOT NULL DEFAULT '', + path_name text NOT NULL DEFAULT '', + generation bigint NOT NULL DEFAULT 1, + quota_source_version bigint, + created_at timestamptz NOT NULL, + updated_at timestamptz NOT NULL, + CONSTRAINT sense_devices_identity_not_blank CHECK ( + btrim(id) <> '' AND btrim(tenant_id) <> '' AND btrim(site_id) <> '' + AND btrim(serial_number) <> '' AND btrim(name) <> '' + ), + CONSTRAINT sense_devices_modality CHECK ( + modality IN ('video', 'radar', 'contact', 'button', 'wearable', 'other') + ), + CONSTRAINT sense_devices_desired_state CHECK (desired_state IN ('disabled', 'enabled')), + CONSTRAINT sense_devices_actual_state CHECK (actual_state IN ('pending', 'online', 'offline', 'failed')), + CONSTRAINT sense_devices_generation_positive CHECK (generation >= 1), + CONSTRAINT sense_devices_quota_version_positive CHECK ( + quota_source_version IS NULL OR quota_source_version >= 1 + ), + UNIQUE (tenant_id, site_id, serial_number) +); +ALTER TABLE sense.devices OWNER TO sense_app; + +CREATE TABLE IF NOT EXISTS sense.device_capabilities ( + device_id text NOT NULL REFERENCES sense.devices(id) ON DELETE CASCADE, + capability text NOT NULL, + PRIMARY KEY (device_id, capability), + CONSTRAINT sense_device_capability_known CHECK ( + capability IN ('video_capture', 'audio_capture', 'spatial_rule', 'telemetry') + ) +); +ALTER TABLE sense.device_capabilities OWNER TO sense_app; + +CREATE TABLE IF NOT EXISTS sense.reconcile_state ( + device_id text PRIMARY KEY REFERENCES sense.devices(id) ON DELETE CASCADE, + failure_count integer NOT NULL DEFAULT 0, + next_attempt_at timestamptz, + last_error_code text, + observed_generation bigint NOT NULL DEFAULT 0, + updated_at timestamptz NOT NULL, + CONSTRAINT sense_reconcile_failure_nonnegative CHECK (failure_count >= 0), + CONSTRAINT sense_reconcile_generation_nonnegative CHECK (observed_generation >= 0) +); +ALTER TABLE sense.reconcile_state OWNER TO sense_app; + +CREATE TABLE IF NOT EXISTS sense.site_quota_projection_state ( + tenant_id text NOT NULL, + site_id text NOT NULL, + source_version bigint NOT NULL, + synced_at timestamptz NOT NULL, + PRIMARY KEY (tenant_id, site_id), + CONSTRAINT sense_quota_projection_version_positive CHECK (source_version >= 1) +); +ALTER TABLE sense.site_quota_projection_state OWNER TO sense_app; + +CREATE INDEX IF NOT EXISTS sense_devices_site_state_idx + ON sense.devices(tenant_id, site_id, desired_state); +CREATE UNIQUE INDEX IF NOT EXISTS sense_devices_path_name_idx + ON sense.devices(path_name) WHERE path_name <> ''; +CREATE INDEX IF NOT EXISTS sense_reconcile_due_idx + ON sense.reconcile_state(next_attempt_at); + +INSERT INTO sense.schema_migrations(version) VALUES (1) +ON CONFLICT (version) DO NOTHING; diff --git a/deploy/postgres/004_privileges.sql b/deploy/postgres/004_privileges.sql new file mode 100644 index 0000000..7cb31e8 --- /dev/null +++ b/deploy/postgres/004_privileges.sql @@ -0,0 +1,21 @@ +-- Cross-schema privileges are intentionally one-way: Bell publishes one view; +-- Sense cannot read Bell source tables or write any Bell object. + +REVOKE ALL ON SCHEMA bell FROM PUBLIC; +REVOKE ALL ON ALL TABLES IN SCHEMA bell FROM PUBLIC; +REVOKE ALL ON ALL FUNCTIONS IN SCHEMA bell FROM PUBLIC; +REVOKE ALL ON SCHEMA sense FROM PUBLIC; +REVOKE ALL ON ALL TABLES IN SCHEMA sense FROM PUBLIC; + +GRANT USAGE ON SCHEMA bell TO sense_app; +GRANT SELECT ON TABLE bell.site_quota_v1 TO sense_app; +REVOKE ALL ON TABLE bell.sites FROM sense_app; +REVOKE ALL ON TABLE bell.schema_migrations FROM sense_app; +REVOKE ALL ON FUNCTION bell.bump_site_version() FROM sense_app; + +GRANT USAGE ON SCHEMA sense TO sense_app; +GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA sense TO sense_app; + +ALTER DEFAULT PRIVILEGES FOR ROLE bell_app IN SCHEMA bell REVOKE ALL ON TABLES FROM PUBLIC; +ALTER DEFAULT PRIVILEGES FOR ROLE bell_app IN SCHEMA bell REVOKE ALL ON FUNCTIONS FROM PUBLIC; +ALTER DEFAULT PRIVILEGES FOR ROLE sense_app IN SCHEMA sense REVOKE ALL ON TABLES FROM PUBLIC; diff --git a/deploy/postgres/README.md b/deploy/postgres/README.md new file mode 100644 index 0000000..4cfca23 --- /dev/null +++ b/deploy/postgres/README.md @@ -0,0 +1,25 @@ +# YoVision PostgreSQL 初始化 + +本目录实现 T-009 的 PostgreSQL `17.10` 初始 schema。SQL 按文件名前缀顺序执行：先创建 NOLOGIN 权限角色，再创建 Bell、Sense 对象，最后收窄跨 schema 权限。应用登录角色及其密码由部署环境或密钥系统创建，并分别加入 `bell_app`、`sense_app`；仓库不保存登录凭据。 + +生产/共享实例必须由管理员先备份并在 YoVision 专用数据库中执行。Sense 进程不会用高权限自动建库或建角色。示例只使用私有环境变量，不把实际 DSN 写入脚本或日志： + +```powershell +$env:YOVISION_PG_ADMIN_DSN = '由管理员私下设置' +Get-ChildItem deploy/postgres/[0-9][0-9][0-9]_*.sql | + Sort-Object Name | + ForEach-Object { & 'D:\pgsql17\bin\psql.exe' -X -v ON_ERROR_STOP=1 -d $env:YOVISION_PG_ADMIN_DSN -f $_.FullName } +``` + +不要把该示例替换成真实连接串后提交。自动验收使用 [`scripts/test_postgres.ps1`](../../scripts/test_postgres.ps1) 启动隔离临时集群，不读取或修改 `D:\pgsql17\data`。 + +## 权限边界 + +- `bell_app` 拥有 `bell.sites`、版本 trigger 和 `bell.site_quota_v1`。 +- `sense_app` 拥有 `sense` schema，只获得 `bell` schema 的 `USAGE` 和配额视图的 `SELECT`。 +- `sense_app` 对 `bell.sites`、Bell migration 表和 trigger function 没有权限。 +- PUBLIC 对两个业务 schema 的表和函数没有权限。 + +## 恢复 + +T-009 不迁移或删除现有 SQLite 数据，默认运行配置也仍为 SQLite。若 PostgreSQL 切换验证失败，停止 PostgreSQL 模式的 Sense 进程并恢复 `SENSE_DB_DRIVER=sqlite` 与原 SQLite DSN。初始 schema 不提供自动 down migration；需要移除时由管理员先备份，再只删除明确确认的 YoVision 专用数据库，禁止对共享实例做模糊或递归清理。 diff --git a/deploy/postgres/tests/assertions.sql b/deploy/postgres/tests/assertions.sql new file mode 100644 index 0000000..59d59cf --- /dev/null +++ b/deploy/postgres/tests/assertions.sql @@ -0,0 +1,65 @@ +\set ON_ERROR_STOP on + +DO $assertions$ +DECLARE + actual_columns text[]; + default_quota integer; + initial_version bigint; + updated_version bigint; +BEGIN + IF current_setting('server_version_num')::integer < 170000 + OR current_setting('server_version_num')::integer >= 180000 THEN + RAISE EXCEPTION 'T-009 requires PostgreSQL 17.x'; + END IF; + + SELECT array_agg(column_name ORDER BY ordinal_position) + INTO actual_columns + FROM information_schema.columns + WHERE table_schema = 'bell' AND table_name = 'site_quota_v1'; + IF actual_columns IS DISTINCT FROM ARRAY[ + 'tenant_id', 'site_id', 'max_video_channels', 'source_version', 'source_updated_at' + ] THEN + RAISE EXCEPTION 'site_quota_v1 signature drift: %', actual_columns; + END IF; + + IF NOT has_table_privilege('sense_app', 'bell.site_quota_v1', 'SELECT') THEN + RAISE EXCEPTION 'sense_app lacks SELECT on quota view'; + END IF; + IF has_table_privilege('sense_app', 'bell.site_quota_v1', 'INSERT') + OR has_table_privilege('sense_app', 'bell.site_quota_v1', 'UPDATE') + OR has_table_privilege('sense_app', 'bell.site_quota_v1', 'DELETE') THEN + RAISE EXCEPTION 'sense_app has forbidden quota-view write privileges'; + END IF; + IF has_table_privilege('sense_app', 'bell.sites', 'SELECT') + OR has_table_privilege('sense_app', 'bell.sites', 'INSERT') + OR has_table_privilege('sense_app', 'bell.sites', 'UPDATE') + OR has_table_privilege('sense_app', 'bell.sites', 'DELETE') THEN + RAISE EXCEPTION 'sense_app has forbidden Bell source-table privileges'; + END IF; + + INSERT INTO bell.sites(tenant_id, id, name) + VALUES ('assertion-tenant', 'assertion-site', 'Assertion Site') + RETURNING max_video_channels, version INTO default_quota, initial_version; + IF default_quota <> 16 OR initial_version <> 1 THEN + RAISE EXCEPTION 'Bell site default quota/version drift'; + END IF; + + UPDATE bell.sites SET name = 'Assertion Site Updated' + WHERE tenant_id = 'assertion-tenant' AND id = 'assertion-site' + RETURNING version INTO updated_version; + IF updated_version <> initial_version + 1 THEN + RAISE EXCEPTION 'Bell site source_version is not monotonic'; + END IF; + + BEGIN + INSERT INTO bell.sites(tenant_id, id, name, max_video_channels) + VALUES ('assertion-tenant', 'invalid-site', 'Invalid', 129); + RAISE EXCEPTION 'quota 129 unexpectedly accepted'; + EXCEPTION + WHEN check_violation THEN NULL; + END; + + DELETE FROM bell.sites + WHERE tenant_id = 'assertion-tenant' AND id = 'assertion-site'; +END +$assertions$; diff --git a/docs/03-tech-stack.md b/docs/03-tech-stack.md index 6967021..3afc0a3 100644 --- a/docs/03-tech-stack.md +++ b/docs/03-tech-stack.md @@ -41,6 +41,15 @@ MediaMTX 官方 `v1.19.3` OpenAPI 固定保存于 `Sense/api/vendor/mediamtx-v1.19.3.openapi.yaml`，SHA-256 为 `a2b58195f1ec76541e124b5de4ee54645e5a3e25f70c4a73acc4a44d6f2b9c52`。下载二进制后必须对照上表或官方同版 `checksums.sha256`，不得使用浮动 `latest` URL。SQLite `v1.56.0` 在本决策日刚发布，T-003 不追新；后续依赖升级单独评审。 +### 1.2 Sense M2 PostgreSQL 冻结版本（T-009） + +| 组件 | 冻结版本 | 许可证 / 校验 | 使用与退出路线 | +| --- | --- | --- | --- | +| PostgreSQL | `17.10` | PostgreSQL License；本机验收二进制 `D:\pgsql17\bin\postgres.exe` 与 `psql.exe` 均报告 `17.10` | 首期一个实例、`sense`/`bell` schema 分离；自动测试用同版二进制启动隔离临时集群，不改现有服务。升级 17.x 小版本须重跑 migration、权限和 repository 集成测试；跨大版本先做备份恢复演练 | +| Go PostgreSQL driver | `github.com/jackc/pgx/v5 v5.10.0` 的 `stdlib` | MIT；module sum `h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=`，go.mod sum `h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=` | 只通过 `database/sql` 和 store port 使用；DSN 只来自进程环境且不记录。退出时可更换 `database/sql` driver，业务域与调和器不依赖 pgx 类型 | + +M2 仍保留 SQLite 作为单机开发与 T-006 回归入口，但生产真相源切换目标是 PostgreSQL。运行时必须显式设置 `SENSE_DB_DRIVER=postgres` 和私有 `SENSE_DB_DSN` 才使用 PostgreSQL；仓库不提供带密码 DSN。初始 schema 由 `deploy/postgres/` 的顺序 migration 创建，不在 Sense 进程启动时使用高权限自动迁移。 + ## 2. 外部项目边界 - MiBeeNvr：只用于 M0 隔离实验室、ONVIF兼容性和交互参考，不作为生产依赖。 @@ -50,7 +59,7 @@ MediaMTX 官方 `v1.19.3` OpenAPI 固定保存于 `Sense/api/vendor/mediamtx-v1. ## 3. 待冻结项 -- Python、PostgreSQL、Savant/DeepStream 的精确版本；Go 与 MediaMTX 已为 Sense M1 冻结，后续阶段可按升级流程调整。 +- Python、Savant/DeepStream 的精确版本；Go、MediaMTX 与 PostgreSQL 已分别为 Sense M1/M2 冻结，后续阶段可按升级流程调整。 - Bell 前端框架和组件库。 - 事件投递 transport 从 HTTP 起步还是直接采用消息总线。 - 目标 GPU/边缘硬件、解码能力和每 worker 的 `max_sources`。 @@ -99,6 +108,7 @@ python scripts/validate_harness_governance.py | Harness 文档/任务/Gitea 模板 | 上述三条 Python 命令 | 任一治理协议、清单或任务 schema 变化 | 不适用 | | `docs/raw/contracts/` | JSON Schema 校验 + 契约代码断言（实现后补命令） | schema/示例/mapper 任一变化 | 生产者与消费者联合评审 | | Sense Go | `go -C Sense generate ./internal/mtx`、`go -C Sense test ./...`、`go -C Sense vet ./...`、`go -C Sense build ./...` | ONVIF、存储、MediaMTX、对账或公共 API 变化 | T-006 使用 1 路指定实机 + 4 路独立合成源；T-007 才要求客户/借用/租赁的真实多路矩阵 | +| PostgreSQL schema/repository | `python -m unittest discover -s tests -p "test_postgres_contract.py"`；Windows 本机再运行 `./scripts/test_postgres.ps1 -PgRoot D:\pgsql17` | migration、权限、配额判定或 PostgreSQL driver 变化 | 不需要摄像头；必须核对临时集群未使用现有 data 目录，现有 5432 listener 前后不变 | | Brain Python | 单元测试、类型/格式检查（命令待项目脚手架冻结） | mapper、判定状态机、模型接口变化 | 命中模型任务时用冻结数据集和目标硬件 | | Bell Go/Web | 后端测试 + 前端 lint/test/build（命令待脚手架冻结） | schema、RBAC、预警状态机或公共 UI 变化 | P0 流程由产品/值班角色验收 | | 容量/分片 | 任务内基准脚本 | 16/64/128 路里程碑 | 目标网络、媒体和 GPU 硬件必需 | diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 22a2ebb..74be670 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -45,7 +45,7 @@ Sense ── 视频流/触发信号 ──> Brain 主流程： -1. Bell 持有站点、Area、配额与 `capture_policy`；首期在同一 PostgreSQL 实例内发布版本化只读视图 `bell.site_quota_v1`，Sense 在新增/启用设备的同一写路径校验并记录所用 `source_version`。未来分库必须发布新版本契约，不能静默改变 v1 语义。 +1. Bell 持有站点、Area、配额与 `capture_policy`；首期在同一 PostgreSQL 实例内发布版本化只读视图 `bell.site_quota_v1`。T-009 已实现 Bell 源表/视图、最小权限和 Sense PostgreSQL repository；Sense 在新增/启用设备的同一事务内以站点 advisory lock 校验配额并记录所用 `source_version`。未来分库必须发布新版本契约，不能静默改变 v1 语义。 2. Sense 维护设备期望态，通过 MediaMTX API 和对账器收敛实际态。 3. Brain 消费视频与触发信号，产生符合 v0.1 的事件。 4. Bell 做 schema 与代码级断言，生成平台 ULID，保存不可变事件。 @@ -81,10 +81,10 @@ Sense ── 视频流/触发信号 ──> Brain ## 7. 一致性与失败处理 -- PostgreSQL 是期望态真相源；MediaMTX、推理 worker 和对象存储是可对账的实际态。 +- PostgreSQL `sense` schema 是生产期望态真相源；SQLite 只保留为 M1 本地开发/回归路径。MediaMTX、推理 worker 和对象存储是可对账的实际态。 - 对账器水平触发、幂等、指数退避、限制并发；部分失败不做跨系统回滚，只持续收敛。 - 孤儿删除必须有 10% 安全闸和人工可观察指标。 -- `bell.site_quota_v1` 行缺失、数值越界、版本回退或读取失败只阻止视频设备新增/启用，不中断已有流；降低配额导致超限时不自动停用，后续准入返回稳定错误并产生运维信号。 +- `bell.site_quota_v1` 行缺失、数值越界、版本回退或读取失败只阻止视频设备新增/启用，不中断已有流；降低配额导致超限时不自动停用，后续准入返回稳定错误并产生运维信号。多 Sense 实例使用 PostgreSQL transaction-scoped advisory lock 串行化同 tenant/site 的计数与写入，不能用进程内锁替代。 - Area 策略投影读取失败只阻止相关设备新增/启用；已有设备保持原状态并产生运维告警。策略变更与已有成像设备冲突时由 Bell 管理端显式处置。 - 高风险设备操作在本地事务内同时写期望态与审计 outbox；异步 relay 可重试、幂等投递到 Bell。T-004 只验证交互归属，不定义或实现接口契约。 - Brain 投递失败落本地队列重试，不阻塞实时推理主链路。 @@ -95,7 +95,7 @@ Sense ── 视频流/触发信号 ──> Brain ## 8. 数据与契约 - Bell 核心实体：Tenant → Site → Area（含 `capture_policy`）以及 Role/Binding/Quota/Audit；Sense 核心实体：Device（含 `modality + capabilities`）→ StreamBinding/Zone，以及带 `source_version`/`synced_at` 的 SiteQuota/AreaPolicyProjection。两个 schema 以稳定逻辑 ID 关联，不跨 schema 写入；配额 v1 的跨 schema 列签名限定为 `tenant_id/site_id/max_video_channels/source_version/source_updated_at`。 -- Sense Control API v1 使用站点作用域路径、认证上下文 tenant、cursor 分页、幂等键与 ETag；敏感连接引用只写不读。正式签名和兼容规则以 [`contracts/`](contracts/) 为准，当前契约冻结不等于 handler 或 PostgreSQL migration 已实现。 +- Sense Control API v1 使用站点作用域路径、认证上下文 tenant、cursor 分页、幂等键与 ETag；敏感连接引用只写不读。正式签名和兼容规则以 [`contracts/`](contracts/) 为准。T-009 只实现 PostgreSQL migration/repository，不等于 handler、认证或 Bell 管理服务已实现。 - 业务实体：Rule → Event → Alert → DeliveryAttempt/Ack；Event 与 Alert 不合并。 - Bell 通知域分为三个聚合：Contact/Team 保存身份、成员关系和已验证通道；OnCallSchedule/ScheduleVersion/ShiftException 保存时区、轮换与例外；EscalationPolicy/Step 通过 `person / team / on_call_schedule` 类型化 `target_ref` 引用目标。三者共享逻辑 ID，不复制手机号、班次或轮换字段。 - 每个 DeliveryAttempt 创建时解析当时生效的排班版本，并保存实际收件人、通道、`schedule_version` 和解析时间快照；之后联系人或排班修改不得回写既有投递事实。 @@ -111,6 +111,7 @@ Sense/cmd + Sense/internal/{device,onvif,mtx,reconcile,probe,trigger,tunnel,auth Brain/{pipeline,models,judge,emit,trigger,contracts} Bell/cmd + Bell/internal/{ingest,event,rule,alert,deliver,feedback,tenant,audit,store} Bell/{web,packs,contracts} +deploy/postgres/{001_roles.sql,002_bell.sql,003_sense.sql,004_privileges.sql,tests} ``` 当前只有空目录占位；真实脚手架必须由对应任务创建。 diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 42a0d2b..c6da134 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -23,7 +23,8 @@ 出口：10 个站点试点，至少一个站点完成 16 路开通/停用；`unconverged = 0` 稳定。 - T-008：冻结 Sense Control API v1 与 Bell `site_quota_v1` 只读投影契约；只建立接口和测试门禁，不等于 handler、Bell 表或 migration 已实现。 -- 按 T-008 契约实现认证 tenant 上下文、设备管理 handler、PostgreSQL repository、Bell 配额源表/migration 与只读投影。 +- T-009：使用 PostgreSQL 17.10 建立 `sense`/`bell` schema、Bell 配额源表/只读视图、最小权限和 Sense PostgreSQL repository；SQLite 保留为 M1 本地路径。 +- 后续任务按 T-008 契约实现认证 tenant 上下文、设备管理 handler、幂等/ETag/cursor 与批量操作；不得把 T-009 数据基础误报为 API 已交付。 - 对账器幂等/退避/并发/10% 安全闸。 - WireGuard 边缘隧道与断网恢复。 - 16 路批量开通、停用和容量基准。 diff --git a/docs/api.md b/docs/api.md index ef776c4..24520b3 100644 --- a/docs/api.md +++ b/docs/api.md @@ -21,7 +21,7 @@ | 调用方 → 提供方 | 用途 | 当前约束 | 状态 | | --- | --- | --- | --- | -| Sense → Bell | 读取站点视频配额 | 同一 PostgreSQL 实例内只读 `bell.site_quota_v1`；默认 16、最大 128；失败时拒绝新增/启用但不影响已有流 | T-008 已冻结 | +| Sense → Bell | 读取站点视频配额 | 同一 PostgreSQL 实例内只读 `bell.site_quota_v1`；默认 16、最大 128；失败时拒绝新增/启用但不影响已有流 | T-008 冻结，T-009 已实现数据路径 | | Bell → Sense | 请求事件证据/pre-roll 切片 | 幂等、按租户授权、异步结果、不得暴露原始凭据 | 待 M3 设计 | | Bell → Brain | outcome/误报反馈 | 原事件不可变；反馈可重试、去重、审计 | 待 M3 设计 | | Sense → Brain | 流绑定与设备型触发 | 分片可路由，触发入口与流控制解耦 | 待 M2/M3 设计 | @@ -43,7 +43,7 @@ - `endpoint_ref`、`credential_ref`、`profile_token` 只写不读；设备 ID 由服务端生成。普通响应和错误不得包含凭据、完整流 URI、token 或 MediaMTX 内部配置。 - v1 不提供删除设备；停用设备保留历史。写入受理只表示期望态已持久化，不能表示实际态已收敛。 -T-008 只冻结契约和仓库门禁，尚未实现 Sense HTTP handler、认证中间件、Bell 表或 PostgreSQL migration。 +T-008 冻结契约；T-009 已实现 PostgreSQL `bell.sites`/配额视图、最小权限和 Sense repository，但尚未实现 Sense HTTP handler、认证中间件或 Bell 管理服务。 ## 4. 待冻结的 Bell 公共 API @@ -68,7 +68,7 @@ Sense 使用 MediaMTX 官方 OpenAPI 生成客户端并加薄封装。业务代 | --- | --- | --- | | ONVIF adapter | `Probe(target)`、`SetSystemDateAndTime(target, time)` | 设备是外部来源；`target` 只含 endpoint ref 与不透明 credential ref。错误稳定映射为认证失败、超时、不可用、响应无效，不记录凭据或完整流地址 | | MediaMTX paths | `CreatePath`、`GetPath`、`EnsurePath`、`DeletePath`、`PathReady` | SQLite 设备台账持有期望态，MediaMTX 只持有运行配置；`EnsurePath` 相同 source 不写、不同 source patch、缺失时 add；当前调和器绝不枚举或删除孤儿 | -| Device repository | 站点、设备、期望态、实际态、调和进度 | SQLite 是 M1 期望态真相源；调和失败次数与下次时间持久化，进程重启不清空退避；配额读取/写入失败时拒绝新增或启用，不关闭已有流 | +| Device repository | 设备、期望态、实际态、调和进度 | SQLite 是 M1 默认开发路径；PostgreSQL 是 M2 生产路径并只读 `bell.site_quota_v1`。调和失败次数与下次时间持久化，进程重启不清空退避；配额读取失败时拒绝新增或启用，不关闭已有流 | MediaMTX 薄封装调用同版官方 OpenAPI 的 `/v3/config/paths/get|add|patch|delete/{name}` 与 `/v3/paths/get/{name}`。生成源、版本和 SHA-256 见 `docs/03-tech-stack.md`；业务包不得直接 import 生成包。 @@ -76,12 +76,12 @@ MediaMTX 薄封装调用同版官方 OpenAPI 的 `/v3/config/paths/get|add|patch - 设备类型由 `modality` 表达物理类别，由多值 `capabilities` 表达视频采集、音频、空间规则或遥测能力，避免把“摄像头”固化为唯一设备模型。 - 视频配额只统计 `desired_state=enabled` 且具有 `video_capture` capability 的设备；站点默认 16、可配置 1～128。禁用设备和非视频传感器不占视频路数。 -- SQLite 表使用 `sense_` 前缀对应未来 PostgreSQL `sense` schema：`sense_sites`、`sense_devices`、`sense_device_capabilities`、`sense_reconcile_state`。标识、唯一性、状态与时间字段语义保持一致；本地表名前缀不是跨系统公共契约。 +- SQLite 表使用 `sense_` 前缀；T-009 PostgreSQL 使用 `sense.devices`、`sense.device_capabilities`、`sense.reconcile_state` 和只记录已观察版本的 `sense.site_quota_projection_state`。PostgreSQL 不建立可写 Site 真相副本，站点与配额只来自 Bell 视图；SQLite 的 `sense_sites` 仅是 M1 本地兼容表，不是跨系统公共契约。 - 摄像头密码不进入设备普通字段。`credential_ref` 只保存外部密钥引用；ONVIF 返回的 stream URI 只在内存中传给 MediaMTX，不写入设备台账或日志。 ### 5.3 Sense 进程 HTTP 面 -T-003 只实现了运维探针：`GET /healthz` 表示进程存活，`GET /readyz` 表示配置、SQLite 打开及 migration 已完成。两者返回 JSON，均不等价于摄像头、MediaMTX path 或 M1 里程碑健康。T-008 已冻结站点作用域的 `/api/v1/sites/{site_id}/devices` 等设备管理契约，但尚未实现对应 handler；当前可运行进程仍只暴露探针，后续实现不得发布 `/api/v1/devices` 等无站点边界的临时接口。 +T-003 只实现了运维探针：`GET /healthz` 表示进程存活，`GET /readyz` 表示所选数据库已打开且 schema/权限前置检查完成。两者返回 JSON，均不等价于摄像头、MediaMTX path 或里程碑健康。T-008 已冻结站点作用域的 `/api/v1/sites/{site_id}/devices` 等设备管理契约，但尚未实现对应 handler；当前可运行进程仍只暴露探针，后续实现不得发布 `/api/v1/devices` 等无站点边界的临时接口。 ## 6. 变更流程 diff --git a/docs/current-state.md b/docs/current-state.md index 39551d6..b4cdd24 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -4,20 +4,22 @@ ## 当前阶段 -- 阶段：M0 指定摄像头型号准入已完成；M1 的“一实机 + 四合成源”实验室软件闭环已通过；M2 已完成首个控制面/配额契约门禁，但五条独立真实上游和生产 SLA 尚未验收。 -- 生产代码：Sense 已包含可构建进程、SQLite 台账、标准 ONVIF SOAP/WS-Security adapter、凭据引用、MediaMTX 生成客户端、对账、探活和 MediaMTX 重启重建；公共设备管理 API 与生产部署能力仍未实现。 +- 阶段：M0 指定摄像头型号准入已完成；M1 的“一实机 + 四合成源”实验室软件闭环已通过；M2 已完成控制面契约和 PostgreSQL 数据基础，但五条独立真实上游和生产 SLA 尚未验收。 +- 生产代码：Sense 已包含可构建进程、SQLite/PostgreSQL repository、标准 ONVIF SOAP/WS-Security adapter、凭据引用、MediaMTX 生成客户端、对账、探活和 MediaMTX 重启重建；公共设备管理 API、认证和完整生产部署仍未实现。 - 默认容量：16 路；单站点本阶段上限 128 路，必须横向分片。 ## 仓库现实 -- `Sense/` 已有 Go module 与 `cmd/sense-api`；`Brain/`、`Bell/` 仍只有目录占位。 +- `Sense/` 已有 Go module 与 `cmd/sense-api`；`Brain/`、`Bell/` 仍只有目录占位，Bell 的首个数据库源表/视图 migration 位于根目录 `deploy/postgres/`。 - Sense 设备模型使用 `modality + capabilities`，SQLite 执行 v1 migration；视频配额默认 16、允许 1～128，17/128/129、新增/启用和“降低配额不关闭已有流”均有测试。 +- T-009 冻结 PostgreSQL `17.10` 和 `pgx/v5 v5.10.0`，实现 `bell`/`sense` schema、NOLOGIN 权限角色、Bell Site 版本 trigger、`bell.site_quota_v1` 和 Sense PostgreSQL repository；同站点并发准入用事务级 advisory lock，配额缺失/越界/版本回退时失败关闭且不改变已有流。 +- Windows 隔离测试使用 `D:\pgsql17\bin` 启动随机回环端口临时集群，migration 可重放、权限断言和 8 组真实 PostgreSQL repository 集成场景通过后自动清理；现有 `D:\pgsql17\data` 和 5432 服务未被读取、停止或修改。 - MediaMTX 固定为独立二进制 `v1.19.3`，官方 OpenAPI 已按 SHA-256 vendoring，并由固定 `oapi-codegen v2.8.0` 生成客户端；手写薄封装有 create/read/delete、幂等 ensure 与探活假 HTTP 测试。 - T-003 对账进度与指数退避持久化，覆盖取消和 SQLite 重启恢复；T-006 增加真实 ONVIF adapter、RTSP router、实验室播种/状态工具、故障代理和五路自动验收。当前仍不枚举/删除孤儿。 - T-006 正式使用 1 台准入实机和 4 个独立合成 publisher 连续观察 `1806.6 s` / 180 次采样，四类恢复均通过，最大与最终 `unconverged` 均为 0；详细证据见 `docs/research/sense-5-stream-integration.md`。 - `docs/raw/01`～`08` 已记录需求、分析、方案、客户场景、事件比对和三系统职责。 - `docs/raw/contracts/event-v0.1.schema.json` 已冻结，并有多份示例与语义说明。 -- `docs/contracts/sense-control-v1.openapi.json` 已冻结站点作用域的设备查询、创建、修改、启停与最多 128 项批量操作；`site-quota-v1.sql` 已冻结 Bell 所有、Sense 只读的配额视图签名。两者当前只有契约测试，HTTP handler、认证、Bell 表和 migration 尚未实现。 +- `docs/contracts/sense-control-v1.openapi.json` 已冻结站点作用域的设备查询、创建、修改、启停与最多 128 项批量操作；`site-quota-v1.sql` 已冻结 Bell 所有、Sense 只读的配额视图签名。T-009 已实现数据库 migration/repository；HTTP handler、认证和 Bell 管理服务尚未实现。 - harness coding 文档、上下文清单、Gitea Issue/PR 模板和治理脚本已接入。 - Gitea 已初始化 12 个协作标签；`status/waiting` 用于依赖或外部条件未满足的未领取任务，实时可领取状态必须从 Gitea 查询，不在本文复制。 - T-002 已关闭架构影响型问题：首期 S2 民办寄宿学校、客户侧私有部署、ONVIF/RTSP、NVIDIA 主路径、自研 Bell、Web/H5 与客户侧证据存储等边界已批准。 @@ -46,6 +48,12 @@ python -m unittest discover -s tests -p "test_*.py" python scripts/validate_harness_governance.py ``` +本机 PostgreSQL 17.10 隔离验证： + +```powershell +./scripts/test_postgres.ps1 -PgRoot D:\pgsql17 +``` + Sense 默认监听 `127.0.0.1:8080`，提供 `/healthz` 与 `/readyz` 运维探针；它们不代表摄像头或 M1 里程碑健康。MediaMTX 获取、校验和独立启动方法见 `Sense/README.md`。 ## 当前 blocker / 待确认 @@ -56,12 +64,13 @@ Sense 默认监听 `127.0.0.1:8080`，提供 `/healthz` 与 `/readyz` 运维探 - S2 真实生产试点的未成年人影像、公共安全视频法规适用性和最终留存政策仍需客户/法务确认，阻塞 M3 上线但不阻塞 M1 实验室骨架。 - 人脸方向已延后至 M5 的 S4 成人园区候选试点；必要性/PIP 影响评估、单独同意与替代方式、合法底库来源和删除流程未完成，阻塞人脸能力上线。 - 短信/语音具体供应商未选；生产前必须选定两条独立投递路径并验证故障切换。 -- Python/PostgreSQL/Savant 的精确版本、目标硬件和 Bell 前端栈尚未冻结；Sense M1 的 Go、SQLite driver、MediaMTX、生成器及生成运行时版本已在 T-003 冻结。 +- Python/Savant 的精确版本、目标硬件和 Bell 前端栈尚未冻结；Sense M1 的 Go、SQLite driver、MediaMTX、生成器及生成运行时版本已在 T-003 冻结，PostgreSQL/pgx 版本已在 T-009 冻结。 +- 本机现有 PostgreSQL 5432 实例使用 SCRAM 且当前开发进程没有管理员密码；T-009 不绕过认证，自动验收使用隔离临时集群。向共享/生产实例安装 migration 前仍需管理员私下提供专用数据库、登录角色与备份方案。 - 代码知识图谱在无业务代码阶段可能为空；工具不可用时使用 `rg` 处理文档与配置。 ## 下一步 -下一步先按 Gitea 流程建立并领取 T-008 的实现后续任务：实现认证 tenant 上下文、Sense 设备管理 handler/PostgreSQL repository，以及 Bell 配额源表、migration 和 `bell.site_quota_v1`。客户授权、借用或租赁条件具备后再执行 T-007 五条独立真实上游现场门禁。T-006 的合成结果不解除 T-007，也不形成容量或生产 SLA 承诺。实时领取状态仍以 Gitea 为准。 +下一步按 Gitea 流程建立 Sense Control API v1 实现任务：基于 T-009 PostgreSQL repository 实现认证 tenant 上下文、设备 handler、幂等收据、ETag/cursor 和批量操作。客户授权、借用或租赁条件具备后再执行 T-007 五条独立真实上游现场门禁。T-006 的合成结果不解除 T-007，也不形成容量或生产 SLA 承诺。实时领取状态仍以 Gitea 为准。 ## 已知风险 diff --git a/docs/tasks/T-009.md b/docs/tasks/T-009.md index 1b3d4b1..a1c7f07 100644 --- a/docs/tasks/T-009.md +++ b/docs/tasks/T-009.md @@ -3,7 +3,7 @@ id: T-009 title: 建立 PostgreSQL 双 Schema 与 Bell 配额投影基础 phase: 2 deps: [T-008] -status: DOING +status: DONE created: 2026-08-07 issue: 31 context_ref: 8a4514076e9921b12bc26380b2aceee9334e947d @@ -60,7 +60,7 @@ T-008 已冻结 Sense Control API v1 和 `bell.site_quota_v1` 只读投影，但 ## 验收要点 -- 任务相关验证：在本机执行 `./scripts/test_postgres.ps1 -PgRoot D:\pgsql17`，必须完成隔离集群启动、migration 重放、权限断言、tenant/配额/并发/版本回退 repository 集成测试和自动清理；运行 `python -m unittest tests/test_postgres_contract.py` 或对应 discovery 命令，静态负例能够拦截 Bell 写权限与契约漂移。 +- 任务相关验证：在本机执行 `./scripts/test_postgres.ps1 -PgRoot D:\pgsql17`，必须完成隔离集群启动、migration 重放、权限断言、tenant/配额/并发/版本回退 repository 集成测试和自动清理；运行 `python -m unittest discover -s tests -p "test_postgres_contract.py"`，静态负例能够拦截 Bell 写权限与契约漂移。 - 完整门禁：运行 `./init.ps1`、`python scripts/validate_agent_context.py`、`python -m unittest discover -s tests -p "test_*.py"`、`python scripts/validate_harness_governance.py`、`go -C Sense test ./...`、`go -C Sense vet ./...`、`go -C Sense build ./...` 和 `git diff --check`，全部通过。 - 人工 / 设备验收：不需要摄像头、UI 或客户现场；任务所有者必须核对隔离集群确实没有使用 `D:\pgsql17\data`，现有服务 PID/端口在测试前后保持运行，且共享实例未新增 YoVision 对象。 - 构建产物：PostgreSQL migration、权限/迁移断言、本机隔离测试入口、Sense PostgreSQL repository 和配置/运维文档；不提交临时数据目录、数据库 dump 或秘密。 @@ -80,6 +80,14 @@ T-008 已冻结 Sense Control API v1 和 `bell.site_quota_v1` 只读投影，但 ## 执行记录 +### 2026-08-07 完成 PostgreSQL 数据基础 + +- 冻结 PostgreSQL `17.10` 与 `pgx/v5 v5.10.0`；新增四段可重放初始化 SQL，创建 NOLOGIN `bell_app`/`sense_app`、`bell`/`sense` schema、Bell Site 单调版本 trigger、T-008 五列配额视图、Sense 设备/能力/调和/投影观察表及最小权限。 +- 新增 PostgreSQL repository 和 `SENSE_DB_DRIVER=sqlite|postgres` 显式选择；默认 SQLite 与 T-006 工具保持不变。PostgreSQL 启动检查 migration 和跨 schema 权限，Sense 对 Bell 源表或配额视图有写权限时拒绝启动。 +- 视频新增/启用在一个事务内使用 tenant/site transaction-scoped advisory lock，读取并记录 `source_version` 后再计数写入；覆盖默认 16、128/129、非视频不占路、并发不超配额、缺失/回退失败关闭、降配不关流、tenant 隔离、相同期望态不增 generation 和全部调和 repository port。 +- `./scripts/test_postgres.ps1 -PgRoot D:\pgsql17` 使用随机回环端口启动隔离 PostgreSQL 17.10，migration 连续执行两遍、SQL 权限断言及 9 个 `TestPostgres*` 测试通过，临时集群停止并清理；脚本核对现有 5432 listener 前后相同且从不引用 `D:\pgsql17\data`。 +- `./init.ps1` 通过；`python -m unittest discover -s tests -p "test_*.py"` 共 33 项通过，其中 PostgreSQL 静态契约 7 项；`go -C Sense test ./...`、`go -C Sense vet ./...`、`go -C Sense build ./...`、上下文/治理校验和 `git diff --check` 均通过。不需要摄像头、UI 或客户现场验收。 + ### 2026-08-07 领取任务 - dispatcher `ila` 将任务分配给 `codex`；`context_ref` 为 `8a4514076e9921b12bc26380b2aceee9334e947d`，claim 为 `claims/T-009`，工作分支为 `agent/codex/T-009`。 diff --git a/scripts/test_postgres.ps1 b/scripts/test_postgres.ps1 new file mode 100644 index 0000000..5311f14 --- /dev/null +++ b/scripts/test_postgres.ps1 @@ -0,0 +1,133 @@ +[CmdletBinding()] +param( + [string]$PgRoot = 'D:\pgsql17' +) + +$ErrorActionPreference = 'Stop' +Set-StrictMode -Version Latest + +$repoRoot = (Resolve-Path -LiteralPath (Join-Path $PSScriptRoot '..')).Path +$pgRootResolved = (Resolve-Path -LiteralPath $PgRoot).Path +$bin = Join-Path $pgRootResolved 'bin' +$initdb = Join-Path $bin 'initdb.exe' +$pgCtl = Join-Path $bin 'pg_ctl.exe' +$psql = Join-Path $bin 'psql.exe' +$createdb = Join-Path $bin 'createdb.exe' +foreach ($required in @($initdb, $pgCtl, $psql, $createdb)) { + if (-not (Test-Path -LiteralPath $required -PathType Leaf)) { + throw "PostgreSQL executable is missing under the selected PgRoot." + } +} + +$postgresExe = Join-Path $bin 'postgres.exe' +$version = (& $postgresExe --version 2>&1 | Out-String).Trim() +if ($LASTEXITCODE -ne 0 -or $version -notmatch 'PostgreSQL\) 17\.10$') { + throw 'T-009 integration requires the frozen PostgreSQL 17.10 binaries.' +} + +$existing5432 = @( + Get-NetTCPConnection -State Listen -LocalPort 5432 -ErrorAction SilentlyContinue | + Select-Object -ExpandProperty OwningProcess -Unique | + Sort-Object +) + +$listener = [System.Net.Sockets.TcpListener]::new( + [System.Net.IPAddress]::Loopback, + 0 +) +$listener.Start() +$port = ([System.Net.IPEndPoint]$listener.LocalEndpoint).Port +$listener.Stop() + +$systemTemp = [IO.Path]::GetFullPath([IO.Path]::GetTempPath()) +$clusterName = 'yovision-t009-pg-' + [guid]::NewGuid().ToString('N') +$dataDir = Join-Path $systemTemp $clusterName +$logPath = Join-Path $systemTemp ($clusterName + '.log') +$started = $false +$stopped = $false +$hadSenseDSN = Test-Path Env:YOVISION_TEST_POSTGRES_DSN +$previousSenseDSN = if ($hadSenseDSN) { (Get-Item Env:YOVISION_TEST_POSTGRES_DSN).Value } else { $null } +$hadAdminDSN = Test-Path Env:YOVISION_TEST_POSTGRES_ADMIN_DSN +$previousAdminDSN = if ($hadAdminDSN) { (Get-Item Env:YOVISION_TEST_POSTGRES_ADMIN_DSN).Value } else { $null } + +function Invoke-Checked { + param( + [Parameter(Mandatory)] [string]$FilePath, + [Parameter(ValueFromRemainingArguments)] [string[]]$Arguments + ) + & $FilePath @Arguments + if ($LASTEXITCODE -ne 0) { + throw "Command failed with exit code $LASTEXITCODE." + } +} + +try { + New-Item -ItemType Directory -Path $dataDir | Out-Null + Invoke-Checked $initdb '-D' $dataDir '-U' 'postgres' '-A' 'trust' '--encoding=UTF8' '--no-locale' '--no-sync' + $serverOptions = "-h 127.0.0.1 -p $port -c listen_addresses=127.0.0.1" + Invoke-Checked $pgCtl '-D' $dataDir '-l' $logPath '-o' $serverOptions '-w' 'start' + $started = $true + + $adminRootDSN = "postgres://postgres@127.0.0.1:$port/postgres?sslmode=disable" + $databaseName = 'yovision_t009' + $adminDatabaseDSN = "postgres://postgres@127.0.0.1:$port/${databaseName}?sslmode=disable" + $senseDSN = "postgres://yovision_t009_sense@127.0.0.1:$port/${databaseName}?sslmode=disable" + + Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminRootDSN '-f' (Join-Path $repoRoot 'deploy\postgres\001_roles.sql') + Invoke-Checked $createdb '-h' '127.0.0.1' '-p' ([string]$port) '-U' 'postgres' $databaseName + + foreach ($pass in 1..2) { + foreach ($name in @('002_bell.sql', '003_sense.sql', '004_privileges.sql')) { + Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminDatabaseDSN '-f' (Join-Path $repoRoot "deploy\postgres\$name") + } + } + Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminRootDSN '-c' 'CREATE ROLE yovision_t009_sense LOGIN IN ROLE sense_app' + Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminDatabaseDSN '-f' (Join-Path $repoRoot 'deploy\postgres\tests\assertions.sql') + + $env:YOVISION_TEST_POSTGRES_DSN = $senseDSN + $env:YOVISION_TEST_POSTGRES_ADMIN_DSN = $adminDatabaseDSN + Invoke-Checked 'go' '-C' (Join-Path $repoRoot 'Sense') 'test' './internal/store' '-run' '^TestPostgres' '-count=1' +} +finally { + if ($started) { + & $pgCtl '-D' $dataDir '-m' 'fast' '-w' 'stop' + $stopped = $LASTEXITCODE -eq 0 + } + if (-not $started -or $stopped) { + $resolvedData = [IO.Path]::GetFullPath($dataDir) + if (-not $resolvedData.StartsWith($systemTemp, [StringComparison]::OrdinalIgnoreCase) -or + [IO.Path]::GetFileName($resolvedData) -notlike 'yovision-t009-pg-*') { + throw "Refusing to clean unexpected temporary path." + } + if (Test-Path -LiteralPath $resolvedData) { + Remove-Item -LiteralPath $resolvedData -Recurse -Force + } + if (Test-Path -LiteralPath $logPath) { + Remove-Item -LiteralPath $logPath -Force + } + } + else { + Write-Warning 'Temporary PostgreSQL did not stop; its data directory was retained for manual recovery.' + } + if ($hadSenseDSN) { + $env:YOVISION_TEST_POSTGRES_DSN = $previousSenseDSN + } + else { + Remove-Item Env:YOVISION_TEST_POSTGRES_DSN -ErrorAction SilentlyContinue + } + if ($hadAdminDSN) { + $env:YOVISION_TEST_POSTGRES_ADMIN_DSN = $previousAdminDSN + } + else { + Remove-Item Env:YOVISION_TEST_POSTGRES_ADMIN_DSN -ErrorAction SilentlyContinue + } + $after5432 = @( + Get-NetTCPConnection -State Listen -LocalPort 5432 -ErrorAction SilentlyContinue | + Select-Object -ExpandProperty OwningProcess -Unique | + Sort-Object + ) + if (($existing5432 -join ',') -ne ($after5432 -join ',')) { + throw 'The existing PostgreSQL listener on port 5432 changed during the isolated test.' + } +} +Write-Output "PostgreSQL $version isolated integration passed; temporary port $port was cleaned." diff --git a/tests/test_postgres_contract.py b/tests/test_postgres_contract.py new file mode 100644 index 0000000..c301947 --- /dev/null +++ b/tests/test_postgres_contract.py @@ -0,0 +1,102 @@ +"""Static safety checks for T-009 PostgreSQL migrations and local harness.""" + +from __future__ import annotations + +import re +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +MIGRATION_ROOT = ROOT / "deploy" / "postgres" + + +def migration_text(name: str) -> str: + return (MIGRATION_ROOT / name).read_text(encoding="utf-8") + + +def normalized(text: str) -> str: + return re.sub(r"\s+", " ", text.lower()).strip() + + +def privilege_findings(text: str) -> list[str]: + value = normalized(text) + findings: list[str] = [] + if "grant select on table bell.site_quota_v1 to sense_app" not in value: + findings.append("Sense SELECT grant on quota view is missing") + if "revoke all on table bell.sites from sense_app" not in value: + findings.append("Bell source table revoke is missing") + if re.search( + r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)" + r"(?:\s*,\s*(?:insert|update|delete|truncate|references|trigger))*" + r"\s+on(?:\s+table)?\s+bell\.", + value, + ): + findings.append("Sense or PUBLIC receives a broad/write Bell privilege") + if re.search(r"grant\s+select\s+on(?:\s+table)?\s+bell\.sites\s+to\s+sense_app", value): + findings.append("Sense receives SELECT on Bell source table") + return findings + + +class PostgresContractTests(unittest.TestCase): + def test_migration_sequence_is_complete_and_stable(self) -> None: + names = [path.name for path in sorted(MIGRATION_ROOT.glob("[0-9][0-9][0-9]_*.sql"))] + self.assertEqual( + ["001_roles.sql", "002_bell.sql", "003_sense.sql", "004_privileges.sql"], + names, + ) + + def test_roles_are_non_login_and_unprivileged(self) -> None: + text = normalized(migration_text("001_roles.sql")) + for role in ("bell_app", "sense_app"): + self.assertRegex( + text, + rf"create role {role} nologin nosuperuser nocreatedb nocreaterole noreplication", + ) + self.assertNotRegex(text, r"\bpassword\b") + + def test_bell_quota_contract_and_version_are_enforced(self) -> None: + text = normalized(migration_text("002_bell.sql")) + expected_signature = ( + "create or replace view bell.site_quota_v1 ( tenant_id, site_id, " + "max_video_channels, source_version, source_updated_at ) as" + ) + self.assertIn(expected_signature, text) + self.assertIn("max_video_channels smallint not null default 16", text) + self.assertIn("check (max_video_channels between 1 and 128)", text) + self.assertIn("new.version := old.version + 1", text) + self.assertIn("alter view bell.site_quota_v1 owner to bell_app", text) + + def test_sense_schema_has_no_writable_site_truth(self) -> None: + text = normalized(migration_text("003_sense.sql")) + self.assertNotRegex(text, r"create table(?: if not exists)? sense\.sites\b") + self.assertIn("create table if not exists sense.devices", text) + self.assertIn("quota_source_version bigint", text) + self.assertIn("create table if not exists sense.site_quota_projection_state", text) + + def test_cross_schema_privileges_are_minimal(self) -> None: + self.assertEqual([], privilege_findings(migration_text("004_privileges.sql"))) + + def test_privilege_validator_rejects_broad_or_source_access(self) -> None: + original = migration_text("004_privileges.sql") + changed = original + "\nGRANT ALL PRIVILEGES ON bell.sites TO sense_app;\n" + self.assertTrue(privilege_findings(changed)) + changed = original + "\nGRANT SELECT ON bell.sites TO sense_app;\n" + self.assertTrue(privilege_findings(changed)) + + def test_local_harness_isolated_from_existing_cluster(self) -> None: + text = (ROOT / "scripts" / "test_postgres.ps1").read_text(encoding="utf-8") + for marker in ( + "initdb.exe", + "pg_ctl.exe", + "127.0.0.1", + "yovision-t009-pg-", + "YOVISION_TEST_POSTGRES_DSN", + "Get-NetTCPConnection", + ): + self.assertIn(marker, text) + self.assertNotIn("D:\\pgsql17\\data", text) + + +if __name__ == "__main__": + unittest.main()