Deprecated State

import { actor } from "rivetkit";

// An actor that still uses the deprecated `state` and `connState` options
const counter = actor({
  state: { count: 0 },
  connState: { joinedAt: 0 },

  onConnect: (c, conn) => {
    conn.state.joinedAt = Date.now();
  },

  actions: {
    increment: (c) => {
      c.state.count += 1;
      return c.state.count;
    },
  },
});

npm install cbor-x

CREATE TABLE _rivet_actor_state (
    id    INTEGER PRIMARY KEY CHECK (id = 1),
    state BLOB NOT NULL
) STRICT

import { actor, UserError } from "rivetkit";
import { db } from "rivetkit/db";

const wallet = actor({
	db: db({
		onMigrate: async (db) => {
			await db.execute(`
        CREATE TABLE IF NOT EXISTS wallet (
          id INTEGER PRIMARY KEY CHECK (id = 1),
          balance INTEGER NOT NULL
        );
      `);
			await db.execute(`
        CREATE TABLE IF NOT EXISTS ledger (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          amount INTEGER NOT NULL,
          created_at INTEGER NOT NULL
        );
      `);
			await db.execute(
				"INSERT OR IGNORE INTO wallet (id, balance) VALUES (1, 0)",
			);
		},
	}),

	actions: {
		balance: async (c) => {
			const [{ balance }] = await c.db.execute<{ balance: number }>(
				"SELECT balance FROM wallet WHERE id = 1",
			);
			return balance;
		},

		// The check and the change run in one statement, so two withdrawals
		// can never both pass the check
		withdraw: async (c, amount: number) => {
			const rows = await c.db.execute<{ balance: number }>(
				"UPDATE wallet SET balance = balance - ? WHERE id = 1 AND balance >= ? RETURNING balance",
				amount,
				amount,
			);
			if (rows.length === 0) {
				throw new UserError("Insufficient balance", {
					code: "insufficient_balance",
				});
			}
			return rows[0].balance;
		},

		// Statements that must commit together run in one transaction
		deposit: async (c, amount: number) => {
			return await c.db.transaction(
				async (tx) => {
					await tx.execute(
						"INSERT INTO ledger (amount, created_at) VALUES (?, ?)",
						amount,
						Date.now(),
					);
					const [{ balance }] = await tx.execute<{ balance: number }>(
						"UPDATE wallet SET balance = balance + ? WHERE id = 1 RETURNING balance",
						amount,
					);
					return balance;
				},
				{ name: "deposit" },
			);
		},
	},
});

import { actor } from "rivetkit";
import { db, type RawAccess } from "rivetkit/db";

// A crash loses at most this much progress
const SNAPSHOT_INTERVAL_MS = 1_000;
const TICK_MS = 50;

type Ball = { id: string; x: number; y: number; vx: number; vy: number };

interface WorldVars {
	balls: Ball[];
	// `onWake` stores its loop here so `onSleep` can wait for it.
	loop: Promise<void> | null;
}

interface WorldContext {
	vars: WorldVars;
	db: RawAccess;
	aborted: boolean;
	abortSignal: AbortSignal;
}

const world = actor({
	db: db({
		onMigrate: async (db) => {
			await db.execute(`
        CREATE TABLE IF NOT EXISTS balls (
          id TEXT PRIMARY KEY,
          x REAL NOT NULL,
          y REAL NOT NULL,
          vx REAL NOT NULL,
          vy REAL NOT NULL
        );
      `);
		},
	}),

	// The live simulation runs in memory, starting from the last snapshot
	createVars: async (c): Promise<WorldVars> => ({
		balls: await c.db.execute<Ball>("SELECT id, x, y, vx, vy FROM balls"),
		loop: null,
	}),

	onWake: (c) => {
		c.vars.loop = runWorld(c).catch((error) => {
			c.log.error({ msg: "world loop failed", error });
		});
	},

	// `c.abortSignal` fires on shutdown, so the loop saves and exits on its own
	onSleep: async (c) => {
		await c.vars.loop;
	},

	actions: {
		addBall: (c, ball: Ball) => {
			c.vars.balls.push(ball);
		},
		balls: (c) => c.vars.balls,
	},
});

