增加浏览器离线通知功能

2026-06-02 20:50:33 +08:00
parent 16e48bef8b
commit d600317ade
20 changed files with 1739 additions and 91 deletions
@@ -14,6 +14,7 @@ import {
  DanmakuFilterConfig,
  Notification,
  MovieRequest,
+  PushSubscriptionRecord,
 } from './types';
 import { AdminConfig } from './admin.types';
 import { MangaReadRecord, MangaShelfItem } from './manga.types';
@@ -25,6 +26,7 @@ import {
  MusicV2PlaylistRecord,
 } from './music-v2';
 import { userInfoCache } from './user-cache';
+import { dispatchWebPushNotification } from './web-push';

 /**
 * Cloudflare D1 存储实现
@@ -81,6 +83,7 @@ export class D1Storage implements IStorage {
    }
  }

+
  // ==================== 播放记录 ====================

  async getPlayRecord(
@@ -1911,6 +1914,133 @@ export class D1Storage implements IStorage {
    }
  }

+
+  async upsertPushSubscription(
+    userName: string,
+    subscription: PushSubscriptionRecord
+  ): Promise<void> {
+    try {
+      const result = await this.db
+        .prepare(`
+          INSERT INTO notification_push_subscriptions (
+            id, username, token_id, endpoint, p256dh, auth, user_agent, enabled,
+            created_at, updated_at, last_success_at, last_failure_at, failure_count
+          ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, 0)
+          ON CONFLICT(endpoint) DO UPDATE SET
+            username = excluded.username,
+            token_id = excluded.token_id,
+            p256dh = excluded.p256dh,
+            auth = excluded.auth,
+            user_agent = excluded.user_agent,
+            enabled = 1,
+            updated_at = excluded.updated_at
+        `)
+        .bind(
+          subscription.id,
+          userName,
+          subscription.tokenId || null,
+          subscription.endpoint,
+          subscription.p256dh,
+          subscription.auth,
+          subscription.userAgent || null,
+          subscription.enabled ? 1 : 0,
+          subscription.createdAt,
+          subscription.updatedAt
+        )
+        .run();
+      if (!result.success) {
+        throw new Error(result.error || '保存浏览器通知订阅失败');
+      }
+    } catch (err) {
+      console.error('D1Storage.upsertPushSubscription error:', err);
+      throw err;
+    }
+  }
+
+  async getEnabledPushSubscriptions(userName: string): Promise<PushSubscriptionRecord[]> {
+    try {
+      const results = await this.db
+        .prepare('SELECT * FROM notification_push_subscriptions WHERE username = ? AND enabled = 1')
+        .bind(userName)
+        .all();
+
+      return (results.results || []).map((row: any) => ({
+        id: row.id as string,
+        username: row.username as string,
+        tokenId: (row.token_id as string | null) || null,
+        endpoint: row.endpoint as string,
+        p256dh: row.p256dh as string,
+        auth: row.auth as string,
+        userAgent: (row.user_agent as string | null) || null,
+        enabled: row.enabled === 1,
+        createdAt: Number(row.created_at),
+        updatedAt: Number(row.updated_at),
+        lastSuccessAt: row.last_success_at ? Number(row.last_success_at) : null,
+        lastFailureAt: row.last_failure_at ? Number(row.last_failure_at) : null,
+        failureCount: Number(row.failure_count || 0),
+      }));
+    } catch (err) {
+      console.error('D1Storage.getEnabledPushSubscriptions error:', err);
+      return [];
+    }
+  }
+
+  async deletePushSubscriptionByEndpoint(userName: string, endpoint: string): Promise<void> {
+    try {
+      await this.db
+        .prepare('DELETE FROM notification_push_subscriptions WHERE username = ? AND endpoint = ?')
+        .bind(userName, endpoint)
+        .run();
+    } catch (err) {
+      console.error('D1Storage.deletePushSubscriptionByEndpoint error:', err);
+    }
+  }
+
+  async deletePushSubscriptionsByTokenId(userName: string, tokenId: string): Promise<void> {
+    try {
+      await this.db
+        .prepare('DELETE FROM notification_push_subscriptions WHERE username = ? AND token_id = ?')
+        .bind(userName, tokenId)
+        .run();
+    } catch (err) {
+      console.error('D1Storage.deletePushSubscriptionsByTokenId error:', err);
+    }
+  }
+
+  async deleteAllPushSubscriptions(userName: string): Promise<void> {
+    try {
+      await this.db
+        .prepare('DELETE FROM notification_push_subscriptions WHERE username = ?')
+        .bind(userName)
+        .run();
+    } catch (err) {
+      console.error('D1Storage.deleteAllPushSubscriptions error:', err);
+    }
+  }
+
+  async updatePushSubscriptionDeliveryStats(
+    userName: string,
+    endpoint: string,
+    success: boolean
+  ): Promise<void> {
+    try {
+      const now = Date.now();
+      if (success) {
+        await this.db
+          .prepare('UPDATE notification_push_subscriptions SET last_success_at = ?, failure_count = 0, updated_at = ? WHERE username = ? AND endpoint = ?')
+          .bind(now, now, userName, endpoint)
+          .run();
+      } else {
+        await this.db
+          .prepare('UPDATE notification_push_subscriptions SET last_failure_at = ?, failure_count = failure_count + 1, updated_at = ? WHERE username = ? AND endpoint = ?')
+          .bind(now, now, userName, endpoint)
+          .run();
+      }
+    } catch (err) {
+      console.error('D1Storage.updatePushSubscriptionDeliveryStats error:', err);
+    }
+  }
+
  // ==================== TVBox订阅token ====================

  async getTvboxSubscribeToken?(userName: string): Promise<string | null> {
@@ -2952,6 +3082,8 @@ export class D1Storage implements IStorage {
          notification.metadata ? JSON.stringify(notification.metadata) : null
        )
        .run();
+
+      await dispatchWebPushNotification(this, userName, notification);
    } catch (err) {
      console.error('D1Storage.addNotification error:', err);
      throw err;
@@ -3059,7 +3191,7 @@ export class D1Storage implements IStorage {
            requested_by, request_count, status, created_at, updated_at,
            fulfilled_at, fulfilled_source, fulfilled_id
          )
-          VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+          VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        `
        )
        .bind(
@@ -16,6 +16,7 @@ import {
  DanmakuFilterConfig,
  Notification,
  MovieRequest,
+  PushSubscriptionRecord,
 } from './types';
 import { AdminConfig } from './admin.types';
 import { MangaReadRecord, MangaShelfItem } from './manga.types';
@@ -26,6 +27,7 @@ import {
  MusicV2PlaylistItem,
  MusicV2PlaylistRecord,
 } from './music-v2';
+import { dispatchWebPushNotification } from './web-push';

 /**
 * Vercel Postgres 存储实现
@@ -1066,13 +1068,137 @@ export class PostgresStorage implements IStorage {
    }
  }

+
+  async upsertPushSubscription(
+    userName: string,
+    subscription: PushSubscriptionRecord
+  ): Promise<void> {
+    try {
+      await this.db
+        .prepare(`
+          INSERT INTO notification_push_subscriptions (
+            id, username, token_id, endpoint, p256dh, auth, user_agent, enabled,
+            created_at, updated_at, last_success_at, last_failure_at, failure_count
+          ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NULL, NULL, 0)
+          ON CONFLICT(endpoint) DO UPDATE SET
+            username = excluded.username,
+            token_id = excluded.token_id,
+            p256dh = excluded.p256dh,
+            auth = excluded.auth,
+            user_agent = excluded.user_agent,
+            enabled = 1,
+            updated_at = excluded.updated_at
+        `)
+        .bind(
+          subscription.id,
+          userName,
+          subscription.tokenId || null,
+          subscription.endpoint,
+          subscription.p256dh,
+          subscription.auth,
+          subscription.userAgent || null,
+          subscription.enabled ? 1 : 0,
+          subscription.createdAt,
+          subscription.updatedAt
+        )
+        .run();
+    } catch (err) {
+      console.error('PostgresStorage.upsertPushSubscription error:', err);
+      throw err;
+    }
+  }
+
+  async getEnabledPushSubscriptions(userName: string): Promise<PushSubscriptionRecord[]> {
+    try {
+      const results = await this.db
+        .prepare('SELECT * FROM notification_push_subscriptions WHERE username = $1 AND enabled = 1')
+        .bind(userName)
+        .all();
+
+      return (results.results || []).map((row: any) => ({
+        id: row.id as string,
+        username: row.username as string,
+        tokenId: (row.token_id as string | null) || null,
+        endpoint: row.endpoint as string,
+        p256dh: row.p256dh as string,
+        auth: row.auth as string,
+        userAgent: (row.user_agent as string | null) || null,
+        enabled: row.enabled === 1,
+        createdAt: Number(row.created_at),
+        updatedAt: Number(row.updated_at),
+        lastSuccessAt: row.last_success_at ? Number(row.last_success_at) : null,
+        lastFailureAt: row.last_failure_at ? Number(row.last_failure_at) : null,
+        failureCount: Number(row.failure_count || 0),
+      }));
+    } catch (err) {
+      console.error('PostgresStorage.getEnabledPushSubscriptions error:', err);
+      return [];
+    }
+  }
+
+  async deletePushSubscriptionByEndpoint(userName: string, endpoint: string): Promise<void> {
+    try {
+      await this.db
+        .prepare('DELETE FROM notification_push_subscriptions WHERE username = $1 AND endpoint = $2')
+        .bind(userName, endpoint)
+        .run();
+    } catch (err) {
+      console.error('PostgresStorage.deletePushSubscriptionByEndpoint error:', err);
+    }
+  }
+
+  async deletePushSubscriptionsByTokenId(userName: string, tokenId: string): Promise<void> {
+    try {
+      await this.db
+        .prepare('DELETE FROM notification_push_subscriptions WHERE username = $1 AND token_id = $2')
+        .bind(userName, tokenId)
+        .run();
+    } catch (err) {
+      console.error('PostgresStorage.deletePushSubscriptionsByTokenId error:', err);
+    }
+  }
+
+  async deleteAllPushSubscriptions(userName: string): Promise<void> {
+    try {
+      await this.db
+        .prepare('DELETE FROM notification_push_subscriptions WHERE username = $1')
+        .bind(userName)
+        .run();
+    } catch (err) {
+      console.error('PostgresStorage.deleteAllPushSubscriptions error:', err);
+    }
+  }
+
+  async updatePushSubscriptionDeliveryStats(
+    userName: string,
+    endpoint: string,
+    success: boolean
+  ): Promise<void> {
+    try {
+      const now = Date.now();
+      if (success) {
+        await this.db
+          .prepare('UPDATE notification_push_subscriptions SET last_success_at = $1, failure_count = 0, updated_at = $2 WHERE username = $3 AND endpoint = $4')
+          .bind(now, now, userName, endpoint)
+          .run();
+      } else {
+        await this.db
+          .prepare('UPDATE notification_push_subscriptions SET last_failure_at = $1, failure_count = failure_count + 1, updated_at = $2 WHERE username = $3 AND endpoint = $4')
+          .bind(now, now, userName, endpoint)
+          .run();
+      }
+    } catch (err) {
+      console.error('PostgresStorage.updatePushSubscriptionDeliveryStats error:', err);
+    }
+  }
+
  // ==================== TVBox订阅token ====================

  async getTvboxSubscribeToken(userName: string): Promise<string | null> {
    try {
      const result = await this.db
        .prepare(
-          'SELECT tvbox_subscribe_token FROM users_v2 WHERE username = $1'
+          'SELECT tvbox_subscribe_token FROM users WHERE username = $1'
        )
        .bind(userName)
        .first();
@@ -1088,7 +1214,7 @@ export class PostgresStorage implements IStorage {
    try {
      await this.db
        .prepare(
-          'UPDATE users_v2 SET tvbox_subscribe_token = $1 WHERE username = $2'
+          'UPDATE users SET tvbox_subscribe_token = $1 WHERE username = $2'
        )
        .bind(token, userName)
        .run();
@@ -1106,7 +1232,7 @@ export class PostgresStorage implements IStorage {
    try {
      const result = await this.db
        .prepare(
-          'SELECT username FROM users_v2 WHERE tvbox_subscribe_token = $1'
+          'SELECT username FROM users WHERE tvbox_subscribe_token = $1'
        )
        .bind(token)
        .first();
@@ -2937,6 +3063,8 @@ export class PostgresStorage implements IStorage {
          notification.metadata ? JSON.stringify(notification.metadata) : null
        )
        .run();
+
+      await dispatchWebPushNotification(this, userName, notification);
    } catch (err) {
      console.error('PostgresStorage.addNotification error:', err);
      throw err;
@@ -3044,7 +3172,7 @@ export class PostgresStorage implements IStorage {
            requested_by, request_count, status, created_at, updated_at,
            fulfilled_at, fulfilled_source, fulfilled_id
          )
-          VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
+          VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
        `
        )
        .bind(
@@ -11,8 +11,9 @@ import {
  MusicV2PlaylistRecord,
 } from './music-v2';
 import { RedisAdapter } from './redis-adapter';
-import { Favorite, IStorage, PlayRecord, SkipConfig } from './types';
+import { Favorite, IStorage, Notification, PlayRecord, PushSubscriptionRecord, SkipConfig } from './types';
 import { userInfoCache } from './user-cache';
+import { dispatchWebPushNotification } from './web-push';

 // 搜索历史最大条数
 const SEARCH_HISTORY_LIMIT = 20;
@@ -2268,6 +2269,8 @@ export abstract class BaseRedisStorage implements IStorage {
        JSON.stringify(notifications)
      )
    );
+
+    await dispatchWebPushNotification(this, userName, notification);
  }

  async markNotificationAsRead(
@@ -2465,6 +2468,106 @@ export abstract class BaseRedisStorage implements IStorage {
    userInfoCache?.delete(userName);
  }

+
+  private pushSubscriptionsKey(userName: string): string {
+    return `u:${userName}:push_subscriptions`;
+  }
+
+  async upsertPushSubscription(
+    userName: string,
+    subscription: PushSubscriptionRecord
+  ): Promise<void> {
+    await this.withRetry(() =>
+      this.adapter.hSet(
+        this.pushSubscriptionsKey(userName),
+        subscription.id,
+        JSON.stringify({ ...subscription, username: userName, updatedAt: Date.now() })
+      )
+    );
+  }
+
+  async getEnabledPushSubscriptions(userName: string): Promise<PushSubscriptionRecord[]> {
+    const all = await this.withRetry(() =>
+      this.adapter.hGetAll(this.pushSubscriptionsKey(userName))
+    );
+    if (!all || typeof all !== 'object') return [];
+
+    return Object.values(all)
+      .map((raw) => {
+        try {
+          return JSON.parse(raw as string) as PushSubscriptionRecord;
+        } catch {
+          return null;
+        }
+      })
+      .filter((item): item is PushSubscriptionRecord => Boolean(item?.enabled));
+  }
+
+  async deletePushSubscriptionByEndpoint(userName: string, endpoint: string): Promise<void> {
+    const subscriptions = await this.getEnabledPushSubscriptions(userName);
+    const target = subscriptions.find((item) => item.endpoint === endpoint);
+    if (!target) return;
+    await this.withRetry(() =>
+      this.adapter.hDel(this.pushSubscriptionsKey(userName), target.id)
+    );
+  }
+
+  async deletePushSubscriptionsByTokenId(userName: string, tokenId: string): Promise<void> {
+    const all = await this.withRetry(() =>
+      this.adapter.hGetAll(this.pushSubscriptionsKey(userName))
+    );
+    if (!all || typeof all !== 'object') return;
+
+    for (const [id, raw] of Object.entries(all)) {
+      try {
+        const subscription = JSON.parse(raw as string) as PushSubscriptionRecord;
+        if (subscription.tokenId === tokenId) {
+          await this.withRetry(() =>
+            this.adapter.hDel(this.pushSubscriptionsKey(userName), id)
+          );
+        }
+      } catch {
+        // ignore malformed record
+      }
+    }
+  }
+
+  async deleteAllPushSubscriptions(userName: string): Promise<void> {
+    await this.withRetry(() => this.adapter.del(this.pushSubscriptionsKey(userName)));
+  }
+
+  async updatePushSubscriptionDeliveryStats(
+    userName: string,
+    endpoint: string,
+    success: boolean
+  ): Promise<void> {
+    const all = await this.withRetry(() =>
+      this.adapter.hGetAll(this.pushSubscriptionsKey(userName))
+    );
+    if (!all || typeof all !== 'object') return;
+
+    for (const [id, raw] of Object.entries(all)) {
+      try {
+        const subscription = JSON.parse(raw as string) as PushSubscriptionRecord;
+        if (subscription.endpoint !== endpoint) continue;
+        const now = Date.now();
+        const next = {
+          ...subscription,
+          updatedAt: now,
+          lastSuccessAt: success ? now : subscription.lastSuccessAt || null,
+          lastFailureAt: success ? subscription.lastFailureAt || null : now,
+          failureCount: success ? 0 : (subscription.failureCount || 0) + 1,
+        };
+        await this.withRetry(() =>
+          this.adapter.hSet(this.pushSubscriptionsKey(userName), id, JSON.stringify(next))
+        );
+        return;
+      } catch {
+        // ignore malformed record
+      }
+    }
+  }
+
  // ---------- TVBox订阅token相关 ----------
  async getTvboxSubscribeToken(userName: string): Promise<string | null> {
    // 直接从数据库读取，不使用缓存
@@ -233,6 +233,26 @@ export interface IStorage {
    userName: string,
    enabled: boolean
  ): Promise<void>;
+  // Web Push订阅相关
+  upsertPushSubscription?(
+    userName: string,
+    subscription: PushSubscriptionRecord
+  ): Promise<void>;
+  getEnabledPushSubscriptions?(userName: string): Promise<PushSubscriptionRecord[]>;
+  deletePushSubscriptionByEndpoint?(
+    userName: string,
+    endpoint: string
+  ): Promise<void>;
+  deletePushSubscriptionsByTokenId?(
+    userName: string,
+    tokenId: string
+  ): Promise<void>;
+  deleteAllPushSubscriptions?(userName: string): Promise<void>;
+  updatePushSubscriptionDeliveryStats?(
+    userName: string,
+    endpoint: string,
+    success: boolean
+  ): Promise<void>;

  // TVBox订阅token相关
  getTvboxSubscribeToken?(userName: string): Promise<string | null>;
@@ -314,6 +334,23 @@ export interface EpisodeFilterConfig {
  reverseMode?: boolean; // 反向模式：开启后仅显示符合规则的集数
 }

+
+export interface PushSubscriptionRecord {
+  id: string;
+  username?: string;
+  tokenId?: string | null;
+  endpoint: string;
+  p256dh: string;
+  auth: string;
+  userAgent?: string | null;
+  enabled: boolean;
+  createdAt: number;
+  updatedAt: number;
+  lastSuccessAt?: number | null;
+  lastFailureAt?: number | null;
+  failureCount?: number;
+}
+
 // 通知类型枚举
 export type NotificationType =
  | 'favorite_update' // 收藏更新
@@ -0,0 +1,547 @@
+/* eslint-disable no-console */
+
+import crypto from 'crypto';
+
+import { HttpsProxyAgent } from 'https-proxy-agent';
+import nodeFetch from 'node-fetch';
+
+import { lockManager } from './lock';
+import { IStorage, Notification, PushSubscriptionRecord } from './types';
+
+const DEFAULT_TTL_SECONDS = 60 * 60 * 24;
+const WEB_PUSH_MAX_RETRIES = 2;
+const SUBJECT_ENV = 'WEB_PUSH_SUBJECT';
+const PROXY_ENV = 'WEB_PUSH_PROXY';
+const BASE_URL_ENV = 'WEB_PUSH_BASEURL';
+const VAPID_KEYS_GLOBAL_CONFIG_KEY = 'web_push_vapid_keys';
+
+interface VapidKeys {
+  publicKey: string;
+  privateKey: string;
+}
+
+export interface WebPushDeliveryResult {
+  endpointHost: string;
+  ok: boolean;
+  status?: number;
+  error?: string;
+  removed?: boolean;
+}
+
+export interface WebPushDispatchResult {
+  configured: boolean;
+  preferenceEnabled: boolean;
+  subscriptionCount: number;
+  deliveries: WebPushDeliveryResult[];
+}
+
+const globalVapidCacheKey = Symbol.for('__MOONTV_WEB_PUSH_VAPID_KEYS__');
+const globalVapidPromiseKey = Symbol.for('__MOONTV_WEB_PUSH_VAPID_KEYS_PROMISE__');
+
+function getCachedVapidKeys(): VapidKeys | null {
+  return ((globalThis as any)[globalVapidCacheKey] as VapidKeys | undefined) || null;
+}
+
+function setCachedVapidKeys(keys: VapidKeys): void {
+  (globalThis as any)[globalVapidCacheKey] = keys;
+}
+
+function getCachedVapidKeysPromise(): Promise<VapidKeys> | null {
+  return ((globalThis as any)[globalVapidPromiseKey] as Promise<VapidKeys> | undefined) || null;
+}
+
+function setCachedVapidKeysPromise(promise: Promise<VapidKeys> | null): void {
+  if (promise) {
+    (globalThis as any)[globalVapidPromiseKey] = promise;
+  } else {
+    delete (globalThis as any)[globalVapidPromiseKey];
+  }
+}
+
+function base64UrlEncode(input: Buffer | Uint8Array | string): string {
+  const buffer = Buffer.isBuffer(input) ? input : Buffer.from(input);
+  return buffer
+    .toString('base64')
+    .replace(/\+/g, '-')
+    .replace(/\//g, '_')
+    .replace(/=+$/g, '');
+}
+
+function base64UrlDecode(input: string): Buffer {
+  const normalized = input.replace(/-/g, '+').replace(/_/g, '/');
+  const padding = '='.repeat((4 - (normalized.length % 4)) % 4);
+  return Buffer.from(normalized + padding, 'base64');
+}
+
+function isCloudflareEnvironment(): boolean {
+  return process.env.CF_PAGES === '1' || process.env.BUILD_TARGET === 'cloudflare';
+}
+
+function getWebPushProxy(): string | null {
+  const proxy = process.env[PROXY_ENV]?.trim();
+  return proxy || null;
+}
+
+function normalizeBaseUrl(input: string): string {
+  return input.trim().replace(/\/+$/, '');
+}
+
+function getWebPushRequestUrl(endpoint: string): string {
+  const baseUrl = process.env[BASE_URL_ENV]?.trim();
+  if (!baseUrl) return endpoint;
+
+  const endpointUrl = new URL(endpoint);
+  const normalizedBase = normalizeBaseUrl(baseUrl);
+
+  // 支持自定义转发服务格式：
+  // - {endpoint}: URL 编码后的完整原始 endpoint，适合放在 query 参数里
+  // - {raw_endpoint}: 未编码的完整原始 endpoint，适合路径重写或代理服务自行解析
+  if (normalizedBase.includes('{raw_endpoint}')) {
+    return normalizedBase.replace('{raw_endpoint}', endpoint);
+  }
+  if (normalizedBase.includes('{endpoint}')) {
+    return normalizedBase.replace('{endpoint}', encodeURIComponent(endpoint));
+  }
+
+  const base = new URL(normalizedBase);
+  const basePath = base.pathname.replace(/\/+$/, '');
+  const endpointPath = endpointUrl.pathname.startsWith('/')
+    ? endpointUrl.pathname
+    : `/${endpointUrl.pathname}`;
+
+  base.pathname = `${basePath}${endpointPath}`.replace(/\/+/g, '/');
+  base.search = endpointUrl.search;
+  return base.toString();
+}
+
+async function fetchWebPushEndpoint(
+  endpoint: string,
+  init: {
+    method: string;
+    headers: Record<string, string>;
+    body: Buffer;
+  }
+): Promise<Response> {
+  const requestUrl = getWebPushRequestUrl(endpoint);
+  const proxy = getWebPushProxy();
+
+  if (isCloudflareEnvironment()) {
+    if (proxy) {
+      console.warn('WEB_PUSH_PROXY is ignored in Cloudflare runtime; use WEB_PUSH_BASEURL instead.');
+    }
+    return fetch(requestUrl, init) as Promise<Response>;
+  }
+
+  const fetchOptions: any = {
+    method: init.method,
+    headers: init.headers,
+    body: init.body,
+  };
+
+  if (proxy) {
+    fetchOptions.agent = new HttpsProxyAgent(proxy, {
+      timeout: 30000,
+      keepAlive: false,
+    });
+  }
+
+  return nodeFetch(requestUrl, fetchOptions) as unknown as Response;
+}
+
+function generateVapidKeys(): VapidKeys {
+  const ecdh = crypto.createECDH('prime256v1');
+  ecdh.generateKeys();
+
+  return {
+    publicKey: base64UrlEncode(ecdh.getPublicKey(undefined, 'uncompressed')),
+    privateKey: base64UrlEncode(ecdh.getPrivateKey()),
+  };
+}
+
+function parseStoredVapidKeys(raw: string | null): VapidKeys | null {
+  if (!raw) return null;
+
+  try {
+    const parsed = JSON.parse(raw) as Partial<VapidKeys>;
+    if (parsed.publicKey && parsed.privateKey) {
+      return { publicKey: parsed.publicKey, privateKey: parsed.privateKey };
+    }
+  } catch (error) {
+    console.error('Failed to parse stored Web Push VAPID keys:', error);
+  }
+
+  return null;
+}
+
+async function loadOrCreateDatabaseVapidKeys(storage: IStorage): Promise<VapidKeys> {
+  if (!storage?.getGlobalValue || !storage?.setGlobalValue) {
+    throw new Error('当前存储类型不支持保存 Web Push VAPID 密钥');
+  }
+
+  const existing = parseStoredVapidKeys(
+    await storage.getGlobalValue(VAPID_KEYS_GLOBAL_CONFIG_KEY)
+  );
+  if (existing) return existing;
+
+  const release = await lockManager.acquire('web-push-vapid-keys');
+  try {
+    const latest = parseStoredVapidKeys(
+      await storage.getGlobalValue(VAPID_KEYS_GLOBAL_CONFIG_KEY)
+    );
+    if (latest) return latest;
+
+    const keys = generateVapidKeys();
+    await storage.setGlobalValue(
+      VAPID_KEYS_GLOBAL_CONFIG_KEY,
+      JSON.stringify(keys)
+    );
+    return keys;
+  } finally {
+    release();
+  }
+}
+
+export async function getVapidKeys(storage: IStorage): Promise<VapidKeys> {
+  const cached = getCachedVapidKeys();
+  if (cached) return cached;
+
+  const cachedPromise = getCachedVapidKeysPromise();
+  if (cachedPromise) return cachedPromise;
+
+  const promise = loadOrCreateDatabaseVapidKeys(storage)
+    .then((keys) => {
+      setCachedVapidKeys(keys);
+      return keys;
+    })
+    .finally(() => setCachedVapidKeysPromise(null));
+
+  setCachedVapidKeysPromise(promise);
+  return promise;
+}
+
+export async function getVapidPublicKey(storage: IStorage): Promise<string | null> {
+  try {
+    return (await getVapidKeys(storage)).publicKey;
+  } catch (error) {
+    console.error('Failed to get Web Push VAPID public key:', error);
+    return null;
+  }
+}
+
+export async function isWebPushConfigured(storage: IStorage): Promise<boolean> {
+  try {
+    await getVapidKeys(storage);
+    return true;
+  } catch (error) {
+    console.error('Web Push VAPID keys are not configured:', error);
+    return false;
+  }
+}
+
+function getVapidSubject(): string {
+  return process.env[SUBJECT_ENV] || process.env.NEXT_PUBLIC_SITE_URL || 'mailto:[email protected]';
+}
+
+function getPublicKeyFromPrivate(privateKeyBase64Url: string): Buffer {
+  const ecdh = crypto.createECDH('prime256v1');
+  ecdh.setPrivateKey(base64UrlDecode(privateKeyBase64Url));
+  return ecdh.getPublicKey(undefined, 'uncompressed');
+}
+
+function createVapidJwt(endpoint: string, privateKeyBase64Url: string): string {
+  const audience = new URL(endpoint).origin;
+  const publicKey = getPublicKeyFromPrivate(privateKeyBase64Url);
+  const x = publicKey.subarray(1, 33);
+  const y = publicKey.subarray(33, 65);
+  const d = base64UrlDecode(privateKeyBase64Url);
+
+  const header = { typ: 'JWT', alg: 'ES256' };
+  const payload = {
+    aud: audience,
+    exp: Math.floor(Date.now() / 1000) + 12 * 60 * 60,
+    sub: getVapidSubject(),
+  };
+
+  const signingInput = `${base64UrlEncode(JSON.stringify(header))}.${base64UrlEncode(JSON.stringify(payload))}`;
+  const key = crypto.createPrivateKey({
+    key: {
+      kty: 'EC',
+      crv: 'P-256',
+      x: base64UrlEncode(x),
+      y: base64UrlEncode(y),
+      d: base64UrlEncode(d),
+    },
+    format: 'jwk',
+  });
+
+  const signature = crypto.sign('sha256', Buffer.from(signingInput), {
+    key,
+    dsaEncoding: 'ieee-p1363',
+  });
+
+  return `${signingInput}.${base64UrlEncode(signature)}`;
+}
+
+function hkdf(secret: Buffer, salt: Buffer, info: Buffer | string, length: number): Buffer {
+  const prk = crypto.createHmac('sha256', salt).update(secret).digest();
+  const infoBuffer = Buffer.isBuffer(info) ? info : Buffer.from(info);
+  const blocks: Buffer[] = [];
+  let previous = Buffer.alloc(0);
+  let counter = 1;
+
+  while (Buffer.concat(blocks).length < length) {
+    previous = crypto
+      .createHmac('sha256', prk)
+      .update(Buffer.concat([previous, infoBuffer, Buffer.from([counter])]))
+      .digest();
+    blocks.push(previous);
+    counter += 1;
+  }
+
+  return Buffer.concat(blocks).subarray(0, length);
+}
+
+function encryptPayload(payload: string, subscription: PushSubscriptionRecord) {
+  const receiverPublicKey = base64UrlDecode(subscription.p256dh);
+  const authSecret = base64UrlDecode(subscription.auth);
+  const salt = crypto.randomBytes(16);
+  const localEcdh = crypto.createECDH('prime256v1');
+  localEcdh.generateKeys();
+  const senderPublicKey = localEcdh.getPublicKey(undefined, 'uncompressed');
+  const sharedSecret = localEcdh.computeSecret(receiverPublicKey);
+
+  const keyInfo = Buffer.concat([
+    Buffer.from('WebPush: info\0'),
+    receiverPublicKey,
+    senderPublicKey,
+  ]);
+  const ikm = hkdf(sharedSecret, authSecret, keyInfo, 32);
+  const cek = hkdf(ikm, salt, 'Content-Encoding: aes128gcm\0', 16);
+  const nonce = hkdf(ikm, salt, 'Content-Encoding: nonce\0', 12);
+
+  const plaintext = Buffer.concat([Buffer.from(payload), Buffer.from([0x02])]);
+  const cipher = crypto.createCipheriv('aes-128-gcm', cek, nonce);
+  const encrypted = Buffer.concat([cipher.update(plaintext), cipher.final()]);
+  const tag = cipher.getAuthTag();
+  const ciphertext = Buffer.concat([encrypted, tag]);
+
+  const recordSize = Buffer.alloc(4);
+  recordSize.writeUInt32BE(4096, 0);
+
+  const body = Buffer.concat([
+    salt,
+    recordSize,
+    Buffer.from([senderPublicKey.length]),
+    senderPublicKey,
+    ciphertext,
+  ]);
+
+  return body;
+}
+
+export function getNotificationClickUrl(notification: Notification): string {
+  const metadata = notification.metadata || {};
+
+  if (notification.type === 'favorite_update' && metadata.source && metadata.id) {
+    const title = encodeURIComponent(String(metadata.title || ''));
+    return `/play?source=${encodeURIComponent(String(metadata.source))}&id=${encodeURIComponent(String(metadata.id))}&title=${title}`;
+  }
+
+  if (notification.type === 'manga_update' && metadata.sourceId && metadata.mangaId) {
+    const params = new URLSearchParams({
+      sourceId: String(metadata.sourceId),
+      mangaId: String(metadata.mangaId),
+      title: String(metadata.title || ''),
+      cover: String(metadata.cover || ''),
+      sourceName: String(metadata.sourceName || ''),
+    });
+    return `/manga/detail?${params.toString()}`;
+  }
+
+  if (notification.type === 'movie_request') {
+    return '/admin';
+  }
+
+  if (notification.type === 'request_fulfilled') {
+    if (metadata.source && metadata.id) {
+      return `/play?source=${encodeURIComponent(String(metadata.source))}&id=${encodeURIComponent(String(metadata.id))}&title=${encodeURIComponent(notification.title)}`;
+    }
+    return '/movie-request';
+  }
+
+  if (notification.type === 'anime_subscription_update') {
+    return '/private-library';
+  }
+
+  return '/';
+}
+
+function buildPayload(notification: Notification): string {
+  return JSON.stringify({
+    notificationId: notification.id,
+    type: notification.type,
+    title: notification.title,
+    message: notification.message,
+    body: notification.message,
+    url: getNotificationClickUrl(notification),
+    timestamp: notification.timestamp,
+  });
+}
+
+async function sendToSubscription(storage: IStorage, subscription: PushSubscriptionRecord, payload: string): Promise<Response> {
+  const keys = await getVapidKeys(storage);
+
+  const jwt = createVapidJwt(subscription.endpoint, keys.privateKey);
+  const encryptedBody = encryptPayload(payload, subscription);
+
+  return fetchWebPushEndpoint(subscription.endpoint, {
+    method: 'POST',
+    headers: {
+      Authorization: `vapid t=${jwt}, k=${keys.publicKey}`,
+      'Content-Encoding': 'aes128gcm',
+      'Content-Type': 'application/octet-stream',
+      TTL: String(DEFAULT_TTL_SECONDS),
+      Urgency: 'normal',
+    },
+    body: encryptedBody,
+  });
+}
+
+function shouldRetryWebPushResponse(response: Response): boolean {
+  if (response.status === 404 || response.status === 410) return false;
+  return !response.ok;
+}
+
+async function sendToSubscriptionWithRetry(
+  storage: IStorage,
+  subscription: PushSubscriptionRecord,
+  payload: string
+): Promise<Response> {
+  let lastError: unknown = null;
+
+  for (let attempt = 0; attempt <= WEB_PUSH_MAX_RETRIES; attempt++) {
+    try {
+      const response = await sendToSubscription(storage, subscription, payload);
+      if (!shouldRetryWebPushResponse(response) || attempt === WEB_PUSH_MAX_RETRIES) {
+        return response;
+      }
+
+      console.warn(
+        `Web Push send failed with ${response.status}, retrying (${attempt + 1}/${WEB_PUSH_MAX_RETRIES})...`
+      );
+    } catch (error) {
+      lastError = error;
+      if (attempt === WEB_PUSH_MAX_RETRIES) break;
+      console.warn(
+        `Web Push send error, retrying (${attempt + 1}/${WEB_PUSH_MAX_RETRIES}):`,
+        error
+      );
+    }
+  }
+
+  throw lastError instanceof Error ? lastError : new Error(String(lastError || 'Web Push send failed'));
+}
+
+export async function dispatchWebPushNotificationWithResult(
+  storage: IStorage,
+  userName: string,
+  notification: Notification
+): Promise<WebPushDispatchResult> {
+  const configured = await isWebPushConfigured(storage);
+  if (!configured || !storage.getEnabledPushSubscriptions) {
+    return {
+      configured,
+      preferenceEnabled: false,
+      subscriptionCount: 0,
+      deliveries: [],
+    };
+  }
+
+  const subscriptions = await storage.getEnabledPushSubscriptions(userName);
+  if (!subscriptions.length) {
+    return {
+      configured,
+      preferenceEnabled: true,
+      subscriptionCount: 0,
+      deliveries: [],
+    };
+  }
+
+  const payload = buildPayload(notification);
+  const settled = await Promise.allSettled(
+    subscriptions.map(async (subscription): Promise<WebPushDeliveryResult> => {
+      const endpointHost = new URL(subscription.endpoint).host;
+      try {
+        const response = await sendToSubscriptionWithRetry(storage, subscription, payload);
+
+        if (response.ok) {
+          await storage.updatePushSubscriptionDeliveryStats?.(userName, subscription.endpoint, true);
+          return { endpointHost, ok: true, status: response.status };
+        }
+
+        if (response.status === 404 || response.status === 410) {
+          await storage.deletePushSubscriptionByEndpoint?.(userName, subscription.endpoint);
+          return { endpointHost, ok: false, status: response.status, removed: true };
+        }
+
+        await storage.updatePushSubscriptionDeliveryStats?.(userName, subscription.endpoint, false);
+        const errorText = await response.text().catch(() => '');
+        console.warn(`Web Push failed (${response.status}) for ${userName}: ${errorText}`);
+        return { endpointHost, ok: false, status: response.status, error: errorText || response.statusText };
+      } catch (error) {
+        await storage.updatePushSubscriptionDeliveryStats?.(userName, subscription.endpoint, false);
+        console.error('Web Push delivery error:', error);
+        return {
+          endpointHost,
+          ok: false,
+          error: error instanceof Error ? error.message : String(error),
+        };
+      }
+    })
+  );
+
+  return {
+    configured,
+    preferenceEnabled: true,
+    subscriptionCount: subscriptions.length,
+    deliveries: settled.map((item) =>
+      item.status === 'fulfilled'
+        ? item.value
+        : { endpointHost: 'unknown', ok: false, error: item.reason instanceof Error ? item.reason.message : String(item.reason) }
+    ),
+  };
+}
+
+export async function dispatchWebPushNotification(
+  storage: IStorage,
+  userName: string,
+  notification: Notification
+): Promise<void> {
+  await dispatchWebPushNotificationWithResult(storage, userName, notification);
+}
+
+export function createPushSubscriptionRecord(input: {
+  username: string;
+  tokenId?: string | null;
+  endpoint: string;
+  p256dh: string;
+  auth: string;
+  userAgent?: string | null;
+}): PushSubscriptionRecord {
+  const now = Date.now();
+  return {
+    id: base64UrlEncode(crypto.createHash('sha256').update(input.endpoint).digest()),
+    username: input.username,
+    tokenId: input.tokenId || null,
+    endpoint: input.endpoint,
+    p256dh: input.p256dh,
+    auth: input.auth,
+    userAgent: input.userAgent || null,
+    enabled: true,
+    createdAt: now,
+    updatedAt: now,
+    lastSuccessAt: null,
+    lastFailureAt: null,
+    failureCount: 0,
+  };
+}