[services] Next.js + Python Queues

services/cross-runtime-queues/.gitignore

@@ -0,0 +1,29 @@
+# frontend
+frontend/.next/
+frontend/node_modules/
+frontend/*.log
+frontend/*.tsbuildinfo
+frontend/next-env.d.ts
+frontend/package-lock.json
+
+# python
+backend/.venv/
+backend/**/__pycache__/
+backend/**/*.py[cod]
+backend/**/*.egg-info/
+backend/.pytest_cache/
+backend/.queue-results.sqlite3
+backend/.queue-results.sqlite3-*
+
+# environment files
+.env
+.env*.local
+frontend/.env
+frontend/.env*.local
+backend/.env
+backend/.env*.local
+
+# common
+.vercel/
+.DS_Store
+.vercel

services/cross-runtime-queues/README.md

@@ -0,0 +1,65 @@
+# Next.js + FastAPI Queues
+
+Minimal example showing [Vercel Services](https://vercel.com/docs/services)
+with [Vercel Queues](https://vercel.com/docs/queues):
+
+- `frontend` (Next.js) mounted at `/`
+- `backend` (FastAPI) mounted at `/api/python`
+
+It demonstrates:
+
+1. Next.js publishing a message for a Python subscriber
+2. Python publishing a message for a Next.js callback
+3. Next.js publishing one message to both runtimes
+
+## How it works
+
+Each producer creates a task before publishing a message. The queue delivers
+the message to the configured subscriber, which stores its result in
+[Vercel Runtime Cache](https://vercel.com/docs/runtime-cache). The frontend
+polls the FastAPI service until every expected subscriber has completed.
+
+The fanout example uses separate consumer groups for the Next.js callback and
+Python subscriber, so both receive a copy of the same message.
+
+## Project structure
+
+```txt
+cross-runtime-queues/
+├── backend/       # FastAPI producer, subscribers, and result store
+├── frontend/      # Next.js UI, producers, and queue callbacks
+└── vercel.json    # Service routing and Next.js queue triggers
+```
+
+## Services config
+
+Configuration in `vercel.json`:
+
+- routes `/(.*)` to `frontend`
+- routes `/api/python/(.*)` to `backend`
+- binds `backend` to `frontend` as `BACKEND_URL`
+- registers the Next.js callbacks as `queue/v2beta` triggers
+
+Python subscribers are configured in `backend/pyproject.toml`.
+
+## Run locally
+
+Install the frontend and Python dependencies:
+
+```bash
+cd frontend
+npm install
+cd ../backend
+python -m venv .venv
+source .venv/bin/activate
+pip install -e .
+cd ..
+```
+
+Start the development server:
+
+```bash
+npx vercel dev
+```
+
+The application is now available at `http://localhost:3000`.

services/cross-runtime-queues/backend/main.py

@@ -0,0 +1,109 @@
+from datetime import datetime, timezone
+from typing import Any, Literal
+from uuid import uuid4
+
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+from vercel.queue import send
+
+from store import result_store
+
+
+class DemoMessage(BaseModel):
+    text: str
+    sentAt: str | None = None
+
+
+class TaskRecord(BaseModel):
+    taskId: str
+    expectedConsumers: list[Literal["nextjs", "python"]]
+    messageId: str | None = None
+
+
+class Completion(BaseModel):
+    result: dict[str, Any]
+
+
+app = FastAPI(title="Cross-runtime Vercel Queues demo")
+
+
+def task_key(task_id: str) -> str:
+    return f"task:{task_id}"
+
+
+def completion_key(task_id: str, consumer: str) -> str:
+    return f"task:{task_id}:completion:{consumer}"
+
+
+@app.put("/api/python/tasks/{task_id}", status_code=204)
+async def put_task(task_id: str, task: TaskRecord) -> None:
+    if task.taskId != task_id:
+        raise HTTPException(status_code=400, detail="Task ID mismatch")
+    await result_store.set(task_key(task_id), task.model_dump())
+
+
+@app.delete("/api/python/tasks/{task_id}", status_code=204)
+async def delete_task(task_id: str) -> None:
+    await result_store.delete(task_key(task_id))
+
+
+@app.put(
+    "/api/python/tasks/{task_id}/completions/{consumer}",
+    status_code=204,
+)
+async def put_completion(
+    task_id: str,
+    consumer: Literal["nextjs", "python"],
+    completion: Completion,
+) -> None:
+    await result_store.set(
+        completion_key(task_id, consumer),
+        completion.model_dump(),
+    )
+
+
+@app.get("/api/python/tasks/{task_id}")
+async def get_task(task_id: str) -> dict[str, Any]:
+    task = await result_store.get(task_key(task_id))
+    if task is None:
+        raise HTTPException(status_code=404, detail="Task not found")
+
+    completions: dict[str, Any] = {}
+    for consumer in task["expectedConsumers"]:
+        completion = await result_store.get(completion_key(task_id, consumer))
+        if completion is not None:
+            completions[consumer] = completion
+
+    return {
+        **task,
+        "status": (
+            "completed"
+            if len(completions) == len(task["expectedConsumers"])
+            else "pending"
+        ),
+        "completions": completions,
+    }
+
+
+@app.post("/api/python/messages/python-to-next")
+async def send_to_next(message: DemoMessage) -> TaskRecord:
+    task_id = str(uuid4())
+    payload = message.model_dump()
+    payload["sentAt"] = payload["sentAt"] or datetime.now(timezone.utc).isoformat()
+    payload["taskId"] = task_id
+    task = TaskRecord(
+        taskId=task_id,
+        expectedConsumers=["nextjs"],
+    )
+
+    await result_store.set(task_key(task_id), task.model_dump())
+    try:
+        message_id = await send("demo-python-to-next", payload)
+    except Exception:
+        await result_store.delete(task_key(task_id))
+        raise
+
+    task.messageId = message_id
+    await result_store.set(task_key(task_id), task.model_dump())
+
+    return task

services/cross-runtime-queues/backend/pyproject.toml

@@ -0,0 +1,16 @@
+[project]
+name = "queues-python-backend"
+version = "0.1.0"
+requires-python = ">=3.11"
+dependencies = [
+  "fastapi",
+  "pydantic",
+  "vercel-cache",
+  "vercel-queue",
+]
+
+[tool.vercel]
+entrypoint = "main:app"
+
+[[tool.vercel.subscribers]]
+entrypoint = "subscribers"

services/cross-runtime-queues/backend/store.py

@@ -0,0 +1,102 @@
+import asyncio
+import json
+import os
+import sqlite3
+import time
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any, Iterator
+
+from vercel.cache import AsyncRuntimeCache
+
+RESULT_TTL_SECONDS = 3600
+LOCAL_DATABASE_PATH = Path(__file__).with_name(".queue-results.sqlite3")
+
+
+class ResultStore:
+    def __init__(self) -> None:
+        self._runtime_cache = AsyncRuntimeCache(namespace="cross-runtime-queues")
+        self._use_runtime_cache = bool(
+            os.getenv("VERCEL") and os.getenv("VERCEL_ENV") != "development"
+        )
+
+    async def set(self, key: str, value: dict[str, Any]) -> None:
+        if self._use_runtime_cache:
+            await self._runtime_cache.set(
+                key,
+                value,
+                {"ttl": RESULT_TTL_SECONDS},
+            )
+            return
+
+        await asyncio.to_thread(self._set_local, key, value)
+
+    async def get(self, key: str) -> dict[str, Any] | None:
+        if self._use_runtime_cache:
+            return await self._runtime_cache.get(key)
+
+        return await asyncio.to_thread(self._get_local, key)
+
+    async def delete(self, key: str) -> None:
+        if self._use_runtime_cache:
+            await self._runtime_cache.delete(key)
+            return
+
+        await asyncio.to_thread(self._delete_local, key)
+
+    @contextmanager
+    def _connect(self) -> Iterator[sqlite3.Connection]:
+        connection = sqlite3.connect(LOCAL_DATABASE_PATH, timeout=5)
+        try:
+            connection.execute(
+                """
+                CREATE TABLE IF NOT EXISTS queue_results (
+                    key TEXT PRIMARY KEY,
+                    value TEXT NOT NULL,
+                    expires_at REAL NOT NULL
+                )
+                """
+            )
+            yield connection
+            connection.commit()
+        finally:
+            connection.close()
+
+    def _set_local(self, key: str, value: dict[str, Any]) -> None:
+        with self._connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO queue_results (key, value, expires_at)
+                VALUES (?, ?, ?)
+                ON CONFLICT(key) DO UPDATE SET
+                    value = excluded.value,
+                    expires_at = excluded.expires_at
+                """,
+                (key, json.dumps(value), time.time() + RESULT_TTL_SECONDS),
+            )
+
+    def _get_local(self, key: str) -> dict[str, Any] | None:
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT value, expires_at FROM queue_results WHERE key = ?",
+                (key,),
+            ).fetchone()
+            if row is None:
+                return None
+
+            value, expires_at = row
+            if expires_at <= time.time():
+                connection.execute(
+                    "DELETE FROM queue_results WHERE key = ?",
+                    (key,),
+                )
+                return None
+
+            return json.loads(value)
+
+    def _delete_local(self, key: str) -> None:
+        with self._connect() as connection:
+            connection.execute("DELETE FROM queue_results WHERE key = ?", (key,))
+
+
+result_store = ResultStore()

services/cross-runtime-queues/backend/subscribers.py

@@ -0,0 +1,50 @@
+from typing import Any
+
+from vercel.queue import Message, subscribe
+
+from store import result_store
+
+
+async def complete_task(
+    message: Message[dict[str, Any]],
+    consumer: str,
+) -> None:
+    task_id = str(message.payload["taskId"])
+    await result_store.set(
+        f"task:{task_id}:completion:{consumer}",
+        {
+            "result": {
+                "messageId": message.message_id,
+                "deliveryCount": message.metadata.delivery_count,
+                "received": message.payload,
+            },
+        },
+    )
+
+
+@subscribe(topic="demo-next-to-python")
+async def receive_from_next(message: Message[dict[str, Any]]) -> None:
+    await complete_task(message, "python")
+    print(
+        "[python consumer] received from Next.js",
+        {
+            "payload": message.payload,
+            "message_id": message.message_id,
+            "delivery_count": message.metadata.delivery_count,
+            "topic": message.metadata.topic,
+        },
+    )
+
+
+@subscribe(topic="demo-fanout")
+async def receive_fanout(message: Message[dict[str, Any]]) -> None:
+    await complete_task(message, "python")
+    print(
+        "[python fanout consumer] received a copy",
+        {
+            "payload": message.payload,
+            "message_id": message.message_id,
+            "consumer_group": message.metadata.consumer_group,
+            "delivery_count": message.metadata.delivery_count,
+        },
+    )

services/cross-runtime-queues/frontend/eslint.config.mjs

@@ -0,0 +1,9 @@
+import { defineConfig, globalIgnores } from "eslint/config";
+import nextVitals from "eslint-config-next/core-web-vitals";
+import nextTypeScript from "eslint-config-next/typescript";
+
+export default defineConfig([
+  ...nextVitals,
+  ...nextTypeScript,
+  globalIgnores([".next/**", "next-env.d.ts"]),
+]);