openvibers/openvibe.community Commits

Keep concurrent Live streams in the feed (#374)

OP OpenVibers committed 2 days ago · 74e634cdfc · 4 files, +32 −14

Changes

ddiff --git a/server/pulse/feed.js b/server/pulse/feed.js
iindex e066bb2..3a643eb 100644
---- a/server/pulse/feed.js
++++ b/server/pulse/feed.js
@@@ -180,12 +180,16 @@ function createLiveNow({ liveUrl, fetchImpl }) {
          try {
              const r = await fetchImpl(`${liveUrl}/api/streams`, { headers: { accept: 'application/json' }, signal: AbortSignal.timeout(3000) });
              const j = r.ok ? await r.json() : null;
--            cache = new Map(((j && j.streams) || []).map((x) => {
++            const byChannel = new Map();
++            for (const x of ((j && j.streams) || [])) {
                  const name = String((x && x.username) || '').toLowerCase();
                  const raw = String((x && x.started_at) || '');
                  const stamp = /^\d{4}-\d\d-\d\d \d\d:\d\d:\d\d$/.test(raw) ? `${raw.replace(' ', 'T')}Z` : raw;
--                return [name, Date.parse(stamp)];
--            }).filter(([name]) => name));
++                if (!name) continue;
++                if (!byChannel.has(name)) byChannel.set(name, []);
++                byChannel.get(name).push(Date.parse(stamp));
++            }
++            cache = byChannel;
          } catch { /* keep the last answer */ }
          return cache;
      };
@@@ -194,8 +198,9 @@ const channelOf = (url) => { try { const u = new URL(url); return /(^|\.)openvib
  const isCurrentLive = (card, live) => {
      const name = channelOf(card.url).toLowerCase();
      if (!live.has(name)) return false;
--    const started = live.get(name), recorded = Date.parse(card.at || '');
--    return !Number.isFinite(started) || !Number.isFinite(recorded) || Math.abs(started - recorded) <= 10 * 60_000;
++    const recorded = Date.parse(card.at || '');
++    return live.get(name).some((started) => !Number.isFinite(started) || !Number.isFinite(recorded)
++        || Math.abs(started - recorded) <= 10 * 60_000);
  };
  
  function searchMatches(item, query) {
ddiff --git a/server/pulse/sources.js b/server/pulse/sources.js
iindex 94e7b82..6e88e1e 100644
---- a/server/pulse/sources.js
++++ b/server/pulse/sources.js
@@@ -284,11 +284,14 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
          let cursor = null, created = 0;
          const live = await getJson(`${liveApi}/api/streams`);
          const onAir = new Map();
++        const liveIds = new Set();
          for (const s of (Array.isArray(live.streams) ? live.streams : [])) {
              const id = Number(s && s.id), channel = clean(s && s.username, 64);
              const started = sqlFrom(s && s.started_at, null);
              if (!Number.isSafeInteger(id) || id < 1 || !/^[A-Za-z0-9_.-]{1,64}$/.test(channel) || !started) continue;
--            onAir.set(channel.toLowerCase(), { id: String(id), started: Date.parse(`${started.replace(' ', 'T')}Z`) });
++            if (!onAir.has(channel.toLowerCase())) onAir.set(channel.toLowerCase(), []);
++            onAir.get(channel.toLowerCase()).push(Date.parse(`${started.replace(' ', 'T')}Z`));
++            liveIds.add(String(id));
              const name = clean(s.display_name || channel, 80);
              const r = await store.upsertItem(db, { source_service: 'live', source_type: 'stream', source_id: String(id),
                  title: `${name} went live: ${clean(s.title, 160) || 'a stream'}`, url: `${liveApi}/@${encodeURIComponent(channel)}`,
@@@ -311,15 +314,16 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
                  const at = sqlFrom(v.created_at, null);
                  if (!at) continue;
                  const channel = clean(v.channel && v.channel.username, 64);
--                const current = onAir.get(channel.toLowerCase());
--                if (type === 'recording' && current && Number.isFinite(current.started)
--                    && Math.abs(Date.parse(`${at.replace(' ', 'T')}Z`) - current.started) <= 10 * MINUTE) continue;
++                const current = onAir.get(channel.toLowerCase()) || [];
++                if (type === 'recording' && current.some((started) => Number.isFinite(started)
++                    && Math.abs(Date.parse(`${at.replace(' ', 'T')}Z`) - started) <= 10 * MINUTE)) continue;
                  seen.add(`${type}:${id}`);
                  let prior = null;
                  if (type === 'recording' && channel) {
--                    prior = await db.prepare(`SELECT actor_subject, source_id FROM pulse_items WHERE source_service = 'live' AND source_type = 'stream'
--                        AND lower(url) = lower(?) AND source_id <> ? AND occurred_at BETWEEN ? AND ? ORDER BY occurred_at DESC LIMIT 1`)
--                        .get(`${liveApi}/@${encodeURIComponent(channel)}`, current && current.id || '', sqlTime(Date.parse(`${at.replace(' ', 'T')}Z`) - 60 * MINUTE), sqlTime(Date.parse(`${at.replace(' ', 'T')}Z`) + 60 * MINUTE));
++                    const candidates = await db.prepare(`SELECT actor_subject, source_id FROM pulse_items WHERE source_service = 'live' AND source_type = 'stream'
++                        AND lower(url) = lower(?) AND occurred_at BETWEEN ? AND ? ORDER BY occurred_at DESC LIMIT 20`)
++                        .all(`${liveApi}/@${encodeURIComponent(channel)}`, sqlTime(Date.parse(`${at.replace(' ', 'T')}Z`) - 60 * MINUTE), sqlTime(Date.parse(`${at.replace(' ', 'T')}Z`) + 60 * MINUTE));
++                    prior = candidates.find((row) => !liveIds.has(row.source_id)) || null;
                  }
                  const r = await store.upsertItem(db, { source_service: 'live', source_type: type, source_id: id,
                      title: `${type === 'recording' ? 'Recording: ' : ''}${clean(v.title, 150) || (type === 'recording' ? 'A stream' : 'A clip')}`,
ddiff --git a/test/feed-api.test.js b/test/feed-api.test.js
iindex 4f0d5f5..69f662e 100644
---- a/test/feed-api.test.js
++++ b/test/feed-api.test.js
@@@ -16,7 +16,9 @@ const FRAME = { cards: [
      ] },
      { site: 'news', title: 'Top stories', items: [{ title: 'Not a kind the feed shows', url: 'https://openvibe.news/a', image: { src: 'https://openvibe.media/t/n.webp' } }] },
  ] };
--const LIVE = { streams: [{ username: 'alex', started_at: new Date().toISOString() }, { username: 'ada', started_at: new Date().toISOString() }] };
++const LIVE = { streams: [{ username: 'alex', started_at: new Date().toISOString() },
++    { username: 'alex', started_at: new Date(Date.now() - 40 * 60_000).toISOString() },
++    { username: 'ada', started_at: new Date().toISOString() }] };
  const feedFetch = async (url) => {
      const u = String(url);
      return { ok: /\/api\/(frame|streams)/.test(u), json: async () => (/\/api\/streams/.test(u) ? LIVE : FRAME) };
@@@ -144,6 +146,10 @@ const at = (minsAgo) => new Date(Date.now() - minsAgo * 60_000).toISOString().re
      });
  
      await check('LIVE only for a channel that is live now', async () => {
++        const feedTools = require('../server/pulse/feed');
++        const current = await feedTools.createLiveNow({ liveUrl: 'https://openvibe.live', fetchImpl: feedFetch })();
++        assert.strictEqual(current.get('alex').length, 2, 'both active stream starts are retained for one channel');
++        assert.ok(feedTools.isCurrentLive({ url: 'https://openvibe.live/@alex', at: new Date(Date.now() - 40 * 60_000).toISOString() }, current));
          const mk = (id, who, at = null) => t.get('/api/v1/pulse/items', { method: 'POST', headers: { 'content-type': 'application/json', authorization: `Bearer ${netSvc}`, 'x-ov-subject': alex.subject_id },
              body: JSON.stringify({ ref: { service: 'live', type: 'stream', id }, title: `${who} went live: ${at ? 'an older stream' : 'a stream'}`, url: `https://openvibe.live/@${who}`, ...(at ? { occurred_at: at } : {}) }) });
          assert.strictEqual((await mk('s1', 'ada')).status, 201);
ddiff --git a/test/pulse-sources.test.js b/test/pulse-sources.test.js
iindex fddd185..d3c9e15 100644
---- a/test/pulse-sources.test.js
++++ b/test/pulse-sources.test.js
@@@ -156,15 +156,18 @@ const day = (ms) => new Date(ms).toISOString().slice(0, 10);
      await check('a current broadcast stays one Live item until it becomes a recording', async () => {
          const db = t.app.locals.db;
          const item = (type, id) => db.prepare('SELECT title FROM pulse_items WHERE source_service = ? AND source_type = ? AND source_id = ?').get('live', type, String(id));
--        liveStreams = [{ id: 999, username: 'alex', display_name: 'Alex', title: 'Night stream', started_at: badgeAt }];
++        liveStreams = [{ id: 999, username: 'alex', display_name: 'Alex', title: 'Night stream', started_at: badgeAt },
++            { id: 998, username: 'alex', display_name: 'Alex', title: 'Earlier live stream', started_at: iso(Date.parse(badgeAt) - 40 * 60_000) }];
          extraContent = [{ id: 125, kind: 'vod', title: 'Earlier broadcast', href: 'https://openvibe.video/watch?v=earlier',
              created_at: iso(Date.parse(badgeAt) - 20 * 60_000), channel: { username: 'alex', display_name: 'Alex' } }];
          await t.app.locals.pulseSources.runOnce({ force: true });
          assert.ok(await item('stream', 999));
++        assert.ok(await item('stream', 998), 'both on-air stream IDs survive an older VOD on the same channel');
          assert.ok(!(await item('recording', 123)), 'a recording for the broadcast is suppressed while it is live');
          liveStreams = [];
          await t.app.locals.pulseSources.runOnce({ force: true });
          assert.ok(!(await item('stream', 999)), 'the ended Live item leaves when the recording arrives');
++        assert.ok(!(await item('stream', 998)));
          assert.ok(await item('recording', 123));
      });