openvibers/openvibe.work Commits
Work: listing pages in OpenVibe.Search (search-feed, work.index_document.*) (#10)
Changes
@@@ -242,7 +242,8 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).<!-- versions:start -->--- openvibe-contracts: v0.127.0++- openvibe-contracts: v0.128.0++- openvibe-publishing: v1.4.0- openvibe-sdk: v0.36.0- openvibe-shared: v2.20.4<!-- versions:end -->@@@ -7,13 +7,14 @@"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9","port": "4960","domain": "openvibe.work",-- "contracts": "openvibe-contracts v0.127.0",++ "contracts": "openvibe-contracts v0.128.0","sdk": "openvibe-sdk v0.36.0","shared": "openvibe-shared v2.20.4","packages": [-- "openvibe-contracts v0.127.0",++ "openvibe-contracts v0.128.0","openvibe-sdk v0.36.0",-- "openvibe-shared v2.20.4"++ "openvibe-shared v2.20.4",++ "openvibe-publishing v1.4.0"],"summary": "OpenVibe.Work gathers job listings from three public job boards (Arbeitnow, Remotive, RemoteOK) under their terms — the source is named and the original listing is linked on every page, and only a short excerpt is kept. It searches them with facets, shows one listing with a schema.org JobPosting whose url is the original, and lets a signed-in person keep a search and see what is new since they last looked. Public reads need no token.","works": [nnew file mode 100644@@@ -0,0 +1,31 @@++-- phase: expand++-- OpenVibe.Work: listing pages in OpenVibe.Search (server/search-index.js, openvibe-publishing/search-feed).++-- event_outbox holds the work.index_document.* events until the relay delivers them to OpenVibe.Events (written in++-- the same transaction as the revision that produced them); work_index_revisions is the index sequencer, the last++-- revision and content hash Search was sent for each page, so an unchanged page is never sent twice.++++CREATE TABLE IF NOT EXISTS event_outbox (++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,++ event_id text NOT NULL UNIQUE,++ envelope jsonb NOT NULL,++ traceparent text,++ created_at bigint NOT NULL,++ attempts integer NOT NULL DEFAULT 0,++ next_attempt_at bigint NOT NULL DEFAULT 0,++ sent_at bigint,++ seq bigint,++ rejected_at bigint,++ last_error text++);++CREATE INDEX IF NOT EXISTS event_outbox_due ON event_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;++CREATE INDEX IF NOT EXISTS event_outbox_sent ON event_outbox (sent_at) WHERE sent_at IS NOT NULL;++++CREATE TABLE IF NOT EXISTS work_index_revisions (++ owner text COLLATE "C" NOT NULL,++ type text COLLATE "C" NOT NULL,++ id text COLLATE "C" NOT NULL,++ revision integer NOT NULL,++ hash text NOT NULL,++ updated_at bigint NOT NULL,++ PRIMARY KEY (owner, type, id)++);@@@ -15,7 +15,8 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.127.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.128.0",++ "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.20.4","pg": "^8.23.0"@@@ -871,9 +872,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.127.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.127.0",-- "integrity": "sha512-ySC+Qh8I3OlN5MULCqv0vfcAGubZQzWM0Fw7EOpnOo76vVnrkwx+m33aTxDum2yQqPkU8Ot5UtMuMwm1Cc1k3g==",++ "version": "0.128.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.128.0",++ "integrity": "sha512-r/WKk3NhdqxEsVxqsVPjg+HTD2So1amRx0BY6EOHhvmwjLuL9j5XJrNDZAfqf/4/P2xZCBzTLrtUqlWxC2EA6Q==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -886,6 +887,24 @@"node": ">=20"}},++ "node_modules/openvibe-publishing": {++ "version": "1.4.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.0",++ "integrity": "sha512-9vByyVqpvy2zVQI4PrJWk5QqCym/YQ/fXZDgrcloKGBqvukKrUn7fg4ZERiJHcgEq+rgDbxbU4rNflZhw62K/Q==",++ "license": "MIT",++ "engines": {++ "node": ">=20"++ },++ "peerDependencies": {++ "openvibe-sdk": ">=0.15.0",++ "openvibe-shared": ">=1.5.0"++ },++ "peerDependenciesMeta": {++ "openvibe-shared": {++ "optional": true++ }++ }++ },"node_modules/openvibe-sdk": {"version": "0.36.0","resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0",@@@ -2,7 +2,7 @@"name": "openvibe-work","version": "0.2.0","private": true,-- "description": "OpenVibe.Work — Find work, with an agent on your side.",++ "description": "OpenVibe.Work \u2014 Find work, with an agent on your side.","main": "server/index.js","scripts": {"start": "node server/index.js",@@@ -23,7 +23,8 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.127.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.128.0",++ "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.20.4","pg": "^8.23.0"@@@ -31,6 +31,7 @@ const { createPageRoutes } = require('./http/pages');const { createServiceReadiness } = require('./observability');const { createCallerLimits } = require('./http/caller-limits');const { createIngest } = require('./ingest');++const { createSearchIndex } = require('./search-index');const accountDataLib = require('./identity/account-data');const { createNetworkSender } = require('openvibe-sdk/account-data');const { assetVersion, send } = require('./render/layout');@@@ -67,7 +68,10 @@ async function createApp(opts = {}) {ctx.limits = createCallerLimits({ config, now: opts.limitsNow || (() => Date.now()), registry: metrics.registry, log, enabled: opts.callerLimits !== false, valkey });// The job-board ingest. Built here so a test can run one source against its stand-in server, but started only by// server/index.js, and only when WORK_INGEST=on: nothing fetches a board in a test or a development run.-- ctx.ingest = createIngest({ config, s, fetchImpl, log });++ // Listing pages in OpenVibe.Search (./search-index.js): swept after every ingest run, and on its own timer once++ // server/index.js starts it. The outbox relay is off until the events URL and the OAuth client secret are set.++ ctx.search = opts.search || createSearchIndex({ config, s, log });++ ctx.ingest = createIngest({ config, s, fetchImpl, log, afterRun: () => ctx.search.afterIngest() });// Account export and deletion (ADR-033, ./identity/account-data.js): the one table that holds a person's rows.// The sender posts to Network's internal export/deletion routes with this service's own client-credentials token;@@@ -15,7 +15,7 @@ const { gracefulStop } = require('openvibe-sdk/service');function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {return gracefulStop({name: 'OpenVibe.Work', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,-- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.ingest.stop(), () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],++ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.ingest.stop(), () => ctx.search.stop(), () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],});}@@@ -29,6 +29,7 @@ async function start() {server.keepAliveTimeout = 65_000;ctx.keys.client.start();ctx.ingest.start();++ ctx.search.start();// Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without WORK_EVENTS_URL and// WORK_EVENTS_SECRET). The consumer itself is mounted in server/app.js.@@@ -36,7 +36,7 @@ async function fetchJson(url, { fetchImpl, timeoutMs, userAgent }) {const TICK_MS = 3_600_000;--function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console }) {++function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console, afterRun = null }) {const timers = new Map();let running = false;@@@ -95,6 +95,8 @@ function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console }} catch (err) {log.warn(`[Ingest] expiry failed: ${(err && err.message) || err}`);}++ // Search catches up with what this run changed (server/search-index.js); never awaited, never fatal.++ if (afterRun && out.some((o) => !o.skipped)) afterRun();return out;} finally { running = false; }}nnew file mode 100644@@@ -0,0 +1,94 @@++'use strict';++++/**++ * Work's listing pages in OpenVibe.Search (openvibe-publishing/search-feed): one search.index-document@1 per job page++ * (/jobs/<id>), sent as work.index_document.upserted|deleted through this service's events outbox. A listing that++ * expired (its page answers 404) becomes a tombstone. The document carries what the page shows of a listing (title,++ * company, location, tags, salary text, the short excerpt) and names the board it came from as provenance, with the++ * original listing's URL: the attribution the boards' terms ask for travels with the document.++ *++ * Listings change in bulk (an ingest run upserts a board's whole page), so Search is brought level by a sweep: after++ * every ingest run, once a few seconds after boot and every hour. The sweep re-sends only what changed (the sequencer++ * in work_index_revisions gives an unchanged document its old revision), so a quiet hour sends nothing.++ *++ * const search = createSearchIndex({ config, s, log }); // outbox: the relay is off until the events URL and the++ * search.start(); search.sweep(); search.stop(); // OAuth client secret are set; rows wait meanwhile++ */++const { createServiceOutbox } = require('openvibe-sdk/events');++const { createSearchFeed } = require('openvibe-publishing/search-feed');++const sources = require('./jobs/sources');++const listing = require('./render/listing');++++const EVENT_TYPES = ['work.index_document.upserted', 'work.index_document.deleted'];++const START_DELAY_MS = 60_000;++const INTERVAL_MS = 3_600_000;++++/** One listing row → what its page shows, as search-feed's document description. */++function describe(row) {++ const l = listing.meta(row);++ const src = sources.byId(row.source);++ const where = [l.location, l.remote ? 'remote' : '', l.job_type].filter(Boolean).join(' · ');++ const excerpt = listing.plainExcerpt(row);++ return {++ listed: !row.expired,++ title: `${l.title} at ${l.company}`,++ summary: [where, l.salary].filter(Boolean).join(' · ') || excerpt.slice(0, 200) || null,++ body: [l.title, l.company, where, l.tags.join(', '), l.salary, excerpt].filter(Boolean).join('\n'),++ facets: { source: row.source, remote: l.remote, type: l.job_type || null, location: l.location || null },++ authorship: { mode: 'imported' },++ provenance: [{++ service: 'work', type: 'board_listing', id: `${row.source}:${row.source_id}`.slice(0, 128),++ label: src ? src.name : row.source, url: row.url, retrievedAt: row.fetched_at,++ }],++ publishedAt: row.posted_at || row.first_seen_at,++ updatedAt: row.fetched_at,++ };++}++++function createSearchIndex({ config, s, log = console, outbox = null }) {++ const out = outbox || createServiceOutbox({++ db: s.db, source: 'work', eventsUrl: config.events.url || null, networkInternalUrl: config.networkInternalUrl,++ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, log, eventTypes: EVENT_TYPES,++ });++ const feed = createSearchFeed({++ owner: 'work', db: s.db, outbox: out, baseUrl: config.baseUrl, now: s.now, log,++ types: {++ job: {++ page: (row) => `/jobs/${row.id}`,++ document: describe,++ rows: (after, limit) => s.db.many('SELECT * FROM work_listings WHERE id > $1 ORDER BY id LIMIT $2', [after, limit]),++ exists: async (ids) => (await s.db.many('SELECT id FROM work_listings WHERE id = ANY($1)', [ids])).map((r) => r.id),++ },++ },++ });++ const timers = [];++++ async function sweep() {++ const res = await feed.sweep();++ const j = res.job;++ if (j.sent || j.removed || j.failed) log.log(`[Search] listings: ${j.sent} sent, ${j.removed} removed, ${j.failed} failed of ${j.seen}`);++ out.kick && out.kick().catch(() => {});++ return res;++ }++ const quietly = () => { sweep().catch((err) => log.warn(`[Search] sweep failed: ${(err && err.message) || err}`)); };++++ function start() {++ if (timers.length) return false;++ out.start();++ const kick = setTimeout(quietly, START_DELAY_MS);++ kick.unref();++ const tick = setInterval(quietly, INTERVAL_MS);++ tick.unref();++ timers.push(kick, tick);++ return true;++ }++++ function stop() {++ for (const t of timers.splice(0)) { clearTimeout(t); clearInterval(t); }++ out.stop();++ }++++ return { feed, outbox: out, describe, sweep, afterIngest: quietly, start, stop, status: () => out.status() };++}++++module.exports = { createSearchIndex, describe, EVENT_TYPES };@@@ -77,6 +77,7 @@ async function boot(opts = {}) {logs: () => captured.join('\n'),async close() {built.ctx.ingest.stop();++ built.ctx.search.stop();await new Promise((r) => server.close(r));built.ctx.keys.client.stop();await testdb.close();nnew file mode 100644@@@ -0,0 +1,109 @@++'use strict';++/**++ * Listing pages in OpenVibe.Search (server/search-index.js): after an ingest run every listing is one++ * work.index_document.upserted in the outbox, exactly the contract (the event payload and search.index-document@1),++ * with the page's URL, the board as provenance and the original listing's URL; a second sweep sends nothing; a listing++ * that changes is re-sent at the next revision; one that expires becomes a tombstone; the ingest run itself triggers++ * the sweep.++ */++const assert = require('assert');++const contracts = require('openvibe-contracts');++const { boot, check, done } = require('./helpers/boot');++++const DAY = 86_400_000;++++(async () => {++ const clock = { t: Date.parse('2026-10-01T06:00:00Z') };++ const t = await boot({ now: () => clock.t, env: { WORK_INGEST_PAGES: '2' } });++ const { s, search, ingest } = t.ctx;++ const outbox = async () => (await s.db.many('SELECT envelope FROM event_outbox ORDER BY id')).map((r) => r.envelope);++ const valid = (env) => {++ const e = contracts.validate('events.event-envelope@1', env);++ assert.ok(e.valid, `envelope: ${JSON.stringify(e.errors)}`);++ const p = contracts.validate(env.event_type, env.payload);++ assert.ok(p.valid, `${env.event_type}: ${JSON.stringify(p.errors)}`);++ if (env.event_type.endsWith('.upserted')) {++ const d = contracts.validate('search.index-document@1', env.payload);++ assert.ok(d.valid, `search.index-document@1: ${JSON.stringify(d.errors)}`);++ }++ assert.strictEqual(env.source, 'work');++ assert.strictEqual(env.visibility, 'internal');++ assert.strictEqual(env.subject.id, env.payload.id);++ return env;++ };++++ try {++ await check('an ingest run puts every listing in the outbox, as the contract says', async () => {++ await ingest.runAll();++ await search.sweep(); // the run's own sweep is fire-and-forget; this one waits (and shares or repeats it)++ const envs = (await outbox()).map(valid);++ const listings = await s.db.many('SELECT * FROM work_listings ORDER BY id');++ assert.strictEqual(listings.length, 8);++ assert.strictEqual(envs.length, 8, 'one document per listing, sent once');++ assert.ok(envs.every((e) => e.event_type === 'work.index_document.upserted'));++ const byId = new Map(envs.map((e) => [e.payload.id, e.payload]));++ const devops = listings.find((r) => r.title.includes('DevOps'));++ const doc = byId.get(devops.id);++ assert.strictEqual(doc.canonical_url, `https://openvibe.work/jobs/${devops.id}`);++ assert.strictEqual(doc.title, `${devops.title} at ${devops.company}`);++ assert.strictEqual(doc.authorship, 'imported');++ assert.strictEqual(doc.visibility, 'public');++ assert.deepStrictEqual(doc.indexability, { decision: 'index', reasons: [] });++ assert.strictEqual(doc.facets.source, 'remoteok');++ assert.strictEqual(doc.facets.remote, true);++ assert.match(doc.body, /Skyward Cloud/);++ assert.match(doc.summary, /\$90,000–\$130,000 a year/);++ assert.deepStrictEqual(doc.provenance.map((p) => [p.service, p.type, p.label, p.url]),++ [['work', 'board_listing', 'RemoteOK', devops.url]], 'the board and the original listing travel with it');++ assert.ok(!JSON.stringify(envs).includes('personal use'), 'RemoteOK\'s legal notice is not a document');++ });++++ await check('a second sweep sends nothing; a changed listing is sent again at the next revision', async () => {++ const before = (await outbox()).length;++ const again = await search.sweep();++ assert.deepStrictEqual(again.job, { seen: 8, sent: 0, removed: 0, failed: 0 });++ assert.strictEqual((await outbox()).length, before);++ const one = await s.db.one("SELECT * FROM work_listings WHERE source = 'arbeitnow' ORDER BY id LIMIT 1");++ await s.db.exec('UPDATE work_listings SET salary = $1 WHERE id = $2', ['€70,000 a year', one.id]);++ const res = await search.sweep();++ assert.strictEqual(res.job.sent, 1);++ const last = valid((await outbox()).pop());++ assert.strictEqual(last.payload.id, one.id);++ assert.strictEqual(last.payload.revision, 2);++ assert.match(last.payload.summary, /€70,000 a year/);++ });++++ await check('a listing that expires (its page answers 404) becomes a tombstone, once', async () => {++ const one = await s.db.one("SELECT * FROM work_listings WHERE source = 'remotive' ORDER BY id LIMIT 1");++ await s.db.exec('UPDATE work_listings SET expired = true WHERE id = $1', [one.id]);++ assert.strictEqual((await t.get(`/jobs/${one.id}`)).status, 404);++ const res = await search.sweep();++ assert.strictEqual(res.job.sent, 1);++ const last = valid((await outbox()).pop());++ assert.strictEqual(last.event_type, 'work.index_document.deleted');++ assert.deepStrictEqual(last.payload, { type: 'job', id: one.id, revision: 2 });++ assert.strictEqual((await search.sweep()).job.sent, 0, 'a tombstone is not re-sent');++ // A row deleted outright is tombstoned too (exists()).++ const gone = await s.db.one("SELECT * FROM work_listings WHERE source = 'arbeitnow' AND expired = false ORDER BY id DESC LIMIT 1");++ await s.db.exec('DELETE FROM work_listings WHERE id = $1', [gone.id]);++ assert.strictEqual((await search.sweep()).job.removed, 1);++ });++++ await check('the ingest run triggers the sweep itself', async () => {++ const before = (await outbox()).length;++ clock.t += 3 * DAY;++ t.boards.add && t.boards.add('remotive', { id: 990001, title: 'Staff platform engineer', company_name: 'Northwind', url: 'https://remotive.com/remote-jobs/software-dev/staff-platform-engineer-990001', candidate_required_location: 'Worldwide', job_type: 'full_time', publication_date: '2026-10-04T08:00:00', description: '<p>Platform work.</p>', tags: ['go'] });++ await ingest.runAll();++ for (let i = 0; i < 100 && (await outbox()).length === before; i += 1) await new Promise((r) => setTimeout(r, 50));++ assert.ok((await outbox()).length > before, 'the run swept: something changed (fetched_at moved), so something was sent');++ });++++ await check('the outbox relay stays off without the events URL and the client secret; rows wait', async () => {++ const st = await search.status();++ assert.strictEqual(st.enabled, false);++ assert.ok(st.pending > 0);++ });++ } finally {++ await done(t);++ }++})();
@@ -242,7 +242,8 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).
@@ -242,7 +242,8 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).
Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
<!-- versions:start -->
<!-- versions:start -->
-- openvibe-contracts: v0.127.0
+- openvibe-contracts: v0.128.0
+- openvibe-publishing: v1.4.0
- openvibe-sdk: v0.36.0
- openvibe-sdk: v0.36.0
- openvibe-shared: v2.20.4
- openvibe-shared: v2.20.4
<!-- versions:end -->
<!-- versions:end -->
@@ -7,13 +7,14 @@
@@ -7,13 +7,14 @@
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"port": "4960",
"port": "4960",
"domain": "openvibe.work",
"domain": "openvibe.work",
- "contracts": "openvibe-contracts v0.127.0",
+ "contracts": "openvibe-contracts v0.128.0",
"sdk": "openvibe-sdk v0.36.0",
"sdk": "openvibe-sdk v0.36.0",
"shared": "openvibe-shared v2.20.4",
"shared": "openvibe-shared v2.20.4",
"packages": [
"packages": [
- "openvibe-contracts v0.127.0",
+ "openvibe-contracts v0.128.0",
"openvibe-sdk v0.36.0",
"openvibe-sdk v0.36.0",
- "openvibe-shared v2.20.4"
+ "openvibe-shared v2.20.4",
+ "openvibe-publishing v1.4.0"
],
],
"summary": "OpenVibe.Work gathers job listings from three public job boards (Arbeitnow, Remotive, RemoteOK) under their terms — the source is named and the original listing is linked on every page, and only a short excerpt is kept. It searches them with facets, shows one listing with a schema.org JobPosting whose url is the original, and lets a signed-in person keep a search and see what is new since they last looked. Public reads need no token.",
"summary": "OpenVibe.Work gathers job listings from three public job boards (Arbeitnow, Remotive, RemoteOK) under their terms — the source is named and the original listing is linked on every page, and only a short excerpt is kept. It searches them with facets, shows one listing with a schema.org JobPosting whose url is the original, and lets a signed-in person keep a search and see what is new since they last looked. Public reads need no token.",
"works": [
"works": [
new file mode 100644
new file mode 100644
@@ -0,0 +1,31 @@
@@ -0,0 +1,31 @@
+-- phase: expand
+-- OpenVibe.Work: listing pages in OpenVibe.Search (server/search-index.js, openvibe-publishing/search-feed).
+-- event_outbox holds the work.index_document.* events until the relay delivers them to OpenVibe.Events (written in
+-- the same transaction as the revision that produced them); work_index_revisions is the index sequencer, the last
+-- revision and content hash Search was sent for each page, so an unchanged page is never sent twice.
+
+CREATE TABLE IF NOT EXISTS event_outbox (
+ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+ event_id text NOT NULL UNIQUE,
+ envelope jsonb NOT NULL,
+ traceparent text,
+ created_at bigint NOT NULL,
+ attempts integer NOT NULL DEFAULT 0,
+ next_attempt_at bigint NOT NULL DEFAULT 0,
+ sent_at bigint,
+ seq bigint,
+ rejected_at bigint,
+ last_error text
+);
+CREATE INDEX IF NOT EXISTS event_outbox_due ON event_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
+CREATE INDEX IF NOT EXISTS event_outbox_sent ON event_outbox (sent_at) WHERE sent_at IS NOT NULL;
+
+CREATE TABLE IF NOT EXISTS work_index_revisions (
+ owner text COLLATE "C" NOT NULL,
+ type text COLLATE "C" NOT NULL,
+ id text COLLATE "C" NOT NULL,
+ revision integer NOT NULL,
+ hash text NOT NULL,
+ updated_at bigint NOT NULL,
+ PRIMARY KEY (owner, type, id)
+);
@@ -15,7 +15,8 @@
@@ -15,7 +15,8 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.127.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.128.0",
+ "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.20.4",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.20.4",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -871,9 +872,9 @@
@@ -871,9 +872,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.127.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.127.0",
- "integrity": "sha512-ySC+Qh8I3OlN5MULCqv0vfcAGubZQzWM0Fw7EOpnOo76vVnrkwx+m33aTxDum2yQqPkU8Ot5UtMuMwm1Cc1k3g==",
+ "version": "0.128.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.128.0",
+ "integrity": "sha512-r/WKk3NhdqxEsVxqsVPjg+HTD2So1amRx0BY6EOHhvmwjLuL9j5XJrNDZAfqf/4/P2xZCBzTLrtUqlWxC2EA6Q==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -886,6 +887,24 @@
@@ -886,6 +887,24 @@
"node": ">=20"
"node": ">=20"
}
}
},
},
+ "node_modules/openvibe-publishing": {
+ "version": "1.4.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.0",
+ "integrity": "sha512-9vByyVqpvy2zVQI4PrJWk5QqCym/YQ/fXZDgrcloKGBqvukKrUn7fg4ZERiJHcgEq+rgDbxbU4rNflZhw62K/Q==",
+ "license": "MIT",
+ "engines": {
+ "node": ">=20"
+ },
+ "peerDependencies": {
+ "openvibe-sdk": ">=0.15.0",
+ "openvibe-shared": ">=1.5.0"
+ },
+ "peerDependenciesMeta": {
+ "openvibe-shared": {
+ "optional": true
+ }
+ }
+ },
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
"version": "0.36.0",
"version": "0.36.0",
"resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0",
"resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0",
@@ -2,7 +2,7 @@
@@ -2,7 +2,7 @@
"name": "openvibe-work",
"name": "openvibe-work",
"version": "0.2.0",
"version": "0.2.0",
"private": true,
"private": true,
- "description": "OpenVibe.Work — Find work, with an agent on your side.",
+ "description": "OpenVibe.Work \u2014 Find work, with an agent on your side.",
"main": "server/index.js",
"main": "server/index.js",
"scripts": {
"scripts": {
"start": "node server/index.js",
"start": "node server/index.js",
@@ -23,7 +23,8 @@
@@ -23,7 +23,8 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.127.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.128.0",
+ "openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.4.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.36.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.20.4",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.20.4",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -31,6 +31,7 @@ const { createPageRoutes } = require('./http/pages');
@@ -31,6 +31,7 @@ const { createPageRoutes } = require('./http/pages');
const { createServiceReadiness } = require('./observability');
const { createServiceReadiness } = require('./observability');
const { createCallerLimits } = require('./http/caller-limits');
const { createCallerLimits } = require('./http/caller-limits');
const { createIngest } = require('./ingest');
const { createIngest } = require('./ingest');
+const { createSearchIndex } = require('./search-index');
const accountDataLib = require('./identity/account-data');
const accountDataLib = require('./identity/account-data');
const { createNetworkSender } = require('openvibe-sdk/account-data');
const { createNetworkSender } = require('openvibe-sdk/account-data');
const { assetVersion, send } = require('./render/layout');
const { assetVersion, send } = require('./render/layout');
@@ -67,7 +68,10 @@ async function createApp(opts = {}) {
@@ -67,7 +68,10 @@ async function createApp(opts = {}) {
ctx.limits = createCallerLimits({ config, now: opts.limitsNow || (() => Date.now()), registry: metrics.registry, log, enabled: opts.callerLimits !== false, valkey });
ctx.limits = createCallerLimits({ config, now: opts.limitsNow || (() => Date.now()), registry: metrics.registry, log, enabled: opts.callerLimits !== false, valkey });
// The job-board ingest. Built here so a test can run one source against its stand-in server, but started only by
// The job-board ingest. Built here so a test can run one source against its stand-in server, but started only by
// server/index.js, and only when WORK_INGEST=on: nothing fetches a board in a test or a development run.
// server/index.js, and only when WORK_INGEST=on: nothing fetches a board in a test or a development run.
- ctx.ingest = createIngest({ config, s, fetchImpl, log });
+ // Listing pages in OpenVibe.Search (./search-index.js): swept after every ingest run, and on its own timer once
+ // server/index.js starts it. The outbox relay is off until the events URL and the OAuth client secret are set.
+ ctx.search = opts.search || createSearchIndex({ config, s, log });
+ ctx.ingest = createIngest({ config, s, fetchImpl, log, afterRun: () => ctx.search.afterIngest() });
// Account export and deletion (ADR-033, ./identity/account-data.js): the one table that holds a person's rows.
// Account export and deletion (ADR-033, ./identity/account-data.js): the one table that holds a person's rows.
// The sender posts to Network's internal export/deletion routes with this service's own client-credentials token;
// The sender posts to Network's internal export/deletion routes with this service's own client-credentials token;
@@ -15,7 +15,7 @@ const { gracefulStop } = require('openvibe-sdk/service');
@@ -15,7 +15,7 @@ const { gracefulStop } = require('openvibe-sdk/service');
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
return gracefulStop({
return gracefulStop({
name: 'OpenVibe.Work', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
name: 'OpenVibe.Work', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.ingest.stop(), () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],
+ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.ingest.stop(), () => ctx.search.stop(), () => ctx.keys.client.stop(), () => ctx.s.close(), ...extra],
});
});
}
}
@@ -29,6 +29,7 @@ async function start() {
@@ -29,6 +29,7 @@ async function start() {
server.keepAliveTimeout = 65_000;
server.keepAliveTimeout = 65_000;
ctx.keys.client.start();
ctx.keys.client.start();
ctx.ingest.start();
ctx.ingest.start();
+ ctx.search.start();
// Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without WORK_EVENTS_URL and
// Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without WORK_EVENTS_URL and
// WORK_EVENTS_SECRET). The consumer itself is mounted in server/app.js.
// WORK_EVENTS_SECRET). The consumer itself is mounted in server/app.js.
@@ -36,7 +36,7 @@ async function fetchJson(url, { fetchImpl, timeoutMs, userAgent }) {
@@ -36,7 +36,7 @@ async function fetchJson(url, { fetchImpl, timeoutMs, userAgent }) {
const TICK_MS = 3_600_000;
const TICK_MS = 3_600_000;
-function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console }) {
+function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console, afterRun = null }) {
const timers = new Map();
const timers = new Map();
let running = false;
let running = false;
@@ -95,6 +95,8 @@ function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console }
@@ -95,6 +95,8 @@ function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console }
} catch (err) {
} catch (err) {
log.warn(`[Ingest] expiry failed: ${(err && err.message) || err}`);
log.warn(`[Ingest] expiry failed: ${(err && err.message) || err}`);
}
}
+ // Search catches up with what this run changed (server/search-index.js); never awaited, never fatal.
+ if (afterRun && out.some((o) => !o.skipped)) afterRun();
return out;
return out;
} finally { running = false; }
} finally { running = false; }
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,94 @@
@@ -0,0 +1,94 @@
+'use strict';
+
+/**
+ * Work's listing pages in OpenVibe.Search (openvibe-publishing/search-feed): one search.index-document@1 per job page
+ * (/jobs/<id>), sent as work.index_document.upserted|deleted through this service's events outbox. A listing that
+ * expired (its page answers 404) becomes a tombstone. The document carries what the page shows of a listing (title,
+ * company, location, tags, salary text, the short excerpt) and names the board it came from as provenance, with the
+ * original listing's URL: the attribution the boards' terms ask for travels with the document.
+ *
+ * Listings change in bulk (an ingest run upserts a board's whole page), so Search is brought level by a sweep: after
+ * every ingest run, once a few seconds after boot and every hour. The sweep re-sends only what changed (the sequencer
+ * in work_index_revisions gives an unchanged document its old revision), so a quiet hour sends nothing.
+ *
+ * const search = createSearchIndex({ config, s, log }); // outbox: the relay is off until the events URL and the
+ * search.start(); search.sweep(); search.stop(); // OAuth client secret are set; rows wait meanwhile
+ */
+const { createServiceOutbox } = require('openvibe-sdk/events');
+const { createSearchFeed } = require('openvibe-publishing/search-feed');
+const sources = require('./jobs/sources');
+const listing = require('./render/listing');
+
+const EVENT_TYPES = ['work.index_document.upserted', 'work.index_document.deleted'];
+const START_DELAY_MS = 60_000;
+const INTERVAL_MS = 3_600_000;
+
+/** One listing row → what its page shows, as search-feed's document description. */
+function describe(row) {
+ const l = listing.meta(row);
+ const src = sources.byId(row.source);
+ const where = [l.location, l.remote ? 'remote' : '', l.job_type].filter(Boolean).join(' · ');
+ const excerpt = listing.plainExcerpt(row);
+ return {
+ listed: !row.expired,
+ title: `${l.title} at ${l.company}`,
+ summary: [where, l.salary].filter(Boolean).join(' · ') || excerpt.slice(0, 200) || null,
+ body: [l.title, l.company, where, l.tags.join(', '), l.salary, excerpt].filter(Boolean).join('\n'),
+ facets: { source: row.source, remote: l.remote, type: l.job_type || null, location: l.location || null },
+ authorship: { mode: 'imported' },
+ provenance: [{
+ service: 'work', type: 'board_listing', id: `${row.source}:${row.source_id}`.slice(0, 128),
+ label: src ? src.name : row.source, url: row.url, retrievedAt: row.fetched_at,
+ }],
+ publishedAt: row.posted_at || row.first_seen_at,
+ updatedAt: row.fetched_at,
+ };
+}
+
+function createSearchIndex({ config, s, log = console, outbox = null }) {
+ const out = outbox || createServiceOutbox({
+ db: s.db, source: 'work', eventsUrl: config.events.url || null, networkInternalUrl: config.networkInternalUrl,
+ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, log, eventTypes: EVENT_TYPES,
+ });
+ const feed = createSearchFeed({
+ owner: 'work', db: s.db, outbox: out, baseUrl: config.baseUrl, now: s.now, log,
+ types: {
+ job: {
+ page: (row) => `/jobs/${row.id}`,
+ document: describe,
+ rows: (after, limit) => s.db.many('SELECT * FROM work_listings WHERE id > $1 ORDER BY id LIMIT $2', [after, limit]),
+ exists: async (ids) => (await s.db.many('SELECT id FROM work_listings WHERE id = ANY($1)', [ids])).map((r) => r.id),
+ },
+ },
+ });
+ const timers = [];
+
+ async function sweep() {
+ const res = await feed.sweep();
+ const j = res.job;
+ if (j.sent || j.removed || j.failed) log.log(`[Search] listings: ${j.sent} sent, ${j.removed} removed, ${j.failed} failed of ${j.seen}`);
+ out.kick && out.kick().catch(() => {});
+ return res;
+ }
+ const quietly = () => { sweep().catch((err) => log.warn(`[Search] sweep failed: ${(err && err.message) || err}`)); };
+
+ function start() {
+ if (timers.length) return false;
+ out.start();
+ const kick = setTimeout(quietly, START_DELAY_MS);
+ kick.unref();
+ const tick = setInterval(quietly, INTERVAL_MS);
+ tick.unref();
+ timers.push(kick, tick);
+ return true;
+ }
+
+ function stop() {
+ for (const t of timers.splice(0)) { clearTimeout(t); clearInterval(t); }
+ out.stop();
+ }
+
+ return { feed, outbox: out, describe, sweep, afterIngest: quietly, start, stop, status: () => out.status() };
+}
+
+module.exports = { createSearchIndex, describe, EVENT_TYPES };
@@ -77,6 +77,7 @@ async function boot(opts = {}) {
@@ -77,6 +77,7 @@ async function boot(opts = {}) {
logs: () => captured.join('\n'),
logs: () => captured.join('\n'),
async close() {
async close() {
built.ctx.ingest.stop();
built.ctx.ingest.stop();
+ built.ctx.search.stop();
await new Promise((r) => server.close(r));
await new Promise((r) => server.close(r));
built.ctx.keys.client.stop();
built.ctx.keys.client.stop();
await testdb.close();
await testdb.close();
new file mode 100644
new file mode 100644
@@ -0,0 +1,109 @@
@@ -0,0 +1,109 @@
+'use strict';
+/**
+ * Listing pages in OpenVibe.Search (server/search-index.js): after an ingest run every listing is one
+ * work.index_document.upserted in the outbox, exactly the contract (the event payload and search.index-document@1),
+ * with the page's URL, the board as provenance and the original listing's URL; a second sweep sends nothing; a listing
+ * that changes is re-sent at the next revision; one that expires becomes a tombstone; the ingest run itself triggers
+ * the sweep.
+ */
+const assert = require('assert');
+const contracts = require('openvibe-contracts');
+const { boot, check, done } = require('./helpers/boot');
+
+const DAY = 86_400_000;
+
+(async () => {
+ const clock = { t: Date.parse('2026-10-01T06:00:00Z') };
+ const t = await boot({ now: () => clock.t, env: { WORK_INGEST_PAGES: '2' } });
+ const { s, search, ingest } = t.ctx;
+ const outbox = async () => (await s.db.many('SELECT envelope FROM event_outbox ORDER BY id')).map((r) => r.envelope);
+ const valid = (env) => {
+ const e = contracts.validate('events.event-envelope@1', env);
+ assert.ok(e.valid, `envelope: ${JSON.stringify(e.errors)}`);
+ const p = contracts.validate(env.event_type, env.payload);
+ assert.ok(p.valid, `${env.event_type}: ${JSON.stringify(p.errors)}`);
+ if (env.event_type.endsWith('.upserted')) {
+ const d = contracts.validate('search.index-document@1', env.payload);
+ assert.ok(d.valid, `search.index-document@1: ${JSON.stringify(d.errors)}`);
+ }
+ assert.strictEqual(env.source, 'work');
+ assert.strictEqual(env.visibility, 'internal');
+ assert.strictEqual(env.subject.id, env.payload.id);
+ return env;
+ };
+
+ try {
+ await check('an ingest run puts every listing in the outbox, as the contract says', async () => {
+ await ingest.runAll();
+ await search.sweep(); // the run's own sweep is fire-and-forget; this one waits (and shares or repeats it)
+ const envs = (await outbox()).map(valid);
+ const listings = await s.db.many('SELECT * FROM work_listings ORDER BY id');
+ assert.strictEqual(listings.length, 8);
+ assert.strictEqual(envs.length, 8, 'one document per listing, sent once');
+ assert.ok(envs.every((e) => e.event_type === 'work.index_document.upserted'));
+ const byId = new Map(envs.map((e) => [e.payload.id, e.payload]));
+ const devops = listings.find((r) => r.title.includes('DevOps'));
+ const doc = byId.get(devops.id);
+ assert.strictEqual(doc.canonical_url, `https://openvibe.work/jobs/${devops.id}`);
+ assert.strictEqual(doc.title, `${devops.title} at ${devops.company}`);
+ assert.strictEqual(doc.authorship, 'imported');
+ assert.strictEqual(doc.visibility, 'public');
+ assert.deepStrictEqual(doc.indexability, { decision: 'index', reasons: [] });
+ assert.strictEqual(doc.facets.source, 'remoteok');
+ assert.strictEqual(doc.facets.remote, true);
+ assert.match(doc.body, /Skyward Cloud/);
+ assert.match(doc.summary, /\$90,000–\$130,000 a year/);
+ assert.deepStrictEqual(doc.provenance.map((p) => [p.service, p.type, p.label, p.url]),
+ [['work', 'board_listing', 'RemoteOK', devops.url]], 'the board and the original listing travel with it');
+ assert.ok(!JSON.stringify(envs).includes('personal use'), 'RemoteOK\'s legal notice is not a document');
+ });
+
+ await check('a second sweep sends nothing; a changed listing is sent again at the next revision', async () => {
+ const before = (await outbox()).length;
+ const again = await search.sweep();
+ assert.deepStrictEqual(again.job, { seen: 8, sent: 0, removed: 0, failed: 0 });
+ assert.strictEqual((await outbox()).length, before);
+ const one = await s.db.one("SELECT * FROM work_listings WHERE source = 'arbeitnow' ORDER BY id LIMIT 1");
+ await s.db.exec('UPDATE work_listings SET salary = $1 WHERE id = $2', ['€70,000 a year', one.id]);
+ const res = await search.sweep();
+ assert.strictEqual(res.job.sent, 1);
+ const last = valid((await outbox()).pop());
+ assert.strictEqual(last.payload.id, one.id);
+ assert.strictEqual(last.payload.revision, 2);
+ assert.match(last.payload.summary, /€70,000 a year/);
+ });
+
+ await check('a listing that expires (its page answers 404) becomes a tombstone, once', async () => {
+ const one = await s.db.one("SELECT * FROM work_listings WHERE source = 'remotive' ORDER BY id LIMIT 1");
+ await s.db.exec('UPDATE work_listings SET expired = true WHERE id = $1', [one.id]);
+ assert.strictEqual((await t.get(`/jobs/${one.id}`)).status, 404);
+ const res = await search.sweep();
+ assert.strictEqual(res.job.sent, 1);
+ const last = valid((await outbox()).pop());
+ assert.strictEqual(last.event_type, 'work.index_document.deleted');
+ assert.deepStrictEqual(last.payload, { type: 'job', id: one.id, revision: 2 });
+ assert.strictEqual((await search.sweep()).job.sent, 0, 'a tombstone is not re-sent');
+ // A row deleted outright is tombstoned too (exists()).
+ const gone = await s.db.one("SELECT * FROM work_listings WHERE source = 'arbeitnow' AND expired = false ORDER BY id DESC LIMIT 1");
+ await s.db.exec('DELETE FROM work_listings WHERE id = $1', [gone.id]);
+ assert.strictEqual((await search.sweep()).job.removed, 1);
+ });
+
+ await check('the ingest run triggers the sweep itself', async () => {
+ const before = (await outbox()).length;
+ clock.t += 3 * DAY;
+ t.boards.add && t.boards.add('remotive', { id: 990001, title: 'Staff platform engineer', company_name: 'Northwind', url: 'https://remotive.com/remote-jobs/software-dev/staff-platform-engineer-990001', candidate_required_location: 'Worldwide', job_type: 'full_time', publication_date: '2026-10-04T08:00:00', description: '<p>Platform work.</p>', tags: ['go'] });
+ await ingest.runAll();
+ for (let i = 0; i < 100 && (await outbox()).length === before; i += 1) await new Promise((r) => setTimeout(r, 50));
+ assert.ok((await outbox()).length > before, 'the run swept: something changed (fetched_at moved), so something was sent');
+ });
+
+ await check('the outbox relay stays off without the events URL and the client secret; rows wait', async () => {
+ const st = await search.status();
+ assert.strictEqual(st.enabled, false);
+ assert.ok(st.pending > 0);
+ });
+ } finally {
+ await done(t);
+ }
+})();