(new server) add websocket handling (#524)

Author: nikgraf

.changeset/lazy-birds-wait.md

@@ -0,0 +1,7 @@
+{
+  "permissions": {
+    "allow": ["Bash(pnpm lint:fix:*)", "Bash(pnpm typecheck:*)", "Bash(pnpm check:*)"],
+    "deny": [],
+    "ask": []
+  }
+}

.claude/settings.local.json

@@ -65,7 +65,7 @@ export class IdentityQuery extends Schema.Class<IdentityQuery>('IdentityQuery')(
 /**
  * Health endpoints
  */
-export const statusEndpoint = HttpApiEndpoint.get('status')`/`.addSuccess(Schema.String);
+export const statusEndpoint = HttpApiEndpoint.get('status')`/status`.addSuccess(Schema.String);
 
 export const healthGroup = HttpApiGroup.make('Health').add(statusEndpoint);
 

apps/server-new/src/http/api.ts

@@ -26,6 +26,7 @@ const Observability = Layer.unwrapEffect(
 
 const layer = server.pipe(
   Layer.provide(Logger.structured),
+  // Layer.provide(Logger.pretty),
   Layer.provide(Observability),
   Layer.provide(PlatformConfigProvider.layerDotEnvAdd('.env')),
   Layer.provide(NodeContext.layer),

apps/server-new/src/index.ts

@@ -2,16 +2,14 @@ import { createServer } from 'node:http';
 import * as HttpApiScalar from '@effect/platform/HttpApiScalar';
 import * as HttpLayerRouter from '@effect/platform/HttpLayerRouter';
 import * as HttpMiddleware from '@effect/platform/HttpMiddleware';
-import * as HttpServerRequest from '@effect/platform/HttpServerRequest';
-import * as HttpServerResponse from '@effect/platform/HttpServerResponse';
 import * as NodeHttpServer from '@effect/platform-node/NodeHttpServer';
 import * as Effect from 'effect/Effect';
 import * as Layer from 'effect/Layer';
-import * as Schedule from 'effect/Schedule';
-import * as Stream from 'effect/Stream';
 import { serverPortConfig } from './config/server.ts';
 import { hypergraphApi } from './http/api.ts';
 import { HandlersLive } from './http/handlers.ts';
+import * as ConnectionsService from './services/connections.ts';
+import { WebSocketLayer } from './websocket.ts';
 
 // Create scalar openapi browser layer at /docs.
 const DocsLayer = HttpApiScalar.layerHttpLayerRouter({
@@ -24,20 +22,6 @@ const ApiLayer = HttpLayerRouter.addHttpApi(hypergraphApi, {
   openapiPath: '/docs/openapi.json',
 }).pipe(Layer.provide(HandlersLive));
 
-// Create websocket layer at /ws.
-const WebSocketLayer = HttpLayerRouter.add(
-  'GET',
-  '/ws',
-  // TODO: Implement actual websocket logic here.
-  Stream.fromSchedule(Schedule.spaced(1000)).pipe(
-    Stream.map(JSON.stringify),
-    Stream.pipeThroughChannel(HttpServerRequest.upgradeChannel()),
-    Stream.decodeText(),
-    Stream.runForEach((_) => Effect.log(_)),
-    Effect.as(HttpServerResponse.empty()),
-  ),
-);
-
 // Merge router layers together and add the cors middleware layer.
 const CorsMiddleware = HttpLayerRouter.middleware(HttpMiddleware.cors());
 const AppLayer = Layer.mergeAll(ApiLayer, DocsLayer, WebSocketLayer).pipe(Layer.provide(CorsMiddleware.layer));
@@ -47,4 +31,7 @@ const HttpServerLayer = serverPortConfig.pipe(
   Layer.unwrapEffect,
 );
 
-export const server = HttpLayerRouter.serve(AppLayer).pipe(Layer.provide(HttpServerLayer));
+export const server = HttpLayerRouter.serve(AppLayer).pipe(
+  Layer.provide(HttpServerLayer),
+  Layer.provide(ConnectionsService.layer),
+);

apps/server-new/src/server.ts

@@ -37,6 +37,16 @@ export class AccountInboxService extends Context.Tag('AccountInboxService')<
       Messages.InboxMessage,
       ResourceNotFoundError | ValidationError | AuthorizationError | DatabaseService.DatabaseError
     >;
+    readonly createAccountInbox: (
+      data: Messages.RequestCreateAccountInbox,
+    ) => Effect.Effect<AccountInboxResult, ValidationError | AuthorizationError | DatabaseService.DatabaseError>;
+    readonly getLatestAccountInboxMessages: (params: {
+      inboxId: string;
+      since: Date;
+    }) => Effect.Effect<Messages.InboxMessage[], DatabaseService.DatabaseError>;
+    readonly listAccountInboxes: (params: {
+      accountAddress: string;
+    }) => Effect.Effect<Messages.AccountInbox[], DatabaseService.DatabaseError>;
   }
 >() {}
 
@@ -258,9 +268,131 @@ export const layer = Effect.gen(function* () {
     return createdMessage;
   });
 
+  const createAccountInbox = Effect.fn('createAccountInbox')(function* (data: Messages.RequestCreateAccountInbox) {
+    const { accountAddress, inboxId, isPublic, authPolicy, encryptionPublicKey, signature } = data;
+
+    // Verify the signature is valid for the corresponding accountAddress
+    const signer = Inboxes.recoverAccountInboxCreatorKey(data);
+    const signerAccount = yield* getAppOrConnectIdentity({
+      accountAddress: data.accountAddress,
+      signaturePublicKey: signer,
+    }).pipe(Effect.mapError(() => new AuthorizationError({ message: 'Invalid signature' })));
+
+    if (signerAccount.accountAddress !== accountAddress) {
+      return yield* Effect.fail(new AuthorizationError({ message: 'Invalid signature' }));
+    }
+
+    // Create the inbox (will throw an error if it already exists)
+    const inbox = yield* use((client) =>
+      client.accountInbox.create({
+        data: {
+          id: inboxId,
+          isPublic,
+          authPolicy,
+          encryptionPublicKey,
+          signatureHex: signature.hex,
+          signatureRecovery: signature.recovery,
+          account: { connect: { address: accountAddress } },
+        },
+      }),
+    );
+
+    return {
+      inboxId: inbox.id,
+      accountAddress,
+      isPublic: inbox.isPublic,
+      authPolicy: inbox.authPolicy as Inboxes.InboxSenderAuthPolicy,
+      encryptionPublicKey: inbox.encryptionPublicKey,
+      signature: {
+        hex: inbox.signatureHex,
+        recovery: inbox.signatureRecovery,
+      },
+    };
+  });
+
+  const getLatestAccountInboxMessages = Effect.fn('getLatestAccountInboxMessages')(function* ({
+    inboxId,
+    since,
+  }: {
+    inboxId: string;
+    since: Date;
+  }) {
+    const messages = yield* use((client) =>
+      client.accountInboxMessage.findMany({
+        where: {
+          accountInboxId: inboxId,
+          createdAt: {
+            gte: since,
+          },
+        },
+        orderBy: {
+          createdAt: 'asc',
+        },
+      }),
+    );
+
+    return messages.map(
+      (msg): Messages.InboxMessage => ({
+        id: msg.id,
+        ciphertext: msg.ciphertext,
+        signature:
+          msg.signatureHex != null && msg.signatureRecovery != null
+            ? {
+                hex: msg.signatureHex,
+                recovery: msg.signatureRecovery,
+              }
+            : undefined,
+        authorAccountAddress: msg.authorAccountAddress ?? undefined,
+        createdAt: msg.createdAt,
+      }),
+    );
+  });
+
+  const listAccountInboxes = Effect.fn('listAccountInboxes')(function* ({
+    accountAddress,
+  }: {
+    accountAddress: string;
+  }) {
+    const inboxes = yield* use((client) =>
+      client.accountInbox.findMany({
+        where: { accountAddress },
+        select: {
+          id: true,
+          isPublic: true,
+          authPolicy: true,
+          encryptionPublicKey: true,
+          account: {
+            select: {
+              address: true,
+            },
+          },
+          signatureHex: true,
+          signatureRecovery: true,
+        },
+      }),
+    );
+
+    return inboxes.map(
+      (inbox): Messages.AccountInbox => ({
+        inboxId: inbox.id,
+        accountAddress: inbox.account.address,
+        isPublic: inbox.isPublic,
+        authPolicy: inbox.authPolicy as Inboxes.InboxSenderAuthPolicy,
+        encryptionPublicKey: inbox.encryptionPublicKey,
+        signature: {
+          hex: inbox.signatureHex,
+          recovery: inbox.signatureRecovery,
+        },
+      }),
+    );
+  });
+
   return {
     listPublicAccountInboxes,
     getAccountInbox,
     postAccountInboxMessage,
+    createAccountInbox,
+    getLatestAccountInboxMessages,
+    listAccountInboxes,
   } as const;
 }).pipe(Layer.effect(AccountInboxService), Layer.provide(DatabaseService.layer), Layer.provide(IdentityService.layer));