Kerem/eth subscribe (#113)

* reflect-metadata

* rpc-methods wip

* rpc-methods wip

* wspair -> wscontext

* remove @Inject when unnecessary

* using autoBindInjectable

* defaultScope=singleton + skipBaseClassChecks

* config factory logging

* zodifying subscription rpc methods

* wdatatojson to util

* EthSubscribeRpcParamsType

* subscriptions/index.ts

* outbound subscription factory

* todo added

* created an eth_unsubscribe method

* created 2 util ws functions

* introducing context.abort

* better logging for the outbound subs factory

* outbound subscription

* ws-context handles subscription sharing + unsubscribes

* docs(changeset): Introducing subscription sharing

* removed todo

* removed @Inject

* linting fix

* todo converted

* more todos

.changeset/metal-tables-knock.md

@@ -0,0 +1,5 @@
+---
+"@whatsgood/nexus": minor
+---
+
+Introducing subscription sharing

packages/nexus/src/auth/authorization-service.ts

@@ -1,4 +1,4 @@
-import { inject, injectable } from "inversify";
+import { injectable } from "inversify";
 import { NexusConfig } from "@src/nexus-config";
 
 const AUTH_KEY_QUERY_PARAM_NAME = "key";
@@ -7,7 +7,7 @@ const AUTH_KEY_QUERY_PARAM_NAME = "key";
 export class AuthorizationService {
   private readonly authKey?: string;
 
-  constructor(@inject(NexusConfig) config: NexusConfig) {
+  constructor(config: NexusConfig) {
     this.authKey = config.authKey;
   }
 

packages/nexus/src/dependency-injection/container.ts

@@ -1,3 +1,7 @@
 import { Container } from "inversify";
 
-export const container = new Container();
+export const container = new Container({
+  autoBindInjectable: true,
+  skipBaseClassChecks: true,
+  defaultScope: "Singleton",
+});

packages/nexus/src/example/index.ts

@@ -1,8 +1,5 @@
 import * as http from "node:http";
-import { Nexus } from "@src/nexus";
-import { NodeProvider } from "@src/node-provider";
-import { CHAIN } from "@src/default-chains";
-import { Chain } from "@src/chain";
+import { Nexus, NodeProvider, CHAIN, Chain } from "@src/index";
 
 // const llamaRpcNodeProvider = new NodeProvider({
 //   name: "llama-rpc",

packages/nexus/src/http/http-controller.ts

@@ -1,5 +1,5 @@
 import type { Logger } from "pino";
-import { inject, injectable } from "inversify";
+import { injectable } from "inversify";
 import { NexusConfig } from "@src/nexus-config";
 import { RpcRequestPayloadSchema } from "@src/rpc-schema";
 import type { RpcResponse } from "@src/rpc-response";
@@ -25,13 +25,11 @@ export class HttpController {
   private readonly logger: Logger;
 
   constructor(
-    @inject(NexusConfig) private readonly config: NexusConfig,
-    @inject(LoggerFactory) private readonly loggerFactory: LoggerFactory,
-    @inject(HttpRelayHandler)
+    private readonly config: NexusConfig,
+    private readonly loggerFactory: LoggerFactory,
     private readonly httpRelayHandler: HttpRelayHandler,
-    @inject(NexusMiddlewareHandler)
     private readonly middlewareHandler: NexusMiddlewareHandler,
-    @inject(EventBus) private readonly eventBus: EventBus
+    private readonly eventBus: EventBus
   ) {
     this.logger = this.loggerFactory.get(HttpController.name);
   }

packages/nexus/src/http/http-relay-handler.ts

@@ -1,4 +1,4 @@
-import { inject, injectable } from "inversify";
+import { injectable } from "inversify";
 import type { NexusRpcContext } from "@src/nexus-rpc-context";
 import { NodeEndpointPoolFactory } from "@src/node-endpoint";
 import type { NodeRpcResponseFailure } from "@src/node-endpoint/node-rpc-response";
@@ -15,7 +15,6 @@ import {
 @injectable()
 export class HttpRelayHandler {
   constructor(
-    @inject(NodeEndpointPoolFactory)
     private readonly nodeEndpointPoolFactory: NodeEndpointPoolFactory
   ) {}
 

packages/nexus/src/logging/logger-factory.ts

@@ -1,10 +1,10 @@
 import type { Logger } from "pino";
-import { inject, injectable } from "inversify";
+import { injectable } from "inversify";
 import { NexusConfig } from "@src/nexus-config";
 
 @injectable()
 export class LoggerFactory {
-  constructor(@inject(NexusConfig) private readonly config: NexusConfig) {}
+  constructor(private readonly config: NexusConfig) {}
 
   public get(name: string, options: Record<string, any> = {}): Logger {
     // TODO: redact node provider url, and start logging providers directly

packages/nexus/src/middleware/nexus-middleware-handler.ts

@@ -1,5 +1,5 @@
 import type { Logger } from "pino";
-import { inject, injectable } from "inversify";
+import { injectable } from "inversify";
 import { errSerialize } from "@src/utils";
 import type { NexusRpcContext } from "@src/nexus-rpc-context";
 import { NexusConfig } from "@src/nexus-config";
@@ -12,10 +12,7 @@ export class NexusMiddlewareHandler {
   private readonly middleware: NexusMiddleware[];
   private readonly logger: Logger;
 
-  constructor(
-    @inject(NexusConfig) config: NexusConfig,
-    @inject(LoggerFactory) loggerFactory: LoggerFactory
-  ) {
+  constructor(config: NexusConfig, loggerFactory: LoggerFactory) {
     this.middleware = [...config.middleware, authMiddleware];
     this.logger = loggerFactory.get(NexusMiddlewareHandler.name);
   }

packages/nexus/src/nexus-config/nexus-config-factory.ts

@@ -33,12 +33,14 @@ export interface NexusConfigOptions {
 export class NexusConfigFactory {
   private readonly options: NexusConfigOptions;
   private readonly envConfig: EnvConfig;
+  private readonly baseLogger: Logger;
   private readonly logger: Logger;
 
   constructor(options?: NexusConfigOptions) {
     this.options = options || {};
     this.envConfig = getEnvConfig(this.getEnv());
-    this.logger = this.getLogger();
+    this.baseLogger = this.getLogger();
+    this.logger = this.baseLogger.child({ name: NexusConfigFactory.name });
   }
 
   private getEnv() {
@@ -63,7 +65,7 @@ export class NexusConfigFactory {
       chains: new Map(uniqueChains.map((chain) => [chain.chainId, chain])),
       relay: this.getRelayConfig(),
       port: this.getPort(),
-      logger: this.logger,
+      logger: this.baseLogger,
       middleware: this.getMiddleware(),
       authKey: this.getAuthKey(),
     });

packages/nexus/src/nexus/nexus.ts

@@ -5,24 +5,17 @@ import type {
 } from "@whatwg-node/server";
 import { createServerAdapter } from "@whatwg-node/server";
 import type { Logger } from "pino";
-import { decorate, inject, injectable } from "inversify";
-import { EventEmitter } from "eventemitter3";
+import { injectable } from "inversify";
 import {
   NexusConfig,
   NexusConfigFactory,
   type NexusConfigOptions,
 } from "@src/nexus-config";
 import { HttpController } from "@src/http";
-import { WsPairHandler, WsRpcServer } from "@src/websockets";
+import { WsContextHandler, WsRpcServer } from "@src/websockets";
 import { LoggerFactory } from "@src/logging";
 import { EventBus } from "@src/events";
 import { container } from "@src/dependency-injection";
-import { HttpRelayHandler } from "@src/http/http-relay-handler";
-import { NodeEndpointPoolFactory } from "@src/node-endpoint";
-import { NexusMiddlewareHandler } from "@src/middleware";
-import { AuthorizationService } from "@src/auth";
-
-decorate(injectable(), EventEmitter); // TODO: put this somewhere else
 
 export type NexusServerInstance = ServerAdapter<unknown, Nexus>;
 
@@ -33,11 +26,11 @@ export class Nexus implements ServerAdapterBaseObject<unknown> {
   public readonly on: EventBus["on"];
 
   constructor(
-    @inject(HttpController) private readonly controller: HttpController,
-    @inject(WsPairHandler) private readonly wsPairHandler: WsPairHandler,
-    @inject(LoggerFactory) loggerFactory: LoggerFactory,
-    @inject(EventBus) eventBus: EventBus,
-    @inject(NexusConfig) config: NexusConfig
+    private readonly controller: HttpController,
+    private readonly wsContextHandler: WsContextHandler,
+    loggerFactory: LoggerFactory,
+    eventBus: EventBus,
+    config: NexusConfig
   ) {
     this.logger = loggerFactory.get(Nexus.name);
     this.on = eventBus.on.bind(eventBus);
@@ -52,8 +45,8 @@ export class Nexus implements ServerAdapterBaseObject<unknown> {
   public ws(httpServer: NodeHttpServer) {
     const wsServer = container.resolve(WsRpcServer);
 
-    wsServer.on("connection", (pair) => {
-      this.wsPairHandler.handleConnection(pair);
+    wsServer.on("connection", (context) => {
+      this.wsContextHandler.handleConnection(context);
     });
 
     httpServer.on("upgrade", wsServer.handleUpgrade.bind(wsServer));
@@ -64,17 +57,6 @@ export class Nexus implements ServerAdapterBaseObject<unknown> {
     const config = nexusConfigFactory.getNexusConfig();
 
     container.bind(NexusConfig).toConstantValue(config);
-    container.bind(HttpController).toSelf().inSingletonScope();
-    container.bind(Nexus).toSelf().inSingletonScope();
-    container.bind(LoggerFactory).toSelf().inSingletonScope();
-    container.bind(WsPairHandler).toSelf().inSingletonScope();
-    container.bind(WsRpcServer).toSelf().inSingletonScope();
-    container.bind(EventBus).toSelf().inSingletonScope();
-    container.bind(HttpRelayHandler).toSelf().inSingletonScope();
-    container.bind(NodeEndpointPoolFactory).toSelf().inSingletonScope();
-    container.bind(NexusMiddlewareHandler).toSelf().inSingletonScope();
-    container.bind(EventEmitter).toSelf().inSingletonScope();
-    container.bind(AuthorizationService).toSelf().inSingletonScope();
 
     const nexus = container.resolve(Nexus);
 

packages/nexus/src/node-endpoint/node-endpoint-pool-factory.ts

@@ -1,4 +1,4 @@
-import { inject, injectable } from "inversify";
+import { injectable } from "inversify";
 import type { Chain } from "@src/chain";
 import type { NodeProvider } from "@src/node-provider";
 import { LoggerFactory } from "@src/logging";
@@ -15,8 +15,8 @@ export class NodeEndpointPoolFactory {
   public readonly ws: Map<Chain, NodeEndpointPool>;
 
   constructor(
-    @inject(NexusConfig) private readonly config: NexusConfig,
-    @inject(LoggerFactory) private readonly loggerFactory: LoggerFactory // TODO: this should not be a property.
+    private readonly config: NexusConfig,
+    private readonly loggerFactory: LoggerFactory
   ) {
     this.nodeProviders = config.nodeProviders;
     this.http = this.createChainToEndpointPoolMap("http");
@@ -63,7 +63,9 @@ export class NodeEndpointPoolFactory {
                 nodeProvider,
               })
           ),
-          loggerFactory: this.loggerFactory,
+          logger: this.loggerFactory.get(NodeEndpointPool.name, {
+            chain,
+          }),
         })
       );
     }

packages/nexus/src/node-endpoint/node-endpoint-pool.ts

@@ -2,7 +2,6 @@ import type { Logger } from "pino";
 import type { Chain } from "@src/chain";
 import type { RpcRequestPayloadType } from "@src/rpc-schema";
 import { generatorOf, take, weightedShuffleGenerator } from "@src/utils";
-import type { LoggerFactory } from "@src/logging";
 import type { NexusConfig } from "@src/nexus-config";
 import type { RelayConfig } from "./relay-config";
 import {
@@ -28,12 +27,12 @@ export class NodeEndpointPool {
     chain: Chain;
     config: NexusConfig;
     nodeEndpoints: NodeEndpoint[];
-    loggerFactory: LoggerFactory;
+    logger: Logger;
   }) {
     this.chain = params.chain;
     this.nodeEndpoints = params.nodeEndpoints;
     this.config = params.config.relay;
-    this.logger = params.loggerFactory.get(NodeEndpointPool.name);
+    this.logger = params.logger;
 
     if (this.config.failure.kind === "cycle-requests") {
       this.maxRelayAttempts = this.config.failure.maxAttempts;

packages/nexus/src/rpc-methods/index.ts

@@ -0,0 +1,57 @@
+import { z } from "zod";
+import {
+  RpcRequestPayloadSchema,
+  RpcResponseSuccessPayloadSchema,
+} from "@src/rpc-schema";
+
+export const eth_subscribe_newHeads = RpcRequestPayloadSchema.extend({
+  method: z.literal("eth_subscribe"),
+  params: z.tuple([
+    z.literal("newHeads"),
+    // TODO: can this method have more params?
+  ]),
+});
+
+export const eth_subscribe_newPendingTransactions =
+  RpcRequestPayloadSchema.extend({
+    method: z.literal("eth_subscribe"),
+    params: z.tuple([
+      z.literal("newPendingTransactions"),
+      // TODO: can this method have more params?
+    ]),
+  });
+
+export const eth_subscribe = z.union([
+  eth_subscribe_newHeads,
+  eth_subscribe_newPendingTransactions,
+]);
+
+export type EthSubscribeRpcPayloadType = z.infer<typeof eth_subscribe>;
+export type EthSubscribeRpcParamsType = EthSubscribeRpcPayloadType["params"];
+
+export const eth_unsubscribe = RpcRequestPayloadSchema.extend({
+  method: z.literal("eth_unsubscribe"),
+  params: z.tuple([z.string()]),
+});
+
+export const eth_subscribeSuccessResponsePayloadSchema =
+  RpcResponseSuccessPayloadSchema.extend({
+    result: z.string(),
+  });
+
+export type EthSubscribeSuccessResponsePayloadType = z.infer<
+  typeof eth_subscribeSuccessResponsePayloadSchema
+>;
+
+export const eth_subscriptionPayloadSchema = z.object({
+  jsonrpc: z.literal("2.0"),
+  method: z.literal("eth_subscription"),
+  params: z.object({
+    subscription: z.string(),
+    result: z.unknown(), // TODO: do better narrowing here
+  }),
+});
+
+export type EthSubscriptionPayloadType = z.infer<
+  typeof eth_subscriptionPayloadSchema
+>;