Ask 273: internal channel-points changes read for Network (#794)

ddiff --git a/package-lock.json b/package-lock.json
iindex 7cf5355e..cbde8bc2 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -24,7 +24,7 @@
          "multer": "^1.4.5-lts.1",
          "node-media-server": "^2.7.1",
          "onnxruntime-node": "^1.20.0",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.399.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.401.0",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.147.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.203.0",
          "pg": "^8.23.1",
@@@ -3373,9 +3373,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.399.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.399.0",
--      "integrity": "sha512-7QqSrMH2BT6ZOJu+xxusjEPICaVKGrr+3ij3tg0leE3sy4jfPwvWVTwaXZSCPM1feoP7heUDnSO4/9QBmjguJw==",
++      "version": "0.401.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.401.0",
++      "integrity": "sha512-NKuWV2l97rd2f7SZ2PkdDRvw3SeY8WrmYByorDSLvadJFFEcdSwXAilsSIa1QA8YqRkzcXi2eHdCGRmb6WZR3w==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex bb936e38..5a68d66e 100644
---- a/package.json
++++ b/package.json
@@@ -44,7 +44,7 @@
      "multer": "^1.4.5-lts.1",
      "node-media-server": "^2.7.1",
      "onnxruntime-node": "^1.20.0",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.399.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.401.0",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.147.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.203.0",
      "pg": "^8.23.1",
ddiff --git a/server/db/database.js b/server/db/database.js
iindex f4256c0b..e3de6b56 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -2222,7 +2222,10 @@ async function applyChannelPoints({ userId, streamerId, delta, key, reason = nul
      if (!userId || !streamerId || !Number.isInteger(delta) || delta === 0) {
          return { applied: false, replayed: false, balance: await getChannelPoints(userId, streamerId) };
      }
--    return await getDb().tx(async () => {
++    const events = require('../events/stream-events');
++    events.init();
++    let announced = false;
++    const result = await getDb().tx(async () => {
          const seen = await get('SELECT user_id, streamer_id, delta FROM channel_points_log WHERE idempotency_key = ?', [key]);
          if (seen) {
              if (seen.user_id !== userId || seen.streamer_id !== streamerId || seen.delta !== delta) {
@@@ -2242,8 +2245,19 @@ async function applyChannelPoints({ userId, streamerId, delta, key, reason = nul
          }
          await run('INSERT INTO channel_points_log (idempotency_key, user_id, streamer_id, delta, reason) VALUES (?, ?, ?, ?, ?)',
              [key, userId, streamerId, delta, reason ? String(reason).slice(0, 200) : null]);
--        return { applied: true, replayed: false, balance: await getChannelPoints(userId, streamerId) };
++        const balance = await getChannelPoints(userId, streamerId);
++        const linked = await get("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ? AND subject_id IS NOT NULL ORDER BY id DESC LIMIT 1", [userId]);
++        if (linked && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(linked.subject_id)) {
++            announced = !!(await events.enqueue({
++                event_type: 'live.balance.changed', actor: { type: 'service', id: 'live' },
++                subject: { type: 'user', id: linked.subject_id }, visibility: 'subject', priority: 'important',
++                payload: { currency: 'channel-points', streamer_id: streamerId, delta, balance, reason: reason || null },
++            }));
++        }
++        return { applied: true, replayed: false, balance };
      });
++    if (announced) events.kick();
++    return result;
  }
  /** Credit channel points for one event (key required). Returns the new balance. */
  async function addChannelPoints(userId, streamerId, amount, key, reason) {
ddiff --git a/server/internal/routes.js b/server/internal/routes.js
iindex c85be390..b52eece4 100644
---- a/server/internal/routes.js
++++ b/server/internal/routes.js
@@@ -9,6 +9,8 @@ const { guard } = require('../net/service-guard');
  const { http } = require('openvibe-contracts');
  const { bindSlot, BindError } = require('../openre/bind');
  
++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++
  // Service-to-service only, loopback only (nothing that came through nginx): each route checks the one capability it
  // performs on a Network service token (server/net/service-guard.js); nothing else gets in (X-Internal-Key: plan T2).
  
@@@ -66,6 +68,52 @@ router.post('/user-avatar', guard('live.avatar.write'), async (req, res) => {
      }
  });
  
++// ── Channel points changes for Network's gain/loss API (ask 273) ────────────────
++// GET /internal/channel-points/changes?network_user_id=<id>&since_id=&streamer_id=&limit=
++// guard('live.channel_points.read'): a Network service token only, loopback only. Reads Live's own
++// channel_points_log (the per-channel ledger; Billing stays the only money ledger) for the person
++// behind the Network account: one move per row, oldest first, with the cursor for the next page.
++router.get('/channel-points/changes', guard('live.channel_points.read'), async (req, res) => {
++    try {
++        const networkId = String(req.query.network_user_id || '');
++        if (!networkId) return res.status(400).json({ ok: false, error: 'network_user_id is required' });
++        // The account is named by its Network subject (usr_…); older links that only carry the numeric
++        // service id still resolve.
++        let linked = SUBJECT_RE.test(networkId)
++            ? await db.getDb().prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id DESC LIMIT 1").get(networkId)
++            : null;
++        if (!linked) linked = await db.getDb().prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND service_user_id = ?").get(networkId);
++        if (!linked) return res.status(404).json({ ok: false, error: 'no linked live account' });
++        const userId = linked.user_id;
++        let streamerId = null;
++        if (req.query.streamer_id != null && req.query.streamer_id !== '') {
++            streamerId = Number(req.query.streamer_id);
++            if (!Number.isInteger(streamerId) || streamerId <= 0) return res.status(400).json({ ok: false, error: 'streamer_id is a user id' });
++        }
++        let sinceId = 0;
++        if (req.query.since_id != null && req.query.since_id !== '') {
++            sinceId = Number(req.query.since_id);
++            if (!Number.isInteger(sinceId) || sinceId < 0) return res.status(400).json({ ok: false, error: 'since_id is a log id' });
++        }
++        const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 100, 1), 500);
++        const where = streamerId ? 'user_id = ? AND streamer_id = ? AND id > ?' : 'user_id = ? AND id > ?';
++        const args = streamerId ? [userId, streamerId, sinceId] : [userId, sinceId];
++        const rows = await db.getDb().prepare(`SELECT id, streamer_id, delta, reason, created_at AS at FROM channel_points_log
++            WHERE ${where} ORDER BY id ASC LIMIT ?`).all(...args, limit + 1);
++        const page = rows.slice(0, limit);
++        const balance = streamerId ? await db.getChannelPoints(userId, streamerId) : null;
++        return res.json({
++            ok: true, user_id: userId, streamer_id: streamerId, balance,
++            moves: page.map((r) => ({ id: r.id, streamer_id: r.streamer_id, at: r.at, delta: r.delta, reason: r.reason || null })),
++            cursor: page.length ? page[page.length - 1].id : sinceId,
++            next_since: rows.length > limit ? page[page.length - 1].id : null,
++        });
++    } catch (err) {
++        console.error('[Internal] channel-points/changes error:', err.message);
++        return res.status(500).json({ ok: false, error: err.message });
++    }
++});
++
  // ── OpenVibe Live on by default for OpenRestream (live.openre.slot.bind) ─────────
  // OpenRestream asks for a slot on the person's channel bound to their OpenRestream stream (server/openre/bind.js).
  router.post('/openre/slots', guard('live.openre.slot.bind'), async (req, res) => {
ddiff --git a/test/channel-points-changes.test.js b/test/channel-points-changes.test.js
nnew file mode 100644
iindex 00000000..94072c85
---- /dev/null
++++ b/test/channel-points-changes.test.js
@@@ -0,0 +1,123 @@
++'use strict';
++
++// Live's channel-points read for Network's gain/loss API (ask 273):
++// GET /internal/channel-points/changes is a Network service token with live.channel_points.read only,
++// loopback only; it reads Live's own channel_points_log for the person behind the Network account.
++const assert = require('assert');
++const crypto = require('crypto');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const express = require('express');
++const { serviceAuth } = require('openvibe-contracts');
++
++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-live-cp-changes-'));
++const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
++fs.writeFileSync(path.join(tmp, 'network.pem'), keys.publicKey);
++process.env.OV_NETWORK_PUBLIC_KEY = path.join(tmp, 'network.pem');
++process.env.OV_NETWORK_URL = 'https://openvibe.network';
++console.log = () => {};
++console.warn = () => {};
++console.error = () => {};
++const out = (...a) => process.stdout.write(`${a.join(' ')}\n`);
++
++const now = () => Math.floor(Date.now() / 1000);
++let jti = 0;
++const token = (cap, { aud = 'openvibe.live', sub = 'svc:network', env } = {}) => serviceAuth.signServiceToken({
++    iss: 'https://openvibe.network', sub, actor_type: 'service', aud: [aud], cap, iat: now(), exp: now() + 300, jti: `tok_test_${String(++jti).padStart(4, '0')}`, ...(env ? { env } : {}),
++}, keys.privateKey);
++
++(async () => {
++    const db = require('../server/db/database');
++    await db.initDb();
++    // A viewer and a streamer, the viewer linked to their Network account.
++    const fan = Number((await db.createUser({ username: 'cpfan', password_hash: 'x', display_name: 'Cp Fan', stream_key: 'cpfansecretkey1' })).lastInsertRowid);
++    const streamer = Number((await db.createUser({ username: 'cpstar', password_hash: 'x', display_name: 'Cp Star', stream_key: 'cpstarsecretkey1' })).lastInsertRowid);
++    const other = Number((await db.createUser({ username: 'cpother', password_hash: 'x', display_name: 'Cp Other', stream_key: 'cpothersecretkey1' })).lastInsertRowid);
++    await db.getDb().prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (?, 'network', ?, ?)").run(fan, 'net-fan-1', 'usr_01JAB2C3D4E5F6G7H8J9K0FAN1');
++    await db.getDb().prepare("INSERT INTO linked_accounts (user_id, service, service_user_id) VALUES (?, 'network', ?)").run(other, 'net-other-1');
++    const events = require('../server/events/stream-events');
++    const original = { init: events.init, enqueue: events.enqueue, kick: events.kick };
++    const sent = [];
++    let kicks = 0;
++    events.init = () => ({});
++    events.enqueue = async (env) => { sent.push(env); return env; };
++    events.kick = () => { kicks++; };
++    await db.applyChannelPoints({ userId: fan, streamerId: streamer, delta: 100, key: 'test:cp:watch:1', reason: 'watch' });
++    await db.applyChannelPoints({ userId: fan, streamerId: streamer, delta: -30, key: 'test:cp:redeem:1', reason: 'redeem reward 7' });
++    await db.applyChannelPoints({ userId: other, streamerId: streamer, delta: 50, key: 'test:cp:watch:2', reason: 'watch' });
++    await db.applyChannelPoints({ userId: fan, streamerId: streamer, delta: 100, key: 'test:cp:watch:1', reason: 'watch' });
++    assert.deepStrictEqual(sent.map((e) => e.payload.delta), [100, -30], 'only applied moves for a linked person announce');
++    assert(sent.every((e) => e.event_type === 'live.balance.changed' && e.visibility === 'subject' && e.subject.id === 'usr_01JAB2C3D4E5F6G7H8J9K0FAN1'));
++    assert.strictEqual(kicks, 2);
++    Object.assign(events, original);
++
++    const app = express();
++    app.use(express.json());
++    app.use('/internal', require('../server/internal/routes'));
++    const server = app.listen(0, '127.0.0.1');
++    if (!server.listening) await new Promise((r) => server.once('listening', r));
++    const origin = `http://127.0.0.1:${server.address().port}`;
++    const get = (p, headers = {}) => fetch(origin + p, { headers });
++    const Bearer = (t) => ({ authorization: `Bearer ${t}` });
++    const q = (o) => `/internal/channel-points/changes?${new URLSearchParams(o)}`;
++
++    let failures = 0;
++    const check = async (name, fn) => {
++        try { await fn(); out('  ✓', name); }
++        catch (e) { failures++; out('  ✗', name, '\n     ', (e.stack || String(e)).split('\n').slice(0, 6).join('\n      ')); }
++    };
++    const good = Bearer(token(['live.channel_points.read']));
++
++    await check('a gain and a loss, oldest first, with the balance', async () => {
++        const r = await get(q({ network_user_id: 'net-fan-1', streamer_id: String(streamer) }), good);
++        assert.strictEqual(r.status, 200);
++        const b = await r.json();
++        assert.strictEqual(b.balance, 70);
++        assert.strictEqual(b.moves.length, 2);
++        assert.strictEqual(b.moves[0].delta, 100);
++        assert.strictEqual(b.moves[0].reason, 'watch');
++        assert.strictEqual(b.moves[1].delta, -30);
++        assert.strictEqual(b.cursor, b.moves[1].id);
++        assert.strictEqual(b.next_since, null);
++    });
++
++    await check('the Network subject resolves the same account', async () => {
++        const rr = await get(q({ network_user_id: 'usr_01JAB2C3D4E5F6G7H8J9K0FAN1', streamer_id: String(streamer) }), good);
++        const b = await rr.json();
++        assert.strictEqual(rr.status, 200, JSON.stringify(b));
++        assert.strictEqual(b.moves.length, 2);
++        assert.strictEqual(b.balance, 70);
++    });
++
++    await check('a cursor pages forward', async () => {
++        const first = await (await get(q({ network_user_id: 'net-fan-1', streamer_id: String(streamer), limit: '1' }), good)).json();
++        assert.strictEqual(first.moves.length, 1);
++        assert.strictEqual(first.next_since, first.moves[0].id);
++        const rest = await (await get(q({ network_user_id: 'net-fan-1', streamer_id: String(streamer), since_id: String(first.next_since) }), good)).json();
++        assert.strictEqual(rest.moves.length, 1);
++        assert.strictEqual(rest.moves[0].delta, -30);
++    });
++
++    await check("another person's points are unreachable: no user parameter beyond the linked account", async () => {
++        const b = await (await get(q({ network_user_id: 'net-other-1', streamer_id: String(streamer) }), good)).json();
++        assert.strictEqual(b.moves.length, 1);
++        assert.strictEqual(b.moves[0].delta, 50, 'only their own row');
++        assert.strictEqual((await get(q({ network_user_id: 'nobody' }), good)).status, 404);
++        assert.strictEqual((await get(q({}), good)).status, 400);
++        assert.strictEqual((await get(q({ network_user_id: 'net-fan-1', streamer_id: 'x' }), good)).status, 400);
++    });
++
++    await check('only the capability opens it', async () => {
++        const p = q({ network_user_id: 'net-fan-1' });
++        assert.strictEqual((await get(p)).status, 401);
++        assert.strictEqual((await get(p, Bearer(token(['live.avatar.write'])))).status, 403);
++        assert.strictEqual((await get(p, Bearer(token(['live.channel_points.read'], { aud: 'openvibe.network' })))).status, 401);
++        assert.strictEqual((await get(p, { ...good, 'x-forwarded-for': '203.0.113.9' })).status, 403);
++    });
++
++    server.close();
++    if (failures) { out(`${failures} failed`); process.exit(1); }
++    out('channel-points changes: all passed');
++    process.exit(0);
++})().catch((e) => { process.stderr.write(String(e && e.stack || e)); process.exit(1); });