跳转到内容
搜索文档
↵选择
↑↓导航
ESC关闭

推送通知

最后更新 查看 MarkdownAgent 设置

从 Agent 发送浏览器推送通知——即使用户已关闭标签页。结合 Agent 的持久化 state（存储推送订阅）、调度（定时投递）与 Web Push API ↗，你可以触达完全离线的用户。

工作原理

Browser                              Agent (Durable Object)
───────                              ──────────────────────
1. Register service worker
2. Subscribe to push (VAPID key)
3. Send subscription to agent ──────► Store in this.state
4. Create reminder ─────────────────► this.schedule(delay, "sendReminder", payload)

   ... user closes tab ...

5.                                    Alarm fires → sendReminder()
                                      web-push sends encrypted payload
                                              │
6. Service worker receives push ◄─────────────┘
7. showNotification()

Agent 在状态中持久存储推送订阅，并用 this.schedule() 在正确时间触发通知。alarm 触发时，Agent 使用 web-push ↗ 库调用推送服务端点。浏览器的 service worker 接收推送事件并显示原生通知。

前提条件

生成 VAPID 密钥

Web Push 需要 VAPID（自愿应用服务器标识，Voluntary Application Server Identification）密钥对。生成方式：

npx web-push generate-vapid-keys

本地开发将密钥存储在 .env 文件中：

VAPID_PUBLIC_KEY=BGxK...
VAPID_PRIVATE_KEY=abc1...
VAPID_SUBJECT=mailto:you@example.com

生产环境使用 wrangler secret put：

wrangler secret put VAPID_PUBLIC_KEY
wrangler secret put VAPID_PRIVATE_KEY
wrangler secret put VAPID_SUBJECT

创建 Agent

Agent 有三项职责：存储推送订阅、调度提醒，以及在 alarm 触发时发送通知。

import { Agent, callable, routeAgentRequest } from "agents";
import webpush from "web-push";

export class ReminderAgent extends Agent {
	initialState = {
		subscriptions: [],
		reminders: [],
	};

	@callable()
	getVapidPublicKey() {
		return this.env.VAPID_PUBLIC_KEY;
	}

	@callable()
	async subscribe(subscription) {
		const exists = this.state.subscriptions.some(
			(s) => s.endpoint === subscription.endpoint,
		);
		if (!exists) {
			this.setState({
				...this.state,
				subscriptions: [...this.state.subscriptions, subscription],
			});
		}
		return { ok: true };
	}

	@callable()
	async unsubscribe(endpoint) {
		this.setState({
			...this.state,
			subscriptions: this.state.subscriptions.filter(
				(s) => s.endpoint !== endpoint,
			),
		});
		return { ok: true };
	}

	@callable()
	async createReminder(message, delaySeconds) {
		const id = crypto.randomUUID();
		const scheduledAt = Date.now() + delaySeconds * 1000;
		const reminder = { id, message, scheduledAt, sent: false };

		this.setState({
			...this.state,
			reminders: [...this.state.reminders, reminder],
		});

		await this.schedule(delaySeconds, "sendReminder", { id, message });
		return reminder;
	}

	async sendReminder(payload) {
		webpush.setVapidDetails(
			this.env.VAPID_SUBJECT,
			this.env.VAPID_PUBLIC_KEY,
			this.env.VAPID_PRIVATE_KEY,
		);

		const deadEndpoints = [];

		await Promise.all(
			this.state.subscriptions.map(async (sub) => {
				try {
					await webpush.sendNotification(
						sub,
						JSON.stringify({
							title: "Reminder",
							body: payload.message,
							tag: `reminder-${payload.id}`,
						}),
					);
				} catch (err) {
					const statusCode =
						err instanceof webpush.WebPushError ? err.statusCode : 0;
					if (statusCode === 404 || statusCode === 410) {
						deadEndpoints.push(sub.endpoint);
					}
				}
			}),
		);

		if (deadEndpoints.length > 0) {
			this.setState({
				...this.state,
				subscriptions: this.state.subscriptions.filter(
					(s) => !deadEndpoints.includes(s.endpoint),
				),
			});
		}

		this.setState({
			...this.state,
			reminders: this.state.reminders.map((r) =>
				r.id === payload.id ? { ...r, sent: true } : r,
			),
		});

		this.broadcast(
			JSON.stringify({
				type: "reminder_sent",
				id: payload.id,
				timestamp: Date.now(),
			}),
		);
	}
}

export default {
	async fetch(request, env) {
		return (
			(await routeAgentRequest(request, env)) ??
			new Response("Not found", { status: 404 })
		);
	},
};
import { Agent, callable, routeAgentRequest } from "agents";
import webpush from "web-push";

type Subscription = {
	endpoint: string;
	expirationTime: number | null;
	keys: {
		p256dh: string;
		auth: string;
	};
};

type Reminder = {
	id: string;
	message: string;
	scheduledAt: number;
	sent: boolean;
};

