effect-cf

Effect-native Cloudflare primitives for Workers, Durable Objects, bindings, KV, D1, Queues, Email, Analytics Engine, Workflows, and Durable Object storage.

Install

effect-cf currently targets Effect 4 beta.

bun add effect-cf "effect@^4.0.0-beta.65"

pnpm add effect-cf "effect@^4.0.0-beta.65"

npm install effect-cf "effect@^4.0.0-beta.65"

Goal

Cloudflare APIs return promises and expose platform-specific bindings. effect-cf wraps those boundaries as Context, Layer, and Effect values so application code stays inside one managed Effect runtime.

Runtime creation belongs at Cloudflare entrypoints, not inside binding helpers.

Exports

• Worker - Worker entrypoint factory, request services, and typed Worker bindings
• DurableObject - Durable Object entrypoint factory and typed namespace helpers
• DurableObjectState / DurableObjectStorage - Effect wrappers for state, alarms, SQL, and embedded KV
• DurableObjectWebSocket - WebSocket upgrade helpers for Durable Objects
• Kv - typed KV namespace helper
• D1 - typed D1 database binding helper with an @effect/sql-d1 backed SQL layer
• R2 - typed R2 bucket binding helper with Effect-wrapped object and multipart operations
• Hyperdrive - typed Hyperdrive binding helper for connection strings and optional Postgres SQL integration
• Images - typed Cloudflare Images binding helper with transformation APIs and optional hosted image operations
• Email - typed Cloudflare Send Email binding helper for send_email bindings
• AnalyticsEngine - typed Cloudflare Analytics Engine write bindings and SQL API query helpers
• Queue - typed Queue producer/consumer tags plus client and error types
• Workflow - typed Workflow entrypoints, steps, starter clients, and instance types
• Rpc - Cloudflare RPC type helpers and scoped disposal utilities
• WorkerConfig - Effect Config helpers backed by Cloudflare env

Worker Example

import { Effect, Layer } from "effect";
import { HttpServerResponse } from "effect/unstable/http";
import { Worker } from "effect-cf";

export default Worker.make(Layer.empty, Effect.succeed(HttpServerResponse.text("ok")));

Durable Object Example

import { Effect, Layer, Schema as S } from "effect";
import { DurableObject, DurableObjectState } from "effect-cf";

export class Counter extends DurableObject.Tag<Counter>()("Counter", {
  get: DurableObject.method({ success: S.Number }),
}) {}

export const CounterDurableObject = Counter.make(Layer.empty, {
  rpc: {
    get: () =>
      Effect.gen(function* () {
        const state = yield* DurableObjectState.DurableObjectState;
        const row = yield* (yield* state.storage.sql.exec<{ count: number }>(
          "SELECT 0 AS count",
        )).one();
        return row.count;
      }),
  },
});

export const CounterLayer = Counter.layer({ binding: "COUNTER" });

export const readCounter = Effect.gen(function* () {
  const counters = yield* Counter;
  return yield* counters.byName("home").get();
});

Define Wrangler bindings and migrations in the consuming application. Durable Object namespace bindings are provided with YourObject.layer({ binding }), and consumers use const namespace = yield* YourObject.

Queue Example

Queues define the message contract once. The same class is the producer service tag and the consumer Worker definition: use YourQueue.layer({ binding }) to provide a Cloudflare producer binding from WorkerEnvironment, and use .make(...) for the consumer Worker entrypoint.

import { Effect, Layer, Schema as S } from "effect";
import { Queue } from "effect-cf";

class AvatarQueue extends Queue.Tag<AvatarQueue>()("AvatarQueue", {
  message: S.Struct({ userId: S.String, imageKey: S.String }),
}) {}

export const AvatarQueueConsumer = AvatarQueue.make(Layer.empty, {
  queue: (batch) =>
    Effect.gen(function* () {
      for (const message of batch.messages) {
        yield* Effect.logInfo("process avatar", message.body.userId);
        yield* message.ack;
      }
    }),
});

