MediaHub publishes its first events: mediahub.video.published and mediahub.picture.uploaded (#83)

ddiff --git a/README.md b/README.md
iindex d78c39d..97216f6 100644
---- a/README.md
++++ b/README.md
@@@ -85,6 +85,12 @@ On other services it calls:
  
  - OpenVibe.Network `identity.subject.resolve` — to turn the usernames a share names into canonical account subjects.
  - OpenVibe.Events `events.subscription.manage` — to create its account export/deletion subscriptions at boot.
++- OpenVibe.Events `events.event.publish` — its own events, in the namespace `mediahub.*` (an event type has no hyphen):
++  `mediahub.video.published` (an upload's first time public and ready: upload, owner, title, url; public; once, by
++  `video_uploads.published_at`) and `mediahub.picture.uploaded` (a signed-in person's picture: picture, owner; internal,
++  without the link or title; signed-out pictures emit nothing). Written to `event_outbox` (migration 0016) in the
++  change's own transaction by [server/events-outbox.js](server/events-outbox.js) and relayed by the SDK; off without
++  `MEDIAHUB_EVENTS_URL`/`EVENTS_URL` and the client secret. OpenVibe.Quest credits "Publish a video" and "Share a picture".
  - OpenVibe.Search `search.document.write` — OpenVibe.Video's public, ready uploads as Search documents (owner
    `media-hub`, type `video`, `/u/<id>`) through the owner API ([server/video/search-index.js](server/video/search-index.js),
    `MEDIAHUB_SEARCH_URL`; a minute after boot, hourly, and 30 s after an edit; a private, hidden or removed upload is
ddiff --git a/migrations/0016_events_outbox.sql b/migrations/0016_events_outbox.sql
nnew file mode 100644
iindex 0000000..936f3ef
---- /dev/null
++++ b/migrations/0016_events_outbox.sql
@@@ -0,0 +1,22 @@
++-- phase: expand
++-- MediaHub publishes its own events (contracts 0.196.0: mediahub.video.published, mediahub.picture.uploaded; Events#91
++-- gives the hyphenated service the namespace mediahub.*). server/events-outbox.js writes them here in the change's own
++-- transaction and the SDK relay sends them to OpenVibe.Events.
++CREATE TABLE IF NOT EXISTS event_outbox (
++    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++    event_id        text NOT NULL UNIQUE,
++    envelope        jsonb NOT NULL,
++    traceparent     text,
++    created_at      bigint NOT NULL,
++    attempts        integer NOT NULL DEFAULT 0,
++    next_attempt_at bigint NOT NULL DEFAULT 0,
++    sent_at         bigint,
++    seq             bigint,
++    rejected_at     bigint,
++    last_error      text
++);
++CREATE INDEX IF NOT EXISTS event_outbox_due ON event_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS event_outbox_sent ON event_outbox (sent_at) WHERE sent_at IS NOT NULL;
++
++-- When an upload was first public and ready (mediahub.video.published is written once, then).
++ALTER TABLE video_uploads ADD COLUMN IF NOT EXISTS published_at text COLLATE "C";
ddiff --git a/migrations/0017_published_backfill.sql b/migrations/0017_published_backfill.sql
nnew file mode 100644
iindex 0000000..45f636d
---- /dev/null
++++ b/migrations/0017_published_backfill.sql
@@@ -0,0 +1,5 @@
++-- phase: migrate
++-- Uploads already public and ready before MediaHub published events count as published then: their
++-- mediahub.video.published is never written late (server/video/uploads.js maybePublished).
++UPDATE video_uploads SET published_at = COALESCE(ready_at, created_at)
++ WHERE published_at IS NULL AND status = 'ready' AND state = 'active' AND visibility = 'public';
ddiff --git a/server/app.js b/server/app.js
iindex 59353d1..32bae09 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -71,16 +71,18 @@ async function createApp(opts = {}) {
      const uploads = createUploads({ s, media, config, service: drive, log });
      const account = createAccount({ db: s.db, config, media, log });
      const playlists = createPlaylists({ s, config, log });
--    const pics = createPics({ s, media, config, log });
++    // MediaHub's own events (server/events-outbox.js): written with the change, relayed to OpenVibe.Events.
++    const eventsOutbox = require('./events-outbox').createEventsOutbox({ s, config, log, fetchImpl: opts.fetchImpl || null });
++    const pics = createPics({ s, media, config, log, events: eventsOutbox });
      const picsDescribe = opts.picsDescribe || require('./pics/describe').createPicDescriber({ s, media, config, fetchImpl, log });
      // OpenVibe.Video uploads (server/video/uploads.js): Media VODs in this tenant.
--    const videoUploads = require('./video/uploads').createUploads({ s, media, config, log });
++    const videoUploads = require('./video/uploads').createUploads({ s, media, config, log, events: eventsOutbox });
      const videoAi = opts.videoAi || require('./video/ai-edit').createAiEdit({ s, media, config, fetchImpl, log });
      const videoCuts = require('./video/cut').createCuts({ s, media, uploads: videoUploads, videoAi, config, log, spawnImpl: opts.spawnImpl });
      const syncKeys = createSyncKeys({ s, config });
      const drops = createDrops({ s, media, config, uploads, log });
      const videoSearch = require('./video/search-index').createVideoSearchIndex({ s, config, videoAi, fetchImpl, log });
--    const ctx = { config, s, store, keys, sso, principal, media, identity, drive, uploads, account, playlists, pics, picsDescribe, videoUploads, videoAi, videoCuts, syncKeys, drops, videoSearch, fetchImpl, log };
++    const ctx = { config, s, store, keys, sso, principal, media, identity, drive, uploads, account, playlists, pics, picsDescribe, eventsOutbox, videoUploads, videoAi, videoCuts, syncKeys, drops, videoSearch, fetchImpl, log };
      // IndexNow for openvibe.video: off without a key; a malformed key warns and stays off rather than stopping the site.
      ctx.indexnow = opts.indexnow || (() => {
          const warn = (...a) => (log.warn || console.warn)(...a);
ddiff --git a/server/events-outbox.js b/server/events-outbox.js
nnew file mode 100644
iindex 0000000..4ebe112
---- /dev/null
++++ b/server/events-outbox.js
@@@ -0,0 +1,62 @@
++'use strict';
++
++/**
++ * MediaHub's own events (contracts 0.196.0), written to event_outbox (migration 0016) in the change's own transaction
++ * and sent to OpenVibe.Events by the SDK relay with MediaHub's service token (events.event.publish). The source is
++ * `media-hub` and the namespace `mediahub.*` (an event type has no hyphen; Events#91).
++ *
++ *   mediahub.video.published   a person's upload is public and ready for the first time (public)
++ *   mediahub.picture.uploaded  a signed-in person's picture on OpenVibe.Pics (internal: no link, no title)
++ *
++ * The relay is off (rows wait) without MEDIAHUB_EVENTS_URL/EVENTS_URL and the OAuth client secret.
++ */
++const contracts = require('openvibe-contracts');
++const { createServiceOutbox } = require('openvibe-sdk/events');
++
++const EVENT_TYPES = ['mediahub.video.published', 'mediahub.picture.uploaded'];
++const subjectOf = (owner) => String(owner || '').replace(/^user:/, '');
++
++function createEventsOutbox({ s, config, log = console, fetchImpl = null }) {
++    const out = createServiceOutbox({
++        db: s.db, source: 'media-hub',
++        eventsUrl: config.events.url || null, networkInternalUrl: config.networkInternalUrl,
++        clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
++        log, eventTypes: EVENT_TYPES, ...(fetchImpl ? { fetch: fetchImpl } : {}),
++        validate: (env) => {
++            const e = contracts.validate('events.event-envelope@1', env);
++            if (!e.valid) return e;
++            return contracts.validate(`${env.event_type}@${env.version}`, env.payload);
++        },
++    });
++    const videoUrl = String(config.video.url).replace(/\/+$/, '');
++
++    return {
++        EVENT_TYPES,
++        /** Inside the transaction that stores the picture; signed-out pictures emit nothing. */
++        async pictureUploaded(row) {
++            const owner = subjectOf(row.owner);
++            if (!/^usr_/.test(owner)) return null;
++            return await out.emit({
++                event_type: 'mediahub.picture.uploaded', version: 1,
++                actor: { type: 'user', id: owner }, subject: { type: 'picture', id: row.id }, visibility: 'internal',
++                payload: { picture: row.id, owner: { type: 'user', id: owner }, uploaded_at: row.created_at },
++            });
++        },
++        /** Inside the transaction that first makes an upload public and ready. */
++        async videoPublished(row, publishedAt) {
++            const owner = subjectOf(row.owner);
++            if (!/^usr_/.test(owner)) return null;
++            return await out.emit({
++                event_type: 'mediahub.video.published', version: 1,
++                actor: { type: 'user', id: owner }, subject: { type: 'video', id: row.id }, visibility: 'public',
++                payload: { upload: row.id, owner: { type: 'user', id: owner }, title: String(row.title || 'Video').slice(0, 120), url: `${videoUrl}/u/${row.id}`, published_at: publishedAt },
++            });
++        },
++        start: () => out.start(),
++        stop: () => out.stop(),
++        kick: () => out.kick(),
++        status: () => out.status(),
++    };
++}
++
++module.exports = { createEventsOutbox, EVENT_TYPES };
ddiff --git a/server/index.js b/server/index.js
iindex ff2fee0..13315b4 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -70,9 +70,11 @@ async function start() {
          cutTimer.unref();
          timers.push(cutTimer);
      }
++    // MediaHub's own events (server/events-outbox.js): the relay is off without the events URL and client secret.
++    ctx.eventsOutbox.start();
      // Public uploads in OpenVibe.Search (server/video/search-index.js; off without MEDIAHUB_SEARCH_URL).
      ctx.videoSearch.start();
--    createLifecycle({ server, ctx, timers, extra: [() => videoIndexNow.stop(), () => ctx.videoSearch.stop(), () => { if (subscriptions) subscriptions.stop(); }] });
++    createLifecycle({ server, ctx, timers, extra: [() => videoIndexNow.stop(), () => ctx.videoSearch.stop(), () => ctx.eventsOutbox.stop(), () => { if (subscriptions) subscriptions.stop(); }] });
      return { server, ctx };
  }
  
ddiff --git a/server/pics/service.js b/server/pics/service.js
iindex e5c332b..0a86912 100644
---- a/server/pics/service.js
++++ b/server/pics/service.js
@@@ -44,7 +44,7 @@ const cleanText = (v, max) => String(v == null ? '' : v).replace(/\r\n?/g, '\n')
  /** A hash with too few or too many set bits is a near-flat picture: its dHash is noise, not an identity. */
  const distinctive = (h) => { const n = images.distance(h, '0000000000000000'); return n >= 8 && n <= 56; };
  
--function createPics({ s, media, config, log = console }) {
++function createPics({ s, media, config, log = console, events = null }) {
      const cfg = config.pics;
      const addressKey = crypto.createHash('sha256').update(`pics-address:${config.oauth.clientSecret || 'dev'}`).digest();
  
@@@ -115,7 +115,8 @@ function createPics({ s, media, config, log = console }) {
          if (!main.ok) return { ok: false, status: 503, code: 'pics.storage', detail: 'The picture could not be stored just now. Try again in a minute.' };
          const thumb = await putObject(prepared.thumb.buffer, { mime: 'image/webp', name: `${id}-thumb.webp`, owner: who && who.requester, anonymous });
          const deleteKey = anonymous ? crypto.randomBytes(18).toString('base64url') : null;
--        const row = await store.insertImage(s, {
++        // The row and its mediahub.picture.uploaded event (signed-in uploads only) commit together.
++        const row = await s.tx(async () => { const r = await store.insertImage(s, {
              id, slug: slug(), owner: anonymous ? null : who.requester,
              delete_key_hash: deleteKey ? images.sha256(Buffer.from(deleteKey)) : null,
              title: base, description: cleanText(description, LIMITS.description),
@@@ -124,6 +125,8 @@ function createPics({ s, media, config, log = console }) {
              bytes: prepared.image.buffer.length, animated: prepared.image.animated,
              sha256: prepared.sha256, source_sha256: prepared.sourceSha256, dhash: prepared.dhash, created_at: s.iso(),
          });
++        if (events && !anonymous) await events.pictureUploaded(r);
++        return r; });
          return { ok: true, image: row, ...(deleteKey ? { deleteKey } : {}) };
      }
  
ddiff --git a/server/video/uploads.js b/server/video/uploads.js
iindex 55ced05..3fc8bf9 100644
---- a/server/video/uploads.js
++++ b/server/video/uploads.js
@@@ -43,7 +43,7 @@ function kindOf(contentType, name) {
      return ext && EXT_KINDS[ext[1]] ? EXT_KINDS[ext[1]] : null;
  }
  
--function createUploads({ s, media, config, log = console }) {
++function createUploads({ s, media, config, log = console, events = null }) {
      const cfg = config.videoUploads;
      const addressKey = crypto.createHash('sha256').update(`video-address:${config.oauth.clientSecret || 'dev'}`).digest();
      const addressId = (ip) => `ip:${crypto.createHmac('sha256', addressKey).update(String(ip || 'unknown')).digest('hex').slice(0, 32)}`;
@@@ -58,6 +58,20 @@ function createUploads({ s, media, config, log = console }) {
          return await byId(id);
      };
  
++    /**
++     * The first time an upload is public, ready and active, published_at is set and mediahub.video.published written in
++     * the same transaction (once: private and public again does not repeat it).
++     */
++    async function maybePublished(row) {
++        if (!events || !row || row.published_at || row.status !== 'ready' || row.state !== 'active' || row.visibility !== 'public') return row;
++        const at = s.iso();
++        return await s.tx(async () => {
++            const r = await s.db.maybe('UPDATE video_uploads SET published_at = $2 WHERE id = $1 AND published_at IS NULL RETURNING *', [row.id, at]);
++            if (r) await events.videoPublished(r, at);
++            return r || row;
++        });
++    }
++
      /** May `who` ({ requester, staff }) see this upload at all? */
      function canSee(row, who) {
          if (!row || row.state === 'removed') return false;
@@@ -148,7 +162,7 @@ function createUploads({ s, media, config, log = console }) {
          const duration = Number(v.duration_seconds);
          const ready = Boolean(v.file_url || v.playback_url) && Number.isFinite(duration) && duration > 0;
          if (!ready) return row;
--        return await patchRow(row.id, { status: 'ready', duration_seconds: duration, thumbnail_url: v.thumbnail_url || null, ready_at: row.ready_at || s.iso(), error: null });
++        return await maybePublished(await patchRow(row.id, { status: 'ready', duration_seconds: duration, thumbnail_url: v.thumbnail_url || null, ready_at: row.ready_at || s.iso(), error: null }));
      }
  
      async function complete(row) {
@@@ -188,7 +202,7 @@ function createUploads({ s, media, config, log = console }) {
          if (input.visibility !== undefined) { if (!VISIBILITY.includes(input.visibility)) throw new UploadError(422, 'upload.visibility', 'Public, unlisted or private.'); patch.visibility = input.visibility; }
          if (!Object.keys(patch).length) return row;
          if (patch.title || patch.visibility) await media.vodUpdate(row.media_vod_id, { ...(patch.title ? { title: patch.title } : {}), ...(patch.visibility ? { visibility: patch.visibility } : {}) });
--        return await patchRow(row.id, patch);
++        return await maybePublished(await patchRow(row.id, patch));
      }
  
      async function remove(row) {
ddiff --git a/test/events-outbox.test.js b/test/events-outbox.test.js
nnew file mode 100644
iindex 0000000..20b7037
---- /dev/null
++++ b/test/events-outbox.test.js
@@@ -0,0 +1,80 @@
++'use strict';
++/**
++ * MediaHub's own events (server/events-outbox.js, contracts 0.196.0): a signed-in person's picture writes
++ * mediahub.picture.uploaded (internal, no link or title) with the row, a signed-out one writes nothing; an upload's first
++ * time public and ready writes mediahub.video.published once (private and public again does not repeat it), and only
++ * then; every envelope is valid (source media-hub, namespace mediahub.*).
++ */
++const assert = require('assert');
++const http = require('http');
++const sharp = require('sharp');
++const { validate } = require('openvibe-contracts');
++const { boot, check, done } = require('./helpers/boot');
++
++function picsClient(t) {
++    return (p, o = {}) => new Promise((resolve, reject) => {
++        const headers = { Host: 'openvibe.pics', 'sec-fetch-site': 'same-origin', ...(o.headers || {}) };
++        if (o.as) headers.cookie = t.signIn(o.as);
++        const body = o.body;
++        if (body != null) headers['content-length'] = String(body.length);
++        const u = new URL(t.base + p);
++        const req = http.request({ hostname: u.hostname, port: u.port, path: u.pathname, method: o.method || 'GET', headers }, (res) => {
++            const chunks = [];
++            res.on('data', (c) => chunks.push(c));
++            res.on('end', () => { const text = Buffer.concat(chunks).toString('utf8'); resolve({ status: res.statusCode, text, json: () => JSON.parse(text) }); });
++        });
++        req.on('error', reject);
++        if (body != null) req.write(body);
++        req.end();
++    });
++}
++
++(async () => {
++    const t0 = await boot({ env: { MEDIAHUB_PICS_ANONYMOUS: '1', MEDIAHUB_VIDEO_URL: 'https://openvibe.video' } });
++    const t = { ...t0, get: picsClient(t0) };
++    const s = t0.ctx.s;
++    const rows = async (type) => (await s.db.many('SELECT envelope FROM event_outbox ORDER BY id')).map((r) => (typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope)).filter((e) => !type || e.event_type === type);
++    try {
++        const kim = t0.network.addUser('kim');
++        const jpeg = await sharp({ create: { width: 300, height: 200, channels: 3, background: '#3a6' } }).jpeg().toBuffer();
++
++        await check('a signed-in picture writes mediahub.picture.uploaded with the row; a signed-out one writes nothing', async () => {
++            const r = await t.get('/api/v1/images', { as: kim, method: 'POST', body: jpeg, headers: { 'content-type': 'image/jpeg', 'x-title': 'secret%20title' } });
++            assert.strictEqual(r.status, 201, r.text);
++            const pic = r.json();
++            const ev = await rows('mediahub.picture.uploaded');
++            assert.strictEqual(ev.length, 1);
++            const e = ev[0];
++            assert.ok(validate('events.event-envelope@1', e).valid, JSON.stringify(validate('events.event-envelope@1', e).errors));
++            assert.ok(validate('mediahub.picture.uploaded@1', e.payload).valid);
++            assert.deepStrictEqual([e.source, e.visibility, e.subject, e.actor], ['media-hub', 'internal', { type: 'picture', id: pic.id }, { type: 'user', id: kim.subject }]);
++            assert.ok(!JSON.stringify(e).includes(pic.slug) && !JSON.stringify(e).includes('secret title'), 'no link and no title');
++            const anon = await t.get('/api/v1/images', { method: 'POST', body: jpeg, headers: { 'content-type': 'image/jpeg' } });
++            assert.strictEqual(anon.status, 201, anon.text);
++            assert.strictEqual((await rows('mediahub.picture.uploaded')).length, 1, 'still one');
++        });
++
++        await check('an upload\'s first time public and ready writes mediahub.video.published once', async () => {
++            const up = t0.ctx.videoUploads;
++            const ins = (id, visibility) => s.db.exec(`INSERT INTO video_uploads (id, owner, owner_username, title, description, visibility, media_vod_id, content_type, size_bytes, part_size, parts_expected, parts_received, status, state, created_at, updated_at, ready_at)
++                VALUES ($1, $2, 'kim', 'A walk on the coast', '', $3, 1, 'video/mp4', 10, 10, 1, 1, 'ready', 'active', $4, $4, $4)`, [id, `user:${kim.subject}`, visibility, s.iso()]);
++            const A = 'vup_01J0000000000000000000000A';
++            await ins(A, 'private');
++            let row = await up.update(await up.byId(A), { title: 'A walk on the coast' });
++            assert.strictEqual((await rows('mediahub.video.published')).length, 0, 'private: not published');
++            row = await up.update(row, { visibility: 'public' });
++            let ev = await rows('mediahub.video.published');
++            assert.strictEqual(ev.length, 1);
++            assert.ok(row.published_at);
++            assert.deepStrictEqual([ev[0].visibility, ev[0].payload.url, ev[0].payload.upload], ['public', `https://openvibe.video/u/${A}`, A]);
++            assert.ok(validate('mediahub.video.published@1', ev[0].payload).valid);
++            row = await up.update(row, { visibility: 'private' });
++            await up.update(row, { visibility: 'public' });
++            ev = await rows('mediahub.video.published');
++            assert.strictEqual(ev.length, 1, 'once');
++        });
++    } finally {
++        await t0.close();
++    }
++    done();
++})();