type ReminderAgentState = {
	subscriptions: Subscription[];
	reminders: Reminder[];
};

export class ReminderAgent extends Agent<Env, ReminderAgentState> {
	initialState: ReminderAgentState = {
		subscriptions: [],
		reminders: [],
	};

	@callable()
	getVapidPublicKey(): string {
		return this.env.VAPID_PUBLIC_KEY;
	}

	@callable()
	async subscribe(subscription: Subscription): Promise<{ ok: boolean }> {
		const exists = this.state.subscriptions.some(
			(s) => s.endpoint === subscription.endpoint,
		);
		if (!exists) {
			this.setState({
				...this.state,
				subscriptions: [...this.state.subscriptions, subscription],
			});
		}
		return { ok: true };
	}

	@callable()
	async unsubscribe(endpoint: string): Promise<{ ok: boolean }> {
		this.setState({
			...this.state,
			subscriptions: this.state.subscriptions.filter(
				(s) => s.endpoint !== endpoint,
			),
		});
		return { ok: true };
	}

	@callable()
	async createReminder(
		message: string,
		delaySeconds: number,
	): Promise<Reminder> {
		const id = crypto.randomUUID();
		const scheduledAt = Date.now() + delaySeconds * 1000;
		const reminder: Reminder = { id, message, scheduledAt, sent: false };

		this.setState({
			...this.state,
			reminders: [...this.state.reminders, reminder],
		});

		await this.schedule(delaySeconds, "sendReminder", { id, message });
		return reminder;
	}

	async sendReminder(payload: { id: string; message: string }) {
		webpush.setVapidDetails(
			this.env.VAPID_SUBJECT,
			this.env.VAPID_PUBLIC_KEY,
			this.env.VAPID_PRIVATE_KEY,
		);

		const deadEndpoints: string[] = [];

		await Promise.all(
			this.state.subscriptions.map(async (sub) => {
				try {
					await webpush.sendNotification(
						sub,
						JSON.stringify({
							title: "Reminder",
							body: payload.message,
							tag: `reminder-${payload.id}`,
						}),
					);
				} catch (err: unknown) {
					const statusCode =
						err instanceof webpush.WebPushError ? err.statusCode : 0;
					if (statusCode === 404 || statusCode === 410) {
						deadEndpoints.push(sub.endpoint);
					}
				}
			}),
		);

		if (deadEndpoints.length > 0) {
			this.setState({
				...this.state,
				subscriptions: this.state.subscriptions.filter(
					(s) => !deadEndpoints.includes(s.endpoint),
				),
			});
		}

		this.setState({
			...this.state,
			reminders: this.state.reminders.map((r) =>
				r.id === payload.id ? { ...r, sent: true } : r,
			),
		});

		this.broadcast(
			JSON.stringify({
				type: "reminder_sent",
				id: payload.id,
				timestamp: Date.now(),
			}),
		);
	}
}

export default {
	async fetch(request: Request, env: Env) {
		return (
			(await routeAgentRequest(request, env)) ??
			new Response("Not found", { status: 404 })
		);
	},
} satisfies ExportedHandler<Env>;

sendReminder 回调处理三件事：通过 web-push 库投递推送通知、清理失效订阅（推送服务在订阅无效时返回 404 或 410），以及向已连接客户端广播以实时更新 UI。

设置 service worker

Service worker 在浏览器中运行，即使没有打开标签页也能接收 push 事件。将此文件放在 public/sw.js，以便从域根路径提供：

self.addEventListener("push", (event) => {
	if (!event.data) return;

	const data = event.data.json();

	event.waitUntil(
		self.registration.showNotification(data.title || "Notification", {
			body: data.body || "",
			icon: data.icon || "/favicon.ico",
			tag: data.tag,
			data: data.data,
		}),
	);
});

self.addEventListener("notificationclick", (event) => {
	event.notification.close();

	event.waitUntil(
		self.clients.matchAll({ type: "window" }).then((windowClients) => {
			for (const client of windowClients) {
				if (
					client.url.includes(self.location.origin) &&
					"focus" in client
				) {
					return client.focus();
				}
			}
			return self.clients.openWindow("/");
		}),
	);
});

push 事件处理程序解析 JSON payload 并显示原生通知。notificationclick 处理程序在用户点击通知时聚焦现有标签页或打开新标签页。

构建客户端

客户端需要：注册 service worker、请求通知权限、使用 VAPID 公钥订阅 push，并将订阅发送给 Agent。

注册 service worker

useEffect(() => {
	if (!("serviceWorker" in navigator) || !("PushManager" in window)) {
		return;
	}
	navigator.serviceWorker.register("/sw.js");
}, []);
useEffect(() => {
	if (!("serviceWorker" in navigator) || !("PushManager" in window)) {
		return;
	}
	navigator.serviceWorker.register("/sw.js");
}, []);