export const AvatarQueueLayer = AvatarQueue.layer({ binding: "AVATAR_QUEUE" });

export const enqueueAvatar = (userId: string, imageKey: string) =>
  Effect.gen(function* () {
    const queue = yield* AvatarQueue;
    yield* queue.send({ userId, imageKey });
  });

Producers should usually use const queue = yield* AvatarQueue and then call queue.send(...), queue.sendBatch(...), or queue.metrics(). The static AvatarQueue.send(...) helpers remain available for concise one-off calls.

Queue handlers run inline failures through Cloudflare's normal retry path. If background work scheduled with WorkerContext.waitUntil(...) should also make the batch retry, use WorkerContext.waitUntilPropagating(...) or waitUntil(..., { mode: "propagate" }); the default waitUntil mode observes and logs failures without rejecting the native waitUntil promise.

R2 Example

R2 bucket tags expose Cloudflare object operations as Effects and map nullable reads to Option.

import { Effect, Layer, Option } from "effect";
import { R2 } from "effect-cf";

class ArtifactBucket extends R2.Tag<ArtifactBucket>()("ArtifactBucket") {}

export const ArtifactBucketLayer = ArtifactBucket.layer({ binding: "ARTIFACT_BUCKET" });

export const writeArtifact = (key: string, body: string) =>
  Effect.gen(function* () {
    const bucket = yield* ArtifactBucket;
    yield* bucket.put(key, body, {
      httpMetadata: { contentType: "application/json" },
    });
  });

export const readArtifact = (key: string) =>
  Effect.gen(function* () {
    const bucket = yield* ArtifactBucket;
    const object = yield* bucket.get(key);
    return Option.isSome(object) ? yield* Effect.promise(() => object.value.text()) : undefined;
  });

Use createMultipartUpload(...) or resumeMultipartUpload(...) for large objects; returned upload handles wrap uploadPart, complete, and abort in Effect.

Hyperdrive Example

Hyperdrive tags expose the binding connectionString directly on the yielded service.

import { Effect } from "effect";
import { Hyperdrive } from "effect-cf";

class AppDatabase extends Hyperdrive.Tag<AppDatabase>()("AppDatabase") {}

export const AppDatabaseLayer = AppDatabase.layer({ binding: "HYPERDRIVE" });

export const databaseUrl = Effect.gen(function* () {
  const hyperdrive = yield* AppDatabase;
  return hyperdrive.connectionString;
});

If your Worker uses Postgres via @effect/sql-pg, install that driver and use the optional subpath integration:

import { Hyperdrive } from "effect-cf";
import * as HyperdrivePg from "effect-cf/hyperdrive-pg";

class AppDatabase extends Hyperdrive.Tag<AppDatabase>()("AppDatabase") {}

export const SqlLive = HyperdrivePg.layer(AppDatabase, { binding: "HYPERDRIVE" });

The Postgres integration builds an Effect PgClient from Hyperdrive's generated connection string with PgClient.makeClient. It intentionally does not expose app-side pool options; Hyperdrive manages the underlying database pool.

Images Example

Images tags expose info, input, optional hosted image operations, and composable transform/draw steps.

import { Effect } from "effect";
import { Images } from "effect-cf";

class AvatarImages extends Images.Tag<AvatarImages>()("AvatarImages") {}

export const AvatarImagesLayer = AvatarImages.layer({ binding: "IMAGES" });

export const resizeAvatar = (image: Images.ImageInputValue) =>
  Effect.gen(function* () {
    const images = yield* AvatarImages;
    const result = yield* images.process(
      Images.transform(Images.empty, { width: 256, height: 256 }),
      {
        stream: image,
        outputOptions: { format: "image/webp" },
      },
    );

    return yield* result.response;
  });

Email Example

Email tags expose Cloudflare Send Email bindings as Effect-wrapped send(...) operations.

import { Effect } from "effect";
import { Email } from "effect-cf";

class TransactionalEmail extends Email.Tag<TransactionalEmail>()("TransactionalEmail") {}

