Merge pull request #1036 from streamich/server-demo-improvements

chore: 🤖 improve demo setup

Author: streamich

packages/channel/docs/index.ts

@@ -1,17 +1,36 @@
 import type {LibPage} from '@jsonjoy.com/ui/src/types/libs';
 
 export const page: LibPage = {
-  name: 'channel',
-  title: 'channel',
+  name: 'Channel',
   type: 'lib',
   subtitle: 'Bidirectional communication channel abstraction over WebSocket, fetch, and more.',
-  children: [],
   pkg: '@jsonjoy.com/channel',
   group: 'sync',
   repo: 'streamich/json-joy',
   repoPath: 'tree/master/packages/channel',
   tech: 'TypeScript',
   techIcon: {set: 'lineicons', icon: 'typescript'},
+  showContentsTable: true,
+  children: [
+    {
+      name: 'Physical channel',
+      subtitle: 'The PhysicalChannel interface, states, and lifecycle events.',
+      // @ts-ignore raw markdown, loaded by the site's webpack raw-loader
+      src: async () => (await import('./physical-channel.md')).default,
+    },
+    {
+      name: 'Transports',
+      subtitle: 'WebSocketChannel, FetchPhysicalChannel, Utf8Channel.',
+      // @ts-ignore
+      src: async () => (await import('./transports.md')).default,
+    },
+    {
+      name: 'Persistent channel',
+      subtitle: 'PersistentPhysicalChannel: auto-reconnect with configurable backoff.',
+      // @ts-ignore
+      src: async () => (await import('./persistent-channel.md')).default,
+    },
+  ],
   // @ts-ignore
   src: async () => (await import('./text.md')).default,
 };

packages/channel/docs/persistent-channel.md

@@ -0,0 +1,81 @@
+## Persistent channel
+
+`PersistentPhysicalChannel<T>` keeps a connection alive across drops. It is **not** itself
+a `PhysicalChannel` --- it owns one at a time and recreates
+it via the `newChannel` factory whenever the previous one closes.
+
+```ts
+import {PersistentPhysicalChannel, WebSocketChannel}
+  from '@jsonjoy.com/channel';
+
+const persistent = new PersistentPhysicalChannel<Uint8Array>({
+  newChannel: () =>
+    new WebSocketChannel({
+      newSocket: () =>
+        new WebSocket('wss://example.com/rx')
+    }),
+});
+
+persistent.start();
+persistent.message$.subscribe((data) => handle(data));
+```
+
+
+## Options
+
+| Option | Default | Description |
+|---|---|---|
+| `newChannel` | required | Called on every (re)connect to build a fresh channel |
+| `minReconnectionDelay` | 1000--2000 ms | Floor for the first backoff |
+| `maxReconnectionDelay` | 10000 ms | Ceiling for the backoff |
+| `reconnectionDelayGrowFactor` | 1.3 | Base of the exponential backoff |
+| `minUptime` | 5000 ms | A connection must stay open this long to reset the retry counter |
+
+
+## Lifecycle
+
+```ts
+persistent.start();   // begin keeping a connection open
+persistent.stop();    // close, dispose, no further use possible
+```
+
+After `stop()`, all subjects are completed and the instance cannot be
+restarted. Construct a new one if you need to reconnect later.
+
+
+## Observable surface
+
+| Stream | Emits |
+|---|---|
+| `active$` | `true` while `start()` was called and not yet stopped |
+| `channel$` | The current underlying `PhysicalChannel`, or `undefined` |
+| `open$` | `true` whenever the active channel is open, `false` otherwise |
+| `message$` | Incoming messages from whichever underlying channel is current |
+| `error$` | Errors from the channel factory or underlying channels |
+
+
+## Sending
+
+```ts
+persistent.send$(payload).subscribe();
+```
+
+`send$` waits for the next open connection, then sends. There is no
+synchronous `send` --- by design, the caller is forced to opt into the
+waiting behavior, which is what you almost always want when reconnect can
+happen at any time.
+
+
+## Backoff
+
+`reconnectDelay()` returns the wait time before the next attempt:
+
+```
+delay = min(
+  maxReconnectionDelay,
+  minReconnectionDelay * reconnectionDelayGrowFactor ** (retries - 1)
+)
+```
+
+If a connection stays open longer than `minUptime`, the retry counter
+resets to 0 --- the next disconnect will reconnect quickly.