订阅 push

从 Agent 获取 VAPID 公钥，然后通过 Push API 订阅：

function base64urlToUint8Array(base64url) {
	const padded = base64url + "=".repeat((4 - (base64url.length % 4)) % 4);
	const binary = atob(padded.replace(/-/g, "+").replace(/_/g, "/"));
	const bytes = new Uint8Array(binary.length);
	for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i);
	return bytes;
}

async function subscribeToPush(agent) {
	const permission = await Notification.requestPermission();
	if (permission !== "granted") return;

	const vapidPublicKey = await agent.call("getVapidPublicKey");
	const reg = await navigator.serviceWorker.ready;
	const subscription = await reg.pushManager.subscribe({
		userVisibleOnly: true,
		applicationServerKey: base64urlToUint8Array(vapidPublicKey).buffer,
	});

	const subJson = subscription.toJSON();
	await agent.call("subscribe", [
		{
			endpoint: subJson.endpoint,
			expirationTime: subJson.expirationTime ?? null,
			keys: subJson.keys,
		},
	]);
}
function base64urlToUint8Array(base64url: string): Uint8Array {
	const padded = base64url + "=".repeat((4 - (base64url.length % 4)) % 4);
	const binary = atob(padded.replace(/-/g, "+").replace(/_/g, "/"));
	const bytes = new Uint8Array(binary.length);
	for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i);
	return bytes;
}

async function subscribeToPush(
	agent: ReturnType<typeof useAgent>,
) {
	const permission = await Notification.requestPermission();
	if (permission !== "granted") return;

	const vapidPublicKey = await agent.call("getVapidPublicKey");
	const reg = await navigator.serviceWorker.ready;
	const subscription = await reg.pushManager.subscribe({
		userVisibleOnly: true,
		applicationServerKey: base64urlToUint8Array(vapidPublicKey).buffer,
	});

	const subJson = subscription.toJSON();
	await agent.call("subscribe", [
		{
			endpoint: subJson.endpoint,
			expirationTime: subJson.expirationTime ?? null,
			keys: subJson.keys,
		},
	]);
}

创建提醒

订阅存储后，创建提醒只需一次 RPC 调用。Agent 处理调度与投递：

await agent.call("createReminder", ["Check the oven", 300]);
await agent.call("createReminder", ["Check the oven", 300]);

Agent 为 300 秒（5 分钟）调度 alarm。触发时推送通知到达——即使用户数分钟前已关闭标签页。

配置

wrangler.jsonc

{
	"name": "push-notifications",
	"compatibility_date": "2026-01-28",
	"compatibility_flags": ["nodejs_compat"],
	"main": "src/server.ts",
	"durable_objects": {
		"bindings": [
			{ "name": "ReminderAgent", "class_name": "ReminderAgent" },
		],
	},
	"migrations": [{ "tag": "v1", "new_sqlite_classes": ["ReminderAgent"] }],
	"assets": {
		"not_found_handling": "single-page-application",
	},
}

nodejs_compat 兼容标志是 web-push 库所必需的。

依赖

npm install agents web-push

生产环境注意事项

订阅过期

推送订阅可能过期或被用户撤销。始终通过从 state 中移除失效订阅来处理 push service 的 404 与 410 响应，如上文 sendReminder 示例所示。

每用户 vs 共享 Agent

大多数应用为每个用户使用一个 Agent（以用户 ID 作为 Agent 名称）。这隔离各用户的订阅与提醒。对于广播式通知（向多用户发送相同消息），共享 Agent 可存储所有订阅，但需注意订阅列表增长时的 state 大小。

结合推送与 WebSocket 广播

对当前连接的客户端使用 this.broadcast()（即时，无 push service 往返），对离线客户端使用 Web Push。上文 sendReminder 示例两者都做——已连接客户端收到实时 WebSocket 消息，离线客户端收到推送通知。

多设备

单个用户可能从多个浏览器或设备订阅。Agent 分别存储每个订阅，sendReminder 遍历全部。每个设备收到各自的推送通知。

失败重试

若 push service 返回 5xx 错误（临时失败），可用 this.schedule() 短延迟重试：

try {
	await webpush.sendNotification(sub, payload);
} catch (err) {
	const statusCode = err instanceof webpush.WebPushError ? err.statusCode : 0;
	if (statusCode >= 500) {
		await this.schedule(60, "retrySendNotification", {
			endpoint: sub.endpoint,
			payload,
		});
	}
}
try {
	await webpush.sendNotification(sub, payload);
} catch (err: unknown) {
	const statusCode =
		err instanceof webpush.WebPushError ? err.statusCode : 0;
	if (statusCode >= 500) {
		await this.schedule(60, "retrySendNotification", {
			endpoint: sub.endpoint,
			payload,
		});
	}
}

后续步骤

调度任务

了解调度与 keepAlive 以支持长时间运行操作。

这篇文档对您有帮助吗？