openvibers/openvibe.work Commits

Arbeitnow can come from OpenVibe.Sources (WORK_ARBEITNOW_FROM=sources; plan F2) (#66)

OP OpenVibers committed 21 days ago · fa6a43246e · 11 files, +287 −23

Changes

ddiff --git a/.env.example b/.env.example
iindex 57bb1bd..1a6ec65 100644
---- a/.env.example
++++ b/.env.example
@@@ -64,6 +64,11 @@ WORK_EVENTS_URL=
# The boards' public APIs. Point these at a stand-in server to test the ingest; nothing else in the service fetches
# anything, and never a URL a caller typed.
#WORK_ARBEITNOW_URL=https://www.arbeitnow.com/api/job-board-api
++# Arbeitnow through OpenVibe.Sources (server/ingest/from-sources.js): with WORK_ARBEITNOW_FROM=sources Work stops reading
++# the board and follows Sources' arbeitnow-job-board items instead (pages 1-3; a job the board stops showing is retired
++# after a week). Needs OV_OAUTH_CLIENT_SECRET (Network grants Work sources.item.read); Sources is on 127.0.0.1:4720.
++#WORK_ARBEITNOW_FROM=sources
++#OV_SOURCES_INTERNAL_URL=http://127.0.0.1:4720
#WORK_REMOTIVE_URL=https://remotive.com/api/remote-jobs
#WORK_REMOTEOK_URL=https://remoteok.com/api
# The User-Agent the boards see. Defaults to OpenVibeWork/0.1 (+<BASE_URL>).
ddiff --git a/README.md b/README.md
iindex 235aa33..d060c74 100644
---- a/README.md
++++ b/README.md
@@@ -145,7 +145,7 @@ Each board is read through its own public API, and its terms are applied on ever
| Source | API | Terms, as applied here |
|---|---|---|
--| **Arbeitnow** | `GET https://www.arbeitnow.com/api/job-board-api?page=N` | Free to use. Arbeitnow is named on every listing and the listing links to its page on arbeitnow.com. Its first pages (configurable) are read. |
++| **Arbeitnow** | `GET https://www.arbeitnow.com/api/job-board-api?page=N` | Free to use. Arbeitnow is named on every listing and the listing links to its page on arbeitnow.com. Its first pages (configurable) are read. With `WORK_ARBEITNOW_FROM=sources` Work reads none: it follows OpenVibe.Sources' `arbeitnow-job-board` items (`server/ingest/from-sources.js`), and a job the board stops showing is retired by Sources after a week. |
| **Remotive** | `GET https://remotive.com/api/remote-jobs?limit=100` | Remotive must be named as the source and the listing linked. The API is called about four times a day — the six-hour interval is exactly that. |
| **RemoteOK** | `GET https://remoteok.com/api` | RemoteOK must be named as the source and the listing linked. The response's first element is a legal notice, not a listing, and is skipped. |
ddiff --git a/migrations/0007_ingest_cursor.sql b/migrations/0007_ingest_cursor.sql
nnew file mode 100644
iindex 0000000..1feb769
---- /dev/null
++++ b/migrations/0007_ingest_cursor.sql
@@@ -0,0 +1,9 @@
++-- phase: expand
++-- Where OpenVibe.Work is in OpenVibe.Sources' change feed (server/ingest/from-sources.js: Arbeitnow through Sources, plan
++-- 2026-10-22 track F2; openvibe-publishing/ingest, prefix work). The DDL is openvibe-publishing lib/ingest.js schema('work').
++CREATE TABLE IF NOT EXISTS work_ingest_cursor (
++ name text COLLATE "C" NOT NULL,
++ cursor bigint NOT NULL,
++ updated_at bigint NOT NULL,
++ PRIMARY KEY (name)
++);
ddiff --git a/server/app.js b/server/app.js
iindex 871985a..5e6b582 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -31,6 +31,7 @@ const { createPageRoutes } = require('./http/pages');
const { createServiceReadiness } = require('./observability');
const { createCallerLimits } = require('./http/caller-limits');
const { createIngest } = require('./ingest');
++const { createFromSources } = require('./ingest/from-sources');
const { createIndexNow } = require('openvibe-shared/indexnow');
const { createSearchIndex } = require('./search-index');
const accountDataLib = require('./identity/account-data');
@@@ -91,7 +92,10 @@ async function createApp(opts = {}) {
return createIndexNow({ host: config.baseUrl, key: '', fetch: fetchImpl, log: warn });
}
})();
-- ctx.ingest = createIngest({ config, s, fetchImpl, log, afterRun: () => ctx.search.afterIngest(), indexnow: ctx.indexnow });
++ // Arbeitnow through OpenVibe.Sources (server/ingest/from-sources.js), off unless WORK_ARBEITNOW_FROM=sources; a test
++ // passes its own Sources stand-in as opts.sourcesClient.
++ ctx.fromSources = createFromSources({ config, s, fetchImpl, log, source: opts.sourcesClient || null, afterRun: () => ctx.search.afterIngest(), indexnow: ctx.indexnow });
++ ctx.ingest = createIngest({ config, s, fetchImpl, log, afterRun: () => ctx.search.afterIngest(), indexnow: ctx.indexnow, fromSources: ctx.fromSources });
// The person's OpenVibe.Actor agent (server/actor.js) and the per-person daily cap on asking it. A test injects
// a stand-in client and clock through opts.actor / opts.assistLimitsNow, so the assist route never calls out.
ddiff --git a/server/config.js b/server/config.js
iindex a35472b..a6a08a2 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -85,6 +85,11 @@ function load(env = process.env) {
arbeitnow: { url: trim(env.WORK_ARBEITNOW_URL || 'https://www.arbeitnow.com/api/job-board-api') },
remotive: { url: trim(env.WORK_REMOTIVE_URL || 'https://remotive.com/api/remote-jobs') },
remoteok: { url: trim(env.WORK_REMOTEOK_URL || 'https://remoteok.com/api') },
++ // OpenVibe.Sources (server/ingest/from-sources.js): WORK_ARBEITNOW_FROM=sources takes Arbeitnow from Sources'
++ // arbeitnow-job-board instead of the board (plan 2026-10-22, F2); it needs Sources (127.0.0.1:4720) and the
++ // OAuth client secret (Network grants Work sources.item.read). Anything else keeps Work's own fetch.
++ internalUrl: trim(env.OV_SOURCES_INTERNAL_URL == null ? 'http://127.0.0.1:4720' : env.OV_SOURCES_INTERNAL_URL),
++ arbeitnowFrom: String(env.WORK_ARBEITNOW_FROM || '').trim().toLowerCase() === 'sources' ? 'sources' : 'board',
},
// The background ingest (server/ingest/): off unless WORK_INGEST=on, so a test and a development run
// never call a job board. Once on, each source is fetched at most every intervalMs (Remotive's terms
ddiff --git a/server/index.js b/server/index.js
iindex 89ba142..054a857 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -16,7 +16,7 @@ const { startTelemetryRetention } = require('openvibe-sdk/telemetry');
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.search.stop(), () => ctx.keys.client.stop(), async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],
++ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.ingest.stop(), () => ctx.fromSources.stop(), () => ctx.search.stop(), () => ctx.keys.client.stop(), async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],
});
}
@@@ -36,6 +36,7 @@ async function start() {
ctx.telemetryTick.unref();
ctx.keys.client.start();
ctx.ingest.start();
++ ctx.fromSources.start();
ctx.search.start();
// Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without WORK_EVENTS_URL and
ddiff --git a/server/ingest/from-sources.js b/server/ingest/from-sources.js
nnew file mode 100644
iindex 0000000..889adda
---- /dev/null
++++ b/server/ingest/from-sources.js
@@@ -0,0 +1,118 @@
++'use strict';
++
++/**
++ * Arbeitnow through OpenVibe.Sources (plan 2026-10-22, track F2). With WORK_ARBEITNOW_FROM=sources, Work stops reading
++ * Arbeitnow's board itself: Sources fetches it (pages 1-3 every six hours, under its robots and terms checks) and keeps
++ * each job as an item (source arbeitnow-job-board, category jobs). Work follows Sources' change feed
++ * (openvibe-publishing/ingest: the cursor in work_ingest_cursor, one transaction per page) and keeps one listing per job
++ * in work_listings, source 'arbeitnow', the row its own fetch made (the same slug, so the same listing id and URL).
++ *
++ * A job Sources retires (the board has not shown it for a week: retire_after_sec) or removes is expired here, like a
++ * listing a board stopped showing. Work's own expiry sweep leaves these listings alone: an unchanged job is no change in
++ * Sources' feed, so Work never hears it was seen again, and Sources' retirement is what says it is gone.
++ *
++ * Pulls 30 seconds after start and then hourly. Off unless WORK_ARBEITNOW_FROM=sources with OV_SOURCES_INTERNAL_URL and
++ * the OAuth client secret (a test passes its own source).
++ */
++const ingest = require('openvibe-publishing/ingest');
++const sources = require('../jobs/sources');
++const store = require('../jobs/store');
++
++const SOURCE_ID = 'arbeitnow';
++const SOURCE_KEY = 'arbeitnow-job-board';
++const CURSOR = 'sources.jobs.after';
++const EVERY_MS = 60 * 60 * 1000;
++const FIRST_MS = 30 * 1000;
++
++/** A Sources item → the parsed row Work's own Arbeitnow parse makes (server/jobs/sources.js), or null. */
++function rawOf(item) {
++ const f = (item && item.fields) || {};
++ const list = (v) => (Array.isArray(v) ? v.filter((x) => typeof x === 'string' && x.trim()).map((x) => x.trim()) : []);
++ return {
++ source_id: item.identity,
++ url: item.canonical_url,
++ title: item.title,
++ company: typeof f.company === 'string' ? f.company.trim() : null,
++ location: typeof f.location === 'string' ? f.location.trim() || null : null,
++ remote: f.remote === true,
++ // Lists arrive with Sources' extra_lists (openvibe-contracts 0.212.0); until then a listing has none.
++ tags: list(f.tags),
++ job_type: list(f.job_types).join(', ') || null,
++ salary: null, // this board gives no salary
++ posted_at: item.published_at || null,
++ description_html: item.summary || '', // Sources keeps a plain-text excerpt of the description
++ };
++}
++
++function createFromSources({ config, s, fetchImpl = globalThis.fetch, log = console, source = null, afterRun = null, indexnow = null }) {
++ const url = String((config.sources && config.sources.internalUrl) || '').replace(/\/+$/, '');
++ const on = config.sources && config.sources.arbeitnowFrom === 'sources';
++ const client = source || (on && url && config.oauth.clientSecret
++ ? ingest.createSourcesClient({ config: { ...config, sources: { internalUrl: url, category: 'jobs' } }, fetchImpl })
++ : null);
++ const cursor = ingest.createChangeCursor(s.db, { prefix: 'work', now: s.now });
++ let timer = null;
++ let first = null;
++ let running = false;
++ let addedIds = [];
++
++ async function apply(item, t) {
++ if (!item || item.source_key !== SOURCE_KEY || item.kind !== 'job') return 'hold';
++ if (item.removed) {
++ await store.expireListing(t, SOURCE_ID, item.identity);
++ return 'removed';
++ }
++ const fetchedAt = (item.provenance && item.provenance.retrieved_at) || s.iso();
++ const row = sources.toRow(rawOf(item), { sourceId: SOURCE_ID, fetchedAt, idOf: store.idOf });
++ if (!row) return 'hold';
++ if (await store.upsertListing(t, row, fetchedAt)) addedIds.push(row.id);
++ return 'applied';
++ }
++
++ /** Follow Sources' change feed to its end and record it as Arbeitnow's fetch. Never throws. */
++ async function pull() {
++ if (!client || running) return { skipped: true };
++ running = true;
++ const startedAt = s.iso();
++ addedIds = [];
++ try {
++ await store.beginFetch(s, SOURCE_ID, startedAt);
++ const out = await ingest.pullChanges({ db: s.db, cursor, source: client, name: CURSOR, apply });
++ const listings = await store.liveCount(s, SOURCE_ID);
++ await store.finishFetch(s, SOURCE_ID, { ok: true, listings, error: null, finishedAt: s.iso() });
++ if (out.applied || out.removed) {
++ log.log(`[Ingest] Arbeitnow through OpenVibe.Sources: ${out.applied} listings updated, ${out.removed} gone (cursor ${out.after})`);
++ // As after a board read: search engines hear about new listing pages, and Search catches up.
++ if (indexnow && addedIds.length) indexnow.pingSoon([...addedIds.map((id) => `${config.baseUrl}/jobs/${encodeURIComponent(id)}`), `${config.baseUrl}/sitemap.xml`]);
++ if (afterRun) afterRun();
++ }
++ return out;
++ } catch (err) {
++ const detail = String((err && err.message) || err).slice(0, 300);
++ await store.finishFetch(s, SOURCE_ID, { ok: false, listings: await store.liveCount(s, SOURCE_ID).catch(() => 0), error: detail, finishedAt: s.iso() }).catch(() => {});
++ log.warn(`[Ingest] Arbeitnow through OpenVibe.Sources failed: ${detail}`);
++ return { error: detail };
++ } finally { running = false; }
++ }
++
++ function start() {
++ if (!client || timer) return false;
++ first = setTimeout(() => { pull().catch(() => {}); }, FIRST_MS);
++ first.unref();
++ timer = setInterval(() => { pull().catch(() => {}); }, EVERY_MS);
++ timer.unref();
++ log.log('[Ingest] Arbeitnow comes from OpenVibe.Sources (arbeitnow-job-board)');
++ return true;
++ }
++
++ function stop() {
++ clearTimeout(first);
++ clearInterval(timer);
++ first = null;
++ timer = null;
++ }
++
++ return { enabled: Boolean(client), pull, start, stop, apply, SOURCE_ID };
++}
++
++module.exports = { createFromSources, rawOf, SOURCE_ID, SOURCE_KEY };
ddiff --git a/server/ingest/index.js b/server/ingest/index.js
iindex 3bc4c3e..592650c 100644
---- a/server/ingest/index.js
++++ b/server/ingest/index.js
@@@ -36,8 +36,10 @@ async function fetchJson(url, { fetchImpl, timeoutMs, userAgent }) {
const TICK_MS = 3_600_000;
--function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console, afterRun = null, indexnow = null }) {
++function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console, afterRun = null, indexnow = null, fromSources = null }) {
const timers = new Map();
++ // A board OpenVibe.Sources feeds (server/ingest/from-sources.js) is not read here, nor swept by the expiry.
++ const viaSources = new Set(fromSources && fromSources.enabled ? [fromSources.SOURCE_ID] : []);
let running = false;
/** Read one source end to end and record what happened. Never throws: the caller is a timer. */
@@@ -90,11 +92,12 @@ function createIngest({ config, s, fetchImpl = globalThis.fetch, log = console,
for (const source of sources.SOURCES) {
const prev = last.get(source.id);
const age = prev && prev.started_at ? s.now() - Date.parse(prev.started_at) : Infinity;
++ if (viaSources.has(source.id)) { out.push({ source: source.id, skipped: 'from_sources' }); continue; }
if (onlyDue && age < config.ingest.intervalMs) { out.push({ source: source.id, skipped: 'not_due' }); continue; }
out.push({ source: source.id, ...(await runSource(source.id)) });
}
try {
-- const gone = await store.expire(s, config.ingest.retentionDays, s.now());
++ const gone = await store.expire(s, config.ingest.retentionDays, s.now(), { except: [...viaSources] });
if (gone) log.log(`[Ingest] expired ${gone} listings not seen for ${config.ingest.retentionDays} days`);
} catch (err) {
log.warn(`[Ingest] expiry failed: ${(err && err.message) || err}`);
ddiff --git a/server/jobs/store.js b/server/jobs/store.js
iindex 2f3d55d..0e93e22 100644
---- a/server/jobs/store.js
++++ b/server/jobs/store.js
@@@ -222,29 +222,44 @@ function newCountSql() {
* Upsert one source's listings. A listing the board still shows is updated in place and its last_seen_at moves; one
* that reappears after expiring comes back (expired = false). Returns { seen, added, updated, addedIds }.
*/
++/** One listing in, through `q` (the database or a transaction). → true when it is new. */
++async function upsertListing(q, row, fetchedAt) {
++ const out = await q.maybe(
++ `INSERT INTO work_listings (id, source, source_id, url, title, company, location, remote, tags, job_type, salary, excerpt, search_text, posted_at, sort_at, fetched_at, first_seen_at, last_seen_at, expired)
++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $16, $16, false)
++ ON CONFLICT (source, source_id) DO UPDATE SET
++ url = EXCLUDED.url, title = EXCLUDED.title, company = EXCLUDED.company, location = EXCLUDED.location,
++ remote = EXCLUDED.remote, tags = EXCLUDED.tags, job_type = EXCLUDED.job_type, salary = EXCLUDED.salary,
++ excerpt = EXCLUDED.excerpt, search_text = EXCLUDED.search_text, posted_at = EXCLUDED.posted_at,
++ sort_at = EXCLUDED.sort_at, fetched_at = EXCLUDED.fetched_at, last_seen_at = EXCLUDED.last_seen_at,
++ expired = false
++ RETURNING (xmax = 0) AS inserted`,
++ [row.id, row.source, row.source_id, row.url, row.title, row.company, row.location, row.remote, row.tags, row.job_type, row.salary, row.excerpt, row.search_text, row.posted_at, row.sort_at, fetchedAt]);
++ return Boolean(out && out.inserted);
++}
++
async function upsertListings(s, rows, { sourceId, fetchedAt }) {
let added = 0;
let updated = 0;
const addedIds = [];
await s.tx(async () => {
for (const row of rows) {
-- const out = await s.db.maybe(
-- `INSERT INTO work_listings (id, source, source_id, url, title, company, location, remote, tags, job_type, salary, excerpt, search_text, posted_at, sort_at, fetched_at, first_seen_at, last_seen_at, expired)
-- VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $16, $16, false)
-- ON CONFLICT (source, source_id) DO UPDATE SET
-- url = EXCLUDED.url, title = EXCLUDED.title, company = EXCLUDED.company, location = EXCLUDED.location,
-- remote = EXCLUDED.remote, tags = EXCLUDED.tags, job_type = EXCLUDED.job_type, salary = EXCLUDED.salary,
-- excerpt = EXCLUDED.excerpt, search_text = EXCLUDED.search_text, posted_at = EXCLUDED.posted_at,
-- sort_at = EXCLUDED.sort_at, fetched_at = EXCLUDED.fetched_at, last_seen_at = EXCLUDED.last_seen_at,
-- expired = false
-- RETURNING (xmax = 0) AS inserted`,
-- [row.id, row.source, row.source_id, row.url, row.title, row.company, row.location, row.remote, row.tags, row.job_type, row.salary, row.excerpt, row.search_text, row.posted_at, row.sort_at, fetchedAt]);
-- if (out && out.inserted) { added++; addedIds.push(row.id); } else updated++;
++ if (await upsertListing(s.db, row, fetchedAt)) { added++; addedIds.push(row.id); } else updated++;
}
});
return { seen: rows.length, added, updated, addedIds };
}
++/** A listing its source says is gone (Sources retired or removed it): expired like one a board stopped showing. */
++const expireListing = (q, source, sourceId) => q.exec(
++ 'UPDATE work_listings SET expired = true WHERE source = $1 AND source_id = $2 AND expired = false', [source, String(sourceId)]);
++
++/** How many live listings a source has (its fetch record's count when Sources feeds it). */
++async function liveCount(s, source) {
++ const r = await s.db.maybe('SELECT count(*)::int AS n FROM work_listings WHERE source = $1 AND expired = false', [source]);
++ return r ? r.n : 0;
++}
++
/** A source's fetch record: started now, finished when it finished, with what happened. */
const beginFetch = (s, sourceId, startedAt) => s.db.query(
`INSERT INTO work_source_fetches (source, started_at, finished_at, ok, listings, error) VALUES ($1, $2, NULL, false, 0, NULL)
@@@ -262,11 +277,12 @@ async function sourceFetches(s) {
/**
* Listings the boards stopped showing go. A listing not seen for `days` is expired rather than deleted: the row keeps
-- * its history, the search and the pages stop offering it, and a board that shows it again brings it back.
++ * its history, the search and the pages stop offering it, and a board that shows it again brings it back. Sources
++ * named in `except` are left alone: OpenVibe.Sources feeds them and says itself when a listing is gone.
*/
--async function expire(s, days, now) {
++async function expire(s, days, now, { except = [] } = {}) {
const cutoff = new Date(now - days * 86_400_000).toISOString();
-- return s.db.exec('UPDATE work_listings SET expired = true WHERE expired = false AND last_seen_at < $1', [cutoff]);
++ return s.db.exec('UPDATE work_listings SET expired = true WHERE expired = false AND last_seen_at < $1 AND NOT (source = ANY($2::text[]))', [cutoff, except]);
}
// ── saved searches ──────────────────────────────────────────
@@@ -335,7 +351,7 @@ function decodeCursor(v) {
}
module.exports = {
-- search, facets, getListing, countsBySource, listRecent, listSalaried, homeSummary, recentIds, upsertListings, beginFetch, finishFetch, sourceFetches,
-- expire, listSaved, createSaved, getSaved, deleteSaved, newCount, touchSaved, toWire, toSavedWire, filtersOf,
++ search, facets, getListing, countsBySource, listRecent, listSalaried, homeSummary, recentIds, upsertListing, upsertListings, expireListing, liveCount,
++ beginFetch, finishFetch, sourceFetches, expire, listSaved, createSaved, getSaved, deleteSaved, newCount, touchSaved, toWire, toSavedWire, filtersOf,
idOf, encodeCursor, decodeCursor, PAGE_SIZE, MAX_PAGE, MAX_SAVED, LISTING_ID,
};
ddiff --git a/test/from-sources.test.js b/test/from-sources.test.js
nnew file mode 100644
iindex 0000000..ff82d85
---- /dev/null
++++ b/test/from-sources.test.js
@@@ -0,0 +1,103 @@
++'use strict';
++/**
++ * Arbeitnow through OpenVibe.Sources (server/ingest/from-sources.js; plan 2026-10-22, track F2), against a Sources
++ * stand-in serving its change feed:
++ * - each job item becomes the listing Work's own fetch made (same slug → same id and URL), with its excerpt, tags and
++ * job types; other sources' items and jobs Work may not publish are held;
++ * - the board is no longer read, and Work's expiry sweep leaves these listings to Sources;
++ * - a job Sources retires (gone_from_source) expires here; the fetch record says what came through;
++ * - without WORK_ARBEITNOW_FROM=sources nothing changes.
++ */
++const assert = require('assert');
++const { boot, check, done } = require('./helpers/boot');
++const store = require('../server/jobs/store');
++const { createFromSources } = require('../server/ingest/from-sources');
++
++const DAY = 86_400_000;
++const feed = [];
++let seq = 0;
++function push(item) { feed.push({ ...item, change_seq: ++seq }); }
++const sourcesClient = {
++ enabled: true,
++ async listItems({ after = 0, limit = 100 } = {}) {
++ const page = feed.filter((i) => i.change_seq > after).slice(0, limit);
++ return { items: page, next_after: page.length ? page[page.length - 1].change_seq : after, more: false };
++ },
++};
++const job = (identity, over = {}) => ({
++ id: `itm_${identity}`, source_key: 'arbeitnow-job-board', category: 'jobs', kind: 'job', identity,
++ canonical_url: `https://www.arbeitnow.com/jobs/companies/acme/${identity}`, title: 'Backend Developer', summary: 'We build payment rails. Node.js and PostgreSQL.',
++ authors: [], published_at: '2026-10-20T08:00:00.000Z', source_updated_at: null,
++ fields: { company: 'Acme GmbH', location: 'Berlin', remote: false, tags: ['Node.js', 'PostgreSQL'], job_types: ['full time'] },
++ revision: 1, provenance: { retrieved_at: '2026-10-22T06:00:00.000Z' }, removed: null, ...over,
++});
++
++(async () => {
++ const clock = { t: Date.parse('2026-10-23T06:00:00Z') };
++ const pinged = [];
++ const indexnow = { enabled: false, pingSoon: (urls) => { pinged.push(...urls); return urls.length; } };
++ const t = await boot({ now: () => clock.t, env: { WORK_ARBEITNOW_FROM: 'sources' }, sourcesClient, indexnow });
++ const s = t.ctx.s;
++ const from = t.ctx.fromSources;
++
++ try {
++ await check('each job item becomes the listing the board fetch made; others are held', async () => {
++ push(job('backend-developer-berlin-123'));
++ // Before Sources maps lists (contracts 0.212.0) an item has none: the listing has no tags or job type.
++ push(job('ops-engineer-remote-456', { title: 'Ops Engineer', summary: 'Keep our clusters healthy.', fields: { company: 'Beta AG', location: 'Remote', remote: true } }));
++ push(job('no-company-789', { fields: { location: 'Munich' } }));
++ push({ ...job('remotive-1'), source_key: 'remotive-feed' });
++ const out = await from.pull();
++ assert.deepStrictEqual([out.applied, out.hold, out.removed], [2, 2, 0]);
++
++ const id = store.idOf('arbeitnow', 'backend-developer-berlin-123');
++ const row = await s.db.maybe('SELECT * FROM work_listings WHERE id = $1', [id]);
++ assert.deepStrictEqual([row.source, row.title, row.company, row.location, row.remote, row.tags, row.job_type, row.expired],
++ ['arbeitnow', 'Backend Developer', 'Acme GmbH', 'Berlin', false, 'Node.js, PostgreSQL', 'full time', false]);
++ assert.strictEqual(row.excerpt, 'We build payment rails. Node.js and PostgreSQL.');
++ assert.strictEqual(row.url, 'https://www.arbeitnow.com/jobs/companies/acme/backend-developer-berlin-123');
++ const ops = await s.db.maybe('SELECT tags, job_type, remote FROM work_listings WHERE source_id = $1', ['ops-engineer-remote-456']);
++ assert.deepStrictEqual([ops.tags, ops.job_type, ops.remote], ['', null, true]);
++ assert.ok(pinged.includes(`https://openvibe.work/jobs/${id}`), 'search engines hear about the new listing page');
++
++ const page = (await t.get('/api/v1/jobs?q=payment')).json();
++ assert.deepStrictEqual(page.jobs.map((j) => [j.id, j.source]), [[id, 'arbeitnow']]);
++ });
++
++ await check('the fetch record says the listings came through', async () => {
++ const rec = (await store.sourceFetches(s)).get('arbeitnow');
++ assert.deepStrictEqual([rec.ok, rec.listings, rec.error], [true, 2, null]);
++ });
++
++ await check('the board is not read, and the expiry sweep leaves these listings to Sources', async () => {
++ const out = await t.ctx.ingest.runAll();
++ assert.deepStrictEqual(out.find((o) => o.source === 'arbeitnow'), { source: 'arbeitnow', skipped: 'from_sources' });
++ clock.t += 60 * DAY;
++ await t.ctx.ingest.runAll();
++ const live = await s.db.many("SELECT source_id FROM work_listings WHERE source = 'arbeitnow' AND expired = false ORDER BY source_id");
++ assert.deepStrictEqual(live.map((r) => r.source_id), ['backend-developer-berlin-123', 'ops-engineer-remote-456']);
++ });
++
++ await check('a job Sources retires expires here and leaves the search', async () => {
++ push(job('backend-developer-berlin-123', { revision: 2, removed: { at: '2026-10-30T06:00:00.000Z', reason: 'gone_from_source' } }));
++ const out = await from.pull();
++ assert.deepStrictEqual([out.applied, out.removed], [0, 1]);
++ const row = await s.db.maybe('SELECT expired FROM work_listings WHERE source_id = $1', ['backend-developer-berlin-123']);
++ assert.strictEqual(row.expired, true);
++ assert.strictEqual((await t.get('/api/v1/jobs?q=payment')).json().jobs.length, 0);
++ assert.strictEqual((await store.sourceFetches(s)).get('arbeitnow').listings, 1);
++ // The cursor moved past it: the next pull changes nothing.
++ assert.deepStrictEqual([(await from.pull()).applied, (await from.pull()).removed], [0, 0]);
++ });
++
++ await check('without WORK_ARBEITNOW_FROM=sources nothing follows Sources', async () => {
++ const config = { ...t.config, sources: { ...t.config.sources, arbeitnowFrom: 'board', internalUrl: 'http://127.0.0.1:1' } };
++ const off = createFromSources({ config, s, log: { log() {}, warn() {} } });
++ assert.strictEqual(off.enabled, false);
++ assert.deepStrictEqual(await off.pull(), { skipped: true });
++ });
++ } finally {
++ await t.close();
++ }
++ done();
++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/helpers/boot.js b/test/helpers/boot.js
iindex 39ab12f..3d7cc5b 100644
---- a/test/helpers/boot.js
++++ b/test/helpers/boot.js
@@@ -45,7 +45,7 @@ async function boot(opts = {}) {
const { createStore } = require('../../server/db');
const testdb = await require('./db').testDb();
const store = createStore(testdb.db, { now: opts.now });
-- const built = await createApp({ config, store, log, telemetrySink: opts.telemetrySink, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, accountSend: opts.accountSend, actor: opts.actor, assistLimitsNow: opts.assistLimitsNow, indexnow: opts.indexnow });
++ const built = await createApp({ config, store, log, telemetrySink: opts.telemetrySink, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, accountSend: opts.accountSend, actor: opts.actor, assistLimitsNow: opts.assistLimitsNow, indexnow: opts.indexnow, sourcesClient: opts.sourcesClient });
await built.ctx.keys.ensure();
const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });
const base = `http://127.0.0.1:${server.address().port}`;