packages/channel/docs/physical-channel.md

@@ -0,0 +1,89 @@
+## PhysicalChannel
+
+`PhysicalChannel<T>` is the interface every concrete transport implements.
+`T` is the message payload type --- either `string`, `Uint8Array`, or the
+union of both.
+
+```ts
+interface PhysicalChannel<T extends string | Uint8Array> {
+  // Reactive surface
+  state$:    BehaviorSubject<ChannelState>;
+  open$:     Observable<PhysicalChannel<T>>;
+  close$:    Observable<[self: PhysicalChannel<T>, event: CloseEventBase]>;
+  error$:    Observable<Error>;
+  message$:  Observable<T>;
+
+  // Imperative surface
+  closed:    boolean;
+  onmessage?: (data: T, isUtf8: boolean) => void;
+  onclose?:   (code: number, reason: string, wasClean: boolean) => void;
+
+  isOpen():  boolean;
+  send(data: T):    number;            // returns bytes buffered, -1 if not ready
+  send$(data: T):   Observable<number>; // waits for open, then sends
+  close(code?, reason?): void;
+  buffer():  number;                   // bytes currently buffered out
+}
+```
+
+
+## States
+
+The `ChannelState` enum reflects the connection lifecycle:
+
+| Value | Meaning |
+|---|---|
+| `CONNECTING` | Initial; not yet open |
+| `OPEN` | Ready to send/receive |
+| `CLOSED` | Terminal; cannot be reopened |
+
+`state$` is a `BehaviorSubject`, so subscribers get the current state
+immediately. `open$` and `close$` are `ReplaySubject(1)` --- subscribing
+after the event still fires the callback.
+
+
+## Two ways to subscribe
+
+`message$` (observable) and `onmessage` (callback) deliver the same data.
+Pick whichever fits the consumer:
+
+```ts
+// Callback flavor
+channel.onmessage = (data, isUtf8) => handle(data);
+
+// RxJS flavor
+channel.message$.subscribe((data) => handle(data));
+```
+
+Same applies to close: `onclose` and `close$` both fire once on disconnect.
+
+
+## `send()` vs `send$()`
+
+- `send(data)` is **fire-and-forget**: writes immediately if open, returns
+  the number of bytes now buffered. Returns `-1` if the channel is not
+  ready --- useful as a quick liveness check.
+- `send$(data)` is **wait-then-send**: returns an Observable that defers
+  until `open$` fires. Errors if the channel closes first.
+
+```ts
+// Fire only if connected right now
+if (channel.isOpen()) channel.send(message);
+
+// Queue until the channel opens (e.g. immediately after construction)
+channel.send$(message).subscribe();
+```
+
+
+## Close event
+
+Both `close$` and `onclose` carry a `CloseEventBase`:
+
+| Field | Description |
+|---|---|
+| `code` | Numeric close code (e.g. 1000 for normal closure) |
+| `reason` | Free-form reason text |
+| `wasClean` | Whether the close handshake completed |
+
+`code` follows WebSocket conventions where applicable but every transport
+chooses its own scheme for non-WebSocket scenarios.

packages/channel/docs/text.md

@@ -1,23 +1,57 @@
-## Communication channel
+## Channel
 