async function runWorld(c: WorldContext): Promise<void> {
	let lastSnapshotAt = Date.now();
	while (!c.aborted) {
		await delay(TICK_MS, c.abortSignal);
		for (const ball of c.vars.balls) {
			ball.x += ball.vx;
			ball.y += ball.vy;
		}

		if (Date.now() - lastSnapshotAt >= SNAPSHOT_INTERVAL_MS) {
			await saveSnapshot(c);
			lastSnapshotAt = Date.now();
		}
	}

	// Save once more on shutdown so a graceful stop loses nothing
	await saveSnapshot(c);
}

// Replace the snapshot in one transaction so a crash never leaves half of one
async function saveSnapshot(c: WorldContext): Promise<void> {
	await c.db.transaction(
		async (tx) => {
			await tx.execute("DELETE FROM balls");
			for (const ball of c.vars.balls) {
				await tx.execute(
					"INSERT INTO balls (id, x, y, vx, vy) VALUES (?, ?, ?, ?, ?)",
					ball.id,
					ball.x,
					ball.y,
					ball.vx,
					ball.vy,
				);
			}
		},
		{ name: "save-snapshot" },
	);
}

// Resolve after `ms`, or as soon as the actor starts shutting down
function delay(ms: number, signal: AbortSignal): Promise<void> {
	return new Promise((resolve) => {
		if (signal.aborted) {
			resolve();
			return;
		}

		const finish = () => {
			clearTimeout(timer);
			signal.removeEventListener("abort", finish);
			resolve();
		};
		const timer = setTimeout(finish, ms);
		signal.addEventListener("abort", finish, { once: true });
	});
}

import { actor } from "rivetkit";

interface Presence {
  name: string;
  joinedAt: number;
}

const room = actor({
  // Live connections are tracked in memory, keyed by connection ID
  createVars: () => ({ presence: new Map<string, Presence>() }),

  onConnect: (c, conn) => {
    const { name } = conn.params as { name: string };
    c.vars.presence.set(conn.id, { name, joinedAt: Date.now() });
  },

  onDisconnect: (c, conn) => {
    c.vars.presence.delete(conn.id);
  },

  actions: {
    whoAmI: (c) => c.vars.presence.get(c.conn.id),
    online: (c) => [...c.vars.presence.values()],
  },
});

import { actor } from "rivetkit";
import { db } from "rivetkit/db";

const room = actor({
  db: db({
    onMigrate: async (db) => {
      await db.execute(`
        CREATE TABLE IF NOT EXISTS presence (
          conn_id TEXT PRIMARY KEY,
          name TEXT NOT NULL,
          joined_at INTEGER NOT NULL
        );
      `);
    },
  }),

  // Hibernatable connections outlive the actor's memory, so store their data in SQLite
  onConnect: async (c, conn) => {
    const { name } = conn.params as { name: string };
    await c.db.execute(
      "INSERT OR REPLACE INTO presence (conn_id, name, joined_at) VALUES (?, ?, ?)",
      conn.id,
      name,
      Date.now(),
    );
  },

  onDisconnect: async (c, conn) => {
    await c.db.execute("DELETE FROM presence WHERE conn_id = ?", conn.id);
  },

  // A connection that closed while the actor slept may never reach
  // `onDisconnect`, so drop rows for connections that are no longer open
  onWake: async (c) => {
    const rows = (await c.db.execute("SELECT conn_id FROM presence")) as {
      conn_id: string;
    }[];
    for (const row of rows) {
      if (!c.conns.has(row.conn_id)) {
        await c.db.execute("DELETE FROM presence WHERE conn_id = ?", row.conn_id);
      }
    }
  },

  actions: {
    whoAmI: async (c) => {
      const rows = (await c.db.execute(
        "SELECT name, joined_at FROM presence WHERE conn_id = ?",
        c.conn.id,
      )) as { name: string; joined_at: number }[];
      return rows[0];
    },
  },
});