openvibers/openvibe.sources Commits

Sources moves to PostgreSQL (ADR-035, roadmap WS-X2, repository-pass group G2), async from end to end.

OP OpenVibers committed 28 Sept 2026 · 6f68dd58f5 · 33 files, +726 −453

- Schema: migrations/0001_initial.sql from the SQLite schema plus the outbox table (gen-migration --ensure).
- server/db.js: openDb(config) (DATABASE_URL through PgBouncer, migrations as the owner, PGlite in development);
  start() takes a migrated handle (the tests). nextSeq is one upsert returning the value.
- Every read and write is awaited (asyncify, which now also awaits the calls a helper makes of an async function it
  was passed, and leaves alone a promise started early and awaited later). The outbox runs in ambient transactions
  (the same module as Search's). SQL: typed IS NULL parameters, GREATEST for a scalar MAX. Gauges are async collects
  (openvibe-shared v1.27.0, which also routes async handler rejections to the error handler). The seed script is async.
- Readiness names the store.
- scripts/migrate-to-postgres.js (runSqliteMigration); rehearsed on a copy of production: the six registered
  sources verified.
- Tests on createTestDb (PGlite; npm run test:pg through PgBouncer): the in-flight test starts its first run without
  awaiting it again, and the scheduler test waits for the source's recorded outcome, which commits just after the run.
  Pins: openvibe-contracts v0.76.0, openvibe-sdk v0.20.0, openvibe-shared v1.27.0. README and STATUS.json follow.

npm test 11/11 on PGlite; npm run test:pg 11/11 through PgBouncer.

Changes

ddiff --git a/.env.example b/.env.example
iindex b7bba8b..8c96093 100644
---- a/.env.example
++++ b/.env.example
@@@ -28,6 +28,14 @@ OV_OAUTH_CLIENT_ID=sources
OV_OAUTH_CLIENT_SECRET=
# Store (SQLite). Production: /var/lib/openvibe-sources/sources.db (systemd StateDirectory).
++# PostgreSQL (ADR-035), written by OpenVibe.Host roles/data/add-service.sh sources: the runtime role through PgBouncer,
++# the owner on a direct connection (migrations and the one-time import), and Valkey (shared limit counters).
++# Without DATABASE_URL, development uses an embedded PGlite database in data/pglite.
++DATABASE_URL=
++DATABASE_DIRECT_URL=
++VALKEY_URL=
++VALKEY_PREFIX=
++# The SQLite file of releases before PostgreSQL: read once by scripts/migrate-to-postgres.js.
SOURCES_DB_PATH=./data/sources.db
# OpenVibe.Events. The outbox relay runs only when EVENTS_URL is set.
ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex 088083d..5289ab5 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -9,6 +9,9 @@ jobs:
uses: OpenVibers/OpenVibe.Shared/.github/workflows/test.yml@423fe519b5fe9594cc81b670e5ba5cb9f21f6176
with:
contracts-service: sources
++ # ADR-035: PostgreSQL 18 + PgBouncer (transaction mode) + Valkey 9 containers (openvibe-sdk
++ # scripts/test-services.sh). npm test runs every file on PGlite; test:pg runs them all through PgBouncer.
++ test: 'eval "$(node_modules/openvibe-sdk/scripts/test-services.sh up)" && npm test && npm run test:pg'
# Secret scan (gitleaks) and dependency audit, shared by every OpenVibers repo (Track Q).
security:
uses: OpenVibers/OpenVibe.Shared/.github/workflows/security.yml@423fe519b5fe9594cc81b670e5ba5cb9f21f6176
ddiff --git a/README.md b/README.md
iindex d870bbe..44a95ed 100644
---- a/README.md
++++ b/README.md
@@@ -31,7 +31,7 @@ npm test # every test/*.test.js against local stub sites; no inter
Node 22 in production (`fnm exec --using=22.22.1 npm test`). Production: `/opt/openvibe.sources`,
env `/etc/openvibe/sources.env`, unit [deploy/systemd/openvibe-sources.service](deploy/systemd/openvibe-sources.service),
--store `/var/lib/openvibe-sources/sources.db`, nginx [deploy/nginx/sources.openvibe.network.conf](deploy/nginx/sources.openvibe.network.conf).
++database `ov_sources` on the host's data role (ADR-035; `sudo /opt/openvibe.host/roles/data/add-service.sh sources`; the one-time move from `/var/lib/openvibe-sources/sources.db` is `scripts/migrate-to-postgres.js`), nginx [deploy/nginx/sources.openvibe.network.conf](deploy/nginx/sources.openvibe.network.conf).
`GET /api/health` is liveness. `GET /api/ready` (openvibe-shared/ready) is 503 only when the
database fails; a Network key that has not loaded and a fetcher that is off, stopped or behind (a
@@@ -125,7 +125,7 @@ Callers use an OpenVibe.Network client-credentials token for audience `openvibe.
`GET /api/v1/items` pages in change order (creations, revisions and — with `include_removed=1` —
removals); resume from `next_after`. Every page carries the status and staleness of the sources
it contains. The capability ids (first proposed in [docs/capabilities-proposal/](docs/capabilities-proposal/))
--are released in `openvibe-contracts` v0.12.0 (this repo pins v0.49.0); [server/auth.js](server/auth.js)
++are released in `openvibe-contracts` v0.76.0 (this repo pins v0.49.0); [server/auth.js](server/auth.js)
decides them with the contracts grant rule.
### Per-actor limits
@@@ -207,6 +207,7 @@ publishes another's output.
## Depends on
++- PostgreSQL 18 (OpenVibe.Host `roles/data/`, ADR-035): every read and write is async through `openvibe-sdk/db`
- OpenVibe.Contracts (service tokens, problem details, ids, the event envelope)
- OpenVibe.Network (signing key; service principal `sources`)
- OpenVibe.Events (outbound events)
@@@ -278,12 +279,11 @@ fast-forward `/opt/openvibe.sources`, install on a lockfile change, restart, wai
| Unit | `openvibe-sources.service` ([deploy/systemd/](deploy/systemd/openvibe-sources.service)) |
| Port | `127.0.0.1:4720` |
| Env file | `/etc/openvibe/sources.env` |
--| Data | `/var/lib/openvibe-sources/sources.db` |
++| Data | PostgreSQL `ov_sources` (`DATABASE_URL` through PgBouncer; migrations on `DATABASE_DIRECT_URL`; schema in [migrations/](migrations/)) |
| Public host | nginx [deploy/nginx/sources.openvibe.network.conf](deploy/nginx/sources.openvibe.network.conf) (below) |
Rollback: ovhost puts the previous sha back by itself when `/api/ready` does not answer 2xx after the
--restart; afterwards `sudo ovhost rollback sources --to <sha>`. Nothing blocks a rollback: the schema
--code only adds tables and columns.
++restart; afterwards `sudo ovhost rollback sources --to <sha>`. Migrations only add tables and columns.
### Public host
ddiff --git a/STATUS.json b/STATUS.json
iindex c6932bb..532b640 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,11 +7,25 @@
"runtime": "node 22, express, better-sqlite3, fast-xml-parser",
"port": 4720,
"domain": "sources.openvibe.network (vhost in deploy/nginx: an internal-service page, health and JSON 404s; the API is loopback-only; README \"Public host\")",
-- "contracts": "openvibe-contracts v0.49.0",
-- "capabilities": ["sources.source.read", "sources.item.read", "sources.source.manage"],
++ "contracts": "openvibe-contracts v0.76.0",
++ "capabilities": [
++ "sources.source.read",
++ "sources.item.read",
++ "sources.source.manage"
++ ],
"capabilitiesRegistered": true,
-- "contractsProposed": ["sources.source@1 (released in openvibe-contracts v0.12.0)", "sources.item@1 (released in openvibe-contracts v0.12.0)"],
-- "adapters": ["rss", "atom", "sitemap", "jsonld", "api", "manual"],
++ "contractsProposed": [
++ "sources.source@1 (released in openvibe-contracts v0.12.0)",
++ "sources.item@1 (released in openvibe-contracts v0.12.0)"
++ ],
++ "adapters": [
++ "rss",
++ "atom",
++ "sitemap",
++ "jsonld",
++ "api",
++ "manual"
++ ],
"seededSources": 6,
"enabledSources": 0,
"deployed": true,
ddiff --git a/deploy/systemd/openvibe-sources.service b/deploy/systemd/openvibe-sources.service
iindex 22356bf..8436423 100644
---- a/deploy/systemd/openvibe-sources.service
++++ b/deploy/systemd/openvibe-sources.service
@@@ -2,8 +2,8 @@
Description=OpenVibe.Sources — source registry and ingestion workers (RSS/Atom, sitemaps, JSON-LD, APIs)
# Network (openvibe-network) signs the tokens this service verifies. Sources starts without it
# and keeps retrying the JWKS fetch; /api/ready stays 503 until the key has loaded.
--After=network.target
--Wants=openvibe-network.service
++After=network.target postgresql.service pgbouncer.service valkey-server.service
++Wants=openvibe-network.service postgresql.service pgbouncer.service valkey-server.service
[Service]
Type=simple
ddiff --git a/migrations/0001_initial.sql b/migrations/0001_initial.sql
nnew file mode 100644
iindex 0000000..6fcd09b
---- /dev/null
++++ b/migrations/0001_initial.sql
@@@ -0,0 +1,147 @@
++-- phase: expand
++-- OpenVibe.Sources on PostgreSQL (ADR-035, roadmap WS-X2): the tables as they were on SQLite (converted by openvibe-sdk
++-- tools/asyncify/sqlite-schema-to-pg: text COLLATE "C" compares like SQLite, integers are bigint, identities keep their ids),
++-- then the openvibe-publishing stores and the openvibe-sdk inbox and outbox. Generated once on 2026-09-28; never edited after it runs.
++
++CREATE TABLE sources (
++ key text COLLATE "C" PRIMARY KEY,
++ name text COLLATE "C" NOT NULL,
++ type text COLLATE "C" NOT NULL, -- rss | atom | sitemap | jsonld | api | manual
++ category text COLLATE "C" NOT NULL, -- news | blog | reviews | deals | coupons | trade
++ homepage_url text COLLATE "C",
++ endpoints text COLLATE "C" NOT NULL DEFAULT '[]',
++ auth text COLLATE "C" NOT NULL DEFAULT '{"mode":"none"}', -- env var NAME only, never a value
++ robots_note text COLLATE "C",
++ terms_note text COLLATE "C",
++ license_note text COLLATE "C",
++ min_interval_ms bigint NOT NULL, -- rate limit: gap between two requests to this source
++ poll_interval_sec bigint NOT NULL,
++ stale_after_sec bigint NOT NULL,
++ max_items bigint NOT NULL,
++ enabled bigint NOT NULL DEFAULT 0,
++ review_required bigint NOT NULL DEFAULT 1,
++ sensitivity text COLLATE "C" NOT NULL DEFAULT 'none',
++ default_indexability text COLLATE "C" NOT NULL DEFAULT 'noindex',
++ search_visibility text COLLATE "C", -- null = items are not sent to OpenVibe.Search
++ created_at bigint NOT NULL,
++ updated_at bigint NOT NULL,
++ updated_by text COLLATE "C",
++ -- runtime state
++ next_due_at bigint NOT NULL DEFAULT 0,
++ not_before bigint NOT NULL DEFAULT 0, -- Retry-After / Crawl-delay
++ last_request_at bigint,
++ last_run_at bigint,
++ last_success_at bigint,
++ last_state text COLLATE "C",
++ consecutive_failures bigint NOT NULL DEFAULT 0
++);
++
++CREATE TABLE endpoint_state (
++ source_key text COLLATE "C" NOT NULL,
++ url text COLLATE "C" NOT NULL,
++ etag text COLLATE "C",
++ last_modified text COLLATE "C",
++ last_status bigint,
++ last_fetch_at bigint,
++ PRIMARY KEY (source_key, url)
++);
++
++-- One row per fetch attempt, whatever happened. state is never empty.
++CREATE TABLE fetch_runs (
++ rid bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ id text COLLATE "C" NOT NULL UNIQUE,
++ source_key text COLLATE "C" NOT NULL,
++ endpoint_url text COLLATE "C",
++ trigger text COLLATE "C" NOT NULL, -- schedule | manual
++ started_at bigint NOT NULL,
++ finished_at bigint NOT NULL,
++ state text COLLATE "C" NOT NULL CHECK (state IN ('ok','not_modified','http_error','timeout','robots_denied','parse_error','rate_limited','disabled')),
++ http_status bigint,
++ error_code text COLLATE "C",
++ detail text COLLATE "C",
++ bytes bigint,
++ raw_body_hash text COLLATE "C",
++ parser_version text COLLATE "C",
++ items_seen bigint NOT NULL DEFAULT 0,
++ items_created bigint NOT NULL DEFAULT 0,
++ items_updated bigint NOT NULL DEFAULT 0,
++ items_unchanged bigint NOT NULL DEFAULT 0,
++ items_skipped bigint NOT NULL DEFAULT 0
++);
++CREATE INDEX idx_fetch_runs_source ON fetch_runs (source_key, rid);
++
++CREATE TABLE items (
++ rid bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ id text COLLATE "C" NOT NULL UNIQUE,
++ source_key text COLLATE "C" NOT NULL,
++ category text COLLATE "C" NOT NULL,
++ identity text COLLATE "C" NOT NULL, -- guid / @id / url as the source gave it
++ kind text COLLATE "C" NOT NULL,
++ canonical_url text COLLATE "C",
++ title text COLLATE "C",
++ summary text COLLATE "C",
++ authors text COLLATE "C" NOT NULL DEFAULT '[]',
++ published_at text COLLATE "C",
++ source_updated_at text COLLATE "C",
++ fields text COLLATE "C" NOT NULL DEFAULT '{}',
++ content_hash text COLLATE "C" NOT NULL,
++ raw_body_hash text COLLATE "C",
++ parser_version text COLLATE "C" NOT NULL,
++ license_note text COLLATE "C",
++ terms_note text COLLATE "C",
++ first_seen_at bigint NOT NULL,
++ retrieved_at bigint NOT NULL, -- last successful fetch that contained it
++ revision bigint NOT NULL DEFAULT 1,
++ last_fetch_run_id text COLLATE "C",
++ entered_by text COLLATE "C", -- manual items: the principal that entered it
++ removed_at bigint,
++ removed_reason text COLLATE "C",
++ change_seq bigint NOT NULL,
++ UNIQUE (source_key, identity)
++);
++CREATE INDEX idx_items_change ON items (change_seq);
++CREATE INDEX idx_items_source ON items (source_key, change_seq);
++CREATE INDEX idx_items_category ON items (category, change_seq);
++
++CREATE TABLE item_revisions (
++ item_id text COLLATE "C" NOT NULL,
++ revision bigint NOT NULL,
++ content_hash text COLLATE "C" NOT NULL,
++ raw_body_hash text COLLATE "C",
++ parser_version text COLLATE "C" NOT NULL,
++ fetch_run_id text COLLATE "C",
++ retrieved_at bigint NOT NULL,
++ snapshot text COLLATE "C" NOT NULL,
++ PRIMARY KEY (item_id, revision)
++);
++
++CREATE TABLE counters (
++ name text COLLATE "C" PRIMARY KEY,
++ value bigint NOT NULL
++);
++
++CREATE TABLE robots_cache (
++ origin text COLLATE "C" PRIMARY KEY,
++ fetched_at bigint NOT NULL,
++ expires_at bigint NOT NULL,
++ status bigint,
++ outcome text COLLATE "C" NOT NULL, -- parsed | unavailable (4xx: allow) | unreachable (deny)
++ rules text COLLATE "C" NOT NULL DEFAULT '[]',
++ crawl_delay_sec double precision,
++ detail text COLLATE "C"
++);
++
++-- server/events/outbox.js ensureSchema()
++CREATE TABLE event_outbox (
++ seq bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ event_id text COLLATE "C" NOT NULL UNIQUE,
++ event_type text COLLATE "C" NOT NULL,
++ envelope text COLLATE "C" NOT NULL,
++ created_at bigint NOT NULL,
++ attempts bigint NOT NULL DEFAULT 0,
++ next_attempt_at bigint NOT NULL DEFAULT 0,
++ sent_at bigint,
++ rejected_at bigint,
++ last_error text COLLATE "C"
++);
++CREATE INDEX idx_event_outbox_due ON event_outbox(sent_at, rejected_at, next_attempt_at);
ddiff --git a/package-lock.json b/package-lock.json
iindex 2ff183c..55416ed 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -13,11 +13,13 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.25.0"
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.20.0",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
++ "pg": "^8.23.0"
},
"devDependencies": {
++ "@electric-sql/pglite": "^0.5.8",
"ajv": "^8.20.0",
"ajv-formats": "^3.0.1"
},
@@@ -25,6 +27,13 @@
"node": ">=22.0.0"
}
},
++ "node_modules/@electric-sql/pglite": {
++ "version": "0.5.8",
++ "resolved": "https://registry.npmjs.org/@electric-sql/pglite/-/pglite-0.5.8.tgz",
++ "integrity": "sha512-n9tsbUOhwx2epK1V0ZG9Ar4SHWUju04dhmzZXiSBXwBoleOvIfals33NAaWgagQVAL4Rbvx/Ptsu3P+pA09f6Q==",
++ "devOptional": true,
++ "license": "Apache-2.0"
++ },
"node_modules/accepts": {
"version": "1.3.8",
"resolved": "https://registry.npmjs.org/accepts/-/accepts-1.3.8.tgz",
@@@ -991,9 +1000,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.49.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
-- "integrity": "sha512-lapsLoBm3ZpndgiOMELdCBh2PJtuuyqj6SUhQsfwJkxnLdQfyhXfRFgiuLh+iQrg4kqtG9611191Mp8zAfk7fQ==",
++ "version": "0.76.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
++ "integrity": "sha512-zTdx776KWrcJKm1JTnf+0+BnsvswdvXy1ldPyse1hSqhnW/SSr98sX696eRetFUhTl8qVM1xSRCykqIUbqItPw==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
@@@ -1007,18 +1016,38 @@
}
},
"node_modules/openvibe-sdk": {
-- "version": "0.12.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0",
-- "integrity": "sha512-iG1h905O1f7y/PnP8xLA6GCVzXEBDfSaBj2tc1ucNreWg2sT/AJWosb8NW0WRMK5WS4sIQ6tMsprmDLj/fjjeQ==",
++ "version": "0.20.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.20.0",
++ "integrity": "sha512-CqZFfVuSv3ZPoyG1LuWCAdvo8TG2ZuUCeGaqTsxDKS3gsymSFV2QXPq/5QUotTd82WShTGUKKhTgcyp9ggUMfQ==",
"license": "MIT",
"engines": {
"node": ">=20"
++ },
++ "peerDependencies": {
++ "@electric-sql/pglite": "^0.5.8",
++ "better-sqlite3": ">=11",
++ "iovalkey": "^0.4.0",
++ "pg": "^8.23.0"
++ },
++ "peerDependenciesMeta": {
++ "@electric-sql/pglite": {
++ "optional": true
++ },
++ "better-sqlite3": {
++ "optional": true
++ },
++ "iovalkey": {
++ "optional": true
++ },
++ "pg": {
++ "optional": true
++ }
}
},
"node_modules/openvibe-shared": {
-- "version": "1.25.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.25.0",
-- "integrity": "sha512-oqZ+PvljrWi8IEbtg6bDYai17ZV2sALFx/ezMptniELSr10TgFUiGUQt+b1IWwiHv0T2Y1U2iSrkirJxMNgjSw==",
++ "version": "1.27.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
++ "integrity": "sha512-/H7dyFw+cURcdjyfbpyY7mNc6F+Pcb9TwkVwZHR+9d3M62q70KDhZLo/gZi71eaYxUjD8lO8tmbXu70EAa9paQ==",
"license": "MIT",
"dependencies": {
"jsonwebtoken": "^9.0.0"
@@@ -1039,6 +1068,134 @@
"integrity": "sha512-A/AGNMFN3c8bOlvV9RreMdrv7jsmF9XIfDeCd87+I8RNg6s78BhJxMu69NEMHBSJFxKidViTEdruRwEk/WIKqA==",
"license": "MIT"
},
++ "node_modules/pg": {
++ "version": "8.23.0",
++ "resolved": "https://registry.npmjs.org/pg/-/pg-8.23.0.tgz",
++ "integrity": "sha512-Ip2EQCngowJLGOfCwkFhPXU7/ljlhn6Rxlmy4XYfL2Y+vyRM59+8uR2xqRWKdYmbXmxCFOAmKxBuSUCdF34qLg==",
++ "license": "MIT",
++ "dependencies": {
++ "pg-connection-string": "^2.14.0",
++ "pg-pool": "^3.14.0",
++ "pg-protocol": "^1.16.0",
++ "pg-types": "2.2.0",
++ "pgpass": "1.0.5"
++ },
++ "engines": {
++ "node": ">= 16.0.0"
++ },
++ "optionalDependencies": {
++ "pg-cloudflare": "^1.4.0"
++ },
++ "peerDependencies": {
++ "pg-native": ">=3.0.1"
++ },
++ "peerDependenciesMeta": {
++ "pg-native": {
++ "optional": true
++ }
++ }
++ },
++ "node_modules/pg-cloudflare": {
++ "version": "1.4.0",
++ "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.0.tgz",
++ "integrity": "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==",
++ "license": "MIT",
++ "optional": true
++ },
++ "node_modules/pg-connection-string": {
++ "version": "2.14.0",
++ "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.0.tgz",
++ "integrity": "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==",
++ "license": "MIT"
++ },
++ "node_modules/pg-int8": {
++ "version": "1.0.1",
++ "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz",
++ "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==",
++ "license": "ISC",
++ "engines": {
++ "node": ">=4.0.0"
++ }
++ },
++ "node_modules/pg-pool": {
++ "version": "3.14.0",
++ "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz",
++ "integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==",
++ "license": "MIT",
++ "peerDependencies": {
++ "pg": ">=8.0"
++ }
++ },
++ "node_modules/pg-protocol": {
++ "version": "1.16.0",
++ "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.16.0.tgz",
++ "integrity": "sha512-sILXutLVjCLjcDuOmvhX5e2Z4cS5qG/6Bu3VkpFwdf/633ElGLpEh9bgmuI5I4sqKqkifQiGyiCcx1HdtrK7tg==",
++ "license": "MIT"
++ },
++ "node_modules/pg-types": {
++ "version": "2.2.0",
++ "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz",
++ "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==",
++ "license": "MIT",
++ "dependencies": {
++ "pg-int8": "1.0.1",
++ "postgres-array": "~2.0.0",
++ "postgres-bytea": "~1.0.0",
++ "postgres-date": "~1.0.4",
++ "postgres-interval": "^1.1.0"
++ },
++ "engines": {
++ "node": ">=4"
++ }
++ },
++ "node_modules/pgpass": {
++ "version": "1.0.5",
++ "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz",
++ "integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==",
++ "license": "MIT",
++ "dependencies": {
++ "split2": "^4.1.0"
++ }
++ },
++ "node_modules/postgres-array": {
++ "version": "2.0.0",
++ "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz",
++ "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==",
++ "license": "MIT",
++ "engines": {
++ "node": ">=4"
++ }
++ },
++ "node_modules/postgres-bytea": {
++ "version": "1.0.1",
++ "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.1.tgz",
++ "integrity": "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==",
++ "license": "MIT",
++ "engines": {
++ "node": ">=0.10.0"
++ }
++ },
++ "node_modules/postgres-date": {
++ "version": "1.0.7",
++ "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz",
++ "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==",
++ "license": "MIT",
++ "engines": {
++ "node": ">=0.10.0"
++ }
++ },
++ "node_modules/postgres-interval": {
++ "version": "1.2.0",
++ "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz",
++ "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==",
++ "license": "MIT",
++ "dependencies": {
++ "xtend": "^4.0.0"
++ },
++ "engines": {
++ "node": ">=0.10.0"
++ }
++ },
"node_modules/prebuild-install": {
"version": "7.1.3",
"resolved": "https://registry.npmjs.org/prebuild-install/-/prebuild-install-7.1.3.tgz",
@@@ -1377,6 +1534,15 @@
"simple-concat": "^1.0.0"
}
},
++ "node_modules/split2": {
++ "version": "4.2.0",
++ "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz",
++ "integrity": "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==",
++ "license": "ISC",
++ "engines": {
++ "node": ">= 10.x"
++ }
++ },
"node_modules/statuses": {
"version": "2.0.2",
"resolved": "https://registry.npmjs.org/statuses/-/statuses-2.0.2.tgz",
@@@ -1516,6 +1682,15 @@
"resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz",
"integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==",
"license": "ISC"
++ },
++ "node_modules/xtend": {
++ "version": "4.0.2",
++ "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz",
++ "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==",
++ "license": "MIT",
++ "engines": {
++ "node": ">=0.4"
++ }
}
}
}
ddiff --git a/package.json b/package.json
iindex c63cc34..c27ee5a 100644
---- a/package.json
++++ b/package.json
@@@ -8,7 +8,8 @@
"start": "node server/index.js",
"dev": "NODE_ENV=development node server/index.js",
"test": "node test/run.js",
-- "seed": "node scripts/seed.js"
++ "seed": "node scripts/seed.js",
++ "test:pg": "SOURCES_TEST_STORE=pg node test/run.js"
},
"keywords": [
"openvibe",
@@@ -27,15 +28,17 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.25.0"
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.20.0",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
++ "pg": "^8.23.0"
},
"engines": {
"node": ">=22.0.0"
},
"devDependencies": {
"ajv": "^8.20.0",
-- "ajv-formats": "^3.0.1"
++ "ajv-formats": "^3.0.1",
++ "@electric-sql/pglite": "^0.5.8"
}
}
ddiff --git a/scripts/migrate-to-postgres.js b/scripts/migrate-to-postgres.js
nnew file mode 100644
iindex 0000000..453ecbe
---- /dev/null
++++ b/scripts/migrate-to-postgres.js
@@@ -0,0 +1,26 @@
++#!/usr/bin/env node
++'use strict';
++/**
++ * The one-time move of Sources's SQLite database (SOURCES_DB_PATH) into its PostgreSQL schema (ADR-035; the procedure is
++ * openvibe-sdk docs/migrating-to-postgresql.md, section 6, carried out by openvibe-sdk/db runSqliteMigration).
++ *
++ * node scripts/migrate-to-postgres.js [--sqlite <file>] [--pglite] [--json]
++ *
++ * Applies migrations/ as the owner (DATABASE_DIRECT_URL), copies every table into emptied tables, verifies row counts
++ * and checksums, and exits 1 unless everything verified. The SQLite file is opened read-only. Every table keeps its
++ * name and columns.
++ */
++require('dotenv').config();
++const { runSqliteMigration } = require('openvibe-sdk/db');
++const configLib = require('../server/config');
++const { MIGRATIONS } = require('../server/db');
++
++const TABLES = {};
++
++if (require.main === module) {
++ const config = configLib.load();
++ runSqliteMigration({ service: 'sources', sqlite: config.dbPath, directUrl: config.db.directUrl, migrations: MIGRATIONS, tables: TABLES })
++ .then((code) => process.exit(code), (err) => { console.error(`migrate-to-postgres failed: ${err.message}`); process.exit(1); });
++}
++
++module.exports = { TABLES };
ddiff --git a/scripts/seed.js b/scripts/seed.js
iindex 7f9b824..ba42436 100644
---- a/scripts/seed.js
++++ b/scripts/seed.js
@@@ -5,7 +5,7 @@
* Never overwrites or enables an existing source. Enable one deliberately after verifying its terms:
* PATCH /api/v1/sources/:key { "enabled": true } (sources.source.manage).
*
-- * node scripts/seed.js [--db ./data/sources.db]
++ * node scripts/seed.js (DATABASE_URL from .env or /etc/openvibe/sources.env; without it, the development PGlite)
*/
require('dotenv').config();
const path = require('path');
@@@ -13,26 +13,25 @@ const { load } = require('../server/config');
const { openDb } = require('../server/db');
const { createRegistry } = require('../server/registry');
--function seed(db, { now = () => Date.now(), by = 'seed' } = {}) {
++async function seed(db, { now = () => Date.now(), by = 'seed' } = {}) {
const seeds = require(path.join(__dirname, '..', 'seeds', 'sources.json')).sources;
const registry = createRegistry({ db, now });
const out = { created: [], kept: [] };
for (const s of seeds) {
-- if (registry.get(s.key)) { out.kept.push(s.key); continue; }
-- registry.create({ ...s, enabled: false }, by);
++ if (await registry.get(s.key)) { out.kept.push(s.key); continue; }
++ await registry.create({ ...s, enabled: false }, by);
out.created.push(s.key);
}
return out;
}
if (require.main === module) {
-- const args = process.argv.slice(2);
-- const i = args.indexOf('--db');
-- const config = load();
-- const db = openDb(i >= 0 ? args[i + 1] : config.dbPath);
-- const r = seed(db);
-- console.log(`seeded ${r.created.length} source(s)${r.created.length ? `: ${r.created.join(', ')}` : ''}; kept ${r.kept.length} existing. All seeds are disabled.`);
-- db.close();
++ (async () => {
++ const db = await openDb(load()); // DATABASE_URL (PostgreSQL); without it, the development PGlite
++ const r = await seed(db);
++ console.log(`seeded ${r.created.length} source(s)${r.created.length ? `: ${r.created.join(', ')}` : ''}; kept ${r.kept.length} existing. All seeds are disabled.`);
++ await db.close();
++ })().catch((err) => { console.error(`[seed] ${err.message}`); process.exit(1); });
}
module.exports = { seed };
ddiff --git a/server/api/items.js b/server/api/items.js
iindex fb04ba6..8dc5e58 100644
---- a/server/api/items.js
++++ b/server/api/items.js
@@@ -54,17 +54,17 @@ function itemsRouter({ db, registry, items, auth, relay, now, limits }) {
const create = [auth.requireCap(CAPS.manage), limits.budget('sources.item.create')];
const remove = [auth.requireCap(CAPS.manage), limits.budget('sources.item.remove')];
-- router.get('/api/v1/items', read, (req, res) => {
++ router.get('/api/v1/items', read, async (req, res) => {
const source = req.query.source ? String(req.query.source) : null;
const category = req.query.category ? String(req.query.category) : null;
if (category && !CATEGORIES.includes(category)) return http.sendProblem(res, 400, 'sources.bad_query', { detail: `category must be one of ${CATEGORIES.join('|')}`, ctx: req.ov });
const after = /^\d{1,15}$/.test(String(req.query.after || '')) ? Number(req.query.after) : 0;
const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 100, 1), 500);
-- const page = items.list({ source, category, after, limit, includeRemoved: req.query.include_removed === '1' });
++ const page = await items.list({ source, category, after, limit, includeRemoved: req.query.include_removed === '1' });
const keys = [...new Set(page.items.map(i => i.source_key))];
const sources = {};
for (const k of keys) {
-- const row = registry.get(k);
++ const row = await registry.get(k);
if (row) {
const h = registry.health(row);
sources[k] = { status: h.status, stale: h.stale, last_success_at: h.last_success_at };
@@@ -73,16 +73,16 @@ function itemsRouter({ db, registry, items, auth, relay, now, limits }) {
res.json({ ...page, sources });
});
-- router.get('/api/v1/items/:id', read, (req, res) => {
-- const item = items.get(String(req.params.id), { revisions: req.query.revisions === '1' });
++ router.get('/api/v1/items/:id', read, async (req, res) => {
++ const item = await items.get(String(req.params.id), { revisions: req.query.revisions === '1' });
if (!item) return http.sendProblem(res, 404, 'sources.not_found', { detail: 'no such item', ctx: req.ov });
-- const row = registry.get(item.source_key);
++ const row = await registry.get(item.source_key);
const h = row ? registry.health(row) : null;
res.json({ item, source: h ? { key: item.source_key, status: h.status, stale: h.stale, last_success_at: h.last_success_at } : null });
});
-- router.post('/api/v1/sources/:key/items', create, (req, res) => {
-- const row = registry.get(req.params.key);
++ router.post('/api/v1/sources/:key/items', create, async (req, res) => {
++ const row = await registry.get(req.params.key);
if (!row) return http.sendProblem(res, 404, 'sources.not_found', { detail: 'no such source', ctx: req.ov });
if (row.type !== 'manual') return http.sendProblem(res, 409, 'sources.not_manual', { detail: 'items of fetched sources come from their fetches only', ctx: req.ov });
const source = registry.fromRow(row);
@@@ -90,21 +90,21 @@ function itemsRouter({ db, registry, items, auth, relay, now, limits }) {
const m = manualItem(req.body);
if (m.error) return http.sendProblem(res, 422, 'sources.bad_item', { detail: m.error, ctx: req.ov });
const t = now();
-- const counts = db.transaction(() => items.ingest(source, [m.item], { runId: null, rawBodyHash: null, parserVersion: MANUAL_VERSION, retrievedAt: t, enteredBy: req.principal.sub }))();
-- db.prepare('UPDATE sources SET last_success_at = ?, last_run_at = ? WHERE key = ?').run(t, t, source.key);
++ const counts = await db.tx(async () => await items.ingest(source, [m.item], { runId: null, rawBodyHash: null, parserVersion: MANUAL_VERSION, retrievedAt: t, enteredBy: req.principal.sub }));
++ await db.prepare('UPDATE sources SET last_success_at = ?, last_run_at = ? WHERE key = ?').run(t, t, source.key);
if (relay) relay.flush().catch(() => {});
-- const saved = db.prepare('SELECT id FROM items WHERE source_key = ? AND identity = ?').get(source.key, m.item.identity);
++ const saved = await db.prepare('SELECT id FROM items WHERE source_key = ? AND identity = ?').get(source.key, m.item.identity);
const outcome = counts.created ? 'created' : counts.updated ? 'updated' : counts.skipped ? 'removed' : 'unchanged';
-- res.status(counts.created ? 201 : 200).json({ outcome, item: items.get(saved.id) });
++ res.status(counts.created ? 201 : 200).json({ outcome, item: await items.get(saved.id) });
});
-- router.delete('/api/v1/items/:id', remove, (req, res) => {
++ router.delete('/api/v1/items/:id', remove, async (req, res) => {
const reason = toText(req.body && req.body.reason, 300);
if (!reason) return http.sendProblem(res, 422, 'sources.reason_required', { detail: 'a removal needs a reason (takedown, licence, error…)', ctx: req.ov });
-- const found = items.get(String(req.params.id));
++ const found = await items.get(String(req.params.id));
if (!found) return http.sendProblem(res, 404, 'sources.not_found', { detail: 'no such item', ctx: req.ov });
-- const source = registry.fromRow(registry.get(found.source_key));
-- const item = items.remove(source, found.id, reason, req.principal.sub);
++ const source = registry.fromRow(await registry.get(found.source_key));
++ const item = await items.remove(source, found.id, reason, req.principal.sub);
if (relay) relay.flush().catch(() => {});
res.json({ item });
});
ddiff --git a/server/api/sources.js b/server/api/sources.js
iindex 896d94b..bd761fa 100644
---- a/server/api/sources.js
++++ b/server/api/sources.js
@@@ -48,67 +48,67 @@ function sourcesRouter({ db, registry, ingest, auth, limits }) {
};
const notFound = (res, req) => http.sendProblem(res, 404, 'sources.not_found', { detail: 'no such source', ctx: req.ov });
-- router.get('/api/v1/sources', read, (req, res) => {
-- res.json({ sources: registry.all().map(registry.view) });
++ router.get('/api/v1/sources', read, async (req, res) => {
++ res.json({ sources: (await registry.all()).map(registry.view) });
});
-- router.get('/api/v1/sources/:key', read, (req, res) => {
-- const row = registry.get(req.params.key);
++ router.get('/api/v1/sources/:key', read, async (req, res) => {
++ const row = await registry.get(req.params.key);
if (!row) return notFound(res, req);
res.json({ source: registry.view(row) });
});
-- router.get('/api/v1/sources/:key/runs', read, (req, res) => {
-- if (!registry.get(req.params.key)) return notFound(res, req);
++ router.get('/api/v1/sources/:key/runs', read, async (req, res) => {
++ if (!await registry.get(req.params.key)) return notFound(res, req);
const before = /^\d{1,15}$/.test(String(req.query.before || '')) ? Number(req.query.before) : Number.MAX_SAFE_INTEGER;
const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 50, 1), 200);
-- const rows = db.prepare('SELECT * FROM fetch_runs WHERE source_key = ? AND rid < ? ORDER BY rid DESC LIMIT ?').all(req.params.key, before, limit);
++ const rows = await db.prepare('SELECT * FROM fetch_runs WHERE source_key = ? AND rid < ? ORDER BY rid DESC LIMIT ?').all(req.params.key, before, limit);
res.json({ runs: rows.map(runView), next_before: rows.length === limit ? rows[rows.length - 1].rid : null });
});
-- router.get('/api/v1/runs', read, (req, res) => {
++ router.get('/api/v1/runs', read, async (req, res) => {
const state = req.query.state ? String(req.query.state) : null;
if (state && state !== 'failed' && !STATES.includes(state)) return http.sendProblem(res, 400, 'sources.bad_query', { detail: `state must be failed or one of ${STATES.join('|')}`, ctx: req.ov });
const after = /^\d{1,15}$/.test(String(req.query.after || '')) ? Number(req.query.after) : 0;
const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 100, 1), 500);
-- const rows = db.prepare(`SELECT * FROM fetch_runs WHERE rid > @after AND (@source IS NULL OR source_key = @source)
-- AND (@state IS NULL OR state = @state OR (@state = 'failed' AND state IN ('http_error','timeout','robots_denied','parse_error','rate_limited')))
++ const rows = await db.prepare(`SELECT * FROM fetch_runs WHERE rid > @after AND (@source::text IS NULL OR source_key = @source)
++ AND (@state::text IS NULL OR state = @state OR (@state = 'failed' AND state IN ('http_error','timeout','robots_denied','parse_error','rate_limited')))
ORDER BY rid LIMIT @limit`).all({ after, source: req.query.source ? String(req.query.source) : null, state, limit });
res.json({ runs: rows.map(runView), next_after: rows.length ? rows[rows.length - 1].rid : after });
});
-- router.get('/api/v1/health', read, (req, res) => {
-- const views = registry.all().map(r => ({ key: r.key, category: r.category, type: r.type, ...registry.health(r) }));
++ router.get('/api/v1/health', read, async (req, res) => {
++ const views = (await registry.all()).map(r => ({ key: r.key, category: r.category, type: r.type, ...registry.health(r) }));
const counts = {};
for (const v of views) counts[v.status] = (counts[v.status] || 0) + 1;
res.json({ counts, attention: views.filter(v => ['failing', 'stale', 'never_fetched'].includes(v.status)) });
});
-- router.post('/api/v1/sources', manage, (req, res) => {
++ router.post('/api/v1/sources', manage, async (req, res) => {
try {
-- const row = registry.create(req.body, req.principal.sub);
++ const row = await registry.create(req.body, req.principal.sub);
res.status(201).json({ source: registry.view(row) });
} catch (err) { fail(res, req, err); }
});
-- router.patch('/api/v1/sources/:key', manage, (req, res) => {
++ router.patch('/api/v1/sources/:key', manage, async (req, res) => {
try {
-- const row = registry.patch(req.params.key, req.body, req.principal.sub);
++ const row = await registry.patch(req.params.key, req.body, req.principal.sub);
if (!row) return notFound(res, req);
res.json({ source: registry.view(row) });
} catch (err) { fail(res, req, err); }
});
-- router.delete('/api/v1/sources/:key', manage, (req, res) => {
++ router.delete('/api/v1/sources/:key', manage, async (req, res) => {
try {
-- const ok = db.transaction(() => {
-- const r = registry.remove(req.params.key);
++ const ok = await db.tx(async () => {
++ const r = await registry.remove(req.params.key);
if (r) {
-- db.prepare('DELETE FROM fetch_runs WHERE source_key = ?').run(req.params.key);
-- db.prepare('DELETE FROM endpoint_state WHERE source_key = ?').run(req.params.key);
++ await db.prepare('DELETE FROM fetch_runs WHERE source_key = ?').run(req.params.key);
++ await db.prepare('DELETE FROM endpoint_state WHERE source_key = ?').run(req.params.key);
}
return r;
-- })();
++ });
if (!ok) return notFound(res, req);
res.status(204).end();
} catch (err) { fail(res, req, err); }
ddiff --git a/server/auth.js b/server/auth.js
iindex 7408a79..4538609 100644
---- a/server/auth.js
++++ b/server/auth.js
@@@ -49,7 +49,7 @@ function createKeyStore({ urls = [], pem = null, fetchImpl = globalThis.fetch, l
return null;
}
-- function start() {
++ async function start() {
if (pem) return Promise.resolve(key);
const attempt = async () => {
const k = await fetchOnce();
@@@ -61,7 +61,7 @@ function createKeyStore({ urls = [], pem = null, fetchImpl = globalThis.fetch, l
};
refreshTimer = setInterval(() => { fetchOnce().catch(() => {}); }, 6 * 60 * 60 * 1000);
refreshTimer.unref?.();
-- return attempt();
++ return await attempt();
}
function stop() {
ddiff --git a/server/config.js b/server/config.js
iindex 3620985..59f6b78 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -44,6 +44,9 @@ function load(env = process.env) {
clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',
},
++ // PostgreSQL (ADR-035): DATABASE_URL serves (PgBouncer), DATABASE_DIRECT_URL migrates (owner role).
++ db: { url: env.DATABASE_URL || '', directUrl: env.DATABASE_DIRECT_URL || '' },
++ // The SQLite file of releases before the switch: read once by scripts/migrate-to-postgres.js.
dbPath: env.SOURCES_DB_PATH || './data/sources.db',
events: {
ddiff --git a/server/db.js b/server/db.js
iindex ff020af..6a7815c 100644
---- a/server/db.js
++++ b/server/db.js
@@@ -1,158 +1,37 @@
'use strict';
/**
-- * The Sources database (SQLite, WAL). One file owned by this service only; created on boot,
-- * idempotently.
++ * The Sources database: PostgreSQL (ADR-035), owned by this service only; the schema is migrations/NNNN_*.sql, applied
++ * at boot.
*/
const fs = require('fs');
const path = require('path');
--const Database = require('better-sqlite3');
--const outbox = require('./events/outbox');
++const { createDb } = require('openvibe-sdk/db');
--const SCHEMA = `
--CREATE TABLE IF NOT EXISTS sources (
-- key TEXT PRIMARY KEY,
-- name TEXT NOT NULL,
-- type TEXT NOT NULL, -- rss | atom | sitemap | jsonld | api | manual
-- category TEXT NOT NULL, -- news | blog | reviews | deals | coupons | trade
-- homepage_url TEXT,
-- endpoints TEXT NOT NULL DEFAULT '[]',
-- auth TEXT NOT NULL DEFAULT '{"mode":"none"}', -- env var NAME only, never a value
-- robots_note TEXT,
-- terms_note TEXT,
-- license_note TEXT,
-- min_interval_ms INTEGER NOT NULL, -- rate limit: gap between two requests to this source
-- poll_interval_sec INTEGER NOT NULL,
-- stale_after_sec INTEGER NOT NULL,
-- max_items INTEGER NOT NULL,
-- enabled INTEGER NOT NULL DEFAULT 0,
-- review_required INTEGER NOT NULL DEFAULT 1,
-- sensitivity TEXT NOT NULL DEFAULT 'none',
-- default_indexability TEXT NOT NULL DEFAULT 'noindex',
-- search_visibility TEXT, -- null = items are not sent to OpenVibe.Search
-- created_at INTEGER NOT NULL,
-- updated_at INTEGER NOT NULL,
-- updated_by TEXT,
-- -- runtime state
-- next_due_at INTEGER NOT NULL DEFAULT 0,
-- not_before INTEGER NOT NULL DEFAULT 0, -- Retry-After / Crawl-delay
-- last_request_at INTEGER,
-- last_run_at INTEGER,
-- last_success_at INTEGER,
-- last_state TEXT,
-- consecutive_failures INTEGER NOT NULL DEFAULT 0
--);
++const MIGRATIONS = path.join(__dirname, '..', 'migrations');
++const DEV_PGLITE = path.join(__dirname, '..', 'data', 'pglite');
--CREATE TABLE IF NOT EXISTS endpoint_state (
-- source_key TEXT NOT NULL,
-- url TEXT NOT NULL,
-- etag TEXT,
-- last_modified TEXT,
-- last_status INTEGER,
-- last_fetch_at INTEGER,
-- PRIMARY KEY (source_key, url)
--);
--
---- One row per fetch attempt, whatever happened. state is never empty.
--CREATE TABLE IF NOT EXISTS fetch_runs (
-- rid INTEGER PRIMARY KEY AUTOINCREMENT,
-- id TEXT NOT NULL UNIQUE,
-- source_key TEXT NOT NULL,
-- endpoint_url TEXT,
-- trigger TEXT NOT NULL, -- schedule | manual
-- started_at INTEGER NOT NULL,
-- finished_at INTEGER NOT NULL,
-- state TEXT NOT NULL CHECK (state IN ('ok','not_modified','http_error','timeout','robots_denied','parse_error','rate_limited','disabled')),
-- http_status INTEGER,
-- error_code TEXT,
-- detail TEXT,
-- bytes INTEGER,
-- raw_body_hash TEXT,
-- parser_version TEXT,
-- items_seen INTEGER NOT NULL DEFAULT 0,
-- items_created INTEGER NOT NULL DEFAULT 0,
-- items_updated INTEGER NOT NULL DEFAULT 0,
-- items_unchanged INTEGER NOT NULL DEFAULT 0,
-- items_skipped INTEGER NOT NULL DEFAULT 0
--);
--CREATE INDEX IF NOT EXISTS idx_fetch_runs_source ON fetch_runs (source_key, rid);
--
--CREATE TABLE IF NOT EXISTS items (
-- rid INTEGER PRIMARY KEY AUTOINCREMENT,
-- id TEXT NOT NULL UNIQUE,
-- source_key TEXT NOT NULL,
-- category TEXT NOT NULL,
-- identity TEXT NOT NULL, -- guid / @id / url as the source gave it
-- kind TEXT NOT NULL,
-- canonical_url TEXT,
-- title TEXT,
-- summary TEXT,
-- authors TEXT NOT NULL DEFAULT '[]',
-- published_at TEXT,
-- source_updated_at TEXT,
-- fields TEXT NOT NULL DEFAULT '{}',
-- content_hash TEXT NOT NULL,
-- raw_body_hash TEXT,
-- parser_version TEXT NOT NULL,
-- license_note TEXT,
-- terms_note TEXT,
-- first_seen_at INTEGER NOT NULL,
-- retrieved_at INTEGER NOT NULL, -- last successful fetch that contained it
-- revision INTEGER NOT NULL DEFAULT 1,
-- last_fetch_run_id TEXT,
-- entered_by TEXT, -- manual items: the principal that entered it
-- removed_at INTEGER,
-- removed_reason TEXT,
-- change_seq INTEGER NOT NULL,
-- UNIQUE (source_key, identity)
--);
--CREATE INDEX IF NOT EXISTS idx_items_change ON items (change_seq);
--CREATE INDEX IF NOT EXISTS idx_items_source ON items (source_key, change_seq);
--CREATE INDEX IF NOT EXISTS idx_items_category ON items (category, change_seq);
--
--CREATE TABLE IF NOT EXISTS item_revisions (
-- item_id TEXT NOT NULL,
-- revision INTEGER NOT NULL,
-- content_hash TEXT NOT NULL,
-- raw_body_hash TEXT,
-- parser_version TEXT NOT NULL,
-- fetch_run_id TEXT,
-- retrieved_at INTEGER NOT NULL,
-- snapshot TEXT NOT NULL,
-- PRIMARY KEY (item_id, revision)
--);
--
--CREATE TABLE IF NOT EXISTS counters (
-- name TEXT PRIMARY KEY,
-- value INTEGER NOT NULL
--);
--
--CREATE TABLE IF NOT EXISTS robots_cache (
-- origin TEXT PRIMARY KEY,
-- fetched_at INTEGER NOT NULL,
-- expires_at INTEGER NOT NULL,
-- status INTEGER,
-- outcome TEXT NOT NULL, -- parsed | unavailable (4xx: allow) | unreachable (deny)
-- rules TEXT NOT NULL DEFAULT '[]',
-- crawl_delay_sec REAL,
-- detail TEXT
--);
--`;
--
--function openDb(dbPath) {
-- if (dbPath !== ':memory:') fs.mkdirSync(path.dirname(path.resolve(dbPath)), { recursive: true });
-- const db = new Database(dbPath);
-- db.pragma('journal_mode = WAL');
-- db.pragma('synchronous = NORMAL');
-- db.pragma('busy_timeout = 5000');
-- db.exec(SCHEMA);
-- outbox.ensureSchema(db);
-- return db;
++/**
++ * The serving handle (ADR-035): DATABASE_URL through PgBouncer; in development without it, an embedded PGlite database
++ * in data/pglite. Migrations run first, as the owner (DATABASE_DIRECT_URL), or on the embedded handle.
++ */
++async function openDb(config, { log = console, registry } = {}) {
++ if (!config.db.url) {
++ if (config.isProduction) throw new Error('DATABASE_URL is not set: production serves from PostgreSQL (OpenVibe.Host roles/data add-service.sh sources)');
++ log.warn(`[sources] DATABASE_URL unset: embedded PGlite database in ${DEV_PGLITE} (development only, one process)`);
++ fs.mkdirSync(DEV_PGLITE, { recursive: true });
++ const db = createDb({ pglite: DEV_PGLITE, service: 'sources', registry, log });
++ await db.migrate({ dir: MIGRATIONS, log });
++ return db;
++ }
++ if (!config.db.directUrl) throw new Error('DATABASE_DIRECT_URL is not set: migrations run with the owner role on a direct connection');
++ const owner = createDb({ url: config.db.directUrl, service: 'sources-migrate', max: 1, log });
++ try { await owner.migrate({ dir: MIGRATIONS, log }); } finally { await owner.close(); }
++ return createDb({ url: config.db.url, service: 'sources', registry, log });
}
/** Next value of a named monotonic counter (call inside a transaction). */
--function nextSeq(db, name) {
-- db.prepare('INSERT INTO counters (name, value) VALUES (?, 1) ON CONFLICT(name) DO UPDATE SET value = value + 1').run(name);
-- return db.prepare('SELECT value FROM counters WHERE name = ?').get(name).value;
++async function nextSeq(db, name) {
++ return await db.prepare('INSERT INTO counters (name, value) VALUES (?, 1) ON CONFLICT (name) DO UPDATE SET value = counters.value + 1 RETURNING counters.value').pluck().get(name);
}
--module.exports = { openDb, nextSeq };
++module.exports = { openDb, nextSeq, MIGRATIONS };
ddiff --git a/server/events/outbox.js b/server/events/outbox.js
iindex e056703..7b3943c 100644
---- a/server/events/outbox.js
++++ b/server/events/outbox.js
@@@ -36,8 +36,8 @@ function createOutbox(db, { source, now = () => Date.now() }) {
const insert = db.prepare('INSERT INTO event_outbox (event_id, event_type, envelope, created_at) VALUES (?, ?, ?, ?)');
/** { event_type, subject, payload, visibility?, priority?, actor?, trace_id? } → envelope */
-- function enqueue({ event_type, subject, payload, visibility = 'internal', priority = 'important', actor, trace_id }) {
-- if (!db.inTransaction) throw new Error('outbox.enqueue() must run inside the transaction that makes the change');
++ async function enqueue({ event_type, subject, payload, visibility = 'internal', priority = 'important', actor, trace_id }) {
++ if (!db.inTransaction()) throw new Error('outbox.enqueue() must run inside the transaction that makes the change');
const ms = now();
const env = {
event_id: ids.newId('event', ms),
@@@ -54,7 +54,7 @@ function createOutbox(db, { source, now = () => Date.now() }) {
if (trace_id && /^[0-9a-f]{32}$/.test(trace_id)) env.trace_id = trace_id;
const v = validate('events.event-envelope@1', env);
if (!v.valid) throw new Error(`outbox: invalid envelope for ${event_type}: ${v.errors.map(e => `${e.path} ${e.message}`).join('; ')}`);
-- insert.run(env.event_id, event_type, JSON.stringify(env), ms);
++ await insert.run(env.event_id, event_type, JSON.stringify(env), ms);
return env;
}
@@@ -71,11 +71,11 @@ function createOutbox(db, { source, now = () => Date.now() }) {
return {
enqueue,
-- pending: () => q.pending.get().n,
-- rejected: () => q.rejectedCount.get().n,
++ pending: async () => (await q.pending.get()).n,
++ rejected: async () => (await q.rejectedCount.get()).n,
/** Every envelope in order (tests and operators). */
-- all: () => q.list.all().map(r => JSON.parse(r.envelope)),
-- prune: (olderThanMs = 7 * 24 * 3600 * 1000) => q.prune.run(now() - olderThanMs).changes,
++ all: async () => (await q.list.all()).map(r => JSON.parse(r.envelope)),
++ prune: async (olderThanMs = 7 * 24 * 3600 * 1000) => (await q.prune.run(now() - olderThanMs)).changes,
_q: q,
};
}
@@@ -109,18 +109,18 @@ function createRelay({ outbox, eventsUrl, intervalMs = 2000, tokenClient, tokenO
return { ok: res.ok, status: res.status, text };
}
-- function mark(rows, fn) { db.transaction(() => { for (const r of rows) fn(r); })(); }
++ async function mark(rows, fn) { await db.tx(async () => { for (const r of rows) await fn(r); }); }
async function publishRows(rows) {
let r;
try {
r = await post(rows);
} catch (err) {
-- mark(rows, row => q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err.message).slice(0, 500), row.seq));
++ await mark(rows, async row => await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err.message).slice(0, 500), row.seq));
return { sent: 0, failed: rows.length, rejected: 0 };
}
if (r.ok) {
-- mark(rows, row => q.sent.run(now(), row.seq));
++ await mark(rows, async row => await q.sent.run(now(), row.seq));
return { sent: rows.length, failed: 0, rejected: 0 };
}
if (permanent(r.status) && rows.length > 1) {
@@@ -134,11 +134,11 @@ function createRelay({ outbox, eventsUrl, intervalMs = 2000, tokenClient, tokenO
}
const msg = `${r.status} ${r.text.slice(0, 300)}`;
if (permanent(r.status)) {
-- mark(rows, row => q.rejected.run(now(), msg, row.seq));
++ await mark(rows, async row => await q.rejected.run(now(), msg, row.seq));
log.warn(`[outbox] event refused by Events: ${msg}`);
return { sent: 0, failed: 0, rejected: rows.length };
}
-- mark(rows, row => q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], msg, row.seq));
++ await mark(rows, async row => await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], msg, row.seq));
return { sent: 0, failed: rows.length, rejected: 0 };
}
@@@ -146,7 +146,7 @@ function createRelay({ outbox, eventsUrl, intervalMs = 2000, tokenClient, tokenO
const total = { sent: 0, failed: 0, rejected: 0 };
if (!eventsUrl || !tokens) return total;
for (;;) {
-- const rows = q.due.all(now(), 50);
++ const rows = await q.due.all(now(), 50);
if (!rows.length) break;
const s = await publishRows(rows);
total.sent += s.sent; total.failed += s.failed; total.rejected += s.rejected;
ddiff --git a/server/index.js b/server/index.js
iindex 6b39ddf..7724190 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -21,9 +21,10 @@ const { createOutbox, createRelay } = require('./events/outbox');
const { createKeyStore, createAuth } = require('./auth');
const { createApp } = require('./app');
--async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fetch, tokenClient, lookupImpl, log = console, listen = true, limitsNow = null } = {}) {
++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, tokenClient, lookupImpl, log = console, listen = true, limitsNow = null } = {}) {
config = config || load();
-- const db = openDb(config.dbPath);
++ // PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
++ const db = givenDb || await openDb(config, { log });
const outbox = createOutbox(db, { source: config.serviceId, now });
const relay = createRelay({
db, outbox, eventsUrl: config.events.url, intervalMs: config.events.relayIntervalMs, fetchImpl, log, now,
@@@ -48,7 +49,7 @@ async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fe
const keyLoaded = keys.start().catch(() => null);
relay.start();
if (config.worker.enabled) scheduler.start();
-- const pruneTimer = setInterval(() => { try { outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
++ const pruneTimer = setInterval(async () => { try { await outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
pruneTimer.unref?.();
let server = null;
@@@ -69,7 +70,7 @@ async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fe
server.closeAllConnections?.();
await new Promise(resolve => server.close(() => resolve()));
}
-- db.close();
++ if (!givenDb) await db.close();
}
return { config, db, registry, items, guard, fetcher, robots, ingest, scheduler, outbox, relay, keys, keyLoaded, auth, app, server, close };
ddiff --git a/server/ingest.js b/server/ingest.js
iindex 0bef613..07b2165 100644
---- a/server/ingest.js
++++ b/server/ingest.js
@@@ -72,8 +72,8 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
} catch { return null; }
}
-- function failedEvent(source, run, failures) {
-- outbox.enqueue({
++ async function failedEvent(source, run, failures) {
++ await outbox.enqueue({
event_type: 'sources.fetch.failed',
subject: { type: 'source', id: source.key },
payload: {
@@@ -84,12 +84,12 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
}
/** Record a run that changed no item (every non-ok outcome, and 304). */
-- function recordNoItems(source, run, { countsAsFailure, failures }) {
-- db.transaction(() => {
-- st.insertRun.run(run);
-- if (run.endpoint_url && run.http_status != null) st.touchEndpoint.run({ k: source.key, u: run.endpoint_url, status: run.http_status, at: run.finished_at });
-- if (countsAsFailure) failedEvent(source, run, failures);
-- })();
++ async function recordNoItems(source, run, { countsAsFailure, failures }) {
++ await db.tx(async () => {
++ await st.insertRun.run(run);
++ if (run.endpoint_url && run.http_status != null) await st.touchEndpoint.run({ k: source.key, u: run.endpoint_url, status: run.http_status, at: run.finished_at });
++ if (countsAsFailure) await failedEvent(source, run, failures);
++ });
return run;
}
@@@ -126,9 +126,9 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
return { run, failed: true };
}
-- const validators = st.endpoint.get(source.key, endpoint.url) || {};
++ const validators = await st.endpoint.get(source.key, endpoint.url) || {};
await space(host, source.min_interval_ms);
-- st.requested.run(now(), source.key);
++ await st.requested.run(now(), source.key);
let res;
try {
res = await fetcher.fetchUrl(requestUrl.toString(), {
@@@ -185,20 +185,20 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
}
// The only path that touches items: the run row, items, validators and events commit together.
-- db.transaction(() => {
-- const counts = items.ingest(source, parsed.items, {
++ await db.tx(async () => {
++ const counts = await items.ingest(source, parsed.items, {
runId: run.id, rawBodyHash: run.raw_body_hash, parserVersion: adapter.version, retrievedAt: run.finished_at,
});
Object.assign(run, {
state: 'ok', items_seen: parsed.items.length, items_created: counts.created, items_updated: counts.updated,
items_unchanged: counts.unchanged, items_skipped: parsed.skipped + counts.skipped,
});
-- st.insertRun.run(run);
-- st.saveValidators.run({
++ await st.insertRun.run(run);
++ await st.saveValidators.run({
k: source.key, u: endpoint.url, etag: res.headers.etag || null, lm: res.headers['last-modified'] || null,
status: res.status, at: run.finished_at,
});
-- })();
++ });
return { run, failed: false, success: true, recorded: true };
}
@@@ -208,7 +208,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
* run(key, { trigger }) → { runs: [...] } | { busy: true } | { manual: true } | null (unknown)
*/
async function run(key, { trigger = 'schedule' } = {}) {
-- const row = registry.get(key);
++ const row = await registry.get(key);
if (!row) return null;
if (row.type === 'manual') return { manual: true };
if (inflight.has(key)) return { busy: true };
@@@ -221,7 +221,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
if (!source.enabled) {
const r = baseRun(source, null, trigger, t);
Object.assign(r, { state: 'disabled', error_code: 'source_disabled', detail: 'the source is disabled' });
-- runs.push(recordNoItems(source, r, { countsAsFailure: false }));
++ runs.push(await recordNoItems(source, r, { countsAsFailure: false }));
return { runs };
}
const sinceLast = row.last_request_at == null ? Infinity : t - row.last_request_at;
@@@ -229,7 +229,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
const r = baseRun(source, null, trigger, t);
const until = Math.max(row.not_before, (row.last_request_at || 0) + source.min_interval_ms);
Object.assign(r, { state: 'rate_limited', error_code: 'local_rate_limit', detail: `next request allowed at ${new Date(until).toISOString()}` });
-- runs.push(recordNoItems(source, r, { countsAsFailure: false }));
++ runs.push(await recordNoItems(source, r, { countsAsFailure: false }));
return { runs, rateLimitedUntil: until };
}
@@@ -251,7 +251,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
if (out.failed) { anyFailure = true; failures = row.consecutive_failures + 1; }
if (out.success) anySuccess = true;
if (out.notBefore) notBefore = Math.max(notBefore, out.notBefore);
-- if (!out.recorded) recordNoItems(source, out.run, { countsAsFailure: out.failed, failures });
++ if (!out.recorded) await recordNoItems(source, out.run, { countsAsFailure: out.failed, failures });
runs.push(out.run);
if (out.run.state === 'rate_limited') break; // the site asked us to stop
}
@@@ -262,7 +262,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, ou
? Math.min(source.poll_interval_sec * 1000 * 2 ** Math.min(newFailures - 1, 10), Math.max(config.worker.maxBackoffMs, source.poll_interval_sec * 1000))
: source.poll_interval_sec * 1000;
const lastState = runs.map(r => r.state).find(s => FAILURE_STATES.has(s)) || runs[runs.length - 1].state;
-- st.finish.run({
++ await st.finish.run({
key, at: finishedAt, state: lastState, failures: newFailures,
success_at: anySuccess ? finishedAt : null, next_due: finishedAt + backoff, not_before: notBefore,
});
ddiff --git a/server/items.js b/server/items.js
iindex bf9ff3e..a1c2c25 100644
---- a/server/items.js
++++ b/server/items.js
@@@ -84,8 +84,8 @@ function createItems({ db, outbox, now = () => Date.now() }) {
return { type: 'item', id: row.id, revision: row.revision };
}
-- function itemEvent(type, row, extra = {}) {
-- outbox.enqueue({
++ async function itemEvent(type, row, extra = {}) {
++ await outbox.enqueue({
event_type: type,
subject: subjectOf(row),
payload: {
@@@ -121,9 +121,9 @@ function createItems({ db, outbox, now = () => Date.now() }) {
return doc;
}
-- function indexEvent(source, row, deleted = false) {
++ async function indexEvent(source, row, deleted = false) {
if (!source.search_visibility) return;
-- outbox.enqueue({
++ await outbox.enqueue({
event_type: deleted ? 'sources.index_document.deleted' : 'sources.index_document.upserted',
subject: subjectOf(row),
payload: deleted ? { type: 'item', id: row.id, revision: row.revision } : indexDocument(row),
@@@ -161,31 +161,31 @@ function createItems({ db, outbox, now = () => Date.now() }) {
* Apply one successful fetch's parsed items. Must run inside the transaction that also writes
* the fetch run row. → { created, updated, unchanged, skipped }
*/
-- function ingest(source, parsedItems, ctx) {
-- if (!db.inTransaction) throw new Error('items.ingest() must run inside the fetch run transaction');
++ async function ingest(source, parsedItems, ctx) {
++ if (!db.inTransaction()) throw new Error('items.ingest() must run inside the fetch run transaction');
const counts = { created: 0, updated: 0, unchanged: 0, skipped: 0 };
for (const it of parsedItems) {
-- const cur = st.byIdentity.get(source.key, it.identity);
++ const cur = await st.byIdentity.get(source.key, it.identity);
const v = values(source, it, ctx);
if (!cur) {
-- const row = { id: `itm_${ids.ulid(ctx.retrievedAt)}`, source_key: source.key, category: source.category, identity: it.identity, first_seen_at: ctx.retrievedAt, entered_by: ctx.enteredBy || null, change_seq: nextSeq(db, 'items'), ...v };
-- st.insert.run(row);
-- const saved = st.byId.get(row.id);
-- st.revision.run(saved.id, 1, saved.content_hash, saved.raw_body_hash, saved.parser_version, ctx.runId || null, ctx.retrievedAt, snapshot(saved));
-- itemEvent('sources.item.created', saved);
-- indexEvent(source, saved);
++ const row = { id: `itm_${ids.ulid(ctx.retrievedAt)}`, source_key: source.key, category: source.category, identity: it.identity, first_seen_at: ctx.retrievedAt, entered_by: ctx.enteredBy || null, change_seq: await nextSeq(db, 'items'), ...v };
++ await st.insert.run(row);
++ const saved = await st.byId.get(row.id);
++ await st.revision.run(saved.id, 1, saved.content_hash, saved.raw_body_hash, saved.parser_version, ctx.runId || null, ctx.retrievedAt, snapshot(saved));
++ await itemEvent('sources.item.created', saved);
++ await indexEvent(source, saved);
counts.created++;
} else if (cur.removed_at) {
counts.skipped++; // removal is sticky
} else if (cur.content_hash !== v.content_hash) {
-- st.update.run({ ...v, id: cur.id, change_seq: nextSeq(db, 'items') });
-- const saved = st.byId.get(cur.id);
-- st.revision.run(saved.id, saved.revision, saved.content_hash, saved.raw_body_hash, saved.parser_version, ctx.runId || null, ctx.retrievedAt, snapshot(saved));
-- itemEvent('sources.item.updated', saved, { previous_content_hash: cur.content_hash });
-- indexEvent(source, saved);
++ await st.update.run({ ...v, id: cur.id, change_seq: await nextSeq(db, 'items') });
++ const saved = await st.byId.get(cur.id);
++ await st.revision.run(saved.id, saved.revision, saved.content_hash, saved.raw_body_hash, saved.parser_version, ctx.runId || null, ctx.retrievedAt, snapshot(saved));
++ await itemEvent('sources.item.updated', saved, { previous_content_hash: cur.content_hash });
++ await indexEvent(source, saved);
counts.updated++;
} else {
-- st.seen.run(ctx.retrievedAt, ctx.runId || null, cur.id);
++ await st.seen.run(ctx.retrievedAt, ctx.runId || null, cur.id);
counts.unchanged++;
}
}
@@@ -193,27 +193,27 @@ function createItems({ db, outbox, now = () => Date.now() }) {
}
/** Remove an item (takedown, licence, error). Sticky; emits sources.item.removed. */
-- function remove(source, id, reason, by) {
-- return db.transaction(() => {
-- const cur = st.byId.get(id);
++ async function remove(source, id, reason, by) {
++ return await db.tx(async () => {
++ const cur = await st.byId.get(id);
if (!cur || cur.source_key !== source.key) return null;
if (cur.removed_at) return view(cur);
const t = now();
-- st.remove.run(t, `${reason}${by ? ` (by ${by})` : ''}`.slice(0, 500), nextSeq(db, 'items'), id);
-- const saved = st.byId.get(id);
-- st.revision.run(saved.id, saved.revision, saved.content_hash, saved.raw_body_hash, saved.parser_version, null, t, JSON.stringify({ removed: true, reason }));
-- itemEvent('sources.item.removed', saved, { reason });
-- indexEvent(source, saved, true);
++ await st.remove.run(t, `${reason}${by ? ` (by ${by})` : ''}`.slice(0, 500), await nextSeq(db, 'items'), id);
++ const saved = await st.byId.get(id);
++ await st.revision.run(saved.id, saved.revision, saved.content_hash, saved.raw_body_hash, saved.parser_version, null, t, JSON.stringify({ removed: true, reason }));
++ await itemEvent('sources.item.removed', saved, { reason });
++ await indexEvent(source, saved, true);
return view(saved);
-- })();
++ });
}
-- function get(id, { revisions = false } = {}) {
-- const row = st.byId.get(id);
++ async function get(id, { revisions = false } = {}) {
++ const row = await st.byId.get(id);
if (!row) return null;
const out = view(row);
if (revisions) {
-- out.revisions = st.revisions.all(id).map(r => ({
++ out.revisions = (await st.revisions.all(id)).map(r => ({
revision: r.revision, content_hash: r.content_hash, raw_body_hash: r.raw_body_hash, parser_version: r.parser_version,
fetch_run_id: r.fetch_run_id, retrieved_at: new Date(r.retrieved_at).toISOString(), snapshot: JSON.parse(r.snapshot),
}));
@@@ -222,9 +222,9 @@ function createItems({ db, outbox, now = () => Date.now() }) {
}
/** Items in change order (created, updated or removed after `after`). */
-- function list({ source = null, category = null, after = 0, limit = 100, includeRemoved = false } = {}) {
-- const rows = db.prepare(`SELECT * FROM items WHERE change_seq > @after
-- AND (@source IS NULL OR source_key = @source) AND (@category IS NULL OR category = @category)
++ async function list({ source = null, category = null, after = 0, limit = 100, includeRemoved = false } = {}) {
++ const rows = await db.prepare(`SELECT * FROM items WHERE change_seq > @after
++ AND (@source::text IS NULL OR source_key = @source) AND (@category::text IS NULL OR category = @category)
AND (@incl = 1 OR removed_at IS NULL)
ORDER BY change_seq LIMIT @limit`).all({ after, source, category, incl: includeRemoved ? 1 : 0, limit: limit + 1 });
const more = rows.length > limit;
ddiff --git a/server/observability.js b/server/observability.js
iindex c259139..2b83647 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -22,20 +22,20 @@ const STATUSES = ['healthy', 'stale', 'failing', 'never_fetched', 'disabled', 'm
function readers(db, { startedAt }) {
// A source waits from when it became due, when it was last changed (a new or re-enabled source
// is due at once, from next_due_at 0) or when this process started, whichever is latest.
-- const due = db.prepare(`SELECT COUNT(*) AS n, MIN(MAX(next_due_at, not_before, updated_at)) AS oldest FROM sources
++ const due = db.prepare(`SELECT COUNT(*) AS n, MIN(GREATEST(next_due_at, not_before, updated_at)) AS oldest FROM sources
WHERE enabled = 1 AND type != 'manual' AND next_due_at <= @now AND not_before <= @now`);
const lastRun = db.prepare('SELECT finished_at FROM fetch_runs ORDER BY rid DESC LIMIT 1');
const lastSuccess = db.prepare('SELECT MAX(last_success_at) AS t FROM sources');
const items = db.prepare('SELECT COUNT(*) AS n, COUNT(removed_at) AS removed FROM items');
return {
/** Enabled fetched sources whose turn has come, and how long the oldest has waited. */
-- queue(t) {
-- const r = due.get({ now: t });
++ async queue(t) {
++ const r = await due.get({ now: t });
return { due: r.n, oldest_wait_ms: r.n ? Math.max(0, t - Math.max(r.oldest, startedAt)) : 0 };
},
-- lastFetchAt: () => { const r = lastRun.get(); return r ? r.finished_at : null; },
-- lastSuccessAt: () => lastSuccess.get().t,
-- items: () => { const r = items.get(); return { current: r.n - r.removed, removed: r.removed }; },
++ lastFetchAt: async () => { const r = await lastRun.get(); return r ? r.finished_at : null; },
++ lastSuccessAt: async () => (await lastSuccess.get()).t,
++ items: async () => { const r = await items.get(); return { current: r.n - r.removed, removed: r.removed }; },
};
}
@@@ -46,16 +46,17 @@ function createSourcesReadiness({ db, keys, config, registry, ingest, scheduler,
service: 'sources',
release,
checks: [
-- { name: 'db', required: true, check: () => { db.prepare('SELECT COUNT(*) AS n FROM sources').get(); return true; } },
++ // A real round trip that names the store (postgresql / pglite), and the schema present.
++ { name: 'db', required: true, check: async () => { const r = await db.ready(); if (!r.ok) return r.error; await db.prepare('SELECT COUNT(*) AS n FROM sources').get(); return { ok: true, detail: r.detail }; } },
{ name: 'network_jwks', required: false, check: () => keys.loaded() || 'Network signing key not loaded yet: no service token can be verified' },
{
name: 'fetcher', required: false,
-- check: () => {
-- const q = read.queue(now());
++ check: async () => {
++ const q = await read.queue(now());
const detail = {
enabled: config.worker.enabled, running: scheduler.running(), in_flight: ingest.inflight().length,
max_concurrent: config.worker.maxConcurrent, due: q.due, oldest_wait_ms: q.oldest_wait_ms,
-- last_fetch_at: iso(read.lastFetchAt()),
++ last_fetch_at: iso(await read.lastFetchAt()),
};
if (!config.worker.enabled) return { ok: false, error: 'the ingestion worker is off (SOURCES_WORKER=off): nothing is fetched', detail };
if (!scheduler.running()) return { ok: false, error: 'the ingestion worker is not running', detail };
@@@ -64,12 +65,12 @@ function createSourcesReadiness({ db, keys, config, registry, ingest, scheduler,
},
},
],
-- details: (body) => {
++ details: async (body) => {
const dbOk = body.checks.db.status === 'ok';
let sources = null;
if (dbOk) {
sources = {};
-- for (const r of registry.all()) {
++ for (const r of await registry.all()) {
const s = registry.health(r).status;
sources[s] = (sources[s] || 0) + 1;
}
@@@ -77,7 +78,7 @@ function createSourcesReadiness({ db, keys, config, registry, ingest, scheduler,
return {
sources,
runs_in_flight: ingest.inflight().length,
-- outbox: dbOk ? { pending: outbox.pending(), rejected: outbox.rejected(), relay: config.events.url ? (relay.running() ? 'running' : 'stopped') : 'off (EVENTS_URL unset)' } : null,
++ outbox: dbOk ? { pending: await outbox.pending(), rejected: await outbox.rejected(), relay: config.events.url ? (relay.running() ? 'running' : 'stopped') : 'off (EVENTS_URL unset)' } : null,
};
},
});
@@@ -89,23 +90,23 @@ function registerSourcesGauges(registry, { db, sources, ingest, outbox, now }) {
const seconds = (ms) => (ms == null ? null : ms / 1000);
registry.gauge({
name: 'sources_sources', help: 'Registered sources by health status', labelNames: ['status'],
-- collect: () => {
++ collect: async () => {
const n = Object.fromEntries(STATUSES.map((s) => [s, 0]));
-- for (const r of sources.all()) { const s = sources.health(r).status; n[s] = (n[s] || 0) + 1; }
++ for (const r of await sources.all()) { const s = sources.health(r).status; n[s] = (n[s] || 0) + 1; }
return Object.entries(n).map(([status, value]) => ({ labels: { status }, value }));
},
});
registry.gauge({
name: 'sources_items', help: 'Items held, current and removed', labelNames: ['state'],
-- collect: () => Object.entries(read.items()).map(([state, value]) => ({ labels: { state }, value })),
++ collect: async () => Object.entries(await read.items()).map(([state, value]) => ({ labels: { state }, value })),
});
-- registry.gauge({ name: 'sources_fetch_due', help: 'Enabled sources whose next fetch is due now (the fetch queue)', collect: () => read.queue(now()).due });
-- registry.gauge({ name: 'sources_fetch_oldest_wait_seconds', help: 'How long the longest-waiting due source has waited', collect: () => read.queue(now()).oldest_wait_ms / 1000 });
++ registry.gauge({ name: 'sources_fetch_due', help: 'Enabled sources whose next fetch is due now (the fetch queue)', collect: async () => (await read.queue(now())).due });
++ registry.gauge({ name: 'sources_fetch_oldest_wait_seconds', help: 'How long the longest-waiting due source has waited', collect: async () => (await read.queue(now())).oldest_wait_ms / 1000 });
registry.gauge({ name: 'sources_fetch_in_flight', help: 'Fetch runs in progress', collect: () => ingest.inflight().length });
// Left out until there is a fetch: a timestamp of 0 would read as "1970", not "never".
-- registry.gauge({ name: 'sources_last_fetch_timestamp_seconds', help: 'When the last fetch run finished (any outcome), Unix seconds', collect: () => seconds(read.lastFetchAt()) });
-- registry.gauge({ name: 'sources_last_success_timestamp_seconds', help: 'When a source was last fetched successfully, Unix seconds', collect: () => seconds(read.lastSuccessAt()) });
-- registry.gauge({ name: 'sources_outbox_pending', help: 'Events waiting in the outbox', collect: () => outbox.pending() });
++ registry.gauge({ name: 'sources_last_fetch_timestamp_seconds', help: 'When the last fetch run finished (any outcome), Unix seconds', collect: async () => seconds(await read.lastFetchAt()) });
++ registry.gauge({ name: 'sources_last_success_timestamp_seconds', help: 'When a source was last fetched successfully, Unix seconds', collect: async () => seconds(await read.lastSuccessAt()) });
++ registry.gauge({ name: 'sources_outbox_pending', help: 'Events waiting in the outbox', collect: async () => await outbox.pending() });
}
module.exports = { createSourcesReadiness, registerSourcesGauges, OVERDUE_MS };
ddiff --git a/server/registry.js b/server/registry.js
iindex 50a2be0..05d7c04 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -205,36 +205,36 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 500 }) {
return { ...fromRow(row), health: health(row), created_at: new Date(row.created_at).toISOString(), updated_at: new Date(row.updated_at).toISOString() };
}
-- function create(input, by) {
++ async function create(input, by) {
const rec = validateSource(input, { maxItemsCap });
-- if (st.get.get(rec.key)) throw new RegistryError(`source ${rec.key} exists`, 'sources.exists');
-- st.insert.run(toRow(rec, by));
-- return st.get.get(rec.key);
++ if (await st.get.get(rec.key)) throw new RegistryError(`source ${rec.key} exists`, 'sources.exists');
++ await st.insert.run(toRow(rec, by));
++ return await st.get.get(rec.key);
}
-- function patch(key, changes, by) {
-- const row = st.get.get(key);
++ async function patch(key, changes, by) {
++ const row = await st.get.get(key);
if (!row) return null;
if (changes && changes.key !== undefined && changes.key !== key) throw new RegistryError('key cannot change');
-- if (changes && changes.type !== undefined && changes.type !== row.type && st.itemCount.get(key).n) {
++ if (changes && changes.type !== undefined && changes.type !== row.type && (await st.itemCount.get(key)).n) {
throw new RegistryError('type cannot change once the source has items (register a new source)');
}
const merged = { ...fromRow(row), ...(changes || {}), key };
const rec = validateSource(merged, { maxItemsCap });
-- st.update.run({ ...toRow(rec, by) });
-- return st.get.get(key);
++ await st.update.run({ ...toRow(rec, by) });
++ return await st.get.get(key);
}
/** Only a source that never produced an item can be deleted; otherwise disable it. */
-- function remove(key) {
-- const row = st.get.get(key);
++ async function remove(key) {
++ const row = await st.get.get(key);
if (!row) return null;
-- if (st.itemCount.get(key).n) throw new RegistryError('the source has items; disable it instead (provenance must stay resolvable)', 'sources.has_items');
-- st.del.run(key);
++ if ((await st.itemCount.get(key)).n) throw new RegistryError('the source has items; disable it instead (provenance must stay resolvable)', 'sources.has_items');
++ await st.del.run(key);
return true;
}
-- return { get: (k) => st.get.get(k), all: () => st.all.all(), create, patch, remove, view, fromRow, health };
++ return { get: async (k) => await st.get.get(k), all: async () => await st.all.all(), create, patch, remove, view, fromRow, health };
}
module.exports = { createRegistry, validateSource, RegistryError, TYPES, CATEGORIES };
ddiff --git a/server/robots.js b/server/robots.js
iindex c273057..0d8155a 100644
---- a/server/robots.js
++++ b/server/robots.js
@@@ -116,12 +116,12 @@ function createRobots({ db, fetcher, agent, ttlMs, maxBytes, now = () => Date.no
// An unreachable robots.txt is retried sooner than a parsed one.
const ttl = row.outcome === 'unreachable' ? Math.min(ttlMs, 15 * 60 * 1000) : ttlMs;
const full = { origin, fetched_at: t, expires_at: t + ttl, ...row };
-- put.run(full);
++ await put.run(full);
return full;
}
async function entry(origin) {
-- const cached = get.get(origin);
++ const cached = await get.get(origin);
if (cached && cached.expires_at > now()) return { row: cached, source: 'cache' };
if (!inflight.has(origin)) inflight.set(origin, load(origin).finally(() => inflight.delete(origin)));
return { row: await inflight.get(origin), source: 'fetched' };
@@@ -141,7 +141,7 @@ function createRobots({ db, fetcher, agent, ttlMs, maxBytes, now = () => Date.no
return { allowed, reason: allowed ? null : `disallowed by ${url.origin}/robots.txt`, crawlDelaySec: row.crawl_delay_sec, source };
}
-- return { check, forget: (origin) => db.prepare('DELETE FROM robots_cache WHERE origin = ?').run(origin) };
++ return { check, forget: async (origin) => await db.prepare('DELETE FROM robots_cache WHERE origin = ?').run(origin) };
}
module.exports = { parseRobots, isAllowed, createRobots };
ddiff --git a/server/scheduler.js b/server/scheduler.js
iindex f16cec0..4d08c49 100644
---- a/server/scheduler.js
++++ b/server/scheduler.js
@@@ -10,10 +10,10 @@ function createScheduler({ db, ingest, config, now = () => Date.now(), log = con
let timer = null;
const running = new Set();
-- function tick() {
++ async function tick() {
const free = config.worker.maxConcurrent - running.size;
if (free <= 0) return [];
-- const keys = due.all({ now: now(), limit: free + running.size }).map(r => r.key).filter(k => !running.has(k)).slice(0, free);
++ const keys = (await due.all({ now: now(), limit: free + running.size })).map(r => r.key).filter(k => !running.has(k)).slice(0, free);
for (const key of keys) {
running.add(key);
ingest.run(key, { trigger: 'schedule' })
@@@ -25,7 +25,7 @@ function createScheduler({ db, ingest, config, now = () => Date.now(), log = con
function start() {
if (timer) return;
-- timer = setInterval(() => { try { tick(); } catch (err) { log.error(`[scheduler] tick: ${err.message}`); } }, config.worker.tickMs);
++ timer = setInterval(async () => { try { await tick(); } catch (err) { log.error(`[scheduler] tick: ${err.message}`); } }, config.worker.tickMs);
timer.unref?.();
}
ddiff --git a/test/actor-limits.test.js b/test/actor-limits.test.js
iindex 80a7db0..34d4219 100644
---- a/test/actor-limits.test.js
++++ b/test/actor-limits.test.js
@@@ -61,15 +61,15 @@ t('the next minute opens the window again', async () => {
t('source changes: 30 a minute per principal; nothing is stored past it; another staff tool passes', async () => {
clock = Date.UTC(2026, 8, 27, 12, 5, 0);
-- const count = () => svc.db.prepare('SELECT COUNT(*) AS n FROM sources').get().n;
++ const count = async () => (await svc.db.prepare('SELECT COUNT(*) AS n FROM sources').get()).n;
for (let i = 0; i < 30; i++) {
const r = await api('POST', '/api/v1/sources', { token: ADMIN, body: sourceDef({ key: `limits-${i}`, enabled: false, endpoints: ['https://example.org/feed'] }) });
assert.strictEqual(r.status, 201, `source ${i + 1}: ${r.text}`);
}
-- const before = count();
++ const before = await count();
const r = await api('POST', '/api/v1/sources', { token: ADMIN, body: sourceDef({ key: 'one-too-many', enabled: false, endpoints: ['https://example.org/feed'] }) });
assert.deepStrictEqual([r.status, r.body.code, r.headers.get('retry-after')], [429, 'rate_limited', '60']);
-- assert.strictEqual(count(), before, 'nothing stored');
++ assert.strictEqual(await count(), before, 'nothing stored');
const other = await api('POST', '/api/v1/sources', { token: STAFF_TOOL, body: sourceDef({ key: 'other-tool', enabled: false, endpoints: ['https://example.org/feed'] }) });
assert.strictEqual(other.status, 201, other.text);
});
ddiff --git a/test/api.test.js b/test/api.test.js
iindex 4dffc65..0d7ef65 100644
---- a/test/api.test.js
++++ b/test/api.test.js
@@@ -66,7 +66,7 @@ t('create, read, fetch, list runs; views never carry a credential value', async
let f = await api('POST', '/api/v1/sources/feed-a/fetch', { token: ADMIN });
assert.strictEqual(f.body.runs[0].state, 'disabled');
await api('PATCH', '/api/v1/sources/feed-a', { token: ADMIN, body: { auth: { mode: 'none' } } });
-- svc.db.prepare('UPDATE sources SET last_request_at = NULL').run();
++ await svc.db.prepare('UPDATE sources SET last_request_at = NULL').run();
f = await api('POST', '/api/v1/sources/feed-a/fetch', { token: ADMIN });
assert.strictEqual(f.status, 200);
assert.deepStrictEqual(f.body.runs.map(r => [r.state, r.items.created]), [['ok', 5]]);
@@@ -116,7 +116,7 @@ t('removal needs a reason, is visible with include_removed, and appears in the c
assert.deepStrictEqual(one.body.item.revisions.map(r => r.revision), [1, 2]);
assert.strictEqual((await api('DELETE', '/api/v1/sources/feed-a', { token: ADMIN })).status, 409, 'a source with items cannot be deleted');
const gone = await api('DELETE', '/api/v1/sources/proposal', { token: ADMIN });
-- assert.strictEqual(gone.status, 204, gone.text + JSON.stringify(svc.db.prepare('SELECT key FROM sources').all()));
++ assert.strictEqual(gone.status, 204, gone.text + JSON.stringify(await svc.db.prepare('SELECT key FROM sources').all()));
});
t('manual sources: items need an evidence URL; fetches are refused', async () => {
ddiff --git a/test/db.js b/test/db.js
nnew file mode 100644
iindex 0000000..53678d9
---- /dev/null
++++ b/test/db.js
@@@ -0,0 +1,11 @@
++'use strict';
++/**
++ * A migrated database for one test boot (ADR-035), from openvibe-sdk/testing: PGlite by default; with
++ * SOURCES_TEST_STORE=pg (npm run test:pg) the PostgreSQL + PgBouncer containers, with roles and a schema of its own.
++ */
++const { createTestDb, pgAvailable } = require('openvibe-sdk/testing');
++const { MIGRATIONS } = require('../server/db');
++
++const testDb = ({ store = process.env.SOURCES_TEST_STORE || 'pglite', max = 4 } = {}) => createTestDb({ migrations: MIGRATIONS, store, service: 'sources', max });
++
++module.exports = { testDb, pgAvailable };
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 828f854..b6bd8e4 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -57,9 +57,11 @@ async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now, lo
SOURCES_TICK_MS: '50',
...env,
});
-- const h = await start({ config, log, lookupImpl, tokenClient, limitsNow, ...(now ? { now } : {}) });
++ // One database per boot (PGlite, or SOURCES_TEST_STORE=pg: the containers), dropped when the boot stops.
++ const testdb = await require('./db').testDb();
++ const h = await start({ config, db: testdb.db, log, lookupImpl, tokenClient, limitsNow, ...(now ? { now } : {}) });
const base = `http://127.0.0.1:${h.server.address().port}`;
-- return { ...h, base, dir, async stop() { await h.close(); } };
++ return { ...h, base, dir, async stop() { await h.close(); await testdb.close(); } };
}
async function request(base, method, p, { token, body, headers = {} } = {}) {
ddiff --git a/test/ingest.test.js b/test/ingest.test.js
iindex 58b7f62..7f82d60 100644
---- a/test/ingest.test.js
++++ b/test/ingest.test.js
@@@ -18,15 +18,15 @@ const FEED_V1 = rss([
{ guid: 'b-1', title: 'Beta', link: 'https://news.example.org/b' },
]);
--function create(def) {
-- svc.registry.create(def, 'test');
++async function create(def) {
++ await svc.registry.create(def, 'test');
return def.key;
}
--const run = (key, trigger = 'manual') => svc.ingest.run(key, { trigger });
--const itemsOf = (key) => svc.db.prepare('SELECT * FROM items WHERE source_key = ? ORDER BY identity').all(key);
--const runsOf = (key) => svc.db.prepare('SELECT * FROM fetch_runs WHERE source_key = ? ORDER BY rid').all(key);
--const events = (type) => svc.outbox.all().filter(e => e.event_type === type);
--const reset = (key) => svc.db.prepare('UPDATE sources SET last_request_at = NULL, not_before = 0 WHERE key = ?').run(key);
++const run = async (key, trigger = 'manual') => await svc.ingest.run(key, { trigger });
++const itemsOf = async (key) => await svc.db.prepare('SELECT * FROM items WHERE source_key = ? ORDER BY identity').all(key);
++const runsOf = async (key) => await svc.db.prepare('SELECT * FROM fetch_runs WHERE source_key = ? ORDER BY rid').all(key);
++const events = async (type) => (await svc.outbox.all()).filter(e => e.event_type === type);
++const reset = async (key) => await svc.db.prepare('UPDATE sources SET last_request_at = NULL, not_before = 0 WHERE key = ?').run(key);
t('boot', async () => {
svc = await boot();
@@@ -44,10 +44,10 @@ t('boot', async () => {
});
t('a good fetch creates items with full provenance and an ok run', async () => {
-- const key = create(sourceDef({ key: 'news-a', endpoints: [`${web.origin}/feed.xml`], license_note: 'headlines only', terms_note: 'fixture terms' }));
++ const key = await create(sourceDef({ key: 'news-a', endpoints: [`${web.origin}/feed.xml`], license_note: 'headlines only', terms_note: 'fixture terms' }));
const out = await run(key);
assert.deepStrictEqual(out.runs.map(r => r.state), ['ok']);
-- const items = itemsOf(key);
++ const items = await itemsOf(key);
assert.strictEqual(items.length, 2);
const a = items[0];
assert.strictEqual(a.identity, 'a-1');
@@@ -61,10 +61,10 @@ t('a good fetch creates items with full provenance and an ok run', async () => {
assert.ok(a.retrieved_at > 0 && a.first_seen_at === a.retrieved_at);
assert.strictEqual(a.published_at, '2026-09-21T10:00:00.000Z');
assert.strictEqual(items[1].published_at, null);
-- const [r] = runsOf(key);
++ const [r] = await runsOf(key);
assert.deepStrictEqual([r.state, r.http_status, r.items_created, r.items_seen], ['ok', 200, 2, 2]);
assert.strictEqual(r.raw_body_hash, a.raw_body_hash);
-- assert.strictEqual(events('sources.item.created').filter(e => e.payload.source_key === key).length, 2);
++ assert.strictEqual((await events('sources.item.created')).filter(e => e.payload.source_key === key).length, 2);
// the fetcher identified itself and asked robots.txt first
const feedReq = web.hits('/feed.xml')[0];
assert.match(feedReq.headers['user-agent'], /^OpenVibeSources\//);
@@@ -72,13 +72,13 @@ t('a good fetch creates items with full provenance and an ok run', async () => {
});
t('conditional GET: the stored ETag is sent and a 304 is not_modified with nothing changed', async () => {
-- reset('news-a');
-- const before = itemsOf('news-a');
++ await reset('news-a');
++ const before = await itemsOf('news-a');
const out = await run('news-a');
assert.deepStrictEqual(out.runs.map(r => r.state), ['not_modified']);
const last = web.hits('/feed.xml').at(-1);
assert.strictEqual(last.headers['if-none-match'], '"v1"');
-- assert.deepStrictEqual(itemsOf('news-a').map(i => [i.revision, i.content_hash, i.retrieved_at]), before.map(i => [i.revision, i.content_hash, i.retrieved_at]));
++ assert.deepStrictEqual((await itemsOf('news-a')).map(i => [i.revision, i.content_hash, i.retrieved_at]), before.map(i => [i.revision, i.content_hash, i.retrieved_at]));
});
t('Last-Modified is used when there is no ETag', async () => {
@@@ -88,15 +88,15 @@ t('Last-Modified is used when there is no ETag', async () => {
'/f': (req) => (req.headers['if-modified-since'] === lm ? { status: 304 } : { headers: { 'Last-Modified': lm }, body: FEED_V1 }),
});
try {
-- const key = create(sourceDef({ endpoints: [`${s.origin}/f`] }));
++ const key = await create(sourceDef({ endpoints: [`${s.origin}/f`] }));
assert.strictEqual((await run(key)).runs[0].state, 'ok');
-- reset(key);
++ await reset(key);
assert.strictEqual((await run(key)).runs[0].state, 'not_modified');
} finally { await s.close(); }
});
t('a changed item is a new revision; the old one stays in item_revisions', async () => {
-- reset('news-a');
++ await reset('news-a');
feedEtag = '"v2"';
feedBody = rss([
{ guid: 'a-1', title: 'Alpha (corrected)', link: 'https://news.example.org/a', date: 'Mon, 21 Sep 2026 10:00:00 +0000', description: 'first, corrected' },
@@@ -105,52 +105,52 @@ t('a changed item is a new revision; the old one stays in item_revisions', async
const out = await run('news-a');
assert.strictEqual(out.runs[0].state, 'ok');
assert.deepStrictEqual([out.runs[0].items_updated, out.runs[0].items_unchanged], [1, 1]);
-- const a = itemsOf('news-a')[0];
++ const a = (await itemsOf('news-a'))[0];
assert.strictEqual(a.revision, 2);
assert.strictEqual(a.title, 'Alpha (corrected)');
-- const revs = svc.db.prepare('SELECT revision, snapshot FROM item_revisions WHERE item_id = ? ORDER BY revision').all(a.id);
++ const revs = await svc.db.prepare('SELECT revision, snapshot FROM item_revisions WHERE item_id = ? ORDER BY revision').all(a.id);
assert.deepStrictEqual(revs.map(r => [r.revision, JSON.parse(r.snapshot).title]), [[1, 'Alpha'], [2, 'Alpha (corrected)']]);
-- const upd = events('sources.item.updated').find(e => e.payload.item_id === a.id);
++ const upd = (await events('sources.item.updated')).find(e => e.payload.item_id === a.id);
assert.ok(upd && upd.payload.previous_content_hash && upd.payload.revision === 2);
});
t('failure never fabricates: http errors, timeouts, parse errors and oversize bodies leave items exactly as they were', async () => {
-- const snapshot = () => JSON.stringify(itemsOf('news-a'));
-- const before = snapshot();
-- const count = svc.db.prepare('SELECT COUNT(*) AS n FROM items').get().n;
++ const snapshot = async () => JSON.stringify(await itemsOf('news-a'));
++ const before = await snapshot();
++ const count = (await svc.db.prepare('SELECT COUNT(*) AS n FROM items').get()).n;
feedStatus = 500;
-- reset('news-a');
++ await reset('news-a');
let out = await run('news-a');
assert.deepStrictEqual([out.runs[0].state, out.runs[0].http_status, out.runs[0].error_code], ['http_error', 500, 'http_500']);
feedStatus = 200;
feedEtag = '"v3"';
feedBody = '<rss><channel><item><title>Broken';
-- reset('news-a');
++ await reset('news-a');
out = await run('news-a');
assert.strictEqual(out.runs[0].state, 'parse_error');
assert.match(out.runs[0].raw_body_hash, /^[0-9a-f]{64}$/, 'the unreadable body is identified by its hash');
feedBody = '<html><body>Service temporarily replaced by a login page</body></html>';
-- reset('news-a');
++ await reset('news-a');
out = await run('news-a');
assert.strictEqual(out.runs[0].state, 'parse_error');
-- assert.strictEqual(snapshot(), before, 'no item created, changed, re-dated or re-hashed');
-- assert.strictEqual(svc.db.prepare('SELECT COUNT(*) AS n FROM items').get().n, count);
++ assert.strictEqual(await snapshot(), before, 'no item created, changed, re-dated or re-hashed');
++ assert.strictEqual((await svc.db.prepare('SELECT COUNT(*) AS n FROM items').get()).n, count);
// after a failed parse the validator was not stored: the next good body is fetched in full
feedBody = FEED_V1;
-- reset('news-a');
++ await reset('news-a');
out = await run('news-a');
assert.strictEqual(out.runs[0].state, 'ok');
assert.notStrictEqual(web.hits('/feed.xml').at(-1).headers['if-none-match'], '"v3"');
-- const failed = events('sources.fetch.failed').filter(e => e.payload.source_key === 'news-a');
++ const failed = (await events('sources.fetch.failed')).filter(e => e.payload.source_key === 'news-a');
assert.deepStrictEqual(failed.map(e => e.payload.state), ['http_error', 'parse_error', 'parse_error']);
assert.deepStrictEqual(failed.map(e => e.payload.consecutive_failures), [1, 2, 3]);
-- assert.strictEqual(svc.registry.get('news-a').consecutive_failures, 0, 'reset by the good run');
++ assert.strictEqual((await svc.registry.get('news-a')).consecutive_failures, 0, 'reset by the good run');
});
t('timeouts and oversize bodies are explicit states', async () => {
@@@ -160,14 +160,14 @@ t('timeouts and oversize bodies are explicit states', async () => {
'/huge': () => ({ body: 'x'.repeat(4096) }),
});
try {
-- const key = create(sourceDef({ endpoints: [`${s.origin}/slow`] }));
++ const key = await create(sourceDef({ endpoints: [`${s.origin}/slow`] }));
const out = await run(key);
assert.deepStrictEqual([out.runs[0].state, out.runs[0].error_code], ['timeout', 'timeout']);
-- assert.strictEqual(itemsOf(key).length, 0);
++ assert.strictEqual((await itemsOf(key)).length, 0);
const small = await boot({ env: { SOURCES_MAX_BYTES: '1024' } });
try {
-- small.registry.create(sourceDef({ key: 'huge', endpoints: [`${s.origin}/huge`] }), 'test');
++ await small.registry.create(sourceDef({ key: 'huge', endpoints: [`${s.origin}/huge`] }), 'test');
const r = await small.ingest.run('huge', { trigger: 'manual' });
assert.deepStrictEqual([r.runs[0].state, r.runs[0].error_code], ['http_error', 'too_large']);
} finally { await small.stop(); }
@@@ -175,20 +175,20 @@ t('timeouts and oversize bodies are explicit states', async () => {
});
t('staleness is computed from the last success and exposed in health', async () => {
-- const key = create(sourceDef({ endpoints: [`${web.origin}/feed.xml`], stale_after_sec: 1 }));
-- let h = svc.registry.health(svc.registry.get(key));
++ const key = await create(sourceDef({ endpoints: [`${web.origin}/feed.xml`], stale_after_sec: 1 }));
++ let h = svc.registry.health(await svc.registry.get(key));
assert.deepStrictEqual([h.status, h.stale], ['never_fetched', true]);
feedEtag = null;
await run(key);
-- h = svc.registry.health(svc.registry.get(key));
++ h = svc.registry.health(await svc.registry.get(key));
assert.deepStrictEqual([h.status, h.stale], ['healthy', false]);
await sleep(1100);
-- h = svc.registry.health(svc.registry.get(key));
++ h = svc.registry.health(await svc.registry.get(key));
assert.deepStrictEqual([h.status, h.stale], ['stale', true]);
feedStatus = 503;
-- reset(key);
++ await reset(key);
await run(key);
-- h = svc.registry.health(svc.registry.get(key));
++ h = svc.registry.health(await svc.registry.get(key));
assert.deepStrictEqual([h.status, h.stale, h.consecutive_failures], ['failing', true, 1]);
feedStatus = 200;
});
@@@ -200,7 +200,7 @@ t('rate limit: requests to a source are spaced by min_interval_ms; an early trig
'/two': () => ({ body: FEED_V1 }),
});
try {
-- const key = create(sourceDef({ endpoints: [`${s.origin}/one`, `${s.origin}/two`], min_interval_ms: 400 }));
++ const key = await create(sourceDef({ endpoints: [`${s.origin}/one`, `${s.origin}/two`], min_interval_ms: 400 }));
const out = await run(key);
assert.deepStrictEqual(out.runs.map(r => r.state), ['ok', 'ok']);
const gap = s.hits('/two')[0].at - s.hits('/one')[0].at;
@@@ -208,7 +208,7 @@ t('rate limit: requests to a source are spaced by min_interval_ms; an early trig
const again = await run(key);
assert.deepStrictEqual([again.runs[0].state, again.runs[0].error_code], ['rate_limited', 'local_rate_limit']);
assert.strictEqual(s.hits('/one').length, 1, 'nothing was requested');
-- assert.ok(!events('sources.fetch.failed').some(e => e.payload.source_key === key), 'our own limiter is not a source failure');
++ assert.ok(!(await events('sources.fetch.failed')).some(e => e.payload.source_key === key), 'our own limiter is not a source failure');
} finally { await s.close(); }
});
@@@ -219,11 +219,11 @@ t('a 429 is rate_limited, stops the run and honours Retry-After', async () => {
'/b': () => ({ body: FEED_V1 }),
});
try {
-- const key = create(sourceDef({ endpoints: [`${s.origin}/a`, `${s.origin}/b`] }));
++ const key = await create(sourceDef({ endpoints: [`${s.origin}/a`, `${s.origin}/b`] }));
const out = await run(key);
assert.deepStrictEqual(out.runs.map(r => [r.state, r.error_code]), [['rate_limited', 'upstream_429']]);
assert.strictEqual(s.hits('/b').length, 0);
-- const row = svc.registry.get(key);
++ const row = await svc.registry.get(key);
assert.ok(row.not_before - Date.now() > 100 * 1000, 'Retry-After respected');
const early = await run(key);
assert.strictEqual(early.runs[0].state, 'rate_limited');
@@@ -234,8 +234,8 @@ t('a 429 is rate_limited, stops the run and honours Retry-After', async () => {
t('one run per source in flight', async () => {
const s = await site({ '/robots.txt': () => ({ status: 404 }), '/slow': () => ({ delayMs: 300, body: FEED_V1 }) });
try {
-- const key = create(sourceDef({ endpoints: [`${s.origin}/slow`] }));
-- const first = run(key);
++ const key = await create(sourceDef({ endpoints: [`${s.origin}/slow`] }));
++ const first = run(key); // left running: the second call must find it in flight
await sleep(50);
const second = await run(key);
assert.deepStrictEqual(second, { busy: true });
@@@ -245,11 +245,11 @@ t('one run per source in flight', async () => {
});
t('disabled sources and missing credentials are recorded as disabled, nothing is requested', async () => {
-- const key = create(sourceDef({ endpoints: [`${web.origin}/feed.xml`], enabled: false }));
++ const key = await create(sourceDef({ endpoints: [`${web.origin}/feed.xml`], enabled: false }));
const n = web.requests.length;
const out = await run(key);
assert.deepStrictEqual([out.runs[0].state, out.runs[0].error_code], ['disabled', 'source_disabled']);
-- const key2 = create(sourceDef({ endpoints: [`${web.origin}/feed.xml`], auth: { mode: 'bearer', env: 'SOURCES_CRED_NOT_SET' } }));
++ const key2 = await create(sourceDef({ endpoints: [`${web.origin}/feed.xml`], auth: { mode: 'bearer', env: 'SOURCES_CRED_NOT_SET' } }));
const out2 = await run(key2);
assert.deepStrictEqual([out2.runs[0].state, out2.runs[0].error_code], ['disabled', 'credential_missing']);
assert.match(out2.runs[0].detail, /SOURCES_CRED_NOT_SET/);
@@@ -265,13 +265,13 @@ t('a credential is read by name, sent only to its origin, and never stored', asy
const secret = 'sk_test_' + 'q'.repeat(24);
const svc2 = await boot({ env: { SOURCES_CRED_EXAMPLE: secret } });
try {
-- svc2.registry.create(sourceDef({ key: 'cred', endpoints: [`${s.origin}/api`], auth: { mode: 'header', env: 'SOURCES_CRED_EXAMPLE', header: 'X-Api-Key' } }), 'test');
++ await svc2.registry.create(sourceDef({ key: 'cred', endpoints: [`${s.origin}/api`], auth: { mode: 'header', env: 'SOURCES_CRED_EXAMPLE', header: 'X-Api-Key' } }), 'test');
const out = await svc2.ingest.run('cred', { trigger: 'manual' });
assert.strictEqual(out.runs[0].state, 'ok');
assert.strictEqual(s.hits('/api')[0].headers['x-api-key'], secret);
assert.strictEqual(other.hits('/landing')[0].headers['x-api-key'], undefined, 'not forwarded across origins');
-- const dump = JSON.stringify(svc2.db.prepare('SELECT * FROM sources').all()) + JSON.stringify(svc2.db.prepare('SELECT * FROM fetch_runs').all())
-- + JSON.stringify(svc2.db.prepare('SELECT * FROM items').all()) + JSON.stringify(svc2.outbox.all());
++ const dump = JSON.stringify(await svc2.db.prepare('SELECT * FROM sources').all()) + JSON.stringify(await svc2.db.prepare('SELECT * FROM fetch_runs').all())
++ + JSON.stringify(await svc2.db.prepare('SELECT * FROM items').all()) + JSON.stringify(await svc2.outbox.all());
assert.ok(!dump.includes(secret), 'the secret value is nowhere in the database or events');
assert.ok(dump.includes('SOURCES_CRED_EXAMPLE'), 'the variable name is recorded');
} finally { await svc2.stop(); await s.close(); await other.close(); }
@@@ -279,17 +279,17 @@ t('a credential is read by name, sent only to its origin, and never stored', asy
t('removal is explicit and sticky; the next fetch does not bring the item back', async () => {
const key = 'news-a';
-- const a = itemsOf(key).find(i => i.identity === 'a-1');
-- const source = svc.registry.fromRow(svc.registry.get(key));
-- const removed = svc.items.remove(source, a.id, 'publisher takedown request', 'svc:network');
++ const a = (await itemsOf(key)).find(i => i.identity === 'a-1');
++ const source = svc.registry.fromRow(await svc.registry.get(key));
++ const removed = await svc.items.remove(source, a.id, 'publisher takedown request', 'svc:network');
assert.ok(removed.removed && /takedown/.test(removed.removed.reason));
-- reset(key);
++ await reset(key);
feedEtag = '"v9"';
feedBody = FEED_V1;
await run(key);
-- const again = svc.items.get(a.id);
++ const again = await svc.items.get(a.id);
assert.ok(again.removed, 'still removed');
-- assert.ok(events('sources.item.removed').some(e => e.payload.item_id === a.id));
++ assert.ok((await events('sources.item.removed')).some(e => e.payload.item_id === a.id));
});
t('JSON-LD and sitemap sources ingest end to end', async () => {
@@@ -299,13 +299,13 @@ t('JSON-LD and sitemap sources ingest end to end', async () => {
'/sitemap.xml': () => ({ headers: { 'Content-Type': 'application/xml' }, body: fixture('sitemap.xml') }),
});
try {
-- const p = create(sourceDef({ type: 'jsonld', category: 'deals', endpoints: [`${s.origin}/p/widget-pro`] }));
++ const p = await create(sourceDef({ type: 'jsonld', category: 'deals', endpoints: [`${s.origin}/p/widget-pro`] }));
assert.strictEqual((await run(p)).runs[0].state, 'ok');
-- const [prod] = itemsOf(p);
++ const [prod] = await itemsOf(p);
assert.strictEqual(prod.kind, 'product');
assert.strictEqual(JSON.parse(prod.fields).offers[0].price, '19.99');
assert.strictEqual(prod.parser_version, 'jsonld@1');
-- const sm = create(sourceDef({ type: 'sitemap', category: 'deals', endpoints: [`${s.origin}/sitemap.xml`] }));
++ const sm = await create(sourceDef({ type: 'sitemap', category: 'deals', endpoints: [`${s.origin}/sitemap.xml`] }));
const out = await run(sm);
assert.deepStrictEqual([out.runs[0].state, out.runs[0].items_created, out.runs[0].items_skipped], ['ok', 2, 1]);
} finally { await s.close(); }
@@@ -314,21 +314,21 @@ t('JSON-LD and sitemap sources ingest end to end', async () => {
t('sources with search_visibility send staff-only, noindex index documents through the outbox', async () => {
const s = await site({ '/robots.txt': () => ({ status: 404 }), '/f': () => ({ body: FEED_V1 }) });
try {
-- const key = create(sourceDef({ endpoints: [`${s.origin}/f`], search_visibility: 'members' }));
++ const key = await create(sourceDef({ endpoints: [`${s.origin}/f`], search_visibility: 'members' }));
await run(key);
-- const docs = events('sources.index_document.upserted').filter(e => e.payload.facets.source === key);
++ const docs = (await events('sources.index_document.upserted')).filter(e => e.payload.facets.source === key);
assert.strictEqual(docs.length, 2);
const d = docs[0].payload;
assert.deepStrictEqual([d.owner, d.type, d.visibility, d.indexability.decision], ['sources', 'item', 'members', 'noindex']);
assert.deepStrictEqual(d.acl.groups, ['role:admin', 'role:global_mod']);
assert.deepStrictEqual(docs[0].subject, { type: 'item', id: d.id, revision: d.revision });
assert.strictEqual(docs[0].visibility, 'internal');
-- const item = itemsOf(key)[0];
-- svc.items.remove(svc.registry.fromRow(svc.registry.get(key)), item.id, 'test', null);
-- const del = events('sources.index_document.deleted').find(e => e.payload.id === item.id);
++ const item = (await itemsOf(key))[0];
++ await svc.items.remove(svc.registry.fromRow(await svc.registry.get(key)), item.id, 'test', null);
++ const del = (await events('sources.index_document.deleted')).find(e => e.payload.id === item.id);
assert.deepStrictEqual(del.payload, { type: 'item', id: item.id, revision: 2 });
// a source without search_visibility sends none
-- assert.ok(!events('sources.index_document.upserted').some(e => e.payload.facets.source === 'news-a'));
++ assert.ok(!(await events('sources.index_document.upserted')).some(e => e.payload.facets.source === 'news-a'));
} finally { await s.close(); }
});
ddiff --git a/test/proposals.test.js b/test/proposals.test.js
iindex c599884..5691de2 100644
---- a/test/proposals.test.js
++++ b/test/proposals.test.js
@@@ -41,7 +41,7 @@ t('the service manifest proposal is a valid registry.service-manifest@1', () =>
assert.deepStrictEqual([...m.capabilities].sort(), Object.values(CAPS).sort());
});
--t('seeds: one real source per category, all disabled, all valid, notes recorded', () => {
++t('seeds: one real source per category, all disabled, all valid, notes recorded', async () => {
const seeds = require('../seeds/sources.json').sources;
assert.deepStrictEqual(seeds.map(s => s.category).sort(), [...CATEGORIES].sort());
for (const s of seeds) {
@@@ -53,12 +53,12 @@ t('seeds: one real source per category, all disabled, all valid, notes recorded'
for (const ep of rec.endpoints) assert.ok(ep.url.startsWith('https://'), ep.url);
assert.ok(sourceSchema(rec), JSON.stringify(sourceSchema.errors));
}
-- const { openDb } = require('../server/db');
-- const db = openDb(':memory:');
-- assert.strictEqual(seed(db).created.length, seeds.length);
-- assert.deepStrictEqual(seed(db).created, [], 'never overwrites');
-- assert.strictEqual(db.prepare('SELECT COUNT(*) AS n FROM sources WHERE enabled = 1').get().n, 0);
-- db.close();
++ const fresh = await require('./db').testDb();
++ const db = fresh.db;
++ assert.strictEqual((await seed(db)).created.length, seeds.length);
++ assert.deepStrictEqual((await seed(db)).created, [], 'never overwrites');
++ assert.strictEqual((await db.prepare('SELECT COUNT(*) AS n FROM sources WHERE enabled = 1').get()).n, 0);
++ await fresh.close();
});
t('API views match sources.source@1 and sources.item@1', async () => {
ddiff --git a/test/robots-ssrf.test.js b/test/robots-ssrf.test.js
iindex 70777f1..7b3a3f0 100644
---- a/test/robots-ssrf.test.js
++++ b/test/robots-ssrf.test.js
@@@ -51,16 +51,16 @@ t('boot', async () => {
});
t('a disallowed endpoint is robots_denied and never requested', async () => {
-- svc.registry.create(sourceDef({ key: 'denied', endpoints: [`${web.origin}/private/feed.xml`, `${web.origin}/public/feed.xml`] }), 'test');
++ await svc.registry.create(sourceDef({ key: 'denied', endpoints: [`${web.origin}/private/feed.xml`, `${web.origin}/public/feed.xml`] }), 'test');
const out = await svc.ingest.run('denied', { trigger: 'manual' });
assert.deepStrictEqual(out.runs.map(r => r.state), ['robots_denied', 'ok']);
assert.strictEqual(web.hits('/private/feed.xml').length, 0);
assert.strictEqual(web.hits('/robots.txt').length, 1, 'robots.txt is cached per origin');
-- assert.ok(svc.outbox.all().some(e => e.event_type === 'sources.fetch.failed' && e.payload.state === 'robots_denied'));
++ assert.ok((await svc.outbox.all()).some(e => e.event_type === 'sources.fetch.failed' && e.payload.state === 'robots_denied'));
});
t('a redirect into a disallowed path is refused before it is followed', async () => {
-- svc.registry.create(sourceDef({ key: 'hop', endpoints: [`${web.origin}/public/moved`] }), 'test');
++ await svc.registry.create(sourceDef({ key: 'hop', endpoints: [`${web.origin}/public/moved`] }), 'test');
const out = await svc.ingest.run('hop', { trigger: 'manual' });
assert.deepStrictEqual([out.runs[0].state, out.runs[0].error_code], ['robots_denied', 'hop_refused']);
assert.strictEqual(web.hits('/private/feed.xml').length, 0);
@@@ -70,8 +70,8 @@ t('robots.txt 5xx or unreachable means complete disallow; 404 means allowed', as
const down = await site({ '/robots.txt': () => ({ status: 503 }), '/feed.xml': () => ({ body: FEED }) });
const none = await site({ '/robots.txt': () => ({ status: 404 }), '/feed.xml': () => ({ body: FEED }) });
try {
-- svc.registry.create(sourceDef({ key: 'robots-down', endpoints: [`${down.origin}/feed.xml`] }), 'test');
-- svc.registry.create(sourceDef({ key: 'robots-none', endpoints: [`${none.origin}/feed.xml`] }), 'test');
++ await svc.registry.create(sourceDef({ key: 'robots-down', endpoints: [`${down.origin}/feed.xml`] }), 'test');
++ await svc.registry.create(sourceDef({ key: 'robots-none', endpoints: [`${none.origin}/feed.xml`] }), 'test');
assert.strictEqual((await svc.ingest.run('robots-down', { trigger: 'manual' })).runs[0].state, 'robots_denied');
assert.strictEqual(down.hits('/feed.xml').length, 0);
assert.strictEqual((await svc.ingest.run('robots-none', { trigger: 'manual' })).runs[0].state, 'ok');
@@@ -81,7 +81,7 @@ t('robots.txt 5xx or unreachable means complete disallow; 404 means allowed', as
t('Crawl-delay spaces requests to the host', async () => {
const s = await site({ '/robots.txt': () => ({ body: 'User-agent: *\nCrawl-delay: 1\n' }), '/a': () => ({ body: FEED }), '/b': () => ({ body: FEED }) });
try {
-- svc.registry.create(sourceDef({ key: 'crawl-delay', endpoints: [`${s.origin}/a`, `${s.origin}/b`] }), 'test');
++ await svc.registry.create(sourceDef({ key: 'crawl-delay', endpoints: [`${s.origin}/a`, `${s.origin}/b`] }), 'test');
await svc.ingest.run('crawl-delay', { trigger: 'manual' });
const gap = s.hits('/b')[0].at - s.hits('/a')[0].at;
assert.ok(gap >= 950, `gap ${gap} ms`);
@@@ -91,12 +91,12 @@ t('Crawl-delay spaces requests to the host', async () => {
t('loopback and private addresses are refused unless explicitly allowlisted', async () => {
const strict = await boot({ env: { SOURCES_ALLOW_PRIVATE_HOSTS: '' } });
try {
-- strict.registry.create(sourceDef({ key: 'loop', endpoints: [`${web.origin}/public/feed.xml`] }), 'test');
++ await strict.registry.create(sourceDef({ key: 'loop', endpoints: [`${web.origin}/public/feed.xml`] }), 'test');
const out = await strict.ingest.run('loop', { trigger: 'manual' });
assert.strictEqual(out.runs[0].state, 'http_error');
assert.ok(['address_refused', 'port_refused'].includes(out.runs[0].error_code), out.runs[0].error_code);
const n = web.requests.length;
-- strict.registry.create(sourceDef({ key: 'meta', endpoints: ['http://169.254.169.254/latest/meta-data/'] }), 'test');
++ await strict.registry.create(sourceDef({ key: 'meta', endpoints: ['http://169.254.169.254/latest/meta-data/'] }), 'test');
const meta = await strict.ingest.run('meta', { trigger: 'manual' });
assert.deepStrictEqual([meta.runs[0].state, meta.runs[0].error_code], ['http_error', 'address_refused']);
assert.strictEqual(web.requests.length, n);
@@@ -109,19 +109,19 @@ t('a public-looking name that resolves to a private address is refused at connec
const strict = await boot({ env: { SOURCES_ALLOW_PRIVATE_HOSTS: '', SOURCES_ALLOWED_PORTS: `80,443,${new URL(web.origin).port}` }, lookupImpl });
try {
const port = new URL(web.origin).port;
-- strict.registry.create(sourceDef({ key: 'rebind', endpoints: [`http://feeds.example.com:${port}/public/feed.xml`] }), 'test');
++ await strict.registry.create(sourceDef({ key: 'rebind', endpoints: [`http://feeds.example.com:${port}/public/feed.xml`] }), 'test');
const n = web.requests.length;
const out = await strict.ingest.run('rebind', { trigger: 'manual' });
assert.deepStrictEqual([out.runs[0].state, out.runs[0].error_code], ['http_error', 'address_refused']);
assert.strictEqual(web.requests.length, n, 'no connection was made');
-- assert.strictEqual(strict.db.prepare('SELECT COUNT(*) AS n FROM items').get().n, 0);
++ assert.strictEqual((await strict.db.prepare('SELECT COUNT(*) AS n FROM items').get()).n, 0);
} finally { await strict.stop(); }
});
t('redirects to an internal address are refused', async () => {
const s = await site({ '/robots.txt': () => ({ status: 404 }), '/r': () => ({ status: 302, headers: { Location: 'http://10.0.0.5/admin' } }) });
try {
-- svc.registry.create(sourceDef({ key: 'redir-internal', endpoints: [`${s.origin}/r`] }), 'test');
++ await svc.registry.create(sourceDef({ key: 'redir-internal', endpoints: [`${s.origin}/r`] }), 'test');
const out = await svc.ingest.run('redir-internal', { trigger: 'manual' });
assert.deepStrictEqual([out.runs[0].state, out.runs[0].error_code], ['http_error', 'address_refused']);
} finally { await s.close(); }
ddiff --git a/test/scheduler-events.test.js b/test/scheduler-events.test.js
iindex 1be9ee8..f454cbf 100644
---- a/test/scheduler-events.test.js
++++ b/test/scheduler-events.test.js
@@@ -10,7 +10,7 @@ const FEED = rss([{ guid: 's1', title: 'Scheduled', link: 'https://example.org/s
async function waitFor(fn, ms = 3000) {
const end = Date.now() + ms;
-- while (Date.now() < end) { const v = fn(); if (v) return v; await sleep(25); }
++ while (Date.now() < end) { const v = await fn(); if (v) return v; await sleep(25); }
throw new Error('timed out waiting');
}
@@@ -19,30 +19,31 @@ t('due, enabled sources are fetched on schedule; disabled and manual ones never
const s = await site({ '/robots.txt': () => ({ status: 404 }), '/f': () => (status === 200 ? { body: FEED } : { status }) });
const svc = await boot({ worker: 'on' });
try {
-- svc.registry.create(sourceDef({ key: 'sched', endpoints: [`${s.origin}/f`], poll_interval_sec: 3600 }), 'test');
-- svc.registry.create(sourceDef({ key: 'off', endpoints: [`${s.origin}/f`], enabled: false }), 'test');
-- svc.registry.create({ key: 'hand', name: 'Manual', type: 'manual', category: 'coupons', terms_note: 'x', enabled: true }, 'test');
-- await waitFor(() => svc.db.prepare("SELECT 1 FROM fetch_runs WHERE source_key = 'sched'").get());
-- const row = svc.registry.get('sched');
++ await svc.registry.create(sourceDef({ key: 'sched', endpoints: [`${s.origin}/f`], poll_interval_sec: 3600 }), 'test');
++ await svc.registry.create(sourceDef({ key: 'off', endpoints: [`${s.origin}/f`], enabled: false }), 'test');
++ await svc.registry.create({ key: 'hand', name: 'Manual', type: 'manual', category: 'coupons', terms_note: 'x', enabled: true }, 'test');
++ // The run row commits with the items; the source's outcome is recorded right after it.
++ await waitFor(async () => (await svc.db.prepare("SELECT 1 FROM fetch_runs WHERE source_key = 'sched'").get()) && (await svc.registry.get('sched')).last_state);
++ const row = await svc.registry.get('sched');
assert.strictEqual(row.last_state, 'ok');
assert.ok(row.next_due_at - Date.now() > 3500 * 1000, 'next poll one interval later');
await sleep(200);
-- assert.strictEqual(svc.db.prepare("SELECT COUNT(*) AS n FROM fetch_runs WHERE source_key IN ('off', 'hand')").get().n, 0);
-- assert.strictEqual(svc.db.prepare("SELECT COUNT(*) AS n FROM fetch_runs WHERE source_key = 'sched'").get().n, 1, 'not refetched before it is due');
++ assert.strictEqual((await svc.db.prepare("SELECT COUNT(*) AS n FROM fetch_runs WHERE source_key IN ('off', 'hand')").get()).n, 0);
++ assert.strictEqual((await svc.db.prepare("SELECT COUNT(*) AS n FROM fetch_runs WHERE source_key = 'sched'").get()).n, 1, 'not refetched before it is due');
// a failing source backs off exponentially
status = 500;
-- svc.db.prepare("UPDATE sources SET next_due_at = 0, last_request_at = NULL WHERE key = 'sched'").run();
-- await waitFor(() => svc.registry.get('sched').consecutive_failures === 1);
-- const f1 = svc.registry.get('sched').next_due_at - svc.registry.get('sched').last_run_at;
-- svc.db.prepare("UPDATE sources SET next_due_at = 0, last_request_at = NULL WHERE key = 'sched'").run();
-- await waitFor(() => svc.registry.get('sched').consecutive_failures === 2);
-- const f2 = svc.registry.get('sched').next_due_at - svc.registry.get('sched').last_run_at;
++ await svc.db.prepare("UPDATE sources SET next_due_at = 0, last_request_at = NULL WHERE key = 'sched'").run();
++ await waitFor(async () => (await svc.registry.get('sched')).consecutive_failures === 1);
++ const f1 = (await svc.registry.get('sched')).next_due_at - (await svc.registry.get('sched')).last_run_at;
++ await svc.db.prepare("UPDATE sources SET next_due_at = 0, last_request_at = NULL WHERE key = 'sched'").run();
++ await waitFor(async () => (await svc.registry.get('sched')).consecutive_failures === 2);
++ const f2 = (await svc.registry.get('sched')).next_due_at - (await svc.registry.get('sched')).last_run_at;
assert.strictEqual(f1, 3600 * 1000);
assert.strictEqual(f2, 2 * 3600 * 1000);
// not_before (Retry-After) holds the scheduler back
-- svc.db.prepare("UPDATE sources SET next_due_at = 0, not_before = ?, last_request_at = NULL WHERE key = 'sched'").run(Date.now() + 60000);
++ await svc.db.prepare("UPDATE sources SET next_due_at = 0, not_before = ?, last_request_at = NULL WHERE key = 'sched'").run(Date.now() + 60000);
const n = s.hits('/f').length;
await sleep(300);
assert.strictEqual(s.hits('/f').length, n);
@@@ -66,10 +67,10 @@ t('events are valid envelopes and reach OpenVibe.Events through the relay', asyn
const tokenClient = { async authHeaders() { return { Authorization: 'Bearer relay' }; }, invalidate() {} };
const svc = await boot({ env: { EVENTS_URL: `http://127.0.0.1:${events.address().port}` }, tokenClient });
try {
-- svc.registry.create(sourceDef({ key: 'ev', endpoints: [`${s.origin}/f`, `${s.origin}/bad`] }), 'test');
++ await svc.registry.create(sourceDef({ key: 'ev', endpoints: [`${s.origin}/f`, `${s.origin}/bad`] }), 'test');
await svc.ingest.run('ev', { trigger: 'manual' });
-- for (let i = 0; i < 5 && svc.outbox.pending(); i++) await svc.relay.flush();
-- assert.strictEqual(svc.outbox.pending(), 0);
++ for (let i = 0; i < 5 && await svc.outbox.pending(); i++) await svc.relay.flush();
++ assert.strictEqual(await svc.outbox.pending(), 0);
assert.deepStrictEqual(received.map(e => e.event_type).sort(), ['sources.fetch.failed', 'sources.item.created']);
for (const e of received) {
const v = validate('events.event-envelope@1', e);
ddiff --git a/test/security-secrets.test.js b/test/security-secrets.test.js
iindex 3e80aab..3a45826 100644
---- a/test/security-secrets.test.js
++++ b/test/security-secrets.test.js
@@@ -42,7 +42,7 @@ const fill = (p, v) => p.replace(/:([A-Za-z0-9_]+)/g, () => encodeURIComponent(v
t('boot with a sentinel client secret and a source whose endpoint carries a key', async () => {
web = await site({ '/robots.txt': () => ({ status: 404 }), '/feed.xml': () => ({ status: 500, body: 'upstream broke' }) });
svc = await boot({ env: { OV_OAUTH_CLIENT_SECRET: SECRET, OV_OAUTH_CLIENT_ID: 'sources', OV_NETWORK_INTERNAL_URL: 'http://127.0.0.1:9' } });
-- svc.registry.create(sourceDef({ key: 'keyed', endpoints: [`${web.origin}/feed.xml?api_key=${FEED_KEY}`] }), 'test');
++ await svc.registry.create(sourceDef({ key: 'keyed', endpoints: [`${web.origin}/feed.xml?api_key=${FEED_KEY}`] }), 'test');
await svc.ingest.run('keyed', { trigger: 'manual' });
});
@@@ -76,8 +76,8 @@ t('every route, as every kind of caller, with real and nonsense ids: the client
assert.deepStrictEqual(refusedApi, [], 'an /api/v1 route answered without a valid service token');
});
--t('the outbox carries neither', () => {
-- const rows = JSON.stringify(svc.outbox.all ? svc.outbox.all() : svc.db.prepare('SELECT * FROM event_outbox').all());
++t('the outbox carries neither', async () => {
++ const rows = JSON.stringify(svc.outbox.all ? await svc.outbox.all() : await svc.db.prepare('SELECT * FROM event_outbox').all());
assert.ok(!rows.includes(SECRET), 'client secret in the outbox');
assert.ok(!rows.includes(FEED_KEY), 'feed key in the outbox');
});
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex b35a73e..06ea9a5 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -79,7 +79,7 @@ t('endpoints spelled as internal addresses (with the stub\'s port) are refused a
let i = 0;
for (const host of ['127.0.0.1', '2130706433', '0177.0.0.1', '0x7f000001', '127.1', '[::ffff:127.0.0.1]', '[::ffff:7f00:1]', '[::1]', '0.0.0.0']) {
const key = `spell-${i++}`;
-- svc.registry.create(sourceDef({ key, endpoints: [`http://${host}:${port}/feed.xml`] }), 'test');
++ await svc.registry.create(sourceDef({ key, endpoints: [`http://${host}:${port}/feed.xml`] }), 'test');
const out = await svc.ingest.run(key, { trigger: 'manual' });
assert.deepStrictEqual([out.runs[0].state, out.runs[0].error_code], ['http_error', 'address_refused'], host);
}
@@@ -102,7 +102,7 @@ t('a name whose DNS answers include any internal address is refused where the co
const n = web.requests.length;
for (const host of Object.keys(table)) {
const key = `dns-${host.split('.')[0]}`;
-- svc.registry.create(sourceDef({ key, endpoints: [`http://${host}:${port}/feed.xml`] }), 'test');
++ await svc.registry.create(sourceDef({ key, endpoints: [`http://${host}:${port}/feed.xml`] }), 'test');
const out = await svc.ingest.run(key, { trigger: 'manual' });
assert.deepStrictEqual([out.runs[0].state, out.runs[0].error_code], ['http_error', 'address_refused'], host);
}
@@@ -120,7 +120,7 @@ t('a redirect to any internal spelling is refused at that hop; the internal targ
for (const to of [`http://127.0.0.1:${port}/feed.xml`, `http://2130706433:${port}/feed.xml`, `http://0x7f.1:${port}/feed.xml`, `http://[::ffff:7f00:1]:${port}/feed.xml`,
`http://[::1]:${port}/feed.xml`, 'http://169.254.169.254/latest/meta-data/', 'http://10.0.0.5/admin', 'file:///etc/passwd', `gopher://127.0.0.1:${port}/_x`]) {
const key = `redir-${i++}`;
-- svc.registry.create(sourceDef({ key, endpoints: [`http://hop.example.com:${port}/to/${encodeURIComponent(to)}`] }), 'test');
++ await svc.registry.create(sourceDef({ key, endpoints: [`http://hop.example.com:${port}/to/${encodeURIComponent(to)}`] }), 'test');
const before = web.hits('/feed.xml').length;
const out = await svc.ingest.run(key, { trigger: 'manual' });
assert.strictEqual(out.runs[0].state, 'http_error', `${to}: ${out.runs[0].state}`);
@@@ -142,7 +142,7 @@ t('a source registered over the API with internal endpoints is stored, and its f
const n = web.requests.length;
const r = await request(svc.base, 'POST', '/api/v1/sources/api-internal/fetch', { token, body: {} });
assert.ok(r.status < 500, r.text.slice(0, 200));
-- for (let i = 0; i < 20 && svc.db.prepare("SELECT COUNT(*) AS n FROM fetch_runs WHERE source_key = 'api-internal'").get().n < 2; i++) await new Promise((res) => setTimeout(res, 50));
++ for (let i = 0; i < 20 && (await svc.db.prepare("SELECT COUNT(*) AS n FROM fetch_runs WHERE source_key = 'api-internal'").get()).n < 2; i++) await new Promise((res) => setTimeout(res, 50));
assert.strictEqual(web.requests.length, n, 'the internal endpoint was fetched');
}
} finally { await svc.stop(); }