-A minimal abstraction for bidirectional communication channels. It wraps
-WebSockets, `fetch`, and other transports behind a single `PhysicalChannel`
-interface backed by RxJS observables. This is the transport layer beneath
-[rpc-calls](/libs/rpc-calls) and [rpc-client](/libs/rpc-client).
+`@jsonjoy.com/channel` is a thin abstraction for bidirectional byte streams.
+Every concrete transport --- WebSocket, `fetch`, in-memory test pipe ---
+implements the same `PhysicalChannel` interface, exposing incoming messages
+and lifecycle events as RxJS observables.
+
+This is the transport layer underneath the Reactive RPC stack
+([`rpc-calls`](/libs/rpc-calls), [`rpc-client`](/libs/rpc-client)) but is
+useful on its own when you just need a uniform shape over WebSocket plus
+auto-reconnect.
 
 
 ## Installation
 
 ```
-npm install @jsonjoy.com/channel
+npm install @jsonjoy.com/channel rxjs
 ```
 
+`rxjs@7` is a peer dependency.
+
+
+## Surface
+
+| Area | Surface |
+|---|---|
+| [PhysicalChannel](/libs/channel/physical-channel) | The interface and lifecycle |
+| [Transports](/libs/channel/transports) | `WebSocketChannel`, `FetchPhysicalChannel`, `Utf8Channel` |
+| [Persistent channel](/libs/channel/persistent-channel) | Auto-reconnect with configurable backoff |
+
+
+## Quick start
+
+```ts
+import {WebSocketChannel} from '@jsonjoy.com/channel';
 
-## Components
+const channel = new WebSocketChannel<Uint8Array>({
+  newSocket: () => new WebSocket('wss://example.com/rx'),
+});
 
-- **`PhysicalChannel`** — the interface every channel implements: `message$`,
-  `error$`, `open$`, `close$`, `state$`, `send()`, `send$()`, `close()`, `buffer()`, `isOpen()`.
-- **`WebSocketChannel`** — wraps a native `WebSocket` into a `PhysicalChannel`.
-- **`FetchChannel`** — wraps a `fetch`-style request/response into a channel.
-- **`Utf8Channel`** — transparently converts binary messages to and from UTF-8.
-- **`PersistentChannel`** — auto-reconnecting wrapper with configurable backoff.
+channel.open$.subscribe(() => channel.send(new Uint8Array([1, 2, 3])));
+channel.message$.subscribe((data) => console.log('recv', data));
+channel.close$.subscribe(([, evt]) => console.log('closed', evt.code));
+```
+
+For auto-reconnect, wrap it in [`PersistentPhysicalChannel`](/libs/channel/persistent-channel):
+
+```ts
+import {PersistentPhysicalChannel} from '@jsonjoy.com/channel';
+
+const persistent = new PersistentPhysicalChannel<Uint8Array>({
+  newChannel: () => new WebSocketChannel({newSocket: () => new WebSocket(url)}),
+});
+
+persistent.start();
+persistent.message$.subscribe((data) => handle(data));
+```

packages/channel/docs/transports.md

