Effect RPC with Cloudflare Workers

config.ts

import { Effect as E, Schema } from "@effect/schema"
import {
  BooleanFromSelfOrString,
  NonEmptyObject,
  ObjectValue,
} from "@o/shared"
import { ConfigProvider, Context, Layer } from "effect"

const WorkerBinding = <T extends object>() =>
  NonEmptyObject.pipe(
    Schema.filter((input): input is T => true, { title: "WorkerBinding" }),
  )

const make = (env: unknown) => {
  const get = <A, I>(key: string, schema: Schema.Schema<A, I>) =>
    ObjectValue(key, schema)
      .annotations({ title: "WorkerConfig" })
      .pipe(Schema.decodeUnknownSync)(env)

  return {
    get: () => {},
    serviceBinding: <T extends object>(key: string) =>
      get(key, WorkerBinding<T>()),
    boolean: (key: string) => get(key, BooleanFromSelfOrString),
  }
}

export class WorkerConfig extends Context.Tag("WorkerConfig")<
  WorkerConfig,
  ReturnType<typeof make>
>() {
  static Live = (env: unknown) => {
    const configProvider = ConfigProvider.fromJson(env)
    const configLayer = Layer.setConfigProvider(configProvider)

    return Layer.effect(this, E.succeed(make(env))).pipe(
      Layer.provide(configLayer),
    )
  }
}

db.ts

import { Statement } from "@effect/sql"
import { D1Client } from "@effect/sql-d1"
import * as SqliteDrizzle from "@effect/sql-drizzle/Sqlite"
import { Config, Effect as E, Layer } from "effect"
import { WorkerConfig } from "./config"

const D1Live = Layer.unwrapEffect(
  E.gen(function* () {
    const config = yield* WorkerConfig
    const db = config.serviceBinding<D1Database>("DB")
    const enableDbResultDebug = config.boolean("ENABLE_DB_RESULT_DEBUG")

    const SqlLogger = Statement.setTransformer((statement) =>
      E.gen(function* () {
        const [query, params] = statement.compile()
        if (enableDbResultDebug)
          yield* E.void.pipe(
            E.withSpan("db.params", { attributes: { query, params } }),
          )
        return statement
      }),
    )

    return D1Client.layer({
      db: Config.succeed(db),
    }).pipe(Layer.provide(SqlLogger))
  }),
)

export const DatabaseLive = SqliteDrizzle.layer.pipe(Layer.provide(D1Live))

effect-dev-tools.ts

import { DevTools } from "@effect/experimental"
import { Socket } from "@effect/platform"
import { Boolean, Context, Effect as E, Layer } from "effect"
import { WorkerConfig } from "./config"

const make = (ctx: ExecutionContext) =>
  Layer.unwrapEffect(
    E.gen(function* () {
      const config = yield* WorkerConfig
      const enable = config.boolean("ENABLE_EFFECT_DEV_TOOLS")

      return Boolean.match(enable, {
        onTrue: () => {
          ctx.waitUntil(E.sleep(100).pipe(E.runPromise))
          return DevTools.layerWebSocket().pipe(
            Layer.provide(Socket.layerWebSocketConstructorGlobal),
          )
        },
        onFalse: () => Layer.empty,
      })
    }),
  )

export class EffectDevTools extends Context.Tag("EffectDevTools")<
  EffectDevTools,
  ReturnType<typeof make>
>() {
  static Live = (ctx: ExecutionContext) =>
    Layer.effect(this, E.succeed(make(ctx)))
}

index.ts

import { HttpApp } from "@effect/platform"
import { Effect as E, Layer, Logger, LogLevel } from "effect"
import { WorkerConfig } from "./helpers/config"
import { DatabaseLive } from "./helpers/db"
import { EffectDevTools } from "./helpers/effect-dev-tools"
import { httpRouter } from "./rpc/router"

const worker: ExportedHandler<Record<string, unknown>> = {
  fetch: async (request, env, ctx) => {
    const MainLayer = DatabaseLive.pipe(
      Layer.tapErrorCause(E.logError),
      Layer.provide(Logger.minimumLogLevel(LogLevel.All)),
      Layer.provide(EffectDevTools.Live(ctx)),
      Layer.provide(WorkerConfig.Live(env)),
    )

    return HttpApp.toWebHandlerLayer(httpRouter, MainLayer).handler(request)
  },
}

export default worker

router.ts

import {
  HttpMiddleware,
  HttpRouter,
  HttpServerResponse,
} from "@effect/platform"
import { Router } from "@effect/rpc"
import { toHttpApp } from "@effect/rpc-http/HttpRouter"
import { Effect as E, flow } from "effect"
export { Schema as S } from "@effect/schema"

class FooRequest extends S.TaggedRequest<FooRequest>()("FooRequest", {
  payload: { id: S.String },
  success: S.String,
  failure: S.Never,
}) {}

export const MY_RPCS = [
  Rpc.effect(FooRequest, ({ id }) =>
    E.gen(function* () {
      return "Hello world " + id
    }),
  ),
] as const

export const httpRouter = HttpRouter.empty.pipe(
  HttpRouter.post(
    "/rpc",
    Router.make(MY_RPCS).pipe(toHttpApp),
  ),
  HttpRouter.options("*", HttpServerResponse.empty()),
  HttpRouter.use(
    flow(
      // TODO: enable cors
      HttpMiddleware.cors(),
      HttpMiddleware.logger,
    ),
  ),
)