Merge pull request #17 from jsonjoy-com/schema

streamich · web-flow · commit 6ffa7baf674a · 2024-10-05T12:11:44.000+02:00
Schema
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
@@ -20,7 +20,7 @@ jobs:
           node-version: ${{ matrix.node-version }}
           cache: yarn
       - run: yarn install --frozen-lockfile
-      - run: yarn prettier:check
+      - run: yarn lint
       - run: yarn format
       - run: yarn test:ci --ci
       - run: PORT=10000 yarn test:e2e --server http1 --suite sample-api
diff --git a/package.json b/package.json
@@ -58,7 +58,7 @@
     "demo:ui:json": "webpack serve --config ./src/__demos__/ui-json/webpack.config.js",
     "demo:ui:text": "webpack serve --config ./src/__demos__/ui-text/webpack.config.js",
     "start:json-crdt-server:http1": "NODE_ENV=production PORT=80 JSON_CRDT_STORE=level pm2 start lib/__demos__/json-crdt-server/main-http1.js",
-    "start": "NODE_ENV=production PORT=80 JSON_CRDT_STORE=level pm2 start lib/__demos__/json-crdt-server/main-http1.js --exp-backoff-restart-delay=100",
+    "start": "ts-node src/__demos__/json-crdt-server/main-http1",
     "coverage": "yarn test --collectCoverage",
     "typedoc": "typedoc",
     "build:pages": "rimraf gh-pages && mkdir -p gh-pages && cp -r typedocs/* gh-pages && cp -r coverage gh-pages/coverage",
diff --git a/src/__demos__/json-crdt-server/__bench__/ping.bench.ts b/src/__demos__/json-crdt-server/__bench__/ping.bench.ts
@@ -23,7 +23,7 @@ const main = async () => {
       newChannel: () =>
         new WebSocketChannel({
           newSocket: () =>
-            new WebSocket('ws://localhost:9999/rpc', [codec.specifier()], {perMessageDeflate: false}) as any,
+            new WebSocket('ws://localhost:9999/rx', [codec.specifier()], {perMessageDeflate: false}) as any,
         }),
     },
   });
diff --git a/src/__demos__/json-crdt-server/main-http1.ts b/src/__demos__/json-crdt-server/main-http1.ts
@@ -1,5 +1,6 @@
 // Run: npx ts-node src/json-crdt-server/main-http1.ts
-// curl localhost:9999/rpc -H 'Content-Type: rpc.rx.compact.json' -d '[1,1,"util.ping"]'
+// curl localhost:9999/rpc -d '{"method": "util.ping", "id": 0}'
+// curl localhost:9999/rx -d '[1,1,"util.ping"]'
 
 import {createCaller, createServices} from './routes';
 import {RpcServer} from '../../server/http1/RpcServer';
diff --git a/src/__demos__/json-crdt-server/main-uws.ts b/src/__demos__/json-crdt-server/main-uws.ts
@@ -1,5 +1,5 @@
 // Run: npx ts-node src/json-crdt-server/main-uws.ts
-// curl localhost:9999/rpc -H 'Content-Type: rpc.rx.compact.json' -d '[1,1,"util.ping"]'
+// curl localhost:9999/rx -d '[1,1,"util.ping"]'
 
 import {App} from 'uWebSockets.js';
 import {RpcApp} from '../../server/uws/RpcApp';
diff --git a/src/__demos__/ui-json/main.tsx b/src/__demos__/ui-json/main.tsx
@@ -7,7 +7,7 @@ import {Model, Patch} from 'json-joy/lib/json-crdt';
 /* tslint:disable no-console */
 
 const repo = new JsonCrdtRepo({
-  wsUrl: 'wss://demo-iasd8921ondk0.jsonjoy.com/rpc',
+  wsUrl: 'wss://demo-iasd8921ondk0.jsonjoy.com/rx',
 });
 const id = 'block-sync-ui-demo-json';
 const session = repo.make(id);