@@ -0,0 +1,86 @@
+## Transports
+
+Three concrete implementations ship with the package:
+
+| Class | Wraps | When to use |
+|---|---|---|
+| `WebSocketChannel` | Browser/Node `WebSocket` | Bidirectional, persistent stream |
+| `FetchPhysicalChannel` | `fetch`-style RPC call | Request/response over HTTP |
+| `Utf8Channel` | Another channel | Force text payloads on a binary transport |
+
+
+## `WebSocketChannel`
+
+Wraps any `WebSocket`-shaped object. The constructor takes a `newSocket`
+factory so the channel can recreate the socket on reconnect (when used with
+[`PersistentPhysicalChannel`](/libs/channel/persistent-channel)).
+
+```ts
+import {WebSocketChannel} from '@jsonjoy.com/channel';
+
+const channel = new WebSocketChannel<Uint8Array>({
+  newSocket: () => new WebSocket('wss://example.com/rx', ['rpc.rx.binary.cbor']),
+});
+```
+
+The `binaryType` is automatically set to `'arraybuffer'`, so incoming binary
+messages arrive as `Uint8Array`. Text frames arrive as `string` with the
+`isUtf8` flag on `onmessage` set to `true`.
+
+`buffer()` returns the WebSocket's native `bufferedAmount`. `close(code,
+reason)` forwards directly to `ws.close(...)`.
+
+
+## `FetchPhysicalChannel`
+
+Adapts a request/response function into the channel interface. Every
+`send()` triggers one `fetch`, and the response is emitted as a single
+incoming message.
+
+```ts
+import {FetchPhysicalChannel} from '@jsonjoy.com/channel';
+
+const channel = new FetchPhysicalChannel({
+  fetch: async (data) => {
+    const res = await fetch('https://example.com/rpc', {
+      method: 'POST',
+      body: data,
+    });
+    return new Uint8Array(await res.arrayBuffer());
+  },
+});
+```
+
+Because there is no persistent stream, the channel reports
+`state$ === OPEN` immediately. `buffer()` always returns `0`. Errors from
+`fetch` are pushed to `error$`.
+
+
+## `Utf8Channel`
+
+A decorator. Wraps any underlying channel and converts payloads to/from
+UTF-8 strings on the way in and out. Use it when you want the rest of your
+code to deal in strings but the transport carries bytes.
+
+```ts
+import {Utf8Channel, WebSocketChannel} from '@jsonjoy.com/channel';
+
+const inner = new WebSocketChannel<Uint8Array>({newSocket: () => new WebSocket(url)});
+const channel = new Utf8Channel(inner);
+
+channel.send('hello'); // encoded to UTF-8 bytes and sent through `inner`
+channel.message$.subscribe((str) => console.log(str)); // already decoded
+```
+
+`Utf8Channel` shares the inner channel's `state$`, `open$`, `close$`, and
+`error$` streams.
+
+
+## Which to pick
+
+| Need | Use |
+|---|---|
+| Streaming bidirectional connection | `WebSocketChannel` |
+| Single-shot HTTP request/response per call | `FetchPhysicalChannel` |
+| String API over a binary channel | `Utf8Channel` |
+| Survive disconnects | wrap any of the above in [`PersistentPhysicalChannel`](/libs/channel/persistent-channel) |

packages/collaborative-presence/src/str.ts

@@ -1,4 +1,4 @@
-import {JsonCrdtDataType} from 'json-joy/lib/json-crdt-patch/constants';
+import {NodeType} from './constants';
 import * as id from './id';
 import type {ITimestampStruct, StrApi, StrNode, Model} from 'json-joy/lib/json-crdt';
 import type {PresenceIdShorthand, PresencePoint, RgaSelection, PresenceCursor} from './types';
@@ -49,7 +49,7 @@ export const toDto = (str: StrApi, selections: StrSelection[]): RgaSelection =>
     }
     cursors.push(cursor);
   }
-  const selection: RgaSelection = ['', '', sid, clock.time, {}, JsonCrdtDataType.str, nodeId, cursors];
+  const selection: RgaSelection = ['', '', sid, clock.time, {}, NodeType.str, nodeId, cursors];
   return selection;
 };
 
@@ -109,7 +109,7 @@ const findOffset = (str: StrNode, tsId: ITimestampStruct, senderSid?: number, pr
 export const fromDto = (model: Model<any>, selection: RgaSelection): StrSelectionStrict[] => {
   const [_documentId, _uiLocationId, sid, time, _meta, type, nodeIdDto, cursors] = selection;
   const result: StrSelectionStrict[] = [];
-  if (type !== JsonCrdtDataType.str) return result;
+  if (type !== NodeType.str) return result;
   const nodeId = id.fromDto(sid, nodeIdDto);
   const str = model.index.get(nodeId) as StrNode | undefined;
   if (!str || str.name() !== 'str') return result;