Realtime WebSocket v2

joripspace realtime-v2 status --project PROJECT
joripspace realtime-v2 activate --project PROJECT
joripspace realtime-v2 docs

import { Hono } from 'hono';
import { createRealtimeServer } from '../vendor/realtime/server';
import { requireSession } from '../middleware/session';
import type { AppEnv } from '../types';
type RealtimeEnv = AppEnv & { Bindings: AppEnv['Bindings'] & {
  JORIPSPACE_REALTIME_V2_SIGNING_KEY: string;
} };
export const realtimeRoutes = new Hono<RealtimeEnv>();
realtimeRoutes.use('*', requireSession);
realtimeRoutes.post('/token', async (c) => {
  const identity = c.get('identity');
  const realtime = createRealtimeServer({
    origin: new URL(c.req.url).origin,
    signingKey: c.env.JORIPSPACE_REALTIME_V2_SIGNING_KEY,
  });
  // Private notification room: do not let the browser choose a user ID or room.
  const room = await realtime.createRoom('user_' + identity.user_id);
  return c.json({ room: room.name, ...await realtime.accessToken(room, {
    subject: identity.user_id, session: identity.session_hash,
    permissions: ['subscribe', 'history'],
  }) });
});

import { realtimeRoutes } from './routes/realtime';
app.route('/api/realtime', realtimeRoutes);

import { createRealtimeServer, messageId } from './vendor/realtime/server';

// Call this from business logic after server-side authorization succeeds.
const realtime = createRealtimeServer({
  origin: new URL(request.url).origin,
  signingKey: env.JORIPSPACE_REALTIME_V2_SIGNING_KEY,
});
const room = await realtime.getRoom('user_' + userId);
const event = { id: messageId(), type: 'notification.created', data: { notification_id: notificationId } };
await realtime.publish(room, event);
// When changing business data too, store the event ID, body and room.generation in an outbox
// and retry the same values until publishing succeeds. A successful database write alone is not a successful publish.

import { RealtimeClient } from '/sdk/realtime/browser.js';
// Download the pinned SDK import and serve it from your own site.
async function getToken() {
  const response = await fetch('/api/realtime/token', { method: 'POST', credentials: 'same-origin' });
  if (!response.ok) throw new Error('Check the login session and permissions.');
  return response.json();
}
const first = await getToken();
const client = new RealtimeClient({
  room: first.room,
  getToken, // the server rechecks the current session on every reconnect and refresh
  onEvent(event) {
    if (event.type !== 'notification.created') return;
    const element = document.querySelector('#last-notification');
    if (!element) throw new Error('A notification display element is required.');
    element.textContent = String(event.data.notification_id);
    // The SDK advances the sequence only after the event is applied successfully.
  },
  onState(state) { document.querySelector('#connection-state').textContent = state; },
  onError() { console.warn('Check the realtime connection status.'); },
});
client.start();
window.addEventListener('pagehide', () => client.stop());

// Server SDK example. Create realtime and room in authenticated server code.
async function readPage(cursor) {
  return realtime.history(room, {
    limit: '50', search: 'notification', ...(cursor ? { cursor } : {}),
  });
}
const page = await readPage(); // return this single page from the application API
// Pass page.cursor to a load-more request. A null value means the scan is complete.
// Reset the cursor when the search term changes. Never log raw messages or tokens.

await realtime.deleteMessage(room, messageIdToDelete);
await realtime.revoke(room, { target: 'session', session: sessionHashToRevoke });

// Separate ES module. This lists raw realtime events, not business-database records.
// HTML: <ul id="event-list"></ul>
import { RealtimeClient } from '/sdk/realtime/browser.js';
const items = new Map();
const deleted = new Set(); // prevent a deletion received during loading from being undone by a late response
const list = document.querySelector('#event-list');
if (!list) throw new Error('An event-list element is required.');
function draw() {
  const rows = [...items.values()].sort((a, b) =>
    BigInt(a.seq) < BigInt(b.seq) ? -1 : BigInt(a.seq) > BigInt(b.seq) ? 1 : 0);
  list.replaceChildren(...rows.map(event => {
    const row = document.createElement('li');
    row.dataset.messageId = event.id;
    row.textContent = event.type + ' ' + JSON.stringify(event.data);
    return row;
  }));
}
function merge(event) {
  if (event.deleted) return remove(event.id);
  if (!deleted.has(event.id)) items.set(event.id, event);
  draw();
}
function remove(id) { deleted.add(id); items.delete(id); draw(); }
async function getToken() {
  const response = await fetch('/api/realtime/token', { method: 'POST', credentials: 'same-origin' });
  if (!response.ok) throw new Error('Failed to issue an access token');
  return response.json();
}
const first = await getToken();
const client = new RealtimeClient({ room: first.room, getToken,
  onEvent: merge, onDelete: remove,
  onError() { console.warn('Check the realtime recovery status.'); },
});
client.start(); // this is a new list, so the SDK recovers history from sequence 0
window.addEventListener('pagehide', () => client.stop());
// Merge a separate history page into the same state with page.events.forEach(merge).
// For large histories, use paginated lists and reload canonical records from the database as described below.

// Example server business logic after authentication, author authorization and input validation.
// Changing the body under an existing message ID is a conflict, not an edit.
const updateEvent = {
  id: messageId(),
  type: 'message.updated', // application-defined event name, not a platform edit command
  data: { entity_id: entityId, version: nextVersion },
};
// Store the business-data update and an outbox containing updateEvent ID, body and room generation
// in the same transaction; the outbox processor retries the publish below.
await realtime.publish(room, updateEvent);
// The receiving screen reloads the latest canonical record from an authorized application API using entity_id.
// Do not overwrite the current screen when a response older than version arrives.