Finish Feed pagination and current live/stream filters (#329)

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 70280f8..2a7d11e 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -2,6 +2,8 @@
  
  ## Unreleased
  
++- Keep every Pulse row when paging the Feed, include trending cards within `limit`, and apply live/stream filters after resolving current live status.
++
  - **The wide Feed** (owner, 11-09: "one huge feed of everything, all the data, SDK and API enabled"). Pulse now carries
    every kind the menu's Feed draws — `wiki` (new and updated articles), `quest` (a finished quest's badge), `chat` (a
    busy public room hour), `event` (Today's calendar) and `member` (a day's new Network accounts, "5 people joined
ddiff --git a/server/pulse/feed.js b/server/pulse/feed.js
iindex 05d749a..ca287f5 100644
---- a/server/pulse/feed.js
++++ b/server/pulse/feed.js
@@@ -20,7 +20,7 @@
   */
  const express = require('express');
  const contracts = require('openvibe-contracts');
--const { run, fail, isoTime } = require('../http/v1');
++const { run, fail, isoTime, sqlTime, encodeCursor } = require('../http/v1');
  
  const KINDS = ['live', 'stream', 'clip', 'video', 'pic', 'post', 'meme', 'game', 'update', 'wiki', 'quest', 'chat', 'event', 'member'];
  const SITE_NAMES = { live: 'Live', 'media-hub': 'Video', video: 'Video', pics: 'Pics', meme: 'Meme', games: 'Games', community: 'Community', blog: 'Blog', wiki: 'Wiki', quest: 'Quest', chat: 'Chat', tools: 'Tools', news: 'News', report: 'Report', today: 'Today', network: 'Network' };
@@@ -178,43 +178,52 @@ function createFeedApi({ pulse, viewers, limits, db, networkUrl = 'https://openv
          if (following && (!req.viewer || req.viewer.kind !== 'user' || !req.viewer.token)) fail(401, 'feed.sign_in_required', 'Sign in to see the people you follow');
          const followed = following ? new Set(await pulse.followedFor(req.viewer)) : null;
  
++        const live = await liveNow();
++        // Copy cached trending cards before marking them: a channel can go live again during the cache TTL.
++        const extras = !req.query.cursor && !q && !following
++            ? (await trending()).map((c) => ({ ...c })) : [];
++        mark(extras, live);
++        const t = extras.filter((c) => (!kind || c.kind === kind) && (!site || c.site === site));
++        // Reserve at most one in four slots for trending, keeping the requested limit for the whole response.
++        const pulseWant = want - Math.min(t.length, Math.floor(want / 4));
          const picked = [];
          let cursor = req.query.cursor ? String(req.query.cursor) : '';
          let next = null;
          // A kind, site or word filters Pulse's pages until the page is full (at most eight pages read per call).
--        for (let pages = 0; pages < MAX_PAGE_READS && picked.length < want; pages++) {
++        for (let pages = 0; pages < MAX_PAGE_READS && picked.length < pulseWant; pages++) {
              const page = await pulse.list({ after: cursor || undefined, limit: 40 }, req.viewer);
--            for (const i of page.items || []) {
++            const rows = page.items || [];
++            next = page.next_cursor || null;
++            for (let index = 0; index < rows.length; index++) {
++                const i = rows[index];
                  const actorSubject = i.actor_subject || (i.actor && i.actor.subject) || null;
                  if (followed && !(actorSubject && followed.has(actorSubject))) continue;
                  const c = card(i);
++                mark([c], live);
                  if (kind && c.kind !== kind) continue;
                  if (site && c.site !== site) continue;
                  if (q && !(c.title.toLowerCase().includes(q) || c.text.toLowerCase().includes(q))) continue;
                  picked.push({ row: i, card: c });
--                if (picked.length >= want) break;
++                if (picked.length >= pulseWant) {
++                    // Resume after the last consumed row, including when Pulse's final page was only partly read.
++                    if (index < rows.length - 1) next = encodeCursor([sqlTime(i.occurred_at), i.id]);
++                    break;
++                }
              }
--            next = page.next_cursor || null;
              if (!next) break;
              cursor = next;
          }
          const counts = await countsFor(db, picked.map((p) => p.row));
          for (const p of picked) p.card.counts = counts.get(`${typeOf(p.row)}:${idOf(p.row)}`) || {};
  
--        // The first page carries the network's trending pictures and videos, one after every three items.
--        if (!req.query.cursor && !q && !following) {
--            const t = (await trending()).filter((x) => (!kind || x.kind === kind) && (!site || x.site === site));
--            if (t.length) {
--                const woven = [];
--                let k = 0;
--                picked.forEach((p, n) => { woven.push(p.card); if ((n + 1) % 3 === 0 && k < t.length) woven.push(t[k++]); });
--                while (woven.length < want && k < t.length) woven.push(t[k++]);
--                mark(woven, await liveNow());
--                return { items: woven, next_cursor: next };
--            }
--        }
--        const items = picked.map((p) => p.card);
--        mark(items, await liveNow());
++        // The first page carries trending pictures and videos, one after every three Pulse items.
++        const items = [];
++        let k = 0;
++        picked.forEach((p, n) => {
++            items.push(p.card);
++            if ((n + 1) % 3 === 0 && k < Math.min(t.length, Math.floor(want / 4))) items.push(t[k++]);
++        });
++        while (items.length < want && k < t.length) items.push(t[k++]);
          return { items, next_cursor: next };
      }));
      router.use((req, res) => contracts.http.sendProblem(res, 404, 'route.not_found', { detail: 'Not found', ctx: req.ov }));
ddiff --git a/test/feed-api.test.js b/test/feed-api.test.js
iindex 279daaf..825b7b4 100644
---- a/test/feed-api.test.js
++++ b/test/feed-api.test.js
@@@ -120,10 +120,16 @@ const at = (minsAgo) => new Date(Date.now() - minsAgo * 60_000).toISOString().re
              body: JSON.stringify({ ref: { service: 'live', type: 'stream', id }, title: `${who} went live: a stream`, url: `https://openvibe.live/@${who}` }) });
          assert.strictEqual((await mk('s1', 'ada')).status, 201);
          assert.strictEqual((await mk('s2', 'bo')).status, 201);
