fix do class inheritance, upgrade Effect to beta.57 (#423)

Author: harrysolovay

.changeset/legal-streets-write.md

@@ -0,0 +1,6 @@
+---
+"effect-workerd": patch
+"liminal": patch
+---
+
+Actor now inherits from DurableObject. Effect updated to beta 57.

effect-workerd/D1.ts

@@ -5,4 +5,4 @@ import * as Binding from "./Binding.ts"
 
 export class D1 extends Context.Service<D1, D1Database>()("effect-workerd/D1") {}
 
-export const layer = Binding.layer(D1, ["exec"], (db) => D1Client.layer({ db }))
+export const layer = Binding.layer(D1, ["prepare", "batch", "exec"], (db) => D1Client.layer({ db }))

effect-workerd/Kv.ts

@@ -41,9 +41,11 @@ export const Kv =
       decodeValue: S.decodeUnknownEffect(S.fromJsonString(S.toCodecJson(value))),
     }
 
-    const layer = Binding.layer(tag, ["put", "get", "delete"])
-
-    return Object.assign(tag, { definition, transcoders, layer })
+    return Object.assign(tag, {
+      definition,
+      transcoders,
+      layer: Binding.layer(tag, ["get", "put", "delete", "list", "getWithMetadata"]),
+    })
   }
 
 export const put = Effect.fnUntraced(function* <Self, Id extends string, D extends KvDefinition>(

effect-workerd/Worker.ts

@@ -26,18 +26,29 @@ export interface WorkerDefinition<PreludeROut, PreludeE, E> {
 }
 
 export const make = <PreludeROut, PreludeE, E>({ handler, prelude }: WorkerDefinition<PreludeROut, PreludeE, E>) => {
-  const runtime = ManagedRuntime.make(
-    Layer.mergeAll(FetchHttpClient.layer, ConfigProvider.layer(ConfigProvider.fromUnknown(env))),
-  )
-
-  const fetch = (request: Request, _env: unknown, ctx: globalThis.ExecutionContext): Promise<Response> =>
-    handler.pipe(
+  let runtime:
+    | undefined
+    | ManagedRuntime.ManagedRuntime<
+        PreludeROut | Layer.Success<typeof HttpServer.layerServices> | HttpClient.HttpClient,
+        PreludeE
+      >
+  const fetch = (request: Request, _env: unknown, ctx: globalThis.ExecutionContext): Promise<Response> => {
+    runtime ??= ManagedRuntime.make(
+      prelude.pipe(
+        Layer.provideMerge(
+          Layer.mergeAll(
+            FetchHttpClient.layer,
+            ConfigProvider.layer(ConfigProvider.fromUnknown(env)),
+            HttpServer.layerServices,
+          ),
+        ),
+      ),
+    )
+    return handler.pipe(
       Effect.tapCause(logCause),
       Effect.catchCause(() => Effect.succeed(HttpServerResponse.empty({ status: 500 }))),
       Effect.map(HttpServerResponse.toWeb),
       Effect.provide([
-        prelude,
-        HttpServer.layerServices,
         Layer.succeed(ExecutionContext, ctx),
         Layer.succeed(NativeRequest, request),
         Layer.succeed(HttpServerRequest.HttpServerRequest, HttpServerRequest.fromWeb(request)),
@@ -51,6 +62,7 @@ export const make = <PreludeROut, PreludeE, E>({ handler, prelude }: WorkerDefin
       Effect.provideService(Layer.CurrentMemoMap, runtime.memoMap),
       runtime.runPromise,
     )
+  }
 
   return { fetch }
 }

konfik

@@ -1,3 +1,4 @@
+import { DurableObject } from "cloudflare:workers"
 import {
   Layer,
   Effect,
@@ -12,6 +13,7 @@ import {
   Array,
   Encoding,
   Option,
+  Cause,
 } from "effect"
 import { Binding, DoState, NativeRequest } from "effect-workerd"
 import { SecWebSocketProtocol, close } from "effect-workerd/socket_util"
@@ -91,8 +93,10 @@ export interface ActorNamespace<
   PreludeE,
   RunROut,
   RunE,
-> extends Context.Service<NamespaceSelf, DurableObjectNamespace> {
-  new (state: globalThis.DurableObjectState<{}>): Context.ServiceClass.Shape<NamespaceId, DurableObjectNamespace>
+> {
+  new (state: DurableObjectState<{}>, env: Cloudflare.Env): DurableObject
+
+  readonly service: Context.ServiceClass<NamespaceSelf, NamespaceId, DurableObjectNamespace>
 
   readonly definition: ActorNamespaceDefinition<
     ActorSelf,
@@ -111,7 +115,14 @@ export interface ActorNamespace<
   readonly upgrade: (
     name: Name["Type"],
     attachments: S.Struct<AttachmentFields>["Type"],
-  ) => Effect.Effect<HttpServerResponse.HttpServerResponse, S.SchemaError, NamespaceSelf | NativeRequest.NativeRequest>
+  ) => Effect.Effect<
+    HttpServerResponse.HttpServerResponse,
+    S.SchemaError | Encoding.EncodingError | Cause.NoSuchElementError,
+    | NamespaceSelf
+    | NativeRequest.NativeRequest
+    | Name["EncodingServices"]
+    | S.Struct<AttachmentFields>["EncodingServices"]
+  >
 
   readonly layer: (binding: string) => Layer.Layer<NamespaceSelf, S.SchemaError, never>
 }
@@ -161,7 +172,7 @@ export const Service =
     RunROut,
     RunE
   > => {
-    const { hibernation, actor, prelude, handlers, layer: runLayer, onConnect } = definition
+    const { hibernation, actor, prelude, handlers, layer, onConnect } = definition
     const {
       definition: {
         name: Name,
@@ -171,7 +182,6 @@ export const Service =
     } = actor
 
     const encodeName = S.encodeEffect(Name)
-    const decodeName = S.decodeUnknownEffect(Name)
 
     const Attachments = S.Struct(AttachmentFields)
     const encodeAttachments = S.encodeEffect(S.toCodecJson(Attachments))
@@ -194,63 +204,56 @@ export const Service =
         encodeAttachments(attachments).pipe(Effect.andThen((v) => Effect.sync(() => socket.serializeAttachment(v)))),
     }
 
-    const tag = class tag extends Context.Service<NamespaceSelf, DurableObjectNamespace>()(id) {
+    class NameDecoded extends Context.Service<NameDecoded, Name["Type"]>()(
+      "liminal/WorkerdActorNamespace/NameDecoded",
+    ) {}
+
+    return class extends DurableObject {
+      static definition = definition
+      static service = Context.Service<NamespaceSelf, DurableObjectNamespace>()(id)
+      static layer = Binding.layer(this.service, ["idFromName", "idFromString", "newUniqueId", "get"])
+
       readonly runtime
       readonly directory = ClientDirectory.make(actor, transport)
 
-      constructor(...args: [never]) {
-        super(...args)
-        const [state, env] = args as never as [state: globalThis.DurableObjectState<{}>, env: unknown]
-
+      constructor(state: DurableObjectState<{}>, env: Cloudflare.Env) {
+        super(state, env)
         if (hibernation) {
           Option.andThen(
             Duration.fromInput(hibernation),
             flow(Duration.toMillis, state.setHibernatableWebSocketEventTimeout),
           )
         }
 
-        const baseLayer = Layer.mergeAll(
-          prelude.pipe(Layer.provideMerge(ConfigProvider.layer(ConfigProvider.fromUnknown(env)))),
+        const Live = Layer.mergeAll(
           FetchHttpClient.layer,
           Layer.succeed(DoState.DoState, state),
           Mutex.layer,
-        )
+          Layer.effect(NameDecoded, S.decodeUnknownEffect(Name)(state.id.name)).pipe(
+            Layer.provideMerge(prelude.pipe(Layer.provideMerge(ConfigProvider.layer(ConfigProvider.fromUnknown(env))))),
+          ),
+        ).pipe(boundLayer("actor"))
 
-        this.runtime = Effect.gen({ self: this }, function* () {
-          this.#name = yield* Effect.tryPromise(() => state.storage.get("__liminal_name")).pipe(
-            Effect.flatMap((v) => (typeof v === "string" ? decodeName(v) : Effect.succeed(undefined))),
-          )
+        const hydrateAttachments = Effect.gen({ self: this }, function* () {
           for (const socket of state.getWebSockets()) {
             const attachments = yield* decodeAttachments(socket.deserializeAttachment())
             yield* this.directory.register(socket, attachments)
           }
-        }).pipe(
-          Effect.tapCause(logCause),
-          span("make_runtime"),
-          Layer.effectDiscard,
-          Layer.provideMerge(baseLayer),
-          boundLayer("actor"),
-          ManagedRuntime.make,
-        )
+        }).pipe(span("hydrateAttachments"), Effect.tapCause(logCause))
+
+        this.runtime = hydrateAttachments.pipe(Layer.effectDiscard, Layer.provideMerge(Live), ManagedRuntime.make)
       }
 
-      #name?: Name["Type"] | undefined
-      fetch(request: Request): Promise<Response> {
+      override fetch(request: Request): Promise<Response> {
         return Effect.gen({ self: this }, function* () {
           const url = new URL(request.url)
-          const name = yield* decodeName(url.searchParams.get("__liminal_name"))
           const attachments = yield* decodeAttachmentsString(url.searchParams.get("__liminal_attachments"))
-          if (!this.#name) {
-            this.#name = name
-            const state = yield* DoState.DoState
-            const encoded = yield* S.encodeEffect(Name)(name)
-            yield* Effect.promise(() => state.storage.put("__liminal_name", encoded))
-          }
           const { 0: webSocket, 1: server } = new WebSocketPair()
           const state = yield* DoState.DoState
           state.acceptWebSocket(server)
           server.send(yield* encodeAuditionSuccess({ _tag: "Audition.Success" }))
           const currentClient = yield* this.directory.register(server, attachments)
+          const name = yield* NameDecoded
           const ActorLive = Layer.succeed(actor, {
             name,
             clients: this.directory.handles,
@@ -260,7 +263,7 @@ export const Service =
             Effect.scoped,
             span("onConnect"),
             Effect.scoped,
-            Effect.provide(Layer.provideMerge(runLayer, ActorLive)),
+            Effect.provide(Layer.provideMerge(layer, ActorLive)),
           )
           yield* debug("ClientRegistered")
           return new Response(null, {
@@ -271,10 +274,10 @@ export const Service =
         }).pipe(Effect.tapCause(logCause), span("fetch"), this.runtime.runPromise)
       }
 
-      webSocketMessage(socket: WebSocket, raw: string | ArrayBuffer) {
+      override webSocketMessage(socket: WebSocket, raw: string | ArrayBuffer) {
         Effect.gen({ self: this }, function* () {
           const currentClient = yield* this.directory.get(socket)
-          const name = yield* Effect.fromNullishOr(this.#name)
+          const name = yield* NameDecoded
           const ActorLive = Layer.succeed(actor, {
             name,
             clients: this.directory.handles,
@@ -308,54 +311,52 @@ export const Service =
             span("handler", { attributes: { _tag } }),
             Effect.andThen((v) => Effect.sync(() => socket.send(v))),
             Effect.scoped,
-            Effect.provide(Layer.provideMerge(runLayer, ActorLive)),
+            Effect.provide(Layer.provideMerge(layer, ActorLive)),
           )
         }).pipe(Effect.scoped, Mutex.task, Effect.tapCause(logCause), span("webSocketMessage"), this.runtime.runFork)
       }
 
-      webSocketClose(socket: WebSocket, _code: number, _reason: string, _wasClean: boolean) {
+      override webSocketClose(socket: WebSocket, _code: number, _reason: string, _wasClean: boolean) {
         this.directory
           .unregister(socket)
           .pipe(Effect.tap(debug("SocketClosed")), Effect.tapCause(logCause), this.runtime.runFork)
       }
 
-      webSocketError(socket: WebSocket, cause: unknown) {
+      override webSocketError(socket: WebSocket, cause: unknown) {
         Effect.gen({ self: this }, function* () {
           yield* debug("SocketErrored", { cause })
           yield* this.directory.unregister(socket)
         }).pipe(Effect.tapCause(logCause), span("SocketErrored", { attributes: { cause } }), this.runtime.runFork)
       }
-    }
-
-    const upgrade = Effect.fnUntraced(function* (name: Name["Type"], attachments: (typeof Attachments)["Type"]) {
-      yield* debug("UpgradeInitiated", { attachments })
-      const namespace = yield* tag
-      const nameEncoded = yield* encodeName(name)
-      const stub = namespace.getByName(nameEncoded)
-      const request = yield* NativeRequest.NativeRequest
-      const protocols = yield* Effect.fromNullishOr(request.headers.get(SecWebSocketProtocol)).pipe(
-        Effect.map(flow(String.split(","), Array.map(String.trim))),
-      )
-      const liminalTokenI = yield* Array.findFirstIndex(protocols, (v) => v === "liminal")
-      const requestClientId = yield* Effect.fromNullishOr(protocols[liminalTokenI + 1]).pipe(
-        Effect.flatMap((v) => Encoding.decodeBase64UrlString(v).asEffect()),
-      )
-      if (requestClientId !== clientId) {
-        return close(
-          yield* encodeAuditionFailure({
-            _tag: "Audition.Failure",
-            expected: clientId,
-            actual: requestClientId,
-          }),
-        )
-      }
-      const url = new URL(request.url)
-      url.searchParams.set("__liminal_name", nameEncoded)
-      url.searchParams.set("__liminal_attachments", yield* encodeAttachmentsString(attachments))
-      return yield* Effect.promise(() => stub.fetch(new Request(url, request))).pipe(Effect.map(HttpServerResponse.raw))
-    }, span("upgrade"))
 
-    const layer = Binding.layer(tag, ["getByName"])
-
-    return Object.assign(tag, { definition, upgrade, layer }) as never
+      static readonly upgrade = (name: Name["Type"], attachments: (typeof Attachments)["Type"]) =>
+        Effect.gen({ self: this }, function* () {
+          yield* debug("UpgradeInitiated", { attachments })
+          const namespace = yield* this.service
+          const nameEncoded = yield* encodeName(name)
+          const stub = namespace.getByName(nameEncoded)
+          const request = yield* NativeRequest.NativeRequest
+          const protocols = yield* Effect.fromNullishOr(request.headers.get(SecWebSocketProtocol)).pipe(
+            Effect.map(flow(String.split(","), Array.map(String.trim))),
+          )
+          const liminalTokenI = yield* Array.findFirstIndex(protocols, (v) => v === "liminal")
+          const requestClientId = yield* Effect.fromNullishOr(protocols[liminalTokenI + 1]).pipe(
+            Effect.flatMap((v) => Encoding.decodeBase64UrlString(v).asEffect()),
+          )
+          if (requestClientId !== clientId) {
+            return close(
+              yield* encodeAuditionFailure({
+                _tag: "Audition.Failure",
+                expected: clientId,
+                actual: requestClientId,
+              }),
+            )
+          }
+          const url = new URL(request.url)
+          url.searchParams.set("__liminal_attachments", yield* encodeAttachmentsString(attachments))
+          return yield* Effect.promise(() => stub.fetch(new Request(url, request))).pipe(
+            Effect.map(HttpServerResponse.raw),
+          )
+        }).pipe(span("upgrade"))
+    }
   }