Introducing the Effect SDK for Rivet Actors

import { Action, Actor } from "@rivetkit/effect";
import { Schema } from "effect";

// A typed error in the action's error channel
export class NegativeAmountError extends Schema.TaggedErrorClass<NegativeAmountError>()(
  "NegativeAmountError",
  { amount: Schema.Number, message: Schema.String },
) {}

export const Increment = Action.make("Increment", {
  payload: { amount: Schema.Number },
  success: Schema.Number,
  error: NegativeAmountError,
});

export const GetCount = Action.make("GetCount", { success: Schema.Number });

// The definition ties the actions together under one actor name.
export const Counter = Actor.make("Counter", {
  actions: [Increment, GetCount],
});

import { Actor, State } from "@rivetkit/effect";
import { Effect, Schema } from "effect";
import { Counter, NegativeAmountError } from "./api.ts";

export const CounterLive = Counter.toLayer(
  Effect.fnUntraced(function* ({ rawRivetkitContext, state }) {
    return Counter.of({
      Increment: Effect.fnUntraced(function* ({ payload }) {
        // Reject before mutating, so the error path leaves state untouched.
        // The failure is a value in the typed error channel, not a throw.
        if (payload.amount < 0) {
          return yield* new NegativeAmountError({
            amount: payload.amount,
            message: `increment amount ${payload.amount} must not be negative`,
          });
        }

        // Persisted state via a `SubscriptionRef`-like API
        const next = yield* State.updateAndGet(state, (current) => ({
          count: current.count + payload.amount,
        })).pipe(Effect.orDie);

        // Broadcast the new value to every connected client.
        rawRivetkitContext.broadcast("newCount", next.count);

        return next.count;
      }),
      GetCount: () => State.get(state).pipe(Effect.map((s) => s.count), Effect.orDie),
    });
  }),
  {
    state: {
      schema: Schema.Struct({ count: Schema.Number }),
      initialValue: () => ({ count: 0 }),
    },
    name: "Counter",
    icon: "calculator",
  },
);

import { Client } from "@rivetkit/effect";
import { Effect } from "effect";
import { Counter } from "./api.ts";

const program = Effect.gen(function* () {
  // A typed accessor: methods are your actions, fully typed
  const counter = (yield* Counter.client).getOrCreate("hello-world");

  yield* Effect.log(`count is now ${yield* counter.Increment({ amount: 1 })}`);
  yield* Effect.log(`count is now ${yield* counter.Increment({ amount: 5 })}`);

  // The declared error arrives as a real tagged instance, caught by tag.
  yield* counter.Increment({ amount: -1 }).pipe(
    Effect.catchTag("NegativeAmountError", (err) =>
      Effect.log(`rejected: ${err.message} (amount ${err.amount})`),
    ),
  );

  yield* Effect.log(`final count: ${yield* counter.GetCount()}`);
}).pipe(Effect.provide(Client.layer({ endpoint: "http://127.0.0.1:6420" })));

import { assert, layer } from "@effect/vitest";
import { Registry } from "@rivetkit/effect";
import { Effect, Layer } from "effect";
import { Counter } from "./api.ts";
import { CounterLive } from "./live.ts";

const TestLayer = Registry.test.pipe(
  Layer.provideMerge(CounterLive),
  Layer.provide(Registry.layer()),
);

layer(TestLayer)("counter", (it) => {
  it.effect("increments and reads the count back", () =>
    Effect.gen(function* () {
      const counter = (yield* Counter.client).getOrCreate("t-increment");
      assert.strictEqual(yield* counter.Increment({ amount: 1 }), 1);
      assert.strictEqual(yield* counter.Increment({ amount: 5 }), 6);
      assert.strictEqual(yield* counter.GetCount(), 6);
    }),
  );
});

Counter.toLayer(
	Effect.fnUntraced(function* ({ state }) {
		// Runs on sleep, not on every call
		yield* Effect.addFinalizer(() => Effect.log("actor sleeping"));

		// Background fiber scoped to the wake; cancelled on sleep
		yield* State.changes(state).pipe(
			Stream.runForEach((s) =>
				Effect.log("state changed", { count: s.count }),
			),
			Effect.forkScoped,
		);

		return Counter.of({
			/* handlers */
		});
	}),
	{ state },
);