export const TransactionalEmailLayer = TransactionalEmail.layer({ binding: "EMAIL" });

export const sendWelcomeEmail = (to: string) =>
  Effect.gen(function* () {
    const email = yield* TransactionalEmail;

    return yield* email.send({
      from: { name: "Example", email: "[email protected]" },
      to,
      subject: "Welcome to Example",
      text: "Welcome to Example",
      html: "<p>Welcome to Example</p>",
    });
  });

Analytics Engine Example

Analytics Engine tags expose Cloudflare dataset bindings as Effect-wrapped writeDataPoint(...) operations. Writes use Cloudflare's native non-blocking runtime API and are validated against Workers Analytics Engine limits before reaching the binding.

import { Effect } from "effect";
import { AnalyticsEngine } from "effect-cf";

class RequestAnalytics extends AnalyticsEngine.Tag<RequestAnalytics>()("RequestAnalytics") {}

export const RequestAnalyticsLayer = RequestAnalytics.layer({
  binding: "REQUEST_ANALYTICS",
  write: {
    onInvalid: "error",
    batchSize: 100,
  },
});

export const recordPageView = (request: Request) =>
  Effect.gen(function* () {
    const analytics = yield* RequestAnalytics;
    const url = new URL(request.url);

    yield* analytics.writeDataPoint({
      indexes: [url.hostname],
      blobs: [url.pathname, request.headers.get("cf-connecting-country") ?? "unknown"],
      doubles: [1],
    });

    yield* analytics.writeBatch(
      [
        {
          indexes: [url.hostname],
          blobs: [url.pathname, "page-view"],
          doubles: [1],
        },
      ],
      { onInvalid: "drop" },
    );
  });

Define the dataset binding in the consuming Worker's wrangler.jsonc with analytics_engine_datasets, then provide RequestAnalytics.layer({ binding: "REQUEST_ANALYTICS" }) from the Worker layer. Invalid writes fail with AnalyticsEngineWriteValidationError by default. Use write: { onInvalid: "drop" } on the layer or { onInvalid: "drop" } per call when dropping invalid points is preferable.

Analytics Engine query tags wrap Cloudflare's HTTP SQL API. Configuration can stay in Effect Config, including a redacted API token, the outbound transport is an Effect HttpClient dependency, and rows can be decoded with Effect schemas at the query boundary. Use fetchLayerConfig(...) as shorthand when the platform fetch-backed client is enough.

import { Effect, Layer, Schema as S } from "effect";
import { FetchHttpClient } from "effect/unstable/http";
import { AnalyticsEngine, WorkerConfig } from "effect-cf";

class AnalyticsQuery extends AnalyticsEngine.QueryTag<AnalyticsQuery>()("AnalyticsQuery") {}

export const AnalyticsQueryLayer = AnalyticsQuery.layerConfig(
  AnalyticsEngine.queryConfig({
    accountId: WorkerConfig.string("CLOUDFLARE_ACCOUNT_ID"),
    apiToken: WorkerConfig.redacted("CLOUDFLARE_API_TOKEN"),
  }),
).pipe(Layer.provide(FetchHttpClient.layer));

const PageView = S.Struct({
  path: S.String,
  views: S.Number,
});

export const topPages = Effect.gen(function* () {
  const analytics = yield* AnalyticsQuery;

  return yield* analytics.queryRows(
    PageView,
    `
      SELECT blob1 AS path, SUM(_sample_interval) AS views
      FROM request_metrics
      GROUP BY path
      ORDER BY views DESC
      LIMIT 10
    `,
  );
});

Workflow Example

Workflow definitions type the payload and result. Runtime handlers can access Workflow.WorkflowEvent, use durable Workflow.step(...), and use normal binding services inside steps.

import { Effect, Layer, Schema as S } from "effect";
import { Workflow } from "effect-cf";

class ExportWorkflow extends Workflow.Tag<ExportWorkflow>()("ExportWorkflow", {
  payload: S.Struct({ segmentId: S.String }),
  result: S.Struct({ objectKey: S.String }),
}) {}

