diff --git a/README.md b/README.md index e0c4626..218e1f4 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,7 @@ Need to deploy your Worker to Cloudflare? Python Workers are in open beta and ha - [**`langchain/`**](langchain) — demonstrates how to use the [LangChain](https://pypi.org/project/langchain/) package with Python Workers. Currently broken. - [**`assets/`**](assets) — An example with an assets binding. - [**`durable-objects/`**](durable-objects) — An example with storing state in a [Durable Object](https://developers.cloudflare.com/durable-objects/). +- [**`hyperdrive-batched-counter/`**](hyperdrive-batched-counter) — batches high-frequency counters with Durable Objects and PostgreSQL through Hyperdrive. - [**`cron/`**](cron) — shows a simple [cron job](https://developers.cloudflare.com/workers/configuration/cron-triggers/). - [**`workers-ai/`**](workers-ai) makes a call [Workers AI](https://developers.cloudflare.com/workers-ai/) to run inference on Cloudflare's Global Network. - [**`vectorize-rag/`**](vectorize-rag) — a RAG example using remote Workers AI and [Vectorize](https://developers.cloudflare.com/vectorize/) bindings. diff --git a/hyperdrive-batched-counter/Dockerfile b/hyperdrive-batched-counter/Dockerfile new file mode 100644 index 0000000..215c228 --- /dev/null +++ b/hyperdrive-batched-counter/Dockerfile @@ -0,0 +1,3 @@ +FROM postgres:16 + +COPY schema.sql /docker-entrypoint-initdb.d/001-schema.sql diff --git a/hyperdrive-batched-counter/README.md b/hyperdrive-batched-counter/README.md new file mode 100644 index 0000000..b218c84 --- /dev/null +++ b/hyperdrive-batched-counter/README.md @@ -0,0 +1,44 @@ +# Batched Counter with Durable Objects and Hyperdrive + +This example shows how to use Durable Objects and Hyperdrive to build a unique counter +for multiple users. + +1. It assigns one Python Durable Object to each room. +2. Reactions that increase the count are kept in an in-memory dictionary for fast aggregation first, + and persisted to Durable Object storage until an alarm flushes them to PostgreSQL through Hyperdrive every five seconds. + +## Prerequisites + +- [Docker Compose](https://docs.docker.com/compose/) - for local PostgreSQL +- [uv](https://docs.astral.sh/uv/getting-started/installation/) + +## Run locally + +Start PostgreSQL, install the Python dependencies, then start the Worker: + +```sh +docker compose up --build -d --wait +uv run pywrangler dev +``` + +Open `http://localhost:8787` to join a room, send reactions, and watch each +five-second batch move from the Durable Object to PostgreSQL. + +```sh +curl -X POST http://localhost:8787/rooms/concert/reactions/heart +curl -X POST http://localhost:8787/rooms/concert/reactions/laugh +curl http://localhost:8787/rooms/concert/stats +``` + +The stats response separates accepted `totals`, PostgreSQL `persisted` totals, and the +currently buffered `pending` counts. + +## Deploy + +Apply `schema.sql` to the production PostgreSQL database first. Then create a Hyperdrive +configuration for that database, replace the placeholder `id` in `wrangler.jsonc` with +its Hyperdrive ID, and run: + +```sh +uv run pywrangler deploy +``` diff --git a/hyperdrive-batched-counter/compose.yaml b/hyperdrive-batched-counter/compose.yaml new file mode 100644 index 0000000..ad594be --- /dev/null +++ b/hyperdrive-batched-counter/compose.yaml @@ -0,0 +1,23 @@ +services: + postgres: + build: + context: . + dockerfile: Dockerfile + environment: + POSTGRES_USER: reactions + POSTGRES_PASSWORD: reactions + POSTGRES_DB: reactions + POSTGRES_HOST_AUTH_METHOD: md5 + POSTGRES_INITDB_ARGS: --auth-host=md5 + ports: + - "5432:5432" + volumes: + - reaction-data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U reactions -d reactions"] + interval: 5s + timeout: 5s + retries: 5 + +volumes: + reaction-data: diff --git a/hyperdrive-batched-counter/default.profraw b/hyperdrive-batched-counter/default.profraw new file mode 100644 index 0000000..e69de29 diff --git a/hyperdrive-batched-counter/package.json b/hyperdrive-batched-counter/package.json new file mode 100644 index 0000000..33fb9d5 --- /dev/null +++ b/hyperdrive-batched-counter/package.json @@ -0,0 +1,13 @@ +{ + "name": "batched-counter", + "version": "0.0.0", + "private": true, + "scripts": { + "deploy": "uv run pywrangler deploy", + "dev": "uv run pywrangler dev", + "start": "uv run pywrangler dev" + }, + "devDependencies": { + "wrangler": "^4.127.1" + } +} diff --git a/hyperdrive-batched-counter/public/app.js b/hyperdrive-batched-counter/public/app.js new file mode 100644 index 0000000..5e53293 --- /dev/null +++ b/hyperdrive-batched-counter/public/app.js @@ -0,0 +1,204 @@ +const reactions = ["heart", "laugh", "fire"]; +const roomPattern = /^[A-Za-z0-9_-]{1,64}$/; +const pollIntervalMs = 1000; + +const state = { + room: null, + stats: emptyStats(), + pollId: null, + statsController: null, + polling: false, + generation: 0, +}; + +const roomForm = document.querySelector("#room-form"); +const roomInput = document.querySelector("#room-input"); +const activeRoom = document.querySelector("#active-room"); +const connection = document.querySelector("#connection"); +const connectionText = document.querySelector("#connection-text"); +const lastSynced = document.querySelector("#last-synced"); +const totalAccepted = document.querySelector("#total-accepted"); +const statsBody = document.querySelector("#stats-body"); +const batchCycle = document.querySelector("#batch-cycle"); +const batchStatus = document.querySelector("#batch-status"); +const appStatus = document.querySelector("#app-status"); +const reactionButtons = [...document.querySelectorAll(".reaction-button")]; + +function emptyCounts() { + return Object.fromEntries(reactions.map((reaction) => [reaction, 0])); +} + +function emptyStats() { + return { totals: emptyCounts(), persisted: emptyCounts(), pending: emptyCounts() }; +} + +function count(value) { + const number = Number(value); + return Number.isFinite(number) ? Math.max(0, Math.trunc(number)) : 0; +} + +function normaliseCounts(counts) { + return Object.fromEntries(reactions.map((reaction) => [reaction, count(counts?.[reaction])])); +} + +function normaliseStats(stats) { + return { + totals: normaliseCounts(stats?.totals), + persisted: normaliseCounts(stats?.persisted), + pending: normaliseCounts(stats?.pending), + }; +} + +function endpoint(room, suffix) { + return `/rooms/${encodeURIComponent(room)}${suffix}`; +} + +function setStatus(message) { + appStatus.textContent = message; +} + +function setConnection(text, status) { + connectionText.textContent = text; + connection.dataset.state = status; +} + +function setControlsDisabled(disabled) { + reactionButtons.forEach((button) => { button.disabled = disabled; }); +} + +function renderStats() { + const accepted = reactions.reduce((total, reaction) => total + state.stats.totals[reaction], 0); + const pending = reactions.reduce((total, reaction) => total + state.stats.pending[reaction], 0); + totalAccepted.textContent = accepted.toLocaleString(); + + reactions.forEach((reaction) => { + const row = statsBody.querySelector(`[data-reaction="${reaction}"]`); + ["totals", "persisted", "pending"].forEach((field) => { + row.querySelector(`[data-field="${field}"]`).textContent = state.stats[field][reaction].toLocaleString(); + }); + }); + + batchCycle.classList.toggle("is-buffered", pending > 0); + batchStatus.textContent = pending > 0 + ? `${pending.toLocaleString()} reaction${pending === 1 ? "" : "s"} waiting for the next flush.` + : "No reactions waiting to be persisted."; +} + +async function requestJson(url, options = {}) { + const response = await fetch(url, options); + let data; + try { + data = await response.json(); + } catch { + throw new Error(`The server returned ${response.status} without JSON.`); + } + if (!response.ok) throw new Error(data?.error || `Request failed with status ${response.status}.`); + return data; +} + +function stopPolling() { + if (state.pollId !== null) window.clearInterval(state.pollId); + if (state.statsController !== null) state.statsController.abort(); + state.pollId = null; + state.statsController = null; + state.polling = false; +} + +async function loadStats(initial = false, generation = state.generation) { + const room = state.room; + if (!room || state.polling) return; + + state.polling = true; + const controller = new AbortController(); + state.statsController = controller; + if (initial) { + setConnection("Connecting…", "loading"); + setStatus(`Joining ${room} and loading its reaction ledger.`); + } + + try { + const stats = await requestJson(endpoint(room, "/stats"), { signal: controller.signal }); + if (state.room !== room || state.generation !== generation) return; + state.stats = normaliseStats(stats); + renderStats(); + lastSynced.textContent = `Last updated ${new Date().toLocaleTimeString()} · polling every second`; + setConnection("Live", "live"); + if (initial) setStatus(`Joined ${room}. Reaction controls are ready.`); + } catch (error) { + if (error.name !== "AbortError" && state.room === room && state.generation === generation) { + setConnection("Connection issue", "error"); + lastSynced.textContent = "Waiting to retry in the next poll."; + setStatus(`Could not load ${room}: ${error.message}`); + } + } finally { + if (state.statsController === controller) { + state.statsController = null; + state.polling = false; + } + } +} + +async function joinRoom(event) { + event.preventDefault(); + if (!roomForm.reportValidity()) return; + + const room = roomInput.value.trim(); + if (!roomPattern.test(room)) { + setStatus("Room names use 1–64 letters, numbers, hyphens, or underscores."); + return; + } + + stopPolling(); + state.generation += 1; + const generation = state.generation; + state.room = room; + state.stats = emptyStats(); + activeRoom.textContent = room; + lastSynced.textContent = "Loading room statistics…"; + renderStats(); + setControlsDisabled(false); + + state.pollId = window.setInterval(() => { + void loadStats(false, generation); + }, pollIntervalMs); + await loadStats(true, generation); +} + +async function sendReaction(reaction, button) { + const room = state.room; + if (!room || !reactions.includes(reaction)) return; + + button.disabled = true; + button.setAttribute("aria-busy", "true"); + setConnection("Sending reaction…", "loading"); + + try { + const result = await requestJson( + endpoint(room, `/reactions/${encodeURIComponent(reaction)}`), + { method: "POST" }, + ); + if (state.room !== room) return; + state.stats = { + ...state.stats, + totals: normaliseCounts(result.totals), + pending: normaliseCounts(result.pending), + }; + renderStats(); + setConnection("Live", "live"); + setStatus(`${reaction[0].toUpperCase()}${reaction.slice(1)} accepted in ${room}.`); + void loadStats(); + } catch (error) { + if (state.room === room) { + setConnection("Connection issue", "error"); + setStatus(`Could not send ${reaction}: ${error.message}`); + } + } finally { + button.removeAttribute("aria-busy"); + if (state.room === room) button.disabled = false; + } +} + +roomForm.addEventListener("submit", joinRoom); +reactionButtons.forEach((button) => { + button.addEventListener("click", () => { void sendReaction(button.dataset.reaction, button); }); +}); diff --git a/hyperdrive-batched-counter/public/index.html b/hyperdrive-batched-counter/public/index.html new file mode 100644 index 0000000..88a3cd3 --- /dev/null +++ b/hyperdrive-batched-counter/public/index.html @@ -0,0 +1,89 @@ + + + + + + + Reaction Counter + + + + + +
+
+
+

Reaction Counter

+
+
+ +
+
+

Channel selection

+

Tune into a room

+
+
+ +
+ + +
+
+

Not joined

+
+ +
+
+
+

On-air controls

+

Send a reaction

+
+

Room: —

+
+
+ + + +
+
+ +
+
+
+

Signal ledger

+

Reaction counts

+
+

Join a room to begin polling.

+
+
All accepted0
+
+ + + + + + + + +
Reaction totals, database-persisted counts, and pending counts
ReactionAcceptedPersistedPending
Heart———
Laugh———
Fire———
+
+
+ +
+
+
+

Batch runway

+
+

No reactions waiting to be persisted.

+
+ +
+ +

+ +
+ + diff --git a/hyperdrive-batched-counter/public/style.css b/hyperdrive-batched-counter/public/style.css new file mode 100644 index 0000000..3c3112d --- /dev/null +++ b/hyperdrive-batched-counter/public/style.css @@ -0,0 +1,109 @@ +:root { + --ink: #f6edcf; + --muted: #b7b09b; + --night: #090c12; + --panel: #181e29; + --panel-light: #222a37; + --line: #3b4657; + --amber: #ffc857; + --cyan: #72e2d1; + --coral: #ff7a6a; + --green: #8ed081; + --danger: #ff9b8d; + --shadow: rgb(0 0 0 / 35%); + --space-1: 0.25rem; + --space-2: 0.5rem; + --space-3: 0.75rem; + --space-4: 1rem; + --space-5: 1.5rem; + --space-6: 2rem; + --space-7: 3rem; + --radius: 0.75rem; + --line-width: 0.0625rem; + --content-width: 68rem; + --cycle-time: 5s; +} + +* { box-sizing: border-box; } + +body { + margin: 0; + color: var(--ink); + background: radial-gradient(circle at 12% 8%, rgb(255 200 87 / 16%), transparent 28rem), var(--night); + font-family: "Courier New", Courier, monospace; + line-height: 1.5; +} + +button, input { font: inherit; } +button { cursor: pointer; } +button:focus-visible, input:focus-visible { outline: var(--line-width) solid var(--amber); outline-offset: var(--space-1); } + +.control-room { width: min(calc(100% - var(--space-6)), var(--content-width)); margin: 0 auto; padding: var(--space-7) 0; } +.masthead { margin-bottom: var(--space-6); } +.masthead-row, .section-heading, .room-input-row { display: flex; gap: var(--space-4); align-items: end; justify-content: space-between; } +h1, h2, p { margin-top: 0; } +h1, h2 { font-family: Georgia, "Times New Roman", serif; font-weight: 400; } +h1 { margin-bottom: 0; font-size: clamp(2.75rem, 8vw, 5.5rem); line-height: 0.9; letter-spacing: -0.06em; } +h1 em { color: var(--amber); } +h2 { margin-bottom: 0; font-size: 1.35rem; } +.masthead-row > p { max-width: 25rem; margin-bottom: var(--space-1); color: var(--muted); } +.eyebrow { margin-bottom: var(--space-2); color: var(--cyan); font-size: 0.75rem; font-weight: 700; letter-spacing: 0.14em; text-transform: uppercase; } + +.panel { margin-bottom: var(--space-4); padding: var(--space-5); border: var(--line-width) solid var(--line); border-radius: var(--radius); background: linear-gradient(135deg, var(--panel-light), var(--panel)); box-shadow: var(--space-2) var(--space-2) 0 var(--shadow); } +.room-panel { display: grid; grid-template-columns: minmax(12rem, 1fr) minmax(16rem, 1.5fr) auto; gap: var(--space-5); align-items: center; } +.room-form label { display: block; margin-bottom: var(--space-1); color: var(--muted); font-size: 0.75rem; font-weight: 700; letter-spacing: 0.08em; text-transform: uppercase; } +.room-input-row { gap: var(--space-2); } +input { width: 100%; min-width: 0; padding: var(--space-3); color: var(--ink); border: var(--line-width) solid var(--line); border-radius: var(--space-1); background: var(--night); } +.join-button, .reaction-button { border: var(--line-width) solid transparent; border-radius: var(--space-1); font-weight: 700; transition: transform 160ms, background-color 160ms, border-color 160ms; } +.join-button { flex: 0 0 auto; padding: var(--space-3) var(--space-4); color: var(--night); background: var(--amber); } +.join-button:hover { transform: translateY(calc(var(--space-1) * -1)); background: var(--ink); } +.help-text, .room-readout, .sync-note { margin: var(--space-2) 0 0; color: var(--muted); font-size: 0.75rem; } +.connection { display: flex; gap: var(--space-2); align-items: center; margin: 0; color: var(--muted); font-size: 0.875rem; white-space: nowrap; } +.connection > span:first-child { width: var(--space-2); height: var(--space-2); border-radius: 50%; background: currentColor; } +.connection[data-state="live"] { color: var(--green); } +.connection[data-state="error"] { color: var(--danger); } + +.reaction-controls { display: grid; grid-template-columns: repeat(3, 1fr); gap: var(--space-3); margin-top: var(--space-5); } +.reaction-button { display: grid; gap: var(--space-2); min-height: 8rem; padding: var(--space-4); color: var(--ink); border-color: var(--line); background: var(--night); } +.reaction-button > span { font-size: clamp(2rem, 7vw, 3.5rem); line-height: 1; } +.reaction-button:hover:not(:disabled) { transform: translateY(calc(var(--space-1) * -1)); border-color: var(--amber); background: var(--panel-light); } +.reaction-button:disabled { cursor: not-allowed; opacity: 0.45; } +.reaction-button[data-reaction="heart"] > span { color: var(--coral); } +.reaction-button[data-reaction="laugh"] > span { color: var(--amber); } +.reaction-button[data-reaction="fire"] > span { color: var(--cyan); } + +.total-readout { display: flex; gap: var(--space-3); align-items: baseline; margin: var(--space-5) 0; padding-bottom: var(--space-4); border-bottom: var(--line-width) solid var(--line); } +.total-readout span { color: var(--muted); font-size: 0.875rem; } +.total-readout strong { color: var(--amber); font-family: Georgia, "Times New Roman", serif; font-size: clamp(3rem, 9vw, 5rem); font-weight: 400; line-height: 0.9; } +.table-wrap { overflow-x: auto; } +table { width: 100%; border-collapse: collapse; font-variant-numeric: tabular-nums; } +th, td { padding: var(--space-3); text-align: right; border-bottom: var(--line-width) solid var(--line); } +th:first-child, td:first-child { text-align: left; } +thead { color: var(--muted); font-size: 0.75rem; letter-spacing: 0.06em; text-transform: uppercase; } +tbody th { font-weight: 700; } +td[data-field="totals"] { color: var(--amber); } +td[data-field="persisted"] { color: var(--cyan); } +td[data-field="pending"] { color: var(--coral); } + +.batch-cycle { background: linear-gradient(135deg, var(--panel), var(--night)); } +.cycle-track { position: relative; display: grid; grid-template-columns: repeat(5, 1fr); gap: var(--space-2); height: var(--space-7); margin: var(--space-5) 0 var(--space-3); padding: var(--space-2); overflow: hidden; border: var(--line-width) solid var(--line); border-radius: var(--space-1); background: var(--night); } +.cycle-track > span:not(.cycle-sweep) { z-index: 1; display: grid; place-items: center; color: var(--muted); border-left: var(--line-width) solid var(--line); font-size: 0.75rem; } +.cycle-track > span:nth-child(2) { border-left: 0; } +.cycle-sweep { position: absolute; inset: 0 auto 0 0; width: 20%; opacity: 0; background: linear-gradient(90deg, transparent, var(--coral), transparent); transform: translateX(-100%); } +.batch-cycle.is-buffered .cycle-sweep { opacity: 0.8; animation: batch-sweep var(--cycle-time) linear infinite; } +.app-status { min-height: var(--space-5); margin: var(--space-4) 0 0; color: var(--muted); font-size: 0.875rem; text-align: center; } +.sr-only { position: absolute; width: var(--space-1); height: var(--space-1); padding: 0; margin: calc(var(--space-1) * -1); overflow: hidden; clip: rect(0, 0, 0, 0); white-space: nowrap; border: 0; } + +@keyframes batch-sweep { to { transform: translateX(500%); } } + +@media (max-width: 42rem) { + .control-room { width: min(calc(100% - var(--space-4)), var(--content-width)); padding: var(--space-6) 0; } + .masthead-row, .section-heading, .room-panel { display: grid; grid-template-columns: 1fr; align-items: start; } + .reaction-controls { gap: var(--space-2); } + .reaction-button { min-height: 6rem; padding: var(--space-3); } + th, td { padding: var(--space-2); font-size: 0.875rem; } +} + +@media (prefers-reduced-motion: reduce) { + *, *::before, *::after { scroll-behavior: auto; transition-duration: 0ms; animation-duration: 0ms; animation-iteration-count: 1; } +} diff --git a/hyperdrive-batched-counter/pyproject.toml b/hyperdrive-batched-counter/pyproject.toml new file mode 100644 index 0000000..e3dd87e --- /dev/null +++ b/hyperdrive-batched-counter/pyproject.toml @@ -0,0 +1,16 @@ +[project] +name = "batched-counter" +version = "0.0.0" +description = "Batched counters with Durable Objects and Hyperdrive" +readme = "README.md" +requires-python = ">=3.14" +dependencies = [ + "pg8000", + "starlette>=0.48.0" +] + +[dependency-groups] +dev = [ + "workers-py", + "workers-runtime-sdk" +] diff --git a/hyperdrive-batched-counter/schema.sql b/hyperdrive-batched-counter/schema.sql new file mode 100644 index 0000000..92f3016 --- /dev/null +++ b/hyperdrive-batched-counter/schema.sql @@ -0,0 +1,9 @@ +SET password_encryption = 'md5'; + +CREATE TABLE reaction_stats ( + room_id TEXT NOT NULL, + reaction TEXT NOT NULL CHECK (reaction IN ('heart', 'laugh', 'fire')), + count BIGINT NOT NULL DEFAULT 0 CHECK (count >= 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (room_id, reaction) +); diff --git a/hyperdrive-batched-counter/src/entry.py b/hyperdrive-batched-counter/src/entry.py new file mode 100644 index 0000000..fa9e9cd --- /dev/null +++ b/hyperdrive-batched-counter/src/entry.py @@ -0,0 +1,38 @@ +# Side effect: register the Durable Object class with the Workers runtime +from room import ReactionRoom # noqa: F401 +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.routing import Route +from workers import asgi + + +def room_stub(request: Request): + room = request.path_params["room"] + rooms = request.scope["env"].REACTION_ROOMS + return room, rooms.get(rooms.idFromName(room)) + + +async def add_reaction(request: Request): + room, stub = room_stub(request) + reaction = request.path_params["reaction"] + return JSONResponse(await stub.add_reaction(room, reaction), status_code=202) + + +async def get_stats(request: Request): + room, stub = room_stub(request) + return JSONResponse(await stub.get_stats(room)) + + +app = Starlette( + routes=[ + Route( + "/rooms/{room}/reactions/{reaction}", + add_reaction, + methods=["POST"], + ), + Route("/rooms/{room}/stats", get_stats, methods=["GET"]), + ] +) + +Default = asgi.entrypoint(app) diff --git a/hyperdrive-batched-counter/src/room.py b/hyperdrive-batched-counter/src/room.py new file mode 100644 index 0000000..cb12fa1 --- /dev/null +++ b/hyperdrive-batched-counter/src/room.py @@ -0,0 +1,169 @@ +from contextlib import closing +from datetime import UTC, datetime, timedelta + +import pg8000 +from workers import DurableObject + +REACTIONS = ("heart", "laugh", "fire") +PENDING_KEY = "pending_reactions" +TOTALS_KEY = "total_reactions" +ROOM_KEY = "room_id" +FLUSH_DELAY = timedelta(seconds=5) + + +class ReactionRoom(DurableObject): + """ + A Durable Object that tracks reaction counts for a chat room. + + Methods: + add_reaction: Add a reaction to the room + get_stats: Get the current stats for the room + """ + + def __init__(self, state, env): + super().__init__(state, env) + self.env = env + # pending: reactions that have not been flushed to the database yet + self.pending: dict[str, int] = dict.fromkeys(REACTIONS, 0) + # totals: total reactions for the room + self.totals: dict[str, int] = dict.fromkeys(REACTIONS, 0) + self.room_id = None + self.loaded = False + + self.ctx.blockConcurrencyWhile(self._load_pending) + + async def add_reaction(self, room_id, reaction): + if self.room_id is not None and self.room_id != room_id: + raise ValueError("Room ID does not match this Durable Object") + + next_pending = self.pending.copy() + next_totals = self.totals.copy() + next_pending[reaction] += 1 + next_totals[reaction] += 1 + self.room_id = room_id + self.pending = next_pending + self.totals = next_totals + + await self._schedule_alarm_if_needed() + await self._save_state(room_id, next_pending, next_totals) + + return { + "accepted": reaction, + "totals": self.totals.copy(), + "pending": self.pending.copy(), + } + + async def get_stats(self, room_id): + persisted = dict.fromkeys(REACTIONS, 0) + with self._connection() as connection: + cursor = connection.cursor() + cursor.execute( + "SELECT reaction, count FROM reaction_stats WHERE room_id = %s", + (room_id,), + ) + for reaction, count in cursor.fetchall(): + persisted[reaction] = int(count) + return { + "totals": self.totals.copy(), + "persisted": persisted, + "pending": self.pending.copy(), + } + + async def _load_pending(self): + """ + If the durable object is removed and recreated, the state will be lost. + Load the data from persistent storage. + """ + if self.loaded: + return + + stored = await self.ctx.storage.get([PENDING_KEY, TOTALS_KEY, ROOM_KEY]) + pending = stored.get(PENDING_KEY, {}) + totals = stored.get(TOTALS_KEY, {}) + self.pending = { + reaction: int(pending.get(reaction, 0)) for reaction in REACTIONS + } + self.totals = {reaction: int(totals.get(reaction, 0)) for reaction in REACTIONS} + self.room_id = stored.get(ROOM_KEY) + self.loaded = True + + async def _save_state(self, room_id, pending, totals): + await self.ctx.storage.put( + { + ROOM_KEY: room_id, + PENDING_KEY: pending, + TOTALS_KEY: totals, + } + ) + + async def _schedule_alarm_if_needed(self): + if await self.ctx.storage.getAlarm() is None: + await self.ctx.storage.setAlarm(datetime.now(UTC) + FLUSH_DELAY) + + async def _schedule_retry(self): + await self.ctx.storage.setAlarm(datetime.now(UTC) + FLUSH_DELAY) + + def _connection(self): + hd = self.env.HYPERDRIVE + return closing( + pg8000.connect( + host=hd.host, + port=int(hd.port), + user=hd.user, + password=hd.password, + database=hd.database, + ssl_context=False, + ) + ) + + async def alarm(self, alarm_info): + # Block concurrency while flushing to prevent race conditions + self.ctx.blockConcurrencyWhile(self._flush) + + async def _flush(self): + batch = self.pending.copy() + if not any(batch.values()): + return + totals = self.totals.copy() + + with self._connection() as connection: + try: + cursor = connection.cursor() + cursor.execute( + """ + INSERT INTO reaction_stats (room_id, reaction, count) + VALUES (%s, %s, %s), (%s, %s, %s), (%s, %s, %s) + ON CONFLICT (room_id, reaction) DO UPDATE SET + count = EXCLUDED.count, + updated_at = CURRENT_TIMESTAMP + """, + ( + self.room_id, + "heart", + totals["heart"], + self.room_id, + "laugh", + totals["laugh"], + self.room_id, + "fire", + totals["fire"], + ), + ) + connection.commit() + except Exception as error: + try: + connection.rollback() + except Exception as rollback_error: + print(f"Reaction rollback failed: {rollback_error}") + print(f"Reaction flush failed: {error}") + await self._schedule_retry() + return + + next_pending = { + reaction: self.pending[reaction] - count + for reaction, count in batch.items() + } + await self._save_state(self.room_id, next_pending, self.totals) + self.pending = next_pending + if any(self.pending.values()): + await self._schedule_alarm_if_needed() diff --git a/hyperdrive-batched-counter/wrangler.jsonc b/hyperdrive-batched-counter/wrangler.jsonc new file mode 100644 index 0000000..fe5debc --- /dev/null +++ b/hyperdrive-batched-counter/wrangler.jsonc @@ -0,0 +1,42 @@ +{ + "$schema": "node_modules/wrangler/config-schema.json", + "name": "batched-counter", + "main": "src/entry.py", + "compatibility_date": "2026-09-02", + "compatibility_flags": [ + "python_workers", + "python_workers_314" + ], + "durable_objects": { + "bindings": [ + { + "name": "REACTION_ROOMS", + "class_name": "ReactionRoom" + } + ] + }, + "migrations": [ + { + "tag": "v1", + "new_sqlite_classes": [ + "ReactionRoom" + ] + } + ], + "hyperdrive": [ + { + "binding": "HYPERDRIVE", + "id": "00000000-0000-0000-0000-000000000001", + "localConnectionString": "postgres://reactions:reactions@localhost:5432/reactions" + } + ], + "assets": { + "directory": "./public", + "run_worker_first": [ + "/rooms/*" + ] + }, + "observability": { + "enabled": true + } +}