diff --git a/src/__demos__/ui-text/main.tsx b/src/__demos__/ui-text/main.tsx
@@ -8,7 +8,7 @@ import {Model, Patch} from 'json-joy/lib/json-crdt';
 
 const main = async () => {
   const repo = new JsonCrdtRepo({
-    wsUrl: 'wss://demo-iasd8921ondk0.jsonjoy.com/rpc',
+    wsUrl: 'wss://demo-iasd8921ondk0.jsonjoy.com/rx',
   });
   const id = 'block-sync-ui-demo-text-3';
   const session = await repo.sessions.load({id: [id], make: {}, remote: {timeout: 1000}});
diff --git a/src/__tests__/e2e/clients.ts b/src/__tests__/e2e/clients.ts
@@ -6,7 +6,7 @@ import {StreamingRpcClient} from '../../common';
 
 export const setupRpcPersistentClient = (codec: RpcCodec) => {
   const port = +(process.env.PORT || 9999);
-  const url = `ws://localhost:${port}/rpc`;
+  const url = `ws://localhost:${port}/rx`;
   const client = new RpcPersistentClient({
     codec,
     channel: {
@@ -25,7 +25,7 @@ export const setupRpcPersistentClient = (codec: RpcCodec) => {
 
 export const setupFetchRpcClient = (codec: RpcCodec) => {
   const port = +(process.env.PORT || 9999);
-  const url = `http://localhost:${port}/rpc`;
+  const url = `http://localhost:${port}/rx`;
   const client = new FetchRpcClient({
     url,
     msgCodec: codec.msg,
@@ -44,7 +44,7 @@ export const setupStreamingRpcClient = (codec: RpcCodec) => {
   const client = new StreamingRpcClient({
     send: async (messages) => {
       const port = +(process.env.PORT || 9999);
-      const url = `http://localhost:${port}/rpc`;
+      const url = `http://localhost:${port}/rx`;
       codec.req.encoder.writer.reset();
       codec.msg.encodeBatch(codec.req, messages);
       const body = codec.req.encoder.writer.flush();
diff --git a/src/__tests__/e2e/demo-client.ts b/src/__tests__/e2e/demo-client.ts
@@ -8,7 +8,7 @@ const host = 'demo-iasd8921ondk0.jsonjoy.com';
 // const host = '127.0.0.1:8080';
 
 export const setupDemoServerPersistentClient = (codec: RpcCodec) => {
-  const url = `ws${secure ? 's' : ''}://${host}/rpc`;
+  const url = `ws${secure ? 's' : ''}://${host}/rx`;
   const client = new RpcPersistentClient({
     codec,
     channel: {
@@ -26,7 +26,7 @@ export const setupDemoServerPersistentClient = (codec: RpcCodec) => {
 };
 
 export const setupDemoServerFetchClient = (codec: RpcCodec) => {
-  const url = `http${secure ? 's' : ''}://${host}/rpc`;
+  const url = `http${secure ? 's' : ''}://${host}/rx`;
   const client = new FetchRpcClient({
     url,
     msgCodec: codec.msg,
@@ -44,7 +44,7 @@ export const setupDemoServerStreamingClient = (codec: RpcCodec) => {
   const contentType = 'application/x.' + protocolSpecifier;
   const client = new StreamingRpcClient({
     send: async (messages) => {
-      const url = `http${secure ? 's' : ''}://${host}/rpc`;
+      const url = `http${secure ? 's' : ''}://${host}/rx`;
       codec.req.encoder.writer.reset();
       codec.msg.encodeBatch(codec.req, messages);
       const body = codec.req.encoder.writer.flush();
diff --git a/src/common/rpc/caller/ObjectValueCaller.ts b/src/common/rpc/caller/ObjectValueCaller.ts
@@ -43,7 +43,7 @@ export class ObjectValueCaller<V extends ObjectValue<ObjectType<any>>, Ctx = unk
   extends RpcCaller<Ctx>
   implements Printable
 {
-  protected readonly router: V;
+  public readonly router: V;
   protected readonly system: TypeSystem;
   protected readonly methods = new Map<string, StaticRpcMethod<Ctx> | StreamingRpcMethod<Ctx>>();
 
diff --git a/src/server/http1/Http1Server.ts b/src/server/http1/Http1Server.ts
@@ -14,13 +14,17 @@ import {Http1ConnectionContext, WsConnectionContext} from './context';
 import {RpcCodecs} from '../../common/codec/RpcCodecs';
 import {RpcMessageCodecs} from '../../common/codec/RpcMessageCodecs';
 import {NullObject} from '@jsonjoy.com/util/lib/NullObject';
+import type {RpcMessageCodec} from '../../common/codec/types';
 
 export type Http1Handler = (ctx: Http1ConnectionContext) => void | Promise<void>;
 export type Http1NotFoundHandler = (res: http.ServerResponse, req: http.IncomingMessage) => void;
 export type Http1InternalErrorHandler = (error: unknown, res: http.ServerResponse, req: http.IncomingMessage) => void;
 
 export class Http1EndpointMatch {
-  constructor(public readonly handler: Http1Handler) {}
+  constructor(
+    public readonly handler: Http1Handler,
+    public readonly msgCodec: RpcMessageCodec,
+  ) {}
 }
 
 export interface Http1EndpointDefinition {
@@ -39,6 +43,11 @@ export interface Http1EndpointDefinition {
    * The handler function.
    */
   handler: Http1Handler;
+
+  /**
+   * Default message codec for this route.
+   */
+  msgCodec?: RpcMessageCodec;
 }
 
 export interface WsEndpointDefinition {
@@ -112,7 +121,7 @@ export class Http1Server implements Printable {
     const method = def.method ? def.method.toUpperCase() : 'GET';
     const route = method + path;
     Number(route);
-    const match = new Http1EndpointMatch(def.handler);
+    const match = new Http1EndpointMatch(def.handler, def.msgCodec ?? this.codecs.messages.jsonRpc2);
     this.httpRouter.add(route, match);
   }
 
@@ -136,6 +145,7 @@ export class Http1Server implements Printable {
       const codecs = this.codecs;
       const ip = this.findIp(req);
       const token = this.findToken(req);
+      const matchData = match.data;
       const ctx = new Http1ConnectionContext(
         req,
         res,
@@ -147,12 +157,12 @@ export class Http1Server implements Printable {
         new NullObject(),
         codecs.value.json,
         codecs.value.json,
-        codecs.messages.compact,
+        matchData.msgCodec,
       );
       const headers = req.headers;
       const contentType = headers['content-type'];
       if (typeof contentType === 'string') setCodecs(ctx, contentType, codecs);
-      const handler = match.data.handler;
+      const handler = matchData.handler;
       await handler(ctx);
     } catch (error) {
       this.oninternalerror(error, res, req);
diff --git a/src/server/http1/RpcServer.ts b/src/server/http1/RpcServer.ts
@@ -10,10 +10,14 @@ import {
   RpcMessageBatchProcessor,
   RpcMessageStreamProcessor,
 } from '../../common';
-import type {WsConnectionContext} from './context';
+import type {Http1ConnectionContext, WsConnectionContext} from './context';
 import type {RpcCaller} from '../../common/rpc/caller/RpcCaller';
 import type {ServerLogger} from './types';
 import type {ConnectionContext} from '../types';
+import {ObjectValueCaller} from '../../common/rpc/caller/ObjectValueCaller';
+import type {ObjectValue} from 'json-joy/lib/json-type-value/ObjectValue';
+import type {ObjectType} from 'json-joy/lib/json-type/type/classes';
+import {gzip} from '@jsonjoy.com/util/lib/compression/gzip';
 
 const DEFAULT_MAX_PAYLOAD = 4 * 1024 * 1024;
 
@@ -97,41 +101,54 @@ export class RpcServer implements Printable {
     });
   }
 
-  public enableHttpRpc(path = '/rpc'): void {
-    const batchProcessor = this.batchProcessor;
-    const logger = this.opts.logger ?? console;
-    this.http1.route({
+  private processHttpRpcRequest = async (ctx: Http1ConnectionContext) => {
+    const res = ctx.res;
+    const body = await ctx.body(DEFAULT_MAX_PAYLOAD);
+    if (!res.socket) return;
+    try {
+      const messageCodec = ctx.msgCodec;
+      const incomingMessages = messageCodec.decodeBatch(ctx.reqCodec, body);
+      try {
+        const outgoingMessages = await this.batchProcessor.onBatch(incomingMessages as IncomingBatchMessage[], ctx);
+        if (!res.socket) return;
+        const resCodec = ctx.resCodec;
+        messageCodec.encodeBatch(resCodec, outgoingMessages);
+        const buf = resCodec.encoder.writer.flush();
+        if (!res.socket) return;
+        res.end(buf);
+      } catch (error) {
+        const logger = this.opts.logger ?? console;
+        logger.error('HTTP_RPC_PROCESSING', error, {messages: incomingMessages});
+        throw RpcError.from(error);
+      }
+    } catch (error) {
+      if (typeof error === 'object' && error)
+        if ((error as any).message === 'Invalid JSON') throw RpcError.badRequest();
+      throw RpcError.from(error);
+    }
+  };
+
+  public enableHttpRpc(path = '/rx'): void {
+    const http1 = this.http1;
+    http1.route({
       method: 'POST',
       path,
-      handler: async (ctx) => {
-        const res = ctx.res;
-        const body = await ctx.body(DEFAULT_MAX_PAYLOAD);
-        if (!res.socket) return;
-        try {
-          const messageCodec = ctx.msgCodec;
-          const incomingMessages = messageCodec.decodeBatch(ctx.reqCodec, body);
-          try {
-            const outgoingMessages = await batchProcessor.onBatch(incomingMessages as IncomingBatchMessage[], ctx);
-            if (!res.socket) return;
-            const resCodec = ctx.resCodec;
-            messageCodec.encodeBatch(resCodec, outgoingMessages);
-            const buf = resCodec.encoder.writer.flush();
-            if (!res.socket) return;
-            res.end(buf);
-          } catch (error) {
-            logger.error('HTTP_RPC_PROCESSING', error, {messages: incomingMessages});
-            throw RpcError.from(error);
-          }
-        } catch (error) {
-          if (typeof error === 'object' && error)
-            if ((error as any).message === 'Invalid JSON') throw RpcError.badRequest();
-          throw RpcError.from(error);
-        }
-      },
+      handler: this.processHttpRpcRequest,
+      msgCodec: http1.codecs.messages.compact,
+    });
+  }
+
+  public enableJsonRcp2HttpRpc(path = '/rpc'): void {
+    const http1 = this.http1;
+    http1.route({
+      method: 'POST',
+      path,
+      handler: this.processHttpRpcRequest,
+      msgCodec: http1.codecs.messages.jsonRpc2,
     });
   }
 
-  public enableWsRpc(path = '/rpc'): void {
+  public enableWsRpc(path = '/rx'): void {
     const opts = this.opts;
     const logger = opts.logger ?? console;
     const caller = opts.caller;
@@ -186,11 +203,45 @@ export class RpcServer implements Printable {
     });
   }
 
+  /**
+   * Exposes JSON Type schema under the GET /schema endpoint.
+   */
+  public enableSchema(path: string = '/schema', method: string = 'GET'): void {
+    const caller = this.opts.caller;
+    let responseBody: Uint8Array = Buffer.from('{}');
+    if (caller instanceof ObjectValueCaller) {
+      const api = caller.router as ObjectValue<ObjectType<any>>;
+      const schema = {
+        value: api.type.getSchema(),
+        types: api.type.system?.exportTypes(),
+      };
+      responseBody = Buffer.from(JSON.stringify(schema));
+    }
+    let responseBodyCompressed: Uint8Array = new Uint8Array(0);
+    gzip(responseBody).then((compressed) => (responseBodyCompressed = compressed));
+    this.http1.route({
+      method,
+      path,
+      handler: (ctx) => {
+        const res = ctx.res;
+        res.writeHead(200, 'OK', {
+          'Content-Type': 'application/json',
+          'Content-Encoding': 'gzip',
+          'Cache-Control': 'public, max-age=3600, immutable',
+          'Content-Length': responseBodyCompressed.length,
+        });
+        res.end(responseBodyCompressed);
+      },
+    });
+  }
+
   public enableDefaults(): void {
     this.enableCors();
     this.enableHttpPing();
     this.enableHttpRpc();
+    this.enableJsonRcp2HttpRpc();
     this.enableWsRpc();
+    this.enableSchema();
   }
 
   // ---------------------------------------------------------------- Printable
diff --git a/yarn.lock b/yarn.lock
@@ -4468,11 +4468,6 @@ popper.js@1.16.1-lts:
   resolved "https://registry.yarnpkg.com/popper.js/-/popper.js-1.16.1-lts.tgz#cf6847b807da3799d80ee3d6d2f90df8a3f50b05"
   integrity sha512-Kjw8nKRl1m+VrSFCoVGPph93W/qrSO7ZkqPpTf7F4bk/sqcfWK019dWBUpE/fBOsOQY1dks/Bmcbfn1heM/IsA==
 
-prettier@^3.2.5:
-  version "3.3.3"
-  resolved "https://registry.yarnpkg.com/prettier/-/prettier-3.3.3.tgz#30c54fe0be0d8d12e6ae61dbb10109ea00d53105"
-  integrity sha512-i2tDNA0O5IrMO757lfrdQZCc2jPNDVntV0m/+4whiDfWaTKfMNgR7Qz0NAeGz/nRqF4m5/6CLzbP4/liHt12Ew==
-
 pretty-error@^4.0.0:
   version "4.0.0"
   resolved "https://registry.yarnpkg.com/pretty-error/-/pretty-error-4.0.0.tgz#90a703f46dd7234adb46d0f84823e9d1cb8f10d6"

Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ export class ObjectValueCaller<V extends ObjectValue<ObjectType<any>>, Ctx = unk`
`43`	`43`	`extends RpcCaller<Ctx>`
`44`	`44`	`implements Printable`
`45`	`45`	`{`
`46`		`- protected readonly router: V;`
	`46`	`+ public readonly router: V;`
`47`	`47`	`protected readonly system: TypeSystem;`
`48`	`48`	`protected readonly methods = new Map<string, StaticRpcMethod<Ctx> \| StreamingRpcMethod<Ctx>>();`
`49`	`49`