export const ExportWorkflowEntrypoint = ExportWorkflow.make(Layer.empty, {
  run: (payload) =>
    Effect.gen(function* () {
      const objectKey = yield* Workflow.step(
        "write-export",
        Effect.succeed(`exports/${payload.segmentId}.json`),
      );

      return { objectKey };
    }),
});

export const ExportWorkflowLayer = ExportWorkflow.layer({ binding: "EXPORT_WORKFLOW" });

Provide ExportWorkflow.layer({ binding: "EXPORT_WORKFLOW" }), then use const workflow = yield* ExportWorkflow or the static ExportWorkflow.create(...), createBatch(...), and get(...) helpers to start and inspect instances.

In definition-backed workflows, the payload argument is the typed decoded payload and is the source of truth. Workflow.WorkflowEvent.payload is also re-provided decoded for convenience; Workflow.WorkflowEvent.raw.payload remains the native Cloudflare event payload.

Durable Object WebSockets

Use initialize for work that should run each time Cloudflare loads a Durable Object instance into memory. Yield state.blockConcurrencyWhile(...) when later events should wait for initialize to finish. If work should happen only once for a Durable Object id, store a sentinel in Durable Object storage.

export const RoomLive = DurableObject.make(layer, {
  initialize: Effect.gen(function* () {
    const state = yield* DurableObjectState.DurableObjectState;
    yield* state.blockConcurrencyWhile(
      Effect.gen(function* () {
        yield* state.storage.put("loadedAt", Date.now());
      }),
    );
  }),
  fetch,
});

Durable Object application sockets should use the hibernation-compatible state API. Accept sockets with DurableObjectWebSocket.acceptUpgrade(...); do not call server.accept() or attach native message listeners in application code.

import { Effect, Schema as S } from "effect";
import { DurableObject, DurableObjectState, DurableObjectWebSocket, Worker } from "effect-cf";

const ConnectionAttachment = S.Struct({
  id: S.String,
  roomId: S.String,
});

const Attachments = DurableObjectWebSocket.attachment(ConnectionAttachment);

export const fetch = Effect.gen(function* () {
  const request = yield* Worker.NativeRequest;

  if (!Worker.isWebSocketUpgrade(request)) {
    return new Response("Expected WebSocket upgrade", { status: 426 });
  }

  const upgrade = yield* DurableObjectWebSocket.acceptUpgrade({ tags: ["room:general"] });
  yield* Attachments.serialize(upgrade.server, {
    id: crypto.randomUUID(),
    roomId: "general",
  });
  yield* upgrade.server.send(JSON.stringify({ type: "ready" }));

  return upgrade.response;
});

DurableWebSocket keeps the native socket available as socket.raw, while send, close, serializeAttachment, and deserializeAttachment return typed Effect failures. Use state.getWebSockets(tag) to retrieve wrapped sockets for broadcast and rehydration.

DurableObject.make lifecycle handlers receive wrapped sockets automatically:

export const RoomLive = DurableObject.make(layer, {
  webSocketMessage: (socket, message) =>
    Effect.gen(function* () {
      yield* socket.send(message);
    }),
});

Schema-backed attachments can rehydrate hibernated sockets:

const restored = yield* Attachments.rehydrate({
  tag: "room:general",
  onInvalid: "ignore-and-close",
});

for (const { socket, attachment } of restored) {
  yield* socket.send(`restored:${attachment.id}`).pipe(Effect.ignore);
}

Worker-to-Durable-Object forwarding should stay native so WebSocket upgrade responses are preserved:

if (Worker.isWebSocketUpgrade(request)) {
  const rooms = yield* ChatRoom;
  return yield* rooms.byName(roomId).fetch(request);
}

Use DurableObjectRpcWebSocket.layer(...) for Effect RPC-over-WebSocket transports. It owns protocol parsing and RPC client bookkeeping; use DurableWebSocket for general application sockets, rooms, presence, and broadcast flows.

License

MIT