openvibers/openvibe.community Commits

Keep historical stream cards ended when channel is live again (#370)

OP OpenVibers committed 2 days ago · 75932364a0 · 5 files, +55 −10

* Match live feed cards to current broadcast time

* Keep active broadcasts ahead of their recordings

Changes

ddiff --git a/server/app.js b/server/app.js
iindex be79676..bc7ba2f 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -816,7 +816,7 @@ async function createApp(opts = {}) {
                  const item = rows[index];
                  const card = feedTools.card(item);
                  if (card.kind === 'live') {
--                    card.live_now = live.has(feedTools.channelOf(card.url).toLowerCase());
++                    card.live_now = feedTools.isCurrentLive(card, live);
                      if (!card.live_now) card.kind = 'stream';
                  }
                  seen.push(card);
ddiff --git a/server/pulse/feed.js b/server/pulse/feed.js
iindex e729005..e066bb2 100644
---- a/server/pulse/feed.js
++++ b/server/pulse/feed.js
@@@ -173,19 +173,30 @@ function createTrending({ networkUrl, fetchImpl }) {
  
  /** Who is live on Live right now (GET /api/streams, cached a minute): a "went live" card says LIVE only for them. */
  function createLiveNow({ liveUrl, fetchImpl }) {
--    let cache = new Set(); let at = 0;
++    let cache = new Map(); let at = 0;
      return async function liveNow() {
          if (Date.now() - at < 60_000) return cache;
          at = Date.now();
          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 Set(((j && j.streams) || []).map((x) => String((x && x.username) || '').toLowerCase()).filter(Boolean));
++            cache = new Map(((j && j.streams) || []).map((x) => {
++                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));
          } catch { /* keep the last answer */ }
          return cache;
      };
  }
  const channelOf = (url) => { try { const u = new URL(url); return /(^|\.)openvibe\.live$/.test(u.hostname) ? (/^\/@?([A-Za-z0-9_.-]{1,40})\/?$/.exec(u.pathname) || [])[1] || '' : ''; } catch { return ''; } };
++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;
++};
  
  function searchMatches(item, query) {
      const q = clean(query, 80).toLowerCase();
@@@ -221,7 +232,7 @@ function createFeedApi({ pulse, viewers, limits, db, networkUrl = 'https://openv
      const trending = createTrending({ networkUrl, fetchImpl });
      const liveNow = createLiveNow({ liveUrl, fetchImpl });
      // A "went live" item is live now only while that channel is on Live's live list.
--    const mark = (cards, live) => cards.forEach((c) => { if (c.kind === 'live') { c.live_now = live.has(channelOf(c.url).toLowerCase()); if (!c.live_now) c.kind = 'stream'; } });
++    const mark = (cards, live) => cards.forEach((c) => { if (c.kind === 'live') { c.live_now = isCurrentLive(c, live); if (!c.live_now) c.kind = 'stream'; } });
  
      router.get('/', run(async (req) => {
          const kind = KINDS.includes(String(req.query.kind || '')) ? String(req.query.kind) : '';
@@@ -321,4 +332,4 @@ function createFeedApi({ pulse, viewers, limits, db, networkUrl = 'https://openv
      return router;
  }
  
--module.exports = { createFeedApi, kindOf, card, countsFor, channelOf, createLiveNow, searchMatches, searchExamples, foldBursts };
++module.exports = { createFeedApi, kindOf, card, countsFor, channelOf, createLiveNow, isCurrentLive, searchMatches, searchExamples, foldBursts };
ddiff --git a/server/pulse/sources.js b/server/pulse/sources.js
iindex 5c7da73..7ce44ff 100644
---- a/server/pulse/sources.js
++++ b/server/pulse/sources.js
@@@ -282,6 +282,19 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
      async function publicContent() {
          const seen = new Set();
          let cursor = null, created = 0;
++        const live = await getJson(`${liveApi}/api/streams`);
++        const onAir = new Map();
++        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(), Date.parse(`${started.replace(' ', 'T')}Z`));
++            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)}`,
++                actor_name: name, origin: 'user', occurred_at: started });
++            if (r.created) created++;
++        }
          for (let page = 0; page < 100; page++) {
              const url = new URL('/api/content/feed', liveApi);
              url.searchParams.set('sort', 'new');
@@@ -295,10 +308,13 @@ function createPulseSources({ db, config = {}, fetchImpl = globalThis.fetch, now
                  const id = clean(v.id, 80);
                  if (!/^[A-Za-z0-9_-]{1,80}$/.test(id)) continue;
                  const type = v.kind === 'vod' ? 'recording' : 'clip';
--                seen.add(`${type}:${id}`);
                  const at = sqlFrom(v.created_at, null);
                  if (!at) continue;
                  const channel = clean(v.channel && v.channel.username, 64);
++                const currentStart = onAir.get(channel.toLowerCase());
++                if (type === 'recording' && Number.isFinite(currentStart)
++                    && Math.abs(Date.parse(`${at.replace(' ', 'T')}Z`) - currentStart) <= 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'
ddiff --git a/test/feed-api.test.js b/test/feed-api.test.js
iindex abecbbd..4f0d5f5 100644
---- a/test/feed-api.test.js
++++ b/test/feed-api.test.js
@@@ -16,7 +16,7 @@ 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' }, { username: 'ada' }] };
++const LIVE = { streams: [{ username: 'alex', started_at: new Date().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) };
@@@ -96,7 +96,7 @@ const at = (minsAgo) => new Date(Date.now() - minsAgo * 60_000).toISOString().re
          assert.strictEqual(r.status, 200, r.text);
          const j = r.json();
          const kinds = new Set(j.items.map((x) => x.kind));
--        for (const k of ['live', 'clip', 'video', 'pic', 'paste', 'meme', 'game', 'update', 'wiki', 'quest', 'chat', 'event', 'member']) assert.ok(kinds.has(k), `kind ${k} (${[...kinds]})`);
++        for (const k of ['stream', 'clip', 'video', 'pic', 'paste', 'meme', 'game', 'update', 'wiki', 'quest', 'chat', 'event', 'member']) assert.ok(kinds.has(k), `kind ${k} (${[...kinds]})`);
          assert.ok('next_cursor' in j);
          for (const x of j.items) for (const k of ['id', 'kind', 'url', 'title', 'text', 'at', 'site', 'site_name', 'actor', 'media', 'counts']) assert.ok(k in x, `${k} on every card`);
          const c = j.items.find((x) => x.url === 'https://openvibe.live/clip/c1');
@@@ -144,14 +144,17 @@ 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 mk = (id, who) => 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: a stream`, url: `https://openvibe.live/@${who}` }) });
++        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);
          assert.strictEqual((await mk('s2', 'bo')).status, 201);
++        assert.strictEqual((await mk('s3', 'alex', new Date(Date.now() - 2 * 86_400_000).toISOString())).status, 201);
          const j = (await t.get('/api/v1/feed?site=live&limit=40')).json();
          const by = Object.fromEntries(j.items.map((x) => [x.url, [x.kind, x.live_now]]));
          assert.deepStrictEqual(by['https://openvibe.live/@ada'], ['live', true], 'ada is on the live list');
          assert.deepStrictEqual(by['https://openvibe.live/@bo'], ['stream', false], 'bo went live earlier: a stream that happened, never LIVE');
++        const older = j.items.find((x) => x.title === 'alex went live: an older stream');
++        assert.deepStrictEqual([older.kind, older.live_now], ['stream', false], 'an old broadcast stays ended while the same channel is live again');
          for (const kind of ['live', 'stream']) {
              const filtered = (await t.get(`/api/v1/feed?kind=${kind}&limit=40`)).json();
              assert.ok(filtered.items.length > 0, `${kind} has a card`);
ddiff --git a/test/pulse-sources.test.js b/test/pulse-sources.test.js
iindex 0305843..1d944c2 100644
---- a/test/pulse-sources.test.js
++++ b/test/pulse-sources.test.js
@@@ -14,6 +14,7 @@ const day = (ms) => new Date(ms).toISOString().slice(0, 10);
      const yesterday = day(Date.now() - 86_400_000);
      const badgeAt = iso(Date.now() - 3_600_000);
      const calls = [];
++    let liveStreams = [];
      let galleryRows = [
          { id: 'pic_01JAB2C3D4E5F6G7H8J9K0MNPQ', title: 'First photo', author: 'Alex', listed: true, published_at: badgeAt,
              links: { page: 'https://openvibe.pics/i/firstpic', thumb: 'https://openvibe.pics/i/firstpic/thumb.webp' }, width: 800, height: 600 },
@@@ -37,6 +38,7 @@ const day = (ms) => new Date(ms).toISOString().slice(0, 10);
          if (u.includes('/api/rundown/streams')) return { ok: true, json: async () => ({ streams: [{ stream_id: 123, live: false,
              title: 'Night stream', started_at: badgeAt, thumbnail_url: 'https://openvibe.media/night.webp',
              episode: { title: 'The night sky', logline: 'A recap', aired_at: badgeAt, audience: 'public' } }], next: null }) };
++        if (u.endsWith('/api/streams')) return { ok: true, json: async () => ({ streams: liveStreams }) };
          if (u.includes('/api/content/feed')) return { ok: true, json: async () => ({ items: [
              { id: 123, kind: 'vod', title: 'Night stream', href: 'https://openvibe.video/watch?v=night', thumbnail_url: 'https://openvibe.media/night.webp', created_at: badgeAt, channel: { username: 'alex', display_name: 'Alex' }, duration_seconds: 500 },
              { id: 124, kind: 'clip', title: 'Night clip', href: 'https://openvibe.video/clip/night', thumbnail_url: 'https://openvibe.media/clip.webp', created_at: badgeAt, channel: { username: 'alex', display_name: 'Alex' }, duration_seconds: 25 },
@@@ -149,5 +151,18 @@ const day = (ms) => new Date(ms).toISOString().slice(0, 10);
          assert.strictEqual(store.pulseTime('2030-01-02T08:00:00+02:00'), '2030-01-02 06:00:00', 'an offset is read, not dropped');
      });
  
++    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 }];
++        await t.app.locals.pulseSources.runOnce({ force: true });
++        assert.ok(await item('stream', 999));
++        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('recording', 123));
++    });
++
      await done(t);
  })();