openvibers/openvibe.community Commits

Add M1 status posts and personal pulse contributions (#304)

OP OpenVibers committed 4 days ago · 8acf7fe4d9 · 13 files, +301 −32

* Add status posts and personal pulse contributions

* Keep status space out of forum listings

* Pulse tells a signed-in reader they can post (viewer.can_post)

The menu's Feed (openvibe-shared ov-home.js) shows its "What's new" composer only when the pulse answer says the
reader may post, so it never offers a post the status route would refuse.

* Pin openvibe-contracts v0.352.0 (the M1 contracts)

Changes

ddiff --git a/migrations/0023_status_space.sql b/migrations/0023_status_space.sql
nnew file mode 100644
iindex 0000000..eea27a2
---- /dev/null
++++ b/migrations/0023_status_space.sql
@@@ -0,0 +1,7 @@
++-- phase: expand
++-- Statuses use ordinary moderated forum threads and opening posts, so reactions and replies reuse the post routes.
++INSERT INTO spaces (slug, name, description, visibility, created_by, members_only_owner, thread_kind, style, votes, reactions, group_id, parent_id, position)
++VALUES ('status', 'Status', 'Short public updates from people on OpenVibe.', 'public', 'system', NULL, 'discussion', 'feed', 0, 1, 2, NULL, 3)
++ON CONFLICT (slug) DO NOTHING;
++ALTER TABLE pulse_items ADD COLUMN IF NOT EXISTS picture_url text COLLATE "C";
++ALTER TABLE pulse_items ADD COLUMN IF NOT EXISTS app_name text COLLATE "C";
ddiff --git a/package-lock.json b/package-lock.json
iindex 548019e..4d72d8f 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -17,7 +17,7 @@
"highlight.js": "^11.11.1",
"iovalkey": "^0.4.0",
"multer": "^2.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.352.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.157.0",
"pg": "^8.23.0"
@@@ -1192,9 +1192,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.338.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",
-- "integrity": "sha512-75FD6IxMsXoud3NFXKDs0/D0rhJtF2D+WL8M92ArIyabkUnM/sZvw6E6L97EqwQ3O7M/kB+uwsNSxGudbil+ng==",
++ "version": "0.352.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.352.0",
++ "integrity": "sha512-TN02BrMU4jzXYZ0rjp5dmLbQI06h5tT6miISS6zS7E7/UFI+fGFidTVtQpZXEYG8gppGEo2UEtZ//hk9kkASyQ==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex fa93264..ccf3687 100644
---- a/package.json
++++ b/package.json
@@@ -29,7 +29,7 @@
"highlight.js": "^11.11.1",
"iovalkey": "^0.4.0",
"multer": "^2.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.338.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.352.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.157.0",
"pg": "^8.23.0"
ddiff --git a/server/app.js b/server/app.js
iindex 51ef256..9d86dc5 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -309,6 +309,8 @@ async function createApp(opts = {}) {
const cors = v1.cors(config.apiCorsOrigins);
app.use('/api/v1/comments', cors, createCommentsApi({ service: comments, viewers, anonWriteLimiter, resolveLimiter, limits }));
app.use('/api/v1/pulse', cors, createPulseApi({ pulse, viewers, limits }));
++ app.use('/api/v1/me/status', cors, require('./status/api').createStatusApi({ forum, pulse, viewers, limits, db, baseUrl: config.baseUrl }));
++ app.use('/api/v1/profiles', cors, require('./status/api').createProfileStatusApi({ viewers, limits, db, baseUrl: config.baseUrl }));
// OpenVibe.Events → Pulse (server/pulse/consumer.js): public activity from Live, Blog, Wiki and News.
const pulseConsumer = require('./pulse/consumer').createPulseConsumer({ db, vipCache: vip && vip.cache, revocations, accountSend: config.oauth && config.oauth.clientSecret ? require('./identity/account-data').createSender({ config }) : null, secrets: String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',').map((s) => s.trim()).filter(Boolean) });
app.locals.pulseConsumer = pulseConsumer;
ddiff --git a/server/forum/store.js b/server/forum/store.js
iindex 5566c2f..0bf7e57 100644
---- a/server/forum/store.js
++++ b/server/forum/store.js
@@@ -21,7 +21,7 @@ async function listSpaces(db, visibilities) {
(SELECT COUNT(*) FROM posts p JOIN threads t ON t.id = p.thread_id WHERE t.space_id = s.id AND t.deleted_at IS NULL AND p.deleted_at IS NULL) AS post_count,
(SELECT MAX(t.last_activity_at) FROM threads t WHERE t.space_id = s.id AND t.deleted_at IS NULL) AS last_activity_at
FROM spaces s LEFT JOIN space_groups g ON g.id = s.group_id LEFT JOIN spaces ps ON ps.id = s.parent_id
-- WHERE s.visibility IN (${visibilities.map(() => '?').join(', ')})
++ WHERE s.visibility IN (${visibilities.map(() => '?').join(', ')}) AND s.slug <> 'status'
ORDER BY g.position IS NULL, g.position, s.position, s.id ASC`).all(...visibilities);
}
ddiff --git a/server/openapi.js b/server/openapi.js
iindex ec445d0..9f231f4 100644
---- a/server/openapi.js
++++ b/server/openapi.js
@@@ -318,28 +318,40 @@ route('post', '/api/v1/comments/{commentId}/votes', {
// ── /api/v1/pulse — the network's public activity (server/pulse/api.js) ──────────────────────
route('get', '/api/v1/pulse', {
-- operationId: 'listPulse', summary: 'What the network is doing, newest first', tags: ['pulse'], security: ANONYMOUS, parameters: [
-- query('origin', { type: 'string', enum: ['user', 'ai', 'system', 'all'] }), query('after', str(), 'The next_cursor from the previous page'),
++ operationId: 'listPulse', summary: 'What the network is doing; followed people first in the last hour for me', tags: ['pulse'], security: ANONYMOUS, parameters: [
++ query('origin', { type: 'string', enum: ['user', 'ai', 'system', 'all'] }), query('for', { type: 'string', enum: ['me'] }),
++ query('cursor', str(), 'The personal next_cursor'), query('after', str(), 'The public next_cursor from the previous page'),
query('limit', { type: 'integer', minimum: 1, maximum: 100 })],
responses: { 200: json('Items and the cursor for the next page', { type: 'object', properties: { items: arrayOf(ref('PulseItem')), next_cursor: nullable('string') } }), ...errors([400, 429]) },
});
route('post', '/api/v1/pulse/items', {
operationId: 'ingestPulseItem', summary: 'Publish one public activity item', tags: ['pulse'], security: SERVICE,
-- description: 'community.pulse.write. Idempotent per item ref: 201 when created, 200 when it was already there.',
-- requestBody: body({ type: 'object', required: ['ref', 'title', 'url'], properties: {
++ description: 'community.pulse.write for services or community.pulse.contribute with a person grant for developer apps. An app sends {id,text,url,picture_url?}; its subject and label come from the token. Idempotent per source id.',
++ requestBody: body({ anyOf: [{ type: 'object', required: ['id', 'text', 'url'], properties: {
++ id: str(), text: str({ maxLength: 300 }), url: str({ format: 'uri' }), picture_url: str({ format: 'uri' }),
++ } }, { type: 'object', required: ['ref', 'title', 'url'], properties: {
ref: { type: 'object', required: ['service', 'type', 'id'], properties: { service: str(), type: str(), id: str() } },
title: str(), url: str({ format: 'uri' }), origin: { type: 'string', enum: ['user', 'ai', 'system'] },
occurred_at: str({ format: 'date-time' }), visibility: { type: 'string', enum: ['public'] },
-- } }),
++ } }] }),
responses: { 200: json('Already there', { type: 'object', properties: { item: ref('PulseItem'), created: { const: false } } }),
201: json('Created', { type: 'object', properties: { item: ref('PulseItem'), created: { const: true } } }), ...errors() },
});
route('delete', '/api/v1/pulse/items/{service}/{type}/{id}', {
operationId: 'retractPulseItem', summary: 'Retract one of the calling service\'s items', tags: ['pulse'], security: SERVICE,
parameters: [path('service', str()), path('type', str()), path('id', str())],
-- description: 'community.pulse.write.',
++ description: 'community.pulse.write for services, community.pulse.contribute for apps.',
responses: { 200: json('Whether it was there'), ...errors() },
});
++route('post', '/api/v1/me/status', {
++ operationId: 'postStatus', summary: 'Post a public status to your profile and Pulse', tags: ['pulse'], security: PERSON,
++ requestBody: body({ type: 'object', required: ['text'], properties: { text: str({ minLength: 1, maxLength: 500 }), link: str({ format: 'uri' }) } }),
++ responses: { 201: json('Status and its opening post id'), ...errors() },
++});
++route('get', '/api/v1/profiles/{subject}/statuses', {
++ operationId: 'profileStatuses', summary: 'A person\'s recent public statuses', tags: ['pulse'], security: ANONYMOUS,
++ parameters: [SUBJECT], responses: { 200: json('Recent statuses'), ...errors([400, 429]) },
++});
// ── /api/v1/spaces — the forum (server/forum/api.js) ─────────────────────────────────────────
route('get', '/api/v1/spaces', {
@@@ -723,8 +735,9 @@ function openapi() {
page_url: str(), can: { type: 'object', properties: { withdraw: { type: 'boolean' }, review: { type: 'boolean' } } },
} },
PulseItem: { type: 'object', properties: {
-- id: str(), ref: { type: 'object', properties: { service: str(), type: str(), id: str() } }, title: str(), url: str(),
++ id: { type: 'integer' }, source: { type: 'object', properties: { service: str(), type: str(), id: str() } }, title: str(), url: str(),
origin: { type: 'string', enum: ['user', 'ai', 'system'] }, occurred_at: str({ format: 'date-time' }), actor: nullable('object'),
++ label: nullable('string'), via: str(), picture_url: str({ format: 'uri' }),
} },
Thread: { type: 'object', required: ['id', 'visibility'], properties: {
id: str(), access_id: str(), ref: { type: 'object' }, visibility: { type: 'string', enum: ['public', 'hidden', 'locked'] },
ddiff --git a/server/pulse/api.js b/server/pulse/api.js
iindex 7a532c7..affb5ee 100644
---- a/server/pulse/api.js
++++ b/server/pulse/api.js
@@@ -15,6 +15,7 @@ const contracts = require('openvibe-contracts');
const { run, serviceCap, jsonBody } = require('../http/v1');
const WRITE = 'community.pulse.write';
++const CONTRIBUTE = 'community.pulse.contribute';
function createPulseApi({ pulse, viewers, limits }) {
const router = express.Router();
@@@ -24,13 +25,21 @@ function createPulseApi({ pulse, viewers, limits }) {
// an item when something public happens on its site (most arrive through Events instead): one a second.
router.use(limits.reads('community.pulse.read'));
const ingest = limits('community.pulse.write', { minute: 60, hour: 1200 });
++ const contribute = limits('community.pulse.contribute', { minute: 10, hour: 100 });
++ const app = (req) => req.viewer.kind === 'service' && String(req.viewer.service).startsWith('app:');
const servicesOnly = (req, res, next) => (req.viewer.kind === 'service' ? next()
: contracts.http.sendProblem(res, 403, 'capability.denied', { detail: `a service token with ${WRITE} is required`, ctx: req.ov }));
-- router.get('/', run(async (req) => await pulse.list(req.query)));
-- router.post('/items', servicesOnly, serviceCap(WRITE), ingest, jsonBody, run(async (req) => await pulse.ingest(req.viewer, req.body || {}), (out) => (out.created ? 201 : 200)));
-- router.delete('/items/:service/:type/:id', servicesOnly, serviceCap(WRITE), ingest, run(async (req) => await pulse.retract(req.viewer, req.params.service, req.params.type, req.params.id)));
++ router.get('/', run(async (req) => await pulse.list(req.query, req.viewer)));
++ router.post('/items', servicesOnly, (req, res, next) => (app(req) ? serviceCap(CONTRIBUTE) : serviceCap(WRITE))(req, res, next),
++ (req, res, next) => (app(req) ? contribute : ingest)(req, res, next), jsonBody,
++ run(async (req) => app(req) ? await pulse.contribute(req.viewer, req.body || {}) : await pulse.ingest(req.viewer, req.body || {}), (out) => (out.created ? 201 : 200)));
++ router.delete('/items/:service/:type/:id', servicesOnly,
++ (req, res, next) => (app(req) ? serviceCap(CONTRIBUTE) : serviceCap(WRITE))(req, res, next),
++ (req, res, next) => (app(req) ? contribute : ingest)(req, res, next),
++ run(async (req) => app(req) ? await pulse.retractContribution(req.viewer, req.params.service, req.params.type, req.params.id)
++ : await pulse.retract(req.viewer, req.params.service, req.params.type, req.params.id)));
router.use((req, res) => contracts.http.sendProblem(res, 404, 'route.not_found', { detail: 'Not found', ctx: req.ov }));
return router;
ddiff --git a/server/pulse/service.js b/server/pulse/service.js
iindex 5067804..289cb5f 100644
---- a/server/pulse/service.js
++++ b/server/pulse/service.js
@@@ -28,10 +28,33 @@ const FUTURE_SKEW_MS = 5 * 60_000;
/** 'svc:live' → 'live' (the service a client-credentials token belongs to). */
const serviceNameOf = (sub) => String(sub || '').replace(/^(svc|app|mod):/, '');
--function createPulse({ db, network = null, config = {} } = {}) {
++function createPulse({ db, network = null, config = {}, fetchImpl = globalThis.fetch } = {}) {
const authors = createAuthors({ db, network });
const base = (config.baseUrl || '').replace(/\/$/, '');
let originsCache = { at: 0, list: null };
++ const followsCache = new Map();
++ const appWrites = new Map();
++
++ async function followedSubjects(v) {
++ const cached = followsCache.get(v.subject);
++ if (cached && Date.now() - cached.at < 30_000) return cached.ids;
++ const ids = [];
++ let cursor = null;
++ do {
++ const url = new URL('/api/v1/me/follows', config.networkInternalUrl);
++ url.searchParams.set('type', 'channel');
++ url.searchParams.set('limit', '200');
++ if (cursor) url.searchParams.set('cursor', cursor);
++ const res = await fetchImpl(url, { headers: { Authorization: `Bearer ${v.token}` }, signal: AbortSignal.timeout(5000) });
++ if (!res.ok) fail(503, 'pulse.follows_unavailable', 'Could not load your follows');
++ const page = await res.json();
++ ids.push(...(page.items || []).map((x) => x.target_id));
++ cursor = page.next_cursor;
++ } while (cursor && ids.length < 5000);
++ if (followsCache.size > 1000) followsCache.delete(followsCache.keys().next().value);
++ followsCache.set(v.subject, { at: Date.now(), ids });
++ return ids;
++ }
/** Record one of Community's own items; never lets a Pulse problem fail the write that caused it. */
async function recordLocal(type, id, { title, path, actor = null, origin = 'user', at = null }) {
@@@ -58,6 +81,8 @@ function createPulse({ db, network = null, config = {} } = {}) {
url: i.url,
origin: i.origin,
label: ai ? 'AI' : (i.origin === 'system' ? 'System' : null),
++ ...(i.source_service.startsWith('app:') ? { via: `via ${i.app_name || i.source_service.slice(4)}` } : {}),
++ ...(i.picture_url ? { picture_url: i.picture_url } : {}),
actor,
occurred_at: isoTime(i.occurred_at),
};
@@@ -65,6 +90,11 @@ function createPulse({ db, network = null, config = {} } = {}) {
return {
ORIGINS,
++ forgetLocal,
++ async statusCreated(thread, text, actor) {
++ await recordLocal('status', thread.id, { title: text.slice(0, 300), path: `/forum/status/${thread.slug}`,
++ actor, origin: 'user', at: thread.created_at });
++ },
// ── hooks for Community's own writes ─────────────────
async pasteCreated(p) {
@@@ -95,7 +125,7 @@ function createPulse({ db, network = null, config = {} } = {}) {
await recordLocal('submission', s.slug, { title: s.title, path: `/submissions/${encodeURIComponent(s.slug)}`, actor: s.author_subject, origin: 'user', at: s.reviewed_at });
},
async submissionGone(slug) { await forgetLocal('submission', slug); },
-- async threadGone(id) { await forgetLocal('thread', id); },
++ async threadGone(id) { await forgetLocal('thread', id); await forgetLocal('status', id); },
async postGone(id) { await forgetLocal('post', id); },
/** POST /api/v1/pulse/items (a service with community.pulse.write). → { item, created } */
@@@ -126,6 +156,43 @@ function createPulse({ db, network = null, config = {} } = {}) {
return { item: shape(out.item, projections), created: out.created };
},
++ async contribute(v, body = {}) {
++ if (!v.subject || v.claims.on_behalf_of !== v.subject || !String(v.service).startsWith('app:')) {
++ fail(403, 'pulse.not_delegated', 'An app needs a grant from the person named in its token');
++ }
++ if (Object.keys(body).some((k) => !['id', 'text', 'url', 'picture_url'].includes(k))) fail(400, 'pulse.invalid', 'Only id, text, url and picture_url are accepted');
++ const id = String(body.id || '');
++ const title = typeof body.text === 'string' ? body.text.replace(/\s+/g, ' ').trim() : '';
++ if (!/^[A-Za-z0-9_-]{1,80}$/.test(id) || !title || title.length > 300) fail(400, 'pulse.invalid', 'id and text are required');
++ let url;
++ try { url = new URL(body.url); } catch { fail(400, 'pulse.invalid_url', 'An https URL is required'); }
++ if (url.protocol !== 'https:' || url.toString().length > 2000) fail(400, 'pulse.invalid_url', 'An https URL is required');
++ let picture = null;
++ if (body.picture_url != null) {
++ try { picture = new URL(body.picture_url); } catch { fail(400, 'pulse.picture_proxy_required', 'A Media picture URL is required'); }
++ if (picture.protocol !== 'https:' || picture.hostname !== 'openvibe.media' || !/^\/(?:t|o)\//.test(picture.pathname) || picture.toString().length > 2000) {
++ fail(400, 'pulse.picture_proxy_required', 'Pictures must use Media HTTPS thumbnail or object URLs');
++ }
++ }
++ const at = Date.now();
++ const times = (appWrites.get(v.service) || []).filter((t) => at - t < 60_000);
++ if (times.length >= 10) fail(429, 'pulse.app_rate_limited', 'At most ten contributions per app per minute');
++ times.push(at); appWrites.set(v.service, times);
++ if (appWrites.size > 5000) appWrites.delete(appWrites.keys().next().value);
++ const out = await store.upsertItem(db, { source_service: v.service, source_type: 'contribution', source_id: `${v.subject}:${id}`,
++ title, url: url.toString(), picture_url: picture && picture.toString(), app_name: String(v.claims.app_name || '').slice(0, 80) || null,
++ actor_subject: v.subject, origin: 'user', occurred_at: sqlTime(at) });
++ const projections = await authors.projectionsFor([v.subject]);
++ return { item: shape(out.item, projections), created: out.created };
++ },
++
++ async retractContribution(v, service, type, id) {
++ if (service !== v.service || type !== 'contribution' || !v.subject || v.claims.on_behalf_of !== v.subject || !id.startsWith(`${v.subject}:`)) {
++ fail(403, 'pulse.foreign_ref', 'An app retracts only its own items for the consenting person');
++ }
++ return { removed: await store.removeItem(db, service, type, id) };
++ },
++
/** DELETE /api/v1/pulse/items/:service/:type/:id — a service retracting its own item. */
async retract(v, service, type, id) {
if (service !== serviceNameOf(v.service)) fail(403, 'pulse.foreign_ref', 'A service retracts only its own items');
@@@ -144,15 +211,28 @@ function createPulse({ db, network = null, config = {} } = {}) {
},
/** GET /api/v1/pulse ?origin=user|ai|system&after=<cursor>&limit= → { items, next_cursor } */
-- async list(q = {}) {
++ async list(q = {}, viewer = null) {
const origin = q.origin == null || q.origin === '' || q.origin === 'all' ? null : String(q.origin);
if (origin && !ORIGINS.includes(origin)) fail(400, 'pulse.invalid_origin', `origin must be one of ${ORIGINS.join(', ')}`);
-- const limit = Math.min(Math.max(parseInt(q.limit, 10) || PAGE, 1), 100);
-- const before = decodeCursor(q.after, 2);
-- const { rows, hasMore } = await store.listItems(db, { origin, before, limit });
++ const mine = q.for === 'me';
++ if (q.for && !mine) fail(400, 'pulse.invalid_for', 'for must be me');
++ if (mine && (!viewer || viewer.kind !== 'user' || !viewer.token)) fail(401, 'pulse.sign_in_required', 'Sign in for your feed');
++ const limit = Math.min(Math.max(parseInt(q.limit, 10) || (mine ? 20 : PAGE), 1), 100);
++ const before = mine ? decodeCursor(q.cursor || q.after, 4) : decodeCursor(q.after, 2);
++ if (mine && before && (!/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/.test(before[0])
++ || ![0, 1].includes(before[1]) || typeof before[2] !== 'string' || !Number.isSafeInteger(before[3]))) {
++ fail(400, 'request.invalid_cursor', 'That cursor is not one this API issued');
++ }
++ const followed = mine ? await followedSubjects(viewer) : [];
++ const hour = mine ? (before ? before[0] : sqlTime(Date.now() - 60 * 60_000)) : null;
++ const { rows, hasMore } = await store.listItems(db, { origin, before: before && (mine ? before.slice(1) : [0, ...before]), limit, followed, hour });
const projections = await authors.projectionsFor(rows.map((r) => r.actor_subject));
const last = rows[rows.length - 1];
-- return { items: rows.map((r) => shape(r, projections)), next_cursor: hasMore && last ? encodeCursor([last.occurred_at, last.id]) : null };
++ return { items: rows.map((r) => shape(r, projections)), next_cursor: hasMore && last
++ ? encodeCursor(mine ? [hour, last.feed_priority, last.occurred_at, last.id] : [last.occurred_at, last.id]) : null,
++ // What this reader may do here: the menu's Feed shows its composer only when this says so (openvibe-shared
++ // ov-home.js), so it never offers a post the status route would refuse.
++ ...(viewer && viewer.kind === 'user' && viewer.subject ? { viewer: { can_post: true } } : {}) };
},
};
}
ddiff --git a/server/pulse/store.js b/server/pulse/store.js
iindex d51479d..ff6ff79 100644
---- a/server/pulse/store.js
++++ b/server/pulse/store.js
@@@ -59,11 +59,11 @@ async function upsertItem(db, it) {
return await db.tx(async () => {
const key = [it.source_service, it.source_type, String(it.source_id)];
if (await find.get(...key)) {
-- await db.prepare('UPDATE pulse_items SET title = ?, url = ? WHERE source_service = ? AND source_type = ? AND source_id = ?').run(it.title, it.url, ...key);
++ await db.prepare('UPDATE pulse_items SET title = ?, url = ?, picture_url = ?, app_name = ? WHERE source_service = ? AND source_type = ? AND source_id = ?').run(it.title, it.url, it.picture_url || null, it.app_name || null, ...key);
return { item: await find.get(...key), created: false };
}
-- await db.prepare(`INSERT INTO pulse_items (source_service, source_type, source_id, title, url, actor_subject, origin, visibility, occurred_at)
-- VALUES (?, ?, ?, ?, ?, ?, ?, 'public', ?)`).run(...key, it.title, it.url, it.actor_subject || null, it.origin, it.occurred_at);
++ await db.prepare(`INSERT INTO pulse_items (source_service, source_type, source_id, title, url, actor_subject, origin, visibility, occurred_at, picture_url, app_name)
++ VALUES (?, ?, ?, ?, ?, ?, ?, 'public', ?, ?, ?)`).run(...key, it.title, it.url, it.actor_subject || null, it.origin, it.occurred_at, it.picture_url || null, it.app_name || null);
return { item: await find.get(...key), created: true };
});
}
@@@ -83,11 +83,13 @@ async function removeItem(db, service, type, id) {
* opts.before [occurred_at, id] of the last item already seen
* → { rows, hasMore }
*/
--async function listItems(db, { origin = null, before = null, limit = 30 } = {}) {
++async function listItems(db, { origin = null, before = null, limit = 30, followed = [], hour = null } = {}) {
++ const priority = "CASE WHEN @hour::text IS NOT NULL AND i.occurred_at >= @hour AND jsonb_exists(@followed::jsonb, i.actor_subject) THEN 1 ELSE 0 END";
const rows = await db.prepare(`
-- SELECT i.* FROM pulse_items i
++ SELECT i.*, ${priority} AS feed_priority FROM pulse_items i
WHERE (@origin::text IS NULL OR i.origin = @origin)
-- AND (@at::text IS NULL OR i.occurred_at < @at OR (i.occurred_at = @at AND i.id < @id::bigint))
++ AND (@at::text IS NULL OR ${priority} < @priority::integer
++ OR (${priority} = @priority::integer AND (i.occurred_at < @at OR (i.occurred_at = @at AND i.id < @id::bigint))))
AND NOT (i.source_service = 'live' AND i.source_type = 'stream' AND EXISTS (
SELECT 1 FROM pulse_items j
WHERE j.source_service = 'live' AND j.source_type = 'stream'
@@@ -100,12 +102,16 @@ async function listItems(db, { origin = null, before = null, limit = 30 } = {})
OR (i.source_type = 'thread' AND EXISTS (SELECT 1 FROM threads t JOIN spaces s ON s.id = t.space_id
WHERE t.id = CAST(i.source_id AS INTEGER) AND t.deleted_at IS NULL AND s.visibility = 'public'
AND s.members_only_owner IS NULL AND t.members_only_owner IS NULL))
++ OR (i.source_type = 'status' AND EXISTS (SELECT 1 FROM threads t JOIN spaces s ON s.id = t.space_id
++ WHERE t.id = CAST(i.source_id AS INTEGER) AND t.deleted_at IS NULL AND s.slug = 'status'
++ AND s.visibility = 'public'))
OR (i.source_type = 'post' AND EXISTS (SELECT 1 FROM posts po JOIN threads t ON t.id = po.thread_id JOIN spaces s ON s.id = t.space_id
WHERE po.id = CAST(i.source_id AS INTEGER) AND po.deleted_at IS NULL AND t.deleted_at IS NULL AND s.visibility = 'public'
AND s.members_only_owner IS NULL AND t.members_only_owner IS NULL))
OR (i.source_type = 'submission' AND EXISTS (SELECT 1 FROM submissions su WHERE su.slug = i.source_id AND su.status = 'accepted')))
-- ORDER BY i.occurred_at DESC, i.id DESC
-- LIMIT @limit`).all({ origin, at: before ? before[0] : null, id: before ? before[1] : null, limit: limit + 1 });
++ ORDER BY feed_priority DESC, i.occurred_at DESC, i.id DESC
++ LIMIT @limit`).all({ origin, followed: JSON.stringify(followed), hour,
++ priority: before ? before[0] : null, at: before ? before[1] : null, id: before ? before[2] : null, limit: limit + 1 });
const hasMore = rows.length > limit;
if (hasMore) rows.pop();
return { rows, hasMore };
ddiff --git a/server/render/pulse.js b/server/render/pulse.js
iindex b986950..3ad64d0 100644
---- a/server/render/pulse.js
++++ b/server/render/pulse.js
@@@ -34,7 +34,8 @@ function itemHtml(it) {
<span class="badge pulse-source" title="${esc(`${it.source.service} · ${it.source.type}`)}">${esc(service)} · ${esc(it.source.type)}</span>
<div class="pulse-main">
<a class="thread-title" href="${esc(it.url)}"${local ? '' : ' rel="nofollow ugc noopener"'}>${esc(it.title)}</a>
-- <p class="thread-meta">${byline} <span class="sep">·</span> ${timeTag(it.occurred_at)}</p>
++ ${it.picture_url ? `<img src="${esc(it.picture_url)}" alt="" loading="lazy" decoding="async">` : ''}
++ <p class="thread-meta">${byline}${it.via ? ` <span class="badge">${esc(it.via)}</span>` : ''} <span class="sep">·</span> ${timeTag(it.occurred_at)}</p>
</div>
</li>`;
}
ddiff --git a/server/status/api.js b/server/status/api.js
nnew file mode 100644
iindex 0000000..5d85646
---- /dev/null
++++ b/server/status/api.js
@@@ -0,0 +1,55 @@
++'use strict';
++const express = require('express');
++const contracts = require('openvibe-contracts');
++const { run, fail, jsonBody } = require('../http/v1');
++
++const HTTPS_LINK = /https:\/\/[^\s<>]+/gi;
++
++function createStatusApi({ forum, pulse, viewers, limits, db, baseUrl }) {
++ const r = express.Router();
++ r.use(contracts.http.middleware());
++ r.use(viewers.middleware());
++ r.use(limits.reads('community.pulse.read'));
++ r.post('/', limits('community.status.write', { minute: 6, hour: 30 }), jsonBody, run(async (req) => {
++ if (req.viewer.kind !== 'user' || !req.viewer.subject) fail(401, 'status.sign_in_required', 'Sign in to post a status');
++ const body = req.body || {};
++ if (Object.keys(body).some((key) => !['text', 'link'].includes(key))) fail(400, 'status.invalid', 'Only text and link are accepted');
++ const text = typeof body.text === 'string' ? body.text.trim() : '';
++ const links = text.match(/https?:\/\/[^\s<>]+/gi) || [];
++ if (body.link != null) links.push(body.link);
++ if (body.link != null) {
++ let parsed;
++ try { parsed = new URL(body.link); } catch { fail(400, 'status.invalid_link', 'A valid https link is required'); }
++ if (parsed.protocol !== 'https:' || !parsed.hostname) fail(400, 'status.invalid_link', 'A valid https link is required');
++ }
++ if (links.length > 1 || links.some((link) => !link.startsWith('https://'))
++ || !contracts.validate('community.status@1', { text, ...(body.link == null ? {} : { link: body.link }) }).valid) {
++ fail(400, 'status.invalid', 'Text must be 1–500 characters and contain at most one https link');
++ }
++ const title = `Status: ${text.replace(/https:\/\/[^\s<>]+/gi, '').trim().slice(0, 110) || 'Shared a link'}`;
++ const { thread, post } = await forum.createThread(req.viewer, 'status', { title, body: text });
++ await pulse.forgetLocal('thread', thread.id);
++ await pulse.statusCreated(thread, text, req.viewer.subject);
++ return { text, link: links[0] || null, id: thread.id, post_id: post.id,
++ url: `${baseUrl.replace(/\/$/, '')}/forum/status/${thread.slug}`, created_at: thread.created_at };
++ }, 201));
++ return r;
++}
++
++function createProfileStatusApi({ db, baseUrl, viewers, limits }) {
++ const r = express.Router();
++ r.use(contracts.http.middleware());
++ r.use(viewers.middleware());
++ r.use(limits.reads('community.pulse.read'));
++ r.get('/:subject/statuses', run(async (req) => {
++ const rows = await db.prepare(`SELECT t.id, t.slug, t.created_at, p.id AS post_id, p.body_markdown AS text
++ FROM threads t JOIN spaces s ON s.id = t.space_id JOIN posts p ON p.thread_id = t.id AND p.is_opening = 1
++ WHERE s.slug = 'status' AND t.author_subject = ? AND t.deleted_at IS NULL AND p.deleted_at IS NULL
++ ORDER BY t.created_at DESC, t.id DESC LIMIT 20`).all(req.params.subject);
++ return { statuses: rows.map((x) => ({ ...x, link: (x.text.match(HTTPS_LINK) || [])[0] || null,
++ url: `${baseUrl.replace(/\/$/, '')}/forum/status/${x.slug}` })) };
++ }));
++ return r;
++}
++
++module.exports = { createStatusApi, createProfileStatusApi };
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex b684e77..8a220a8 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -35,6 +35,7 @@ const INVENTORY = {
'server/pastes/review.js': [2, 'OpenVibe.Vision /api/v1/review (OV_VISION_URL, fixed path) with a service token; only the paste\'s Media URL is sent'],
'server/observability.js': [1, '/api/ready probes of the configured Media health URL'],
'server/pulse/consumer.js': [2, 'Events subscriptions API (EVENTS_URL) with a service token'],
++ 'server/pulse/service.js': [1, 'Network /api/v1/me/follows at the configured networkInternalUrl, with the viewer token and a bounded cursor'],
'server/relay/discord-gateway.js': [1, 'DISCORD_GATEWAY_URL (owner) or the resume_gateway_url Discord\'s READY names'],
'server/relay/discord.js': [1, 'a Discord webhook URL read from an environment variable the owner allowlisted; mappings name the variable, never a URL'],
'server/relay/events-worker.js': [3, 'Events pull API (EVENTS_URL) with a service token'],
ddiff --git a/test/status.test.js b/test/status.test.js
nnew file mode 100644
iindex 0000000..c8658b2
---- /dev/null
++++ b/test/status.test.js
@@@ -0,0 +1,95 @@
++'use strict';
++const assert = require('node:assert/strict');
++const { ids } = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/app');
++
++(async () => {
++ const t = await boot({ authority: 'community', appOpts: { forumLimits: { threads: { cooldownSec: 0 } } } });
++ const person = t.network.addUser({ network_user_id: 91, username: 'statusowner', display_name: 'Status Owner' });
++ const token = t.network.sign({ id: 91, subject_id: person.subject_id, username: 'statusowner', role: 'user' });
++ const reader = t.network.addUser({ network_user_id: 93, username: 'statusreader', display_name: 'Status Reader' });
++ const readerToken = t.network.sign({ id: 93, subject_id: reader.subject_id, username: 'statusreader', role: 'user' });
++ const post = (body, auth = token) => t.get('/api/v1/me/status', { method: 'POST',
++ headers: { 'content-type': 'application/json', ...(auth ? { authorization: `Bearer ${auth}` } : {}) }, body: JSON.stringify(body) });
++ await check('status is a moderated public opening post and pulse item', async () => {
++ assert.equal((await post({ text: 'Hello world' }, null)).status, 401);
++ assert.equal((await post({ text: 'bad http://example.test' })).status, 400);
++ assert.equal((await post({ text: 'x'.repeat(501) })).status, 400);
++ const r = await post({ text: 'Hello world' });
++ assert.equal(r.status, 201, r.text);
++ const status = r.json();
++ assert.equal(status.text, 'Hello world');
++ assert.ok(status.post_id && status.url.includes('/forum/status/'));
++ const pulse = await t.db.prepare("SELECT * FROM pulse_items WHERE source_service = 'community' AND source_type = 'status' AND source_id = ?").get(String(status.id));
++ assert.equal(pulse.actor_subject, person.subject_id);
++ const profile = await t.get(`/api/v1/profiles/${person.subject_id}/statuses`);
++ assert.equal(profile.status, 200);
++ assert.equal(profile.json().statuses[0].text, 'Hello world');
++ const reaction = await t.get(`/api/v1/posts/${status.post_id}/reactions`, { method: 'POST',
++ headers: { authorization: `Bearer ${readerToken}`, 'content-type': 'application/json' }, body: JSON.stringify({ reaction: 'friendly' }) });
++ assert.equal(reaction.status, 200, reaction.text);
++ const slug = status.url.split('/').pop();
++ const reply = await t.get(`/api/v1/spaces/status/threads/${slug}/posts`, { method: 'POST',
++ headers: { authorization: `Bearer ${readerToken}`, 'content-type': 'application/json' }, body: JSON.stringify({ body: 'Nice update' }) });
++ assert.equal(reply.status, 201, reply.text);
++ const deleted = await t.get(`/api/v1/spaces/status/threads/${slug}`, { method: 'DELETE',
++ headers: { authorization: `Bearer ${token}` } });
++ assert.equal(deleted.status, 200, deleted.text);
++ assert.equal(await t.db.prepare("SELECT * FROM pulse_items WHERE source_service = 'community' AND source_type = 'status' AND source_id = ?").get(String(status.id)), undefined);
++ });
++ await check('a granted app contributes for its person and can retract only that person\'s item', async () => {
++ const { createPulse } = require('../server/pulse/service');
++ const pulse = createPulse({ db: t.db, config: { baseUrl: 'https://openvibe.community' } });
++ const app = { kind: 'service', service: 'app:app_test', subject: person.subject_id,
++ claims: { on_behalf_of: person.subject_id, app_name: 'Demo App' } };
++ await assert.rejects(pulse.contribute(app, { id: 'bad', text: 'Bad picture', url: 'https://example.test/thing',
++ picture_url: 'https://example.test/image.jpg' }));
++ const out = await pulse.contribute(app, { id: 'hello', text: 'Made a new thing', url: 'https://example.test/thing',
++ picture_url: 'https://openvibe.media/t/demo.jpg' });
++ assert.equal(out.created, true);
++ assert.equal(out.item.via, 'via Demo App');
++ assert.equal(out.item.picture_url, 'https://openvibe.media/t/demo.jpg');
++ assert.equal(out.item.actor.subject, person.subject_id);
++ assert.equal((await pulse.retractContribution(app, out.item.source.service, 'contribution', out.item.source.id)).removed, 1);
++ await assert.rejects(pulse.retractContribution({ ...app, subject: 'usr_01J00000000000000000000000',
++ claims: { on_behalf_of: 'usr_01J00000000000000000000000' } }, out.item.source.service, 'contribution', out.item.source.id));
++ });
++ await check('the HTTP route accepts a person-granted app token and shows its signed app name', async () => {
++ const appId = ids.newId('app');
++ const appToken = t.network.signService({ sub: `app:${appId}`, actorType: 'app', cap: ['community.pulse.contribute'],
++ extra: { project_id: ids.newId('project'), env: 'production', on_behalf_of: person.subject_id, app_name: 'Demo App' } });
++ const response = await t.get('/api/v1/pulse/items', { method: 'POST', headers: { authorization: `Bearer ${appToken}`, 'content-type': 'application/json' },
++ body: JSON.stringify({ id: 'demo-1', text: 'New demo', url: 'https://example.test/demo' }) });
++ assert.equal(response.status, 201, response.text);
++ assert.equal(response.json().item.via, 'via Demo App');
++ assert.equal(response.json().item.actor.subject, person.subject_id);
++ const source = response.json().item.source;
++ const deleted = await t.get(`/api/v1/pulse/items/${encodeURIComponent(source.service)}/contribution/${encodeURIComponent(source.id)}`, {
++ method: 'DELETE', headers: { authorization: `Bearer ${appToken}` } });
++ assert.equal(deleted.status, 200, deleted.text);
++ assert.equal(deleted.json().removed, 1);
++ });
++ await check('followed people lead a personal page and its cursor keeps the order', async () => {
++ const { createPulse } = require('../server/pulse/service');
++ const store = require('../server/pulse/store');
++ const followed = t.network.addUser({ network_user_id: 92, username: 'followed', display_name: 'Followed' });
++ const now = new Date();
++ const at = (ms) => new Date(now.getTime() - ms).toISOString().replace('T', ' ').slice(0, 19);
++ await store.upsertItem(t.db, { source_service: 'live', source_type: 'clip', source_id: 'older-followed',
++ title: 'followed', url: 'https://openvibe.live/c/f', actor_subject: followed.subject_id, origin: 'user', occurred_at: at(30_000) });
++ await store.upsertItem(t.db, { source_service: 'live', source_type: 'clip', source_id: 'newer-other',
++ title: 'other', url: 'https://openvibe.live/c/o', actor_subject: person.subject_id, origin: 'user', occurred_at: at(10_000) });
++ const pulse = createPulse({ db: t.db, config: { networkInternalUrl: 'https://network.test' },
++ fetchImpl: async () => ({ ok: true, json: async () => ({ items: [{ target_id: followed.subject_id }], next_cursor: null }) }) });
++ const viewer = { kind: 'user', subject: person.subject_id, token: 'user-token' };
++ const first = await pulse.list({ for: 'me', limit: 1 }, viewer);
++ assert.equal(first.items[0].source.id, 'older-followed');
++ assert.deepEqual(first.viewer, { can_post: true }, 'a signed-in reader is told they can post (the menu shows its composer)');
++ assert.equal((await pulse.list({ limit: 1 }, null)).viewer, undefined, 'a guest is not');
++ const second = await pulse.list({ for: 'me', limit: 1, cursor: first.next_cursor }, viewer);
++ assert.notEqual(second.items[0].id, first.items[0].id);
++ await assert.rejects(pulse.list({ for: 'me', cursor: Buffer.from(JSON.stringify(['bad', 1, 'time', 2])).toString('base64url') }, viewer));
++ });
++ await t.close();
++ done();
++})().catch((err) => { console.error(err); process.exitCode = 1; });