openvibers/openvibe.media Commits

Media: one view API for every kind (previews, engaged viewers, counts on pages) (#426)

OP OpenVibers committed 21 h ago · fa67707a4d · 15 files, +823 −17

* Media: one view API for every kind — previews, engaged viewers, counts on items

* Merge origin/main into views-media; pin contracts 0.416.0 and sdk 0.158.0

* Docs: name contracts 0.416.0 and sdk 0.158.0

Changes

ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex 2f7dd18..c147842 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -3,6 +3,7 @@ on:
push:
branches: [main]
pull_request:
++ workflow_dispatch:
jobs:
# install, node --check, npm test (with test/n-1.test.js: the previous release's pages and SDK calls,
# and its SQL, against this one, from test/fixtures/n-1) and openvibe-contracts-check: the shared test
ddiff --git a/migrations/0029_unified_views.sql b/migrations/0029_unified_views.sql
nnew file mode 100644
iindex 0000000..db5e9a3
---- /dev/null
++++ b/migrations/0029_unified_views.sql
@@@ -0,0 +1,69 @@
++-- phase: expand
++-- One view store for every content kind (the owner, 11-12: "add view count, and count views even if you're just
++-- watching on the feed … then engaged viewers when you actually watch more of the video"). `service` + `kind` +
++-- `item_id` name the item in its owner's space (media/vod, media/clip, live/stream, network/feed…), so Media can
++-- count a Feed item and a Live stream without owning their rows. `creator_subject` and `site` are the optional
++-- Report dimensions the caller sends (a canonical subject or channel username, and a site id).
++--
++-- Privacy: content_view_actors stores a keyed hash of the person's canonical subject or anonymous session id,
++-- never an IP address and never a user agent. No other table here stores an identifier.
++--
++-- content_view_counts is the item total (a reviewable number the API reads); content_view_days is the same
++-- numbers by UTC day for Report; content_view_actors is the 30-minute window, cumulative watch time and the one
++-- engagement per viewer.
++
++CREATE TABLE content_view_counts (
++ service text COLLATE "C" NOT NULL,
++ kind text COLLATE "C" NOT NULL,
++ item_id text COLLATE "C" NOT NULL,
++ site text COLLATE "C",
++ creator_subject text COLLATE "C",
++ views bigint NOT NULL DEFAULT 0,
++ engaged bigint NOT NULL DEFAULT 0,
++ previews bigint NOT NULL DEFAULT 0,
++ live_peak bigint,
++ updated_at text COLLATE "C" DEFAULT ov_now(),
++ PRIMARY KEY (service, kind, item_id)
++);
++
++CREATE TABLE content_view_days (
++ service text COLLATE "C" NOT NULL,
++ kind text COLLATE "C" NOT NULL,
++ item_id text COLLATE "C" NOT NULL,
++ day text COLLATE "C" NOT NULL,
++ site text COLLATE "C",
++ creator_subject text COLLATE "C",
++ views bigint NOT NULL DEFAULT 0,
++ engaged bigint NOT NULL DEFAULT 0,
++ previews bigint NOT NULL DEFAULT 0,
++ live_peak bigint,
++ PRIMARY KEY (service, kind, item_id, day)
++);
++
++CREATE TABLE content_view_actors (
++ service text COLLATE "C" NOT NULL,
++ kind text COLLATE "C" NOT NULL,
++ item_id text COLLATE "C" NOT NULL,
++ actor text COLLATE "C" NOT NULL,
++ first_ms bigint NOT NULL,
++ last_view_ms bigint,
++ last_seen_ms bigint,
++ watched_seconds double precision NOT NULL DEFAULT 0,
++ engaged_ms bigint,
++ live boolean NOT NULL DEFAULT false,
++ PRIMARY KEY (service, kind, item_id, actor)
++);
++
++CREATE INDEX idx_content_view_days_day ON content_view_days(day);
++CREATE INDEX idx_content_view_days_creator ON content_view_days(creator_subject, day);
++CREATE INDEX idx_content_view_days_site ON content_view_days(site, day);
++CREATE INDEX idx_content_view_actors_seen ON content_view_actors(service, kind, item_id, last_seen_ms);
++
++-- The counts every page already had keep working: a VOD's and a clip's current view_count becomes its starting
++-- view total (the legacy visit counter keeps deduping for six hours and now adds to the same total).
++INSERT INTO content_view_counts (service, kind, item_id, views)
++SELECT 'media', 'vod', id::text, COALESCE(view_count, 0) FROM vods
++ON CONFLICT (service, kind, item_id) DO NOTHING;
++INSERT INTO content_view_counts (service, kind, item_id, views)
++SELECT 'media', 'clip', id::text, COALESCE(view_count, 0) FROM clips
++ON CONFLICT (service, kind, item_id) DO NOTHING;
ddiff --git a/server/auth.js b/server/auth.js
iindex 09874a8..278281f 100644
---- a/server/auth.js
++++ b/server/auth.js
@@@ -411,6 +411,16 @@ async function optionalIdentity(req, _res, next) {
next();
}
++/**
++ * The canonical subject of a signed-in viewer, or null. Used only by the public view API to dedupe a person's views
++ * across sessions: it is not an app-local user id and no route may use it as one (see tenantAuth's note above).
++ */
++async function viewerSubject(token) {
++ const payload = await verifyUserJwt(token);
++ if (!payload) return null;
++ return String(payload.subject_id || payload.sub || '') || null;
++}
++
/** CORS for tenant API routes — reflects allow-listed origins per app. */
function tenantCors(req, res, next) {
const origin = req.headers.origin;
@@@ -442,6 +452,7 @@ module.exports = {
networkKeys,
tenantCors,
optionalIdentity,
++ viewerSubject,
// verifyUserJwt stays internal: it proves a token is a genuine Network credential,
// but its subject is a Network id and must never be used as an app-local user id.
startJwksRefresh,
ddiff --git a/server/events.js b/server/events.js
iindex 5c78449..19d2f8c 100644
---- a/server/events.js
++++ b/server/events.js
@@@ -122,6 +122,23 @@ async function record(outcome, appId, data) {
return await enqueue(eventType, appId, subject, data, eventType === 'media.storage.recovered' ? 'low' : 'important');
}
++/**
++ * Queue one counted view or first engagement (media.viewed / media.engaged) inside the transaction that updates the
++ * counters. Its payload is the Contracts projection (ref, how, times, site, creator, concurrent) and carries no IP,
++ * user agent or raw session id. Low priority: a view is never an alert.
++ */
++async function recordView(eventType, subject, payload) {
++ if (!outbox) return null;
++ return await outbox.enqueue(inTx(), {
++ event_type: eventType,
++ actor: { type: 'service', id: 'media' },
++ subject,
++ visibility: 'internal',
++ priority: 'low',
++ payload,
++ });
++}
++
/** Queue one placement decision inside its transaction; disabled and sandbox outboxes return null. */
async function recordPlacement(eventType, appId, subject, payload, { priority = 'low' } = {}) {
if (!PLACEMENT_TYPES.has(eventType)) throw new Error(`unknown placement event ${eventType}`);
@@@ -349,4 +366,5 @@ function _reset() {
stats.queued = 0; stats.lastError = null;
}
--module.exports = { init, initWriter, record, publish, recordPlacement, recordIndexDocument, recordModeration, recordJob, recordUsage, checkUsage, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, PLACEMENT_TYPES, JOB_TRANSITIONS, _reset };
++module.exports = {
++ recordView, init, initWriter, record, publish, recordPlacement, recordIndexDocument, recordModeration, recordJob, recordUsage, checkUsage, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, PLACEMENT_TYPES, JOB_TRANSITIONS, _reset };
ddiff --git a/server/index.js b/server/index.js
iindex b02a93e..9e9bea4 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -256,7 +256,10 @@ const ready = (async () => {
}
});
-- if (!drill.enabled) setInterval(async () => { try { const n = await require('./views/service').prune(30); if (n) console.log(`[Views] pruned ${n} stale visit row(s)`); } catch { /* */ } }, 12 * 3600 * 1000);
++ if (!drill.enabled) setInterval(async () => {
++ try { const n = await require('./views/service').prune(30); if (n) console.log(`[Views] pruned ${n} stale visit row(s)`); } catch { /* */ }
++ try { const n = await require('./views/unified').prune(30); if (n) console.log(`[Views] pruned ${n} stale session row(s)`); } catch { /* */ }
++ }, 12 * 3600 * 1000);
// The object explorer (WS-G task 12): a signed-in person's own objects and usage, and the operator views
// (server/me/routes.js). /api/v2/me sits ahead of /api/v2/:app/…, so "me" is never a tenant there.
const me = require('./me/routes').createMeRoutes({ auth: userAuthClient });
@@@ -307,6 +310,10 @@ const ready = (async () => {
app.use('/a', attachments.publicRouter);
if (!drill.enabled) app.locals.attachmentsCleanup = attachments.startCleanup();
}
++ // The one view API (server/views/unified.js): every content kind, from a browser beacon on any site. It sits
++ // before the tenant routes and takes no app key, cookie or user id — only a Network person token when the viewer
++ // has one. /api/v1/:app/views below is the older app-facing counter and keeps answering (N-1).
++ app.use('/api/v1/views', require('./views/api'));
app.use('/api/v1/:app/views', require('./views/routes'));
// GET /api/v1/:app/watch/:watch_id: which item a watch id is (server/watch-routes.js, plan track U).
app.use('/api/v1/:app/watch', require('./watch-routes'));
ddiff --git a/server/public/pages.js b/server/public/pages.js
iindex c74c053..381425a 100644
---- a/server/public/pages.js
++++ b/server/public/pages.js
@@@ -155,6 +155,9 @@ function renderWatchPage(kind, record, readiness = null) {
const mime = VIDEO_MIME[path.extname(record.file_path || '').toLowerCase()] || 'video/mp4';
const uploadDate = isoDate(record.created_at);
const duration = isoDuration(record.duration_seconds);
++ // The one view store when the page route attached it; a crawl (or a legacy row) falls back to view_count.
++ const viewTotal = record.counts && record.counts.views != null ? Number(record.counts.views) : (record.view_count != null ? Number(record.view_count) : null);
++ const engaged = record.counts && record.counts.engaged != null ? Number(record.counts.engaged) : null;
const videoObject = {
'@type': 'VideoObject',
@@@ -169,7 +172,7 @@ function renderWatchPage(kind, record, readiness = null) {
author: ownerUrl ? { '@type': 'Person', name: by, url: ownerUrl } : { '@type': 'Organization', name: 'OpenVibe AI', url: NETWORK_URL },
creator: ownerUrl ? { '@type': 'Person', name: by, url: ownerUrl } : { '@type': 'Organization', name: 'OpenVibe AI', url: NETWORK_URL },
...(sourceVod ? { isBasedOn: sourceVod } : {}),
-- ...(record.view_count ? { interactionStatistic: { '@type': 'InteractionCounter', interactionType: 'https://schema.org/WatchAction', userInteractionCount: record.view_count } } : {}),
++ ...(viewTotal ? { interactionStatistic: { '@type': 'InteractionCounter', interactionType: 'https://schema.org/WatchAction', userInteractionCount: viewTotal } } : {}),
publisher: { '@type': 'Organization', name: 'OpenVibe', url: NETWORK_URL },
isFamilyFriendly: true,
};
@@@ -177,7 +180,7 @@ function renderWatchPage(kind, record, readiness = null) {
const metaBits = [
isVod ? 'VOD' : ai ? 'AI clip' : 'Clip',
record.duration_seconds ? fmtDuration(record.duration_seconds) : null,
-- record.view_count != null ? `${Number(record.view_count).toLocaleString('en-US')} views` : null,
++ viewTotal != null ? `${viewTotal.toLocaleString('en-US')} view${viewTotal === 1 ? '' : 's'}${engaged ? ` · ${engaged.toLocaleString('en-US')} engaged` : ''}` : null,
fmtDate(record.created_at) || null,
by ? (ownerUrl ? `by <a href="${esc(ownerUrl)}">${esc(by)}</a>` : `by ${esc(by)}`) : null,
].filter(Boolean);
@@@ -249,7 +252,8 @@ function renderWatchPage(kind, record, readiness = null) {
summary: description, url: canonical,
facts: [
['Duration', record.duration_seconds ? fmtDuration(record.duration_seconds) : null],
-- ['Views', record.view_count != null ? Number(record.view_count).toLocaleString('en-US') : null],
++ ['Views', viewTotal != null ? viewTotal.toLocaleString('en-US') : null],
++ ['Engaged viewers', engaged != null ? engaged.toLocaleString('en-US') : null],
].filter(([, v]) => v != null),
updated: uploadDate || undefined,
} : undefined,
ddiff --git a/server/public/routes.js b/server/public/routes.js
iindex c3a5b7c..1c9b132 100644
---- a/server/public/routes.js
++++ b/server/public/routes.js
@@@ -296,6 +296,9 @@ async function serveMediaRecord(kind, record, req, res) {
if ((record.app_id || 'live') === 'live' && !record.watch_id) {
record.watch_id = await require('../watch-ids').ensure(kind, record.id);
}
++ // The one view store's numbers for the page (server/views/unified.js); a legacy row with no store row
++ // still shows its view_count, and the store is where the new API's views and engagement land.
++ record.counts = await require('../views/unified').countsFor({ service: 'media', kind, id: String(record.id) });
res.set('Cache-Control', cache.htmlHeaders(visibility === 'public' ? { maxAge: 60, swr: 0 } : { private: true }));
return res.type('html').send(pages.renderWatchPage(kind, record, await require('../objects/readiness').forRow(record)));
}
ddiff --git a/server/views/api.js b/server/views/api.js
nnew file mode 100644
iindex 0000000..18850e6
---- /dev/null
++++ b/server/views/api.js
@@@ -0,0 +1,88 @@
++'use strict';
++
++/**
++ * OpenVibe.Media — the public view API (mounted at /api/v1/views, before the tenant routes).
++ *
++ * POST /api/v1/views { events: [view, …] } (or one view object) up to 20
++ * view = { ref: { service, kind, id, site?, creator? }, how: preview|watch|live, seconds,
++ * of_seconds?, session? }
++ * → { results: [{ accepted, counted, engaged, reason? }] }
++ * GET /api/v1/views/counts?service=&kind=&id= → { ref, counts }
++ * ?service=&kind=&ids=1,2,3 → { counts: { "service:kind:id": counts } }
++ * ?refs=<JSON [{service,kind,id}]> → the same, at most 50 refs
++ * GET /api/v1/views/aggregate?scope=item|creator|site&id=&days=&service=&kind=
++ * → { scope, id, days, from, to, rows, totals } for Report
++ *
++ * This is a browser beacon: CORS * and no cookie, no IP or user agent stored. It answers 200 for an event it did not
++ * count (a bot, a repeated view inside the window, a rate limit) and never says whether an item exists.
++ */
++const express = require('express');
++const views = require('./unified');
++
++const router = express.Router();
++
++function cors(req, res) {
++ res.set({
++ 'Access-Control-Allow-Origin': '*',
++ 'Access-Control-Allow-Headers': 'Authorization, Content-Type',
++ 'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
++ 'Access-Control-Max-Age': '600',
++ 'Cross-Origin-Resource-Policy': 'cross-origin',
++ Vary: 'Origin',
++ });
++}
++
++router.use((req, res, next) => { cors(req, res); if (req.method === 'OPTIONS') return res.status(204).end(); return next(); });
++
++router.post('/', async (req, res) => {
++ res.set('Cache-Control', 'no-store');
++ const body = req.body && typeof req.body === 'object' ? req.body : {};
++ const list = Array.isArray(body) ? body : Array.isArray(body.events) ? body.events : body.ref ? [body] : null;
++ if (!list || !list.length) return res.status(400).json({ error: 'Send { events: [ … ] } with at least one view', code: 'views.batch_size' });
++ if (list.length > views.MAX_BATCH) return res.status(400).json({ error: `At most ${views.MAX_BATCH} views in one call`, code: 'views.batch_size' });
++ const out = await views.record(list, { req });
++ if (out.error) return res.status(400).json({ error: 'At most 20 views in one call', code: out.error });
++ return res.json(out);
++});
++
++router.get('/counts', async (req, res) => {
++ res.set('Cache-Control', 'public, max-age=30');
++ const refsParam = String(req.query.refs || '');
++ if (refsParam) {
++ let refs;
++ try { refs = JSON.parse(refsParam); } catch { return res.status(400).json({ error: 'refs must be JSON', code: 'views.refs' }); }
++ if (!Array.isArray(refs) || refs.length > 50) return res.status(400).json({ error: 'refs is an array of at most 50 refs', code: 'views.refs' });
++ const valid = [];
++ for (const raw of refs) { const r = views.normalizeRef(raw); if (r.ok) valid.push(r.ref); }
++ return res.json({ counts: await views.countsMany(valid) });
++ }
++ const service = String(req.query.service || '');
++ const kind = String(req.query.kind || '');
++ const ids = String(req.query.ids || '').split(',').map((s) => s.trim()).filter(Boolean).slice(0, 200);
++ if (!service || !kind) return res.status(400).json({ error: 'service and kind are required', code: 'views.ref' });
++ if (req.query.id != null || (!ids.length && req.query.id != null)) {
++ const r = views.normalizeRef({ service, kind, id: req.query.id });
++ if (!r.ok) return res.status(400).json({ error: 'a valid ref is required', code: r.error });
++ return res.json({ ref: r.ref, counts: await views.countsFor(r.ref) });
++ }
++ if (!ids.length) return res.status(400).json({ error: 'id or ids is required', code: 'views.ref' });
++ const refs = ids.map((id) => ({ service, kind, id })).filter((r) => views.normalizeRef(r).ok);
++ return res.json({ counts: await views.countsMany(refs) });
++});
++
++router.get('/aggregate', async (req, res) => {
++ res.set('Cache-Control', 'public, max-age=60');
++ const scope = String(req.query.scope || 'item');
++ const out = await views.aggregate({
++ scope,
++ id: req.query.id != null ? String(req.query.id) : null,
++ days: req.query.days,
++ service: req.query.service != null ? String(req.query.service) : null,
++ kind: req.query.kind != null ? String(req.query.kind) : null,
++ limit: req.query.limit,
++ });
++ if (out.error) return res.status(400).json({ error: 'scope is item, creator or site', code: out.error });
++ return res.json(out);
++});
++
++module.exports = router;
ddiff --git a/server/views/service.js b/server/views/service.js
iindex 3ffab82..35d003f 100644
---- a/server/views/service.js
++++ b/server/views/service.js
@@@ -100,12 +100,14 @@ async function recordView(type, id, o = {}) {
if (!row) {
await db.run('INSERT INTO content_visits (content_type, content_id, visitor) VALUES (?, ?, ?) ON CONFLICT DO NOTHING', [type, id, visitor]);
await db.run(`UPDATE ${t.table} SET ${t.raw} = COALESCE(${t.raw},0) + 1, unique_views = COALESCE(unique_views,0) + 1 WHERE id = ?`, [id]);
++ await require('./unified').addLegacyView(type, id);
return { counted: true, unique: true, ...await counts(type, id) };
}
const lastMs = new Date(String(row.last_at).replace(' ', 'T') + 'Z').getTime();
if (cd > 0 && Date.now() - lastMs < cd * 1000) return { counted: false, unique: false, reason: 'cooldown', ...await counts(type, id) };
await db.run('UPDATE content_visits SET last_at = ov_now(), visits = visits + 1 WHERE content_type = ? AND content_id = ? AND visitor = ?', [type, id, visitor]);
await db.run(`UPDATE ${t.table} SET ${t.raw} = COALESCE(${t.raw},0) + 1 WHERE id = ?`, [id]);
++ await require('./unified').addLegacyView(type, id);
return { counted: true, unique: false, ...await counts(type, id) };
} catch (e) {
console.warn('[Views] record failed:', e.message);
ddiff --git a/server/views/unified.js b/server/views/unified.js
nnew file mode 100644
iindex 0000000..87bb5e0
---- /dev/null
++++ b/server/views/unified.js
@@@ -0,0 +1,421 @@
++'use strict';
++
++/**
++ * OpenVibe.Media — the one view store for every content kind.
++ *
++ * POST /api/v1/views { events: [{ ref: { service, kind, id, site?, creator? }, how, seconds,
++ * of_seconds?, session? }, …] } (up to 20)
++ * GET /api/v1/views/counts?service=&kind=&id= → { ref, counts }
++ * or ?refs=<JSON [{service,kind,id}]> → { counts: { "service:kind:id": counts } }
++ * GET /api/v1/views/aggregate?scope=item|creator|site&id=&days=&service=&kind=
++ *
++ * A view is one per person (or anonymous session) per item per 30 minutes. A preview counts from one second;
++ * a watch or live heartbeat counts when playback started (seconds >= 1). Engagement is reached at 30 seconds or a
++ * third of the item's length (whichever comes first) for a watch, or 60 seconds of a live stream — once per person
++ * and item, ever. `seconds` is cumulative for the session and item: a retry, a seek or a heartbeat sends the same or
++ * a larger number and never counts twice.
++ *
++ * Privacy: the item's counters keep no identifier. The actor table keeps a keyed hash of the person's canonical
++ * subject or the anonymous session id; an IP address is used only for the rate limit and a user agent only to
++ * refuse a declared crawler. Neither is stored, logged or put in an event.
++ *
++ * Events: `media.viewed` for each counted view and `media.engaged` for the first engagement, enqueued in the same
++ * transaction through OpenVibe.Events (Contracts media.viewed@1 / media.engaged@1).
++ */
++const crypto = require('crypto');
++const db = require('../db/database');
++const events = require('../events');
++const legacy = require('./service');
++const { clientIp } = require('../client-ip');
++
++const WINDOW_MS = 30 * 60_000; // one view per person/session per item
++const LIVE_WINDOW_MS = 90_000; // a live viewer is "watching now" for this long
++const MAX_BATCH = 20;
++const MAX_SECONDS = 86_400;
++const RATE_PER_MIN = 180; // events per actor
++const IP_RATE_PER_MIN = 600; // events per address (protection, not a metric)
++const BOT_UA = /bot\b|crawl|spider|slurp|facebookexternalhit|preview|discordbot|twitterbot|whatsapp|telegram|curl\/|wget\/|python-requests|go-http-client|headless|phantom|selenium/i;
++const SERVICE_RE = /^[a-z][a-z0-9-]{1,39}$/;
++const KIND_RE = /^[a-z][a-z0-9_]{1,39}$/;
++const SESSION_RE = /^[A-Za-z0-9._~-]{8,128}$/;
++const MEDIA_KINDS = new Set(['vod', 'clip', 'recording']);
++
++const iso = (ms) => new Date(ms).toISOString();
++const dayOf = (ms) => new Date(ms).toISOString().slice(0, 10);
++const n = (v) => (Number.isFinite(Number(v)) ? Number(v) : 0);
++
++/** `service:kind:id` — one key across the store, the API and a feed's counts lookup. */
++function keyOf(ref) { return `${ref.service}:${ref.kind}:${ref.id}`; }
++
++/** Validate a ref. Returns { ok, ref, error }. */
++function normalizeRef(input) {
++ if (!input || typeof input !== 'object' || Array.isArray(input)) return { ok: false, error: 'ref.invalid' };
++ const service = String(input.service || '');
++ const kind = String(input.kind || '');
++ const id = input.id == null ? '' : String(input.id);
++ if (!SERVICE_RE.test(service)) return { ok: false, error: 'ref.service' };
++ if (!KIND_RE.test(kind)) return { ok: false, error: 'ref.kind' };
++ if (!id || id.length > 200) return { ok: false, error: 'ref.id' };
++ const ref = { service, kind, id };
++ if (input.site != null) {
++ if (!/^[a-z][a-z0-9-]{0,39}$/.test(String(input.site))) return { ok: false, error: 'ref.site' };
++ ref.site = String(input.site);
++ }
++ if (input.creator != null) {
++ if (String(input.creator).length > 200) return { ok: false, error: 'ref.creator' };
++ ref.creator = String(input.creator);
++ }
++ return { ok: true, ref };
++}
++
++/** Validate one event. Returns { ok, event, error }. */
++function normalizeEvent(input) {
++ if (!input || typeof input !== 'object' || Array.isArray(input)) return { ok: false, error: 'event.invalid' };
++ const r = normalizeRef(input.ref);
++ if (!r.ok) return r;
++ const how = String(input.how || '');
++ if (!['preview', 'watch', 'live'].includes(how)) return { ok: false, error: 'how' };
++ const seconds = Number(input.seconds);
++ if (!Number.isFinite(seconds) || seconds < 0 || seconds > MAX_SECONDS) return { ok: false, error: 'seconds' };
++ const event = { ref: r.ref, how, seconds };
++ if (input.of_seconds != null) {
++ const of = Number(input.of_seconds);
++ if (!Number.isFinite(of) || of <= 0 || of > MAX_SECONDS) return { ok: false, error: 'of_seconds' };
++ event.of_seconds = of;
++ }
++ if (input.session != null) {
++ if (!SESSION_RE.test(String(input.session))) return { ok: false, error: 'session' };
++ event.session = String(input.session);
++ }
++ return { ok: true, event };
++}
++
++/** The seconds that make a watch engaged. */
++function thresholdFor(how, ofSeconds) {
++ if (how === 'live') return 60;
++ if (Number.isFinite(ofSeconds) && ofSeconds > 0) return Math.max(1, Math.min(30, ofSeconds / 3));
++ return 30;
++}
++
++// A keyed actor id: no raw anonymous session id and no raw canonical subject is stored with a view.
++const _hashes = new Map();
++async function actorHash(kind, value) {
++ const cacheKey = `${kind}:${value}`;
++ const kept = _hashes.get(cacheKey);
++ if (kept) return kept;
++ const secret = await legacy.hashSecret();
++ const hash = crypto.createHmac('sha256', String(secret || '')).update(cacheKey).digest('hex').slice(0, 32);
++ const out = `${kind === 'person' ? 'p' : 's'}:${hash}`;
++ if (_hashes.size > 5000) _hashes.clear();
++ _hashes.set(cacheKey, out);
++ return out;
++}
++
++/** The actor for a request: the signed-in person's Network subject, else their anonymous session id. */
++async function actorFor(req, session) {
++ const auth = String(req.get('authorization') || '');
++ if (/^Bearer [A-Za-z0-9._~+/=-]{10,4096}$/.test(auth)) {
++ const subject = await require('../auth').viewerSubject(auth.slice(7));
++ if (subject) return { actor: await actorHash('person', subject), person: true };
++ }
++ if (session) return { actor: await actorHash('session', session), person: false };
++ return null;
++}
++
++// Per-actor and per-address event budgets in memory (a sliding minute). They protect the database; the counts
++// themselves are deduped by the window.
++const _hits = new Map();
++function overBudget(map, key, limit, now) {
++ const cut = now - 60_000;
++ const arr = (map.get(key) || []).filter((t) => t > cut);
++ if (arr.length >= limit) { map.set(key, arr); return true; }
++ arr.push(now);
++ map.set(key, arr);
++ if (map.size > 5000) for (const [k, v] of map) if (!v.some((t) => t > cut)) map.delete(k);
++ return false;
++}
++
++async function creatorOf(ref) {
++ if (ref.creator) return ref.creator;
++ if (ref.service !== 'media' || !MEDIA_KINDS.has(ref.kind)) return null;
++ const table = ref.kind === 'clip' ? 'clips' : 'vods';
++ try {
++ const row = await db.get(`SELECT o.owner_subject AS creator FROM ${table} t JOIN media_objects o ON o.id = t.object_id WHERE t.id = ?`, [ref.id]);
++ return row && row.creator ? String(row.creator) : null;
++ } catch { return null; }
++}
++
++async function bumpCounts(ref, { counted, engaged, preview, livePeak, creator }) {
++ const site = ref.site || ref.service;
++ await db.run(`INSERT INTO content_view_counts (service, kind, item_id, site, creator_subject, views, engaged, previews, live_peak)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT(service, kind, item_id) DO UPDATE SET
++ views = content_view_counts.views + excluded.views,
++ engaged = content_view_counts.engaged + excluded.engaged,
++ previews = content_view_counts.previews + excluded.previews,
++ site = COALESCE(content_view_counts.site, excluded.site),
++ creator_subject = COALESCE(content_view_counts.creator_subject, excluded.creator_subject),
++ live_peak = CASE WHEN excluded.live_peak IS NULL THEN content_view_counts.live_peak
++ ELSE GREATEST(COALESCE(content_view_counts.live_peak, 0), excluded.live_peak) END,
++ updated_at = ov_now()`,
++ [ref.service, ref.kind, ref.id, site, creator, counted ? 1 : 0, engaged ? 1 : 0, counted && preview ? 1 : 0, livePeak == null ? null : livePeak]);
++}
++
++async function bumpDay(ref, { counted, engaged, preview, livePeak, creator }, ms) {
++ if (!counted && !engaged && livePeak == null) return;
++ const site = ref.site || ref.service;
++ await db.run(`INSERT INTO content_view_days (service, kind, item_id, day, site, creator_subject, views, engaged, previews, live_peak)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT(service, kind, item_id, day) DO UPDATE SET
++ views = content_view_days.views + excluded.views,
++ engaged = content_view_days.engaged + excluded.engaged,
++ previews = content_view_days.previews + excluded.previews,
++ site = COALESCE(content_view_days.site, excluded.site),
++ creator_subject = COALESCE(content_view_days.creator_subject, excluded.creator_subject),
++ live_peak = CASE WHEN excluded.live_peak IS NULL THEN content_view_days.live_peak
++ ELSE GREATEST(COALESCE(content_view_days.live_peak, 0), excluded.live_peak) END`,
++ [ref.service, ref.kind, ref.id, dayOf(ms), site, creator, counted ? 1 : 0, engaged ? 1 : 0,
++ counted && preview ? 1 : 0, livePeak == null ? null : livePeak]);
++}
++
++/**
++ * Record one normalized event for one actor. Runs in its own transaction so the counters and the event are one
++ * change. → { counted, engaged, reason? }
++ */
++async function recordOne(event, actor, now = Date.now()) {
++ const ref = event.ref;
++ const how = event.how;
++ const seconds = event.seconds;
++ const preview = how === 'preview';
++ const creator = await creatorOf(ref);
++ const day = dayOf(now);
++ return await db.getDb().tx(async () => {
++ const prior = await db.get(
++ 'SELECT last_view_ms, watched_seconds, engaged_ms, live FROM content_view_actors WHERE service = ? AND kind = ? AND item_id = ? AND actor = ?',
++ [ref.service, ref.kind, ref.id, actor]);
++ const counted = seconds >= 1 && !(prior && prior.last_view_ms != null && now - Number(prior.last_view_ms) < WINDOW_MS);
++ let engaged = false;
++ const threshold = thresholdFor(how, event.of_seconds);
++ const watched = Math.max(Number(prior && prior.watched_seconds) || 0, seconds);
++ if (how !== 'preview' && seconds >= 1 && !(prior && prior.engaged_ms != null) && watched >= threshold) engaged = true;
++
++ if (!counted && !engaged) {
++ // Still remember that a live viewer is watching now (peak concurrency), and the cumulative seconds.
++ if (how === 'live') {
++ await db.run(`INSERT INTO content_view_actors (service, kind, item_id, actor, first_ms, last_view_ms, last_seen_ms, watched_seconds, live)
++ VALUES (?, ?, ?, ?, ?, NULL, ?, ?, true)
++ ON CONFLICT(service, kind, item_id, actor) DO UPDATE SET last_seen_ms = excluded.last_seen_ms,
++ watched_seconds = GREATEST(content_view_actors.watched_seconds, excluded.watched_seconds), live = true`,
++ [ref.service, ref.kind, ref.id, actor, now, now, watched]);
++ } else {
++ await db.run(`INSERT INTO content_view_actors (service, kind, item_id, actor, first_ms, last_seen_ms, watched_seconds)
++ VALUES (?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT(service, kind, item_id, actor) DO UPDATE SET last_seen_ms = excluded.last_seen_ms,
++ watched_seconds = GREATEST(content_view_actors.watched_seconds, excluded.watched_seconds)`,
++ [ref.service, ref.kind, ref.id, actor, now, now, watched]);
++ }
++ return { counted: false, engaged: false, reason: seconds < 1 ? 'too-short' : 'window' };
++ }
++
++ let livePeak = null;
++ await db.run(`INSERT INTO content_view_actors (service, kind, item_id, actor, first_ms, last_view_ms, last_seen_ms, watched_seconds, engaged_ms, live)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT(service, kind, item_id, actor) DO UPDATE SET
++ last_view_ms = CASE WHEN ? THEN excluded.last_view_ms ELSE content_view_actors.last_view_ms END,
++ last_seen_ms = excluded.last_seen_ms,
++ watched_seconds = GREATEST(content_view_actors.watched_seconds, excluded.watched_seconds),
++ engaged_ms = COALESCE(content_view_actors.engaged_ms, excluded.engaged_ms),
++ live = excluded.live`,
++ [ref.service, ref.kind, ref.id, actor, now, counted ? now : null, now, watched, engaged ? now : null, how === 'live' ? true : false, counted ? 1 : 0]);
++
++ if (how === 'live') {
++ const seen = await db.get(
++ 'SELECT COUNT(*) AS n FROM content_view_actors WHERE service = ? AND kind = ? AND item_id = ? AND live = true AND last_seen_ms >= ?',
++ [ref.service, ref.kind, ref.id, now - LIVE_WINDOW_MS]);
++ livePeak = Math.max(1, n(seen && seen.n));
++ }
++
++ await bumpCounts(ref, { counted, engaged, preview, livePeak, creator });
++ await bumpDay(ref, { counted, engaged, preview, livePeak, creator }, now);
++
++ const subject = { type: ref.kind, id: ref.id };
++ if (counted) {
++ await events.recordView('media.viewed', subject, {
++ ref: { ...ref, site: ref.site || ref.service, creator },
++ how,
++ occurred_at: iso(now),
++ seconds,
++ of_seconds: event.of_seconds || null,
++ concurrent: livePeak,
++ });
++ }
++ if (engaged) {
++ await events.recordView('media.engaged', subject, {
++ ref: { ...ref, site: ref.site || ref.service, creator },
++ how,
++ occurred_at: iso(now),
++ seconds: watched,
++ threshold,
++ of_seconds: event.of_seconds || null,
++ concurrent: livePeak,
++ });
++ }
++ return { counted, engaged, reason: counted || engaged ? undefined : 'window' };
++ });
++}
++
++/**
++ * Record a batch of raw events. → { results: [{ accepted, counted, engaged, reason? }], accepted, counted }
++ * Invalid events are refused one by one; a caller never learns whether an item exists.
++ */
++async function record(rawEvents, { req, now = Date.now() } = {}) {
++ const list = Array.isArray(rawEvents) ? rawEvents : [rawEvents];
++ if (!list.length || list.length > MAX_BATCH) return { error: 'views.batch_size' };
++ const ua = String((req && req.get('user-agent')) || '');
++ const ip = req ? clientIp(req) : 'unknown';
++ if (BOT_UA.test(ua)) return { results: list.map((raw) => ({ accepted: false, counted: false, engaged: false, reason: 'bot' })) };
++ if (overBudget(_hits, `ip:${ip}`, IP_RATE_PER_MIN, now)) return { results: list.map(() => ({ accepted: false, counted: false, engaged: false, reason: 'rate-limited' })) };
++
++ const results = [];
++ for (const raw of list) {
++ const v = normalizeEvent(raw);
++ if (!v.ok) { results.push({ accepted: false, counted: false, engaged: false, reason: v.error }); continue; }
++ const who = await actorFor(req, v.event.session);
++ if (!who) { results.push({ accepted: false, counted: false, engaged: false, reason: 'session' }); continue; }
++ if (overBudget(_hits, `actor:${who.actor}`, RATE_PER_MIN, now)) {
++ results.push({ accepted: false, counted: false, engaged: false, reason: 'rate-limited' });
++ continue;
++ }
++ try {
++ const out = await recordOne(v.event, who.actor, now);
++ results.push({ accepted: true, ...out });
++ } catch (err) {
++ console.warn('[Views] record failed:', err.message);
++ results.push({ accepted: true, counted: false, engaged: false, reason: 'error' });
++ }
++ }
++ return { results };
++}
++
++// ── Counts and aggregates ────────────────────────────────────
++
++function countsOfRow(row) {
++ if (!row) return { views: 0, engaged: 0, previews: 0, preview_share: null, peak_concurrent: null };
++ const views = n(row.views);
++ return {
++ views,
++ engaged: n(row.engaged),
++ previews: n(row.previews),
++ preview_share: views > 0 ? Math.round((n(row.previews) / views) * 1000) / 1000 : null,
++ peak_concurrent: row.live_peak == null ? null : n(row.live_peak),
++ };
++}
++
++async function countsFor(ref) {
++ const row = await db.get(
++ 'SELECT views, engaged, previews, live_peak FROM content_view_counts WHERE service = ? AND kind = ? AND item_id = ?',
++ [ref.service, ref.kind, ref.id]);
++ return countsOfRow(row);
++}
++
++async function countsMany(refs) {
++ const out = {};
++ const groups = new Map();
++ for (const ref of refs) {
++ const k = `${ref.service}\u0000${ref.kind}`;
++ if (!groups.has(k)) groups.set(k, []);
++ groups.get(k).push(ref);
++ }
++ for (const [k, list] of groups) {
++ const [service, kind] = k.split('\u0000');
++ const ids = [...new Set(list.map((r) => r.id))];
++ for (let i = 0; i < ids.length; i += 200) {
++ const chunk = ids.slice(i, i + 200);
++ const rows = await db.all(
++ `SELECT item_id, views, engaged, previews, live_peak FROM content_view_counts
++ WHERE service = ? AND kind = ? AND item_id IN (${chunk.map(() => '?').join(',')})`,
++ [service, kind, ...chunk]);
++ const found = new Map(rows.map((r) => [String(r.item_id), r]));
++ for (const id of chunk) out[`${service}:${kind}:${id}`] = countsOfRow(found.get(String(id)));
++ }
++ }
++ return out;
++}
++
++/**
++ * Report's read: per item, per creator or per site, over a window. → { scope, id, days, from, to, rows, totals }
++ * A row is { id, service?, kind?, item_id?, site, creator, views, engaged, previews, preview_share, peak_concurrent }.
++ */
++async function aggregate({ scope = 'item', id = null, days = 30, service = null, kind = null, limit = 100 } = {}) {
++ if (!['item', 'creator', 'site'].includes(scope)) return { error: 'views.scope' };
++ const d = Math.max(1, Math.min(365, parseInt(days, 10) || 30));
++ const to = dayOf(Date.now());
++ const from = dayOf(Date.now() - (d - 1) * 86_400_000);
++ const where = ['day >= ?', 'day <= ?'];
++ const params = [from, to];
++ if (scope === 'creator') { where.push('creator_subject = ?'); params.push(String(id || '')); }
++ if (scope === 'site') { where.push('site = ?'); params.push(String(id || '')); }
++ if (scope === 'item') {
++ where.push('service = ?', 'kind = ?', 'item_id = ?');
++ params.push(String(service || ''), String(kind || ''), String(id || ''));
++ }
++ const group = scope === 'item' ? 'service, kind, item_id' : scope === 'creator' ? 'creator_subject' : 'site';
++ const select = scope === 'item'
++ ? 'service, kind, item_id, MAX(site) AS site, MAX(creator_subject) AS creator'
++ : scope === 'creator' ? 'creator_subject, MAX(site) AS site, MAX(creator_subject) AS creator' : 'site, MAX(creator_subject) AS creator';
++ const rows = await db.all(
++ `SELECT ${select},
++ SUM(views)::bigint AS views, SUM(engaged)::bigint AS engaged, SUM(previews)::bigint AS previews,
++ MAX(live_peak) AS live_peak
++ FROM content_view_days WHERE ${where.join(' AND ')}
++ GROUP BY ${group} ORDER BY views DESC, ${group} ASC LIMIT ?`,
++ [...params, Math.max(1, Math.min(500, parseInt(limit, 10) || 100))]);
++ const out = rows.map((r) => {
++ const views = n(r.views);
++ const base = {
++ id: scope === 'item' ? `${r.service}:${r.kind}:${r.item_id}` : scope === 'creator' ? (r.creator_subject || null) : r.site,
++ site: r.site || null,
++ creator: r.creator || null,
++ views,
++ engaged: n(r.engaged),
++ previews: n(r.previews),
++ preview_share: views > 0 ? Math.round((n(r.previews) / views) * 1000) / 1000 : null,
++ peak_concurrent: r.live_peak == null ? null : n(r.live_peak),
++ };
++ if (scope === 'item') { base.service = r.service; base.kind = r.kind; base.item_id = r.item_id; }
++ return base;
++ });
++ const totals = {
++ views: out.reduce((a, r) => a + r.views, 0),
++ engaged: out.reduce((a, r) => a + r.engaged, 0),
++ previews: out.reduce((a, r) => a + r.previews, 0),
++ peak_concurrent: out.reduce((a, r) => (r.peak_concurrent != null ? Math.max(a, r.peak_concurrent) : a), 0) || null,
++ };
++ totals.preview_share = totals.views > 0 ? Math.round((totals.previews / totals.views) * 1000) / 1000 : null;
++ return { scope, id: id || null, days: d, from, to, rows: out, totals };
++}
++
++/** The legacy /v /c visit counter adds to the same total, so a page's number never splits in two. */
++async function addLegacyView(type, id) {
++ const kind = type === 'clip' ? 'clip' : type === 'file' ? 'file' : 'vod';
++ const ref = { service: 'media', kind, id: String(id) };
++ const now = Date.now();
++ try {
++ const creator = await creatorOf(ref);
++ await bumpCounts(ref, { counted: true, engaged: false, preview: false, livePeak: null, creator });
++ await bumpDay(ref, { counted: true, engaged: false, preview: false, livePeak: null, creator }, now);
++ } catch { /* a page view that is not stored is not an error */ }
++}
++
++/** Forget the 30-minute windows (kept small; the totals and the engagement stay). */
++async function prune(days = 30) {
++ try {
++ return (await db.run('DELETE FROM content_view_actors WHERE first_ms < ?', [Date.now() - days * 86_400_000])).changes;
++ } catch { return 0; }
++}
++
++module.exports = {
++ record, normalizeEvent, normalizeRef, countsFor, countsMany, countsOfRow, aggregate, addLegacyView, prune,
++ thresholdFor, keyOf, MAX_BATCH, WINDOW_MS, LIVE_WINDOW_MS,
++ _internals: { actorHash, overBudget, _hits },
++};
ddiff --git a/server/vod/clips-routes.js b/server/vod/clips-routes.js
iindex 9084d89..7d9c950 100644
---- a/server/vod/clips-routes.js
++++ b/server/vod/clips-routes.js
@@@ -67,7 +67,7 @@ const clipUpload = multer({
// the clip.ready / clip.failed payloads leave them out, as vodPublic's do. `hls` selects the playlist
// fields: undefined = discover them for this row (a single-item read), false = omit them (an event payload
// must never carry a signed URL into the outbox), an object = the answer a list route already batched.
--async function clipPublic(clip, { readiness = false, hls } = {}) {
++async function clipPublic(clip, { readiness = false, hls, counts = false } = {}) {
if (!clip) return null;
const out = {
unique_views: clip.unique_views || 0,
@@@ -123,6 +123,8 @@ async function clipPublic(clip, { readiness = false, hls } = {}) {
if (hls === undefined) Object.assign(out, await require('../objects/hls').discovery(clip.object_id) || {});
else if (hls) Object.assign(out, hls);
if (readiness) out.readiness = await require('../objects/readiness').forRow(clip);
++ // The one view store's numbers (server/views/unified.js); see vodPublic's note on why this is opt-in.
++ if (counts) out.counts = await require('../views/unified').countsFor({ service: 'media', kind: 'clip', id: String(clip.id) });
return out;
}
@@@ -471,7 +473,11 @@ router.get('/', tenantAuth({ verb: 'list', allowUser: true }), async (req, res)
const clips = await db.listClips(req.appId, filters);
const total = await db.countClips(req.appId, filters);
const hls = await require('../objects/hls').discoveryMany(clips);
-- res.json({ clips: await Promise.all(clips.map(async c => await clipPublic(c, { readiness: true, hls: hls.get(String(c.object_id)) || null }))), total, limit, offset, hasMore: offset + clips.length < total });
++ const outClips = await Promise.all(clips.map(async c => await clipPublic(c, { readiness: true, hls: hls.get(String(c.object_id)) || null })));
++ // One batched read of the view store for the page the app renders (never one query a card).
++ const viewCounts = await require('../views/unified').countsMany(outClips.map((c) => ({ service: 'media', kind: 'clip', id: String(c.id) })));
++ for (const c of outClips) c.counts = viewCounts[`media:clip:${c.id}`];
++ res.json({ clips: outClips, total, limit, offset, hasMore: offset + clips.length < total });
} catch (err) {
console.error('[Clips] List error:', err.message);
res.status(500).json({ error: 'Failed to list clips' });
@@@ -507,7 +513,7 @@ router.get('/:id', tenantAuth({ verb: 'read', allowUser: true }), async (req, re
}
// Bare object, mirroring GET /vods/:id per CONTRACTS.md.
// Detail responses carry the transcript too (lists stay light).
-- res.json({ ...await clipPublic(clip, { readiness: true }), ai_transcript: clip.ai_transcript || null });
++ res.json({ ...await clipPublic(clip, { readiness: true, counts: true }), ai_transcript: clip.ai_transcript || null });
} catch (err) {
res.status(500).json({ error: 'Failed to get clip' });
}
@@@ -661,7 +667,7 @@ router.put('/:id', tenantAuth({ verb: 'write' }), async (req, res) => {
// Made public (or already public): its watch page and the sitemap are indexable now.
await require('../indexnow-notify').pingWatch('clip', await db.getClipById(clip.id));
}
-- res.json({ clip: await clipPublic(await db.getClipById(clip.id, req.appId), { readiness: true }) });
++ res.json({ clip: await clipPublic(await db.getClipById(clip.id, req.appId), { readiness: true, counts: true }) });
} catch (err) {
res.status(500).json({ error: 'Failed to update clip' });
}
ddiff --git a/server/vod/finalize.js b/server/vod/finalize.js
iindex c34e75c..0cf231d 100644
---- a/server/vod/finalize.js
++++ b/server/vod/finalize.js
@@@ -41,7 +41,7 @@ function isFinalizing(vodId) {
// `hls` selects the playlist fields: undefined = discover
// them for this row (a single-item read), false = omit them (an event payload must never carry a signed
// URL into the outbox), an object = the answer the list route already batched (objects/hls.discoveryMany).
--async function vodPublic(vod, { readiness = false, hls } = {}) {
++async function vodPublic(vod, { readiness = false, hls, counts = false } = {}) {
if (!vod) return null;
const out = {
unique_views: vod.unique_views || 0,
@@@ -102,6 +102,9 @@ async function vodPublic(vod, { readiness = false, hls } = {}) {
if (hls === undefined) Object.assign(out, await require('../objects/hls').discovery(vod.object_id) || {});
else if (hls) Object.assign(out, hls);
if (readiness) out.readiness = await require('../objects/readiness').forRow(vod);
++ // The one view store's numbers (server/views/unified.js): views, engaged viewers, preview share and live peak.
++ // Off by default: an event payload (vod.ready) must not carry a number that is not its state change.
++ if (counts) out.counts = await require('../views/unified').countsFor({ service: 'media', kind: 'vod', id: String(vod.id) });
return out;
}
ddiff --git a/server/vod/recordings.js b/server/vod/recordings.js
iindex 39a1379..af36319 100644
---- a/server/vod/recordings.js
++++ b/server/vod/recordings.js
@@@ -248,14 +248,18 @@ const fail = (res, err) => {
router.get('/object/:id', tenantAuth({ verb: 'read' }), async (req, res) => {
try {
const rec = await get(req.appId, req.params.id);
-- return rec ? res.json(rec) : res.status(404).json({ error: 'Recording not found', code: 'media.recording.not_found' });
++ if (!rec) return res.status(404).json({ error: 'Recording not found', code: 'media.recording.not_found' });
++ rec.counts = rec.vod_id ? await require('../views/unified').countsFor({ service: 'media', kind: 'vod', id: String(rec.vod_id) }) : null;
++ return res.json(rec);
} catch (err) { return fail(res, err); }
});
router.get('/:key', tenantAuth({ verb: 'read' }), async (req, res) => {
try {
const rec = await byKey(req.appId, req.params.key);
-- return rec ? res.json(rec) : res.status(404).json({ error: 'Recording not found', code: 'media.recording.not_found' });
++ if (!rec) return res.status(404).json({ error: 'Recording not found', code: 'media.recording.not_found' });
++ rec.counts = rec.vod_id ? await require('../views/unified').countsFor({ service: 'media', kind: 'vod', id: String(rec.vod_id) }) : null;
++ return res.json(rec);
} catch (err) { return fail(res, err); }
});
ddiff --git a/server/vod/routes.js b/server/vod/routes.js
iindex 40e0b3a..9ae6fd0 100644
---- a/server/vod/routes.js
++++ b/server/vod/routes.js
@@@ -274,7 +274,7 @@ async function _finalizeHandler(req, res) {
segmentPath: chunkRec && chunkRec.currentSegmentPath !== chunkRec.filePath ? chunkRec.currentSegmentPath : null,
});
if (!result) return res.status(409).json({ error: 'Nothing to finalize (finalization already in progress or VOD discarded)' });
-- res.json({ vod: await vodPublic(result, { readiness: true }) });
++ res.json({ vod: await vodPublic(result, { readiness: true, counts: true }) });
} catch (err) {
console.error('[VOD] Finalize error:', err.message);
res.status(500).json({ error: 'Failed to finalize VOD' });
@@@ -315,8 +315,12 @@ router.get('/', tenantAuth({ verb: 'list', allowUser: true }), async (req, res)
const recs = await require('./recordings').summaries(vods.map((v) => v.id));
const sharing = sharingOf(req);
const audiences = sharing ? await sharing.audiencesOf('vod', vods.map((v) => v.id)) : new Map();
-- res.json({ vods: (await Promise.all(vods.map(async v => ({ ...await vodPublic({ ...v, audience: audiences.get(Number(v.id)) }, { readiness: true, hls: hls.get(String(v.object_id)) || null }),
-- ...(recs.has(Number(v.id)) && { recording: recs.get(Number(v.id)) }) })))), total, limit, offset, hasMore: offset + vods.length < total });
++ const outVods = (await Promise.all(vods.map(async v => ({ ...await vodPublic({ ...v, audience: audiences.get(Number(v.id)) }, { readiness: true, hls: hls.get(String(v.object_id)) || null }),
++ ...(recs.has(Number(v.id)) && { recording: recs.get(Number(v.id)) }) }))));
++ // One batched read of the view store for the page the app renders (never one query a card).
++ const viewCounts = await require('../views/unified').countsMany(outVods.map((v) => ({ service: 'media', kind: 'vod', id: String(v.id) })));
++ for (const v of outVods) v.counts = viewCounts[`media:vod:${v.id}`];
++ res.json({ vods: outVods, total, limit, offset, hasMore: offset + vods.length < total });
} catch (err) {
console.error('[VOD] List error:', err.message);
res.status(500).json({ error: 'Failed to list VODs' });
@@@ -380,7 +384,7 @@ router.get('/:id', tenantAuth({ verb: 'read', allowUser: true }), async (req, re
vod.audience = await sharing.audienceOf('vod', vod.id);
if (vod.audience === 'people') vod.share_members = await membersOf('vod', vod.id);
}
-- res.json({ ...await vodPublic(vod, { readiness: true }), ai_transcript: vod.ai_transcript || null,
++ res.json({ ...await vodPublic(vod, { readiness: true, counts: true }), ai_transcript: vod.ai_transcript || null,
...(rec && { recording: { ...rec.recording, this_part: rec.part } }) });
} catch (err) {
console.error('[VOD] Get error:', err.message);
@@@ -419,7 +423,7 @@ router.put('/:id', tenantAuth({ verb: 'write' }), async (req, res) => {
if (visibility !== undefined || title !== undefined) await require('./recordings').syncPart(vod.id, { visibility, title });
// Made public (or already public): its watch page and the sitemap are indexable now.
if (visibility !== undefined) await require('../indexnow-notify').pingWatch('vod', await db.getVodById(vod.id));
-- res.json({ vod: await vodPublic(await db.getVodById(vod.id, req.appId), { readiness: true }) });
++ res.json({ vod: await vodPublic(await db.getVodById(vod.id, req.appId), { readiness: true, counts: true }) });
} catch (err) {
console.error('[VOD] Update error:', err.message);
res.status(500).json({ error: 'Failed to update VOD' });
ddiff --git a/test/views-api.test.js b/test/views-api.test.js
nnew file mode 100644
iindex 0000000..46b13af
---- /dev/null
++++ b/test/views-api.test.js
@@@ -0,0 +1,165 @@
++'use strict';
++// The one view API (server/views/unified.js, POST /api/v1/views):
++// - a preview counts a view from 1 s, once per person/session per item per 30 minutes;
++// - engagement comes at 30 s, a third of the length when known, or 60 s of a live stream;
++// - bots never count, a batch is up to 20 events, and no IP address, user agent or raw session id is stored;
++// - every kind (a Feed item included) gets counts.views/counts.engaged, and Report reads per item/creator/site.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++
++(async () => {
++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-views-'));
++ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
++ process.env.VOD_PATH = dir('vods');
++ process.env.CLIPS_PATH = dir('clips');
++ process.env.FILES_PATH = dir('files');
++ process.env.OBJECTS_PATH = dir('objects');
++ process.env.THUMBNAILS_PATH = dir('thumbnails');
++ process.env.MEDIA_PUBLIC_URL = 'https://media.test';
++ delete process.env.EVENTS_URL;
++ const realLog = console.log;
++
++ const db = require('../server/db/database');
++ await db.initDb();
++ const views = require('../server/views/unified');
++
++ const express = require('express');
++ const app = express();
++ app.use(express.json());
++ app.use('/api/v1/views', require('../server/views/api'));
++ const server = http.createServer(app);
++ await new Promise((r) => server.listen(0, '127.0.0.1', r));
++ const base = `http://127.0.0.1:${server.address().port}/api/v1/views`;
++ const post = async (body, { ua } = {}) => {
++ const res = await fetch(base, { method: 'POST', headers: { 'content-type': 'application/json', ...(ua ? { 'user-agent': ua } : {}) }, body: JSON.stringify(body) });
++ return { status: res.status, json: await res.json().catch(() => null) };
++ };
++ const get = async (q) => {
++ const res = await fetch(`${base}${q}`);
++ return { status: res.status, json: await res.json().catch(() => null) };
++ };
++ const preview = (id, session, extra = {}) => ({ ref: { service: 'media', kind: 'clip', id: String(id), ...extra }, how: 'preview', seconds: 2, session });
++ let failures = 0;
++ const check = async (name, fn) => {
++ try { await fn(); console.log(` ok - ${name}`); }
++ catch (e) { failures++; console.log(` FAIL - ${name}: ${e.message}`); }
++ };
++ const countsOf = async (service, kind, id) => (await get(`/counts?service=${service}&kind=${kind}&id=${id}`)).json.counts;
++
++ await check('a preview counts from 1 s and repeats inside 30 minutes do not', async () => {
++ let r = await post({ events: [preview(101, 'anon_session_101')] });
++ assert.strictEqual(r.status, 200);
++ assert.deepStrictEqual(r.json.results[0].counted, true);
++ r = await post({ events: [preview(101, 'anon_session_101')] });
++ assert.strictEqual(r.json.results[0].counted, false);
++ assert.strictEqual(r.json.results[0].reason, 'window');
++ assert.strictEqual((await countsOf('media', 'clip', '101')).views, 1);
++ });
++
++ await check('0.5 s is a preview that did not count', async () => {
++ const r = await post({ events: [{ ...preview(102, 'anon_session_102'), seconds: 0.5 }] });
++ assert.strictEqual(r.json.results[0].counted, false);
++ assert.strictEqual(r.json.results[0].reason, 'too-short');
++ assert.strictEqual((await countsOf('media', 'clip', '102')).views, 0);
++ });
++
++ await check('the window passes after 30 minutes and the same session counts again', async () => {
++ const raw = db.getDb();
++ await raw.prepare('UPDATE content_view_actors SET last_view_ms = last_view_ms - ? WHERE item_id = ?').run(30 * 60 * 1000 + 1, '101');
++ const r = await post({ events: [preview(101, 'anon_session_101')] });
++ assert.strictEqual(r.json.results[0].counted, true);
++ assert.strictEqual((await countsOf('media', 'clip', '101')).views, 2);
++ });
++
++ await check('engagement: 30 s, a third of the length, once per viewer', async () => {
++ let r = await post({ events: [{ ref: { service: 'media', kind: 'vod', id: '201' }, how: 'watch', seconds: 30, session: 'anon_session_201' }] });
++ assert.strictEqual(r.json.results[0].counted, true);
++ assert.strictEqual(r.json.results[0].engaged, true);
++ r = await post({ events: [{ ref: { service: 'media', kind: 'vod', id: '201' }, how: 'watch', seconds: 31, session: 'anon_session_201' }] });
++ assert.strictEqual(r.json.results[0].engaged, false, 'engaged only once');
++ assert.strictEqual((await countsOf('media', 'vod', '201')).engaged, 1);
++
++ r = await post({ events: [{ ref: { service: 'media', kind: 'vod', id: '202' }, how: 'watch', seconds: 15, of_seconds: 45, session: 'anon_session_202' }] });
++ assert.strictEqual(r.json.results[0].engaged, true, 'a third of 45 s is 15 s');
++ });
++
++ await check('live: 60 s engages and two watchers make a peak of at least 2', async () => {
++ let r = await post({ events: [{ ref: { service: 'live', kind: 'stream', id: '301' }, how: 'live', seconds: 59, session: 'anon_session_301' }] });
++ assert.strictEqual(r.json.results[0].engaged, false);
++ r = await post({ events: [{ ref: { service: 'live', kind: 'stream', id: '301' }, how: 'live', seconds: 60, session: 'anon_session_301' }] });
++ assert.strictEqual(r.json.results[0].engaged, true, 'the same session watches past 60 s');
++ await post({ events: [{ ref: { service: 'live', kind: 'stream', id: '301' }, how: 'live', seconds: 61, session: 'anon_session_302' }] });
++ const c = await countsOf('live', 'stream', '301');
++ assert.strictEqual(c.engaged, 2);
++ assert.ok(c.peak_concurrent >= 2, `peak ${c.peak_concurrent}`);
++ });
++
++ await check('a declared crawler is never counted', async () => {
++ const before = await countsOf('media', 'clip', '101');
++ const r = await post({ events: [preview(101, 'anon_session_bot')] }, { ua: 'Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)' });
++ assert.strictEqual(r.json.results[0].counted, false);
++ assert.strictEqual(r.json.results[0].reason, 'bot');
++ assert.strictEqual((await countsOf('media', 'clip', '101')).views, before.views);
++ });
++
++ await check('a batch records each event; more than 20 is refused', async () => {
++ let r = await post({ events: [preview(401, 'anon_session_401'), preview(402, 'anon_session_402')] });
++ assert.strictEqual(r.json.results.length, 2);
++ assert.ok(r.json.results.every((x) => x.counted));
++ const tooMany = Array.from({ length: 21 }, (_, i) => preview(500 + i, `anon_session_50${i}`));
++ r = await post({ events: tooMany });
++ assert.strictEqual(r.status, 400);
++ assert.strictEqual(r.json.code, 'views.batch_size');
++ });
++
++ await check('every kind gets counts: a Feed item, a creator and a site', async () => {
++ await post({ events: [{
++ ref: { service: 'network', kind: 'feed', id: 'p:abc', site: 'network', creator: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ' },
++ how: 'preview', seconds: 3, session: 'anon_session_feed',
++ }] });
++ const c = await countsOf('network', 'feed', 'p:abc');
++ assert.strictEqual(c.views, 1);
++ assert.strictEqual(c.preview_share, 1);
++ const byCreator = await get('/aggregate?scope=creator&id=usr_01JAB2C3D4E5F6G7H8J9K0MNPQ&days=7');
++ assert.ok(byCreator.json.rows.some((row) => row.views >= 1 && row.creator === 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ'));
++ const bySite = await get('/aggregate?scope=site&id=network&days=7');
++ assert.ok(bySite.json.rows.some((row) => row.views >= 1));
++ });
++
++ await check('the legacy /v /c counter adds to the same total', async () => {
++ const before = (await countsOf('media', 'vod', '999')).views;
++ await views.addLegacyView('vod', 999);
++ assert.strictEqual((await countsOf('media', 'vod', '999')).views, before + 1);
++ });
++
++ await check('a Media VOD page read carries counts.views and counts.engaged', async () => {
++ const raw = db.getDb();
++ const row = await raw.prepare("INSERT INTO vods (app_id, title, view_count) VALUES ('live', 'Counted', 0) RETURNING *").get();
++ await views.addLegacyView('vod', row.id);
++ const out = await require('../server/vod/finalize').vodPublic(row, { counts: true });
++ assert.strictEqual(out.counts.views, 1);
++ assert.strictEqual(out.counts.engaged, 0);
++ });
++
++ await check('privacy: no IP, user agent or raw session id is stored', async () => {
++ const raw = db.getDb();
++ const rows = [
++ ...await raw.prepare('SELECT * FROM content_view_actors').all(),
++ ...await raw.prepare('SELECT * FROM content_view_counts').all(),
++ ...await raw.prepare('SELECT * FROM content_view_days').all(),
++ ];
++ const text = JSON.stringify(rows);
++ assert.ok(!text.includes('anon_session_101'), 'a raw anonymous session id');
++ assert.ok(!text.includes('127.0.0.1'), 'an IP address');
++ assert.ok(!text.toLowerCase().includes('googlebot'), 'a user agent');
++ });
++
++ server.close();
++ await db.close();
++ console.log = realLog;
++ if (failures) { console.error(`${failures} view check(s) failed`); process.exit(1); }
++ console.log('views: ok');
++})().catch((error) => { console.error(error); process.exit(1); });