--        const j = (await t.get('/api/v1/feed?kind=live&limit=40')).json();
++        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');
++        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`);
++            assert.ok(filtered.items.every((x) => x.kind === kind), `${kind} filters the displayed kind`);
++        }
++
      });
  
      await check('a kind filters the stream, a site filters the stream, q searches title and text', async () => {
@@@ -164,6 +170,39 @@ const at = (minsAgo) => new Date(Date.now() - minsAgo * 60_000).toISOString().re
          assert.ok(!write.headers.get('access-control-allow-origin'), 'a write from elsewhere gets no CORS');
      });
  
++    await check('small pages retain every Pulse item, including the final partially read page', async () => {
++        const all = (await t.get('/api/v1/feed?limit=40')).json().items.filter((x) => x.id.startsWith('p:'));
++        for (const limit of [1, 4]) {
++            const seen = []; let cursor = ''; const cursors = new Set();
++            do {
++                const r = await t.get(`/api/v1/feed?limit=${limit}${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`);
++                assert.strictEqual(r.status, 200, r.text);
++                const page = r.json();
++                assert.ok(page.items.length <= limit, 'trending counts toward the page limit');
++                seen.push(...page.items.filter((x) => x.id.startsWith('p:')).map((x) => x.id));
++                cursor = page.next_cursor;
++                if (cursor) { assert.ok(!cursors.has(cursor), 'cursor advances'); cursors.add(cursor); }
++                assert.ok(cursors.size < 40, 'pagination terminates');
++            } while (cursor);
++            assert.deepStrictEqual(seen, all.map((x) => x.id), `limit=${limit}: no missing or repeated rows`);
++        }
++    });
++
++    await check('filtered pages keep scanning after an empty bounded page', async () => {
++        // More than the eight 40-row scans in one call, ahead of the existing Wiki fixture.
++        await db.prepare(`INSERT INTO pulse_items (source_service, source_type, source_id, title, url, origin, visibility, occurred_at)
++            SELECT 'feed-scan-test', 'event', n::text, 'Unrelated event', 'https://openvibe.today/', 'system', 'public', ?
++            FROM generate_series(1, 321) n`).run(at(1));
++        try {
++            const first = (await t.get('/api/v1/feed?site=wiki&limit=1')).json();
++            assert.deepStrictEqual(first.items, []);
++            assert.ok(first.next_cursor, 'empty bounded scan is not the end');
++            const next = (await t.get(`/api/v1/feed?site=wiki&limit=1&cursor=${encodeURIComponent(first.next_cursor)}`)).json();
++            assert.strictEqual(next.items.length, 1);
++            assert.strictEqual(next.items[0].site, 'wiki');
++        } finally { await db.prepare("DELETE FROM pulse_items WHERE source_service = 'feed-scan-test'").run(); }
++    });
++
      // ── POST /api/v1/posts with pictures: the Feed shows it at once ──
      const upload = async (jwt) => {
          const fd = new FormData();