// Atomic read-modify-write against persisted, single-writer state
const next = yield* State.updateAndGet(state, (s) => ({ count: s.count + 1 }));

// A live Stream of committed changes, backed by real persistence
yield* State.changes(state).pipe(Stream.runForEach(render), Effect.forkScoped);

```ts counter.test.ts
import { layer } from "@effect/vitest";

layer(TestLayer)("counter", (it) => {
	it.effect("increments", () =>
		Effect.gen(function* () {
			const counter = (yield* Counter.client).getOrCreate("t");
			assert.strictEqual(yield* counter.Increment({ amount: 5 }), 5);
		}),
	);
});

yield* room.SendMessage({ sender: "Mallory", text: "hello" }).pipe(
  Effect.catchTag("MemberNotInRoomError", (err) =>
    Effect.logWarning("message rejected", { name: err.name }),
  ),
);

yield* room.SendMessage({ sender: "Admin", text: "Deploying now" }).pipe(
  Effect.withSpan("broadcastAnnouncement", {
    attributes: { room: room.key },
  }),
);

// Inside ChatRoom's wake scope
const moderatorClient = yield* Moderator.client;

SendMessage: Effect.fnUntraced(function* ({ payload }) {
  const moderator = moderatorClient.getOrCreate([...address.key, "main"]);
  // BannedWordsError from Review flows through SendMessage's error channel
  yield* moderator.Review({ text: payload.text });
  // ...persist and broadcast
}),

import { db } from "rivetkit/db";

export const ChatRoomLive = ChatRoom.toLayer(
	Effect.fnUntraced(function* ({ rawRivetkitContext, state }) {
		return ChatRoom.of({
			GetHistory: () =>
				Effect.tryPromise(() =>
					rawRivetkitContext.db.execute(
						"SELECT id, sender, text FROM messages ORDER BY id",
					),
				).pipe(Effect.orDie),
		});
	}),
	{
		state,
		db: db({
			onMigrate: async (client) => {
				await client.execute(`
					CREATE TABLE IF NOT EXISTS messages (
						id INTEGER PRIMARY KEY AUTOINCREMENT,
						sender TEXT NOT NULL,
						text TEXT NOT NULL
					)
				`);
			},
		}),
	},
);

SendMessage: Effect.fnUntraced(function* ({ payload }) {
  yield* persist(payload);
  rawRivetkitContext.broadcast("newMessage", {
    sender: payload.sender,
    text: payload.text,
  });
}),

// Dispatch an action to run later, by name, with the same payload a client sends
rawRivetkitContext.schedule.after(1_000, "SendMessage", {
	sender: "Admin",
	text: `Welcome to the room, ${name}!`,
});

import { Actor, State } from "@rivetkit/effect";
import { Context, DateTime, Effect, Layer, Schema, Stream } from "effect";
import { db } from "rivetkit/db";
import { Moderator } from "../moderator/api.ts";
import { ChatRoom, MemberNotInRoomError } from "./api.ts";

// A custom service used like any other Effect program
export class RoomPolicy extends Context.Service<
	RoomPolicy,
	{
		readonly requireMember: (
			members: ReadonlyArray<{ readonly name: string }>,
			name: string,
		) => Effect.Effect<void, MemberNotInRoomError>;
	}
>()("RoomPolicy") {}

export const ChatRoomLive = ChatRoom.toLayer(
	Effect.fnUntraced(function* ({ rawRivetkitContext, state }) {
		const address = yield* Actor.CurrentAddress;
		const roomPolicy = yield* RoomPolicy;
		const moderatorClient = yield* Moderator.client;

		// Finalizers run on sleep
		yield* Effect.addFinalizer(() =>
			Effect.log("room sleeping", { actorId: address.actorId }),
		);

		// A live, durable view of state changes, cancelled on sleep
		yield* State.changes(state).pipe(
			Stream.runForEach((s) =>
				Effect.log("members", { count: s.members.length }),
			),
			Effect.forkScoped,
		);

		// Compose persisted state with a service-owned domain guard
		const ensureMember = (name: string) =>
			State.get(state).pipe(
				Effect.orDie,
				Effect.flatMap((s) =>
					roomPolicy.requireMember(s.members, name),
				),
			);

		return ChatRoom.of({
			Join: Effect.fnUntraced(function* ({ payload }) {
				const next = yield* State.updateAndGet(state, (s) => ({
					...s,
					members: [...s.members, { name: payload.name }],
				})).pipe(Effect.orDie);
				rawRivetkitContext.broadcast("memberJoined", {
					name: payload.name,
				});
				return { memberCount: next.members.length };
			}),
			SendMessage: Effect.fnUntraced(function* ({ payload }) {
				yield* ensureMember(payload.sender);

				// Actor-to-actor RPC. A BannedWordsError here flows through
				// SendMessage's declared error channel.
				const moderator = moderatorClient.getOrCreate([
					...address.key,
					"main",
				]);
				yield* moderator.Review({ text: payload.text });

				yield* Effect.tryPromise(() =>
					rawRivetkitContext.db.execute(
						"INSERT INTO messages (sender, text) VALUES (?, ?)",
						payload.sender,
						payload.text,
					),
				).pipe(Effect.orDie);

				rawRivetkitContext.broadcast("newMessage", payload);
			}),
			Archive: () => Effect.sync(() => rawRivetkitContext.destroy()),
		});
	}),
	{
		state: {
			schema: Schema.Struct({
				members: Schema.Array(Schema.Struct({ name: Schema.String })),
			}),
			initialValue: () => ({ members: [] }),
		},
		db: db({
			onMigrate: async (client) => {
				await client.execute(`
					CREATE TABLE IF NOT EXISTS messages (
						id INTEGER PRIMARY KEY AUTOINCREMENT,
						sender TEXT NOT NULL,
						text TEXT NOT NULL
					)
				`);
			},
		}),
		name: "Chat Room",
		icon: "comments",
	},
);

import { Actor, State } from "@rivetkit/effect";
import { Effect, Schema } from "effect";
import { LanguageModel } from "effect/unstable/ai";
import { Agent, Message } from "./api.ts";

export const AgentLive = Agent.toLayer(
  Effect.fnUntraced(function* ({ state }) {
    return Agent.of({
      SendMessage: Effect.fnUntraced(function* ({ payload }) {
        // Persist the user turn before calling the model, so the message
        // it replies to survives a restart mid-call.
        const history = yield* State.updateAndGet(state, (turns) => [
          ...turns,
          { role: "user", content: payload.content } satisfies Message,
        ]).pipe(Effect.orDie);

        // The handler requires the `LanguageModel` service but never builds
        // it. Sending the whole history on every call is the agent's memory.
        const response = yield* LanguageModel.generateText({
          prompt: toPrompt(history),
        }).pipe(Effect.orDie);

        // Persist the assistant turn.
        yield* State.update(state, (turns) => [
          ...turns,
          { role: "assistant", content: response.text } satisfies Message,
        ]).pipe(Effect.orDie);

        return response.text;
      }),
      GetHistory: () => State.get(state).pipe(Effect.orDie),
    });
  }),
  {
    // `Message` is Schema.Struct({ role, content }); state is the running log.
    state: { schema: Schema.Array(Message), initialValue: () => [] },
    name: "Agent",
    icon: "robot",
  },
);

Counter.toLayer(
	Effect.fnUntraced(function* ({ rawRivetkitContext, state }) {
		return Counter.of({
			Increment: Effect.fnUntraced(function* ({ payload }) {
				const next = yield* State.updateAndGet(state, (s) => ({
					count: s.count + payload.amount,
				}));
				rawRivetkitContext.broadcast("newCount", next.count); // untyped, full access
				return next.count;
			}),
		});
	}),
	{ state },
);

npm install rivetkit @rivetkit/effect effect @effect/platform-node