消息推送

2026-03-10 16:39:50 +08:00
parent ee9fabd806
commit e67016a6f4
18 changed files with 1176 additions and 49 deletions
--- a/server/.runtime/delivery-backend.pids.json
+++ b/server/.runtime/delivery-backend.pids.json
@@ -0,0 +1,5 @@
+{
+    "push-server":  27380,
+    "webhook-receiver":  13140,
+    "notify-worker":  37804
+}
--- a/server/NOTIFY_WORKER_README.md
+++ b/server/NOTIFY_WORKER_README.md
@@ -0,0 +1,144 @@
+# notify-worker（物流事件 -> 消息入队）说明
+
+面向读者：后端/运维/联调同学。
+
+一句话：`notify-worker` 是常驻 worker，用于从 `notify_queue` 消费“新物流事件通知”，并生成（upsert）到 `express_notifications`，让后续的 `push-server` consumer 去实际下发推送。
+
+---
+
+## 1. 它解决什么问题？
+
+- **解耦**：Webhook 入库只保证事实表（`platform_express_*`）一致；消息生成用 worker 异步做，避免在 webhook 请求里做过多业务计算。
+- **可追溯/可补偿**：队列表 `notify_queue` 和消息表 `express_notifications` 让每次处理结果可落库（成功/跳过/失败原因）。
+- **幂等**：对同一事件（同一 `dedupe_key`）写入消息时，使用稳定的 `message_id` 进行 upsert，避免重复生成消息。
+
+在整体链路中的位置：
+
+1) webhook-receiver 写入 `platform_express_tracking_events`
+2) 事件被推入 `notify_queue`（由触发器/任务/上游逻辑实现）
+3) **notify-worker 消费 `notify_queue` → 写入 `express_notifications`**
+4) push-server consumer 轮询 `express_notifications` → 调用 `CLOUD_FUNC_URL` 下发
+
+---
+
+## 2. 用什么做的（技术栈/依赖）
+
+- **运行时**：Node.js（建议 18+，以确保 `fetch` 可用；旧版本会回退到 `node-fetch`）
+- **通信方式**：直接调用 Supabase PostgREST REST API（`$SUPA_URL/rest/v1/...`）
+- **鉴权**：默认仅发送 `apikey` header；仅当 `SUPA_USE_BEARER=true` 才附加 `Authorization: Bearer ...`
+- **核心依赖**：
+  - `crypto`：计算稳定的 `message_id`（SHA-256）
+  - `fetch` / `node-fetch`：HTTP 调用 Supabase REST
+
+它不是一个 HTTP Server（不监听端口），而是一个“循环轮询的后台进程”。
+
+---
+
+## 3. 输入/输出（读写哪些表）
+
+### 3.1 输入：`notify_queue`
+`notify-worker` 会拉取：
+- `processed_at IS NULL` 的记录
+- 按 `created_at ASC` 排序
+- 每次最多 `BATCH_SIZE` 条
+
+并在处理完成后回写：
+- `processed_at`（处理时间）
+- `process_status`：`queued` / `skipped` / `failed`
+- `last_error`（失败/跳过原因，截断保存）
+
+> 说明：字段命名以当前脚本实现为准，你的表结构需要包含这些列。
+
+### 3.2 查询依赖
+为把“物流事件”映射到收件人（user/merchant），脚本会查询：
+- `platform_express_waybills`：取 `order_id/order_no, carrier, tracking_no`
+- `ml_orders`：取 `user_id, merchant_id`
+
+### 3.3 输出：`express_notifications`
+对每个收件人写入 1 条通知（同一事件会写 user 与 merchant 两条，若都存在）：
+- `aud`：`user` 或 `merchant`
+- `recipient_id`
+- `order_id / waybill_id / tracking_no / carrier`
+- `message_id`：稳定哈希生成（`aud|waybill_id|dedupe_key`）
+- `dedupe_key`：`waybill_id|aud|<queue.dedupe_key>`（截断到 256）
+- `event_text_safe`：当前实现直接复用队列行的 `event_text`
+- `status_code / event_time / payload`
+
+写入方式是 **upsert**：`POST express_notifications?on_conflict=message_id`，并使用 `resolution=merge-duplicates`。
+
+注意：数据库侧必须有 **普通** 唯一索引/约束 `express_notifications(message_id)`；不要用 `WHERE message_id IS NOT NULL` 的部分唯一索引，否则会触发 `42P10`。
+若你已遇到该错误，执行修复脚本：`pages/mall/delivery/doc/需求文档/20260310_fix_express_notifications_on_conflict_message_id.sql`。
+
+---
+
+## 4. 配置与启动
+
+### 4.1 配置加载优先级
+脚本会复用 `server/load-config.js`，加载顺序（只填充未设置的 env）：
+1) 系统环境变量
+2) `CONFIG_FILE` / `CONFIG_PATH` 指定的 `.env` 或 `.json`
+3) 同目录 `server/notify-worker.config.json`
+4) `server/.env` / `server/config.json` / `server/config.json.example`
+
+### 4.2 关键环境变量
+- `SUPA_URL`（必需）
+- `SERVICE_ROLE_KEY` 或 `SUPA_KEY`（必需，推荐 service_role）
+- `SUPA_USE_BEARER`（可选，默认 false）
+
+worker 行为：
+- `NOTIFY_WORKER_POLL_MS`（默认 2000）
+- `NOTIFY_WORKER_BATCH_SIZE`（默认 20）
+- `RUN_ONCE=true`（只跑一轮就退出，适合手动验证/CI）
+
+### 4.3 启动命令
+在仓库根目录：
+
+```powershell
+# 常驻运行
+node server/notify-worker.js
+
+# 只跑一轮（用于验证）
+$env:RUN_ONCE='true'
+node server/notify-worker.js
+```
+
+如果你的配置放在 JSON：
+
+```powershell
+$env:CONFIG_FILE=(Resolve-Path .\server\notify-worker.config.json)
+node server/notify-worker.js
+```
+
+---
+
+## 5. 管理者“一页摘要”（给非一线同学）
+
+### 业务价值
+- 让“物流事件入库”与“消息生成/推送”解耦，降低 webhook 链路复杂度和故障扩散。
+- 通过队列表落库，可审计、可重放、可定位卡点。
+
+### 边界
+- notify-worker **只负责生成消息入队**（`express_notifications`），不负责实际推送下发。
+- 推送通道成功率与重试投递由 push-server consumer + 云函数负责。
+
+### 关键依赖
+- Supabase/Postgres 表：`notify_queue`、`platform_express_waybills`、`ml_orders`、`express_notifications`
+- 上游必须保证：有机制把新事件推入 `notify_queue`（触发器/任务/服务均可）
+
+### 主要风险与控制点
+- **幂等/重复消息**：依赖 `dedupe_key` 的稳定性；上游写队列时应构造稳定的 dedupe_key（不要用 Date.now 这种随机值）。
+- **权限/密钥**：需要 `service_role` 读写多表，必须只放服务器环境变量。
+- **数据合规**：`event_text_safe` 应为清洗/脱敏后的文案；若上游传入的是原文，需在生成前做脱敏。
+
+### 建议监控指标
+- `notify_queue` 未处理数量（`processed_at IS NULL`）是否持续增长
+- 每分钟处理量、`process_status=failed` 比例与 Top `last_error`
+- 从队列入库到消息入队的延迟（created_at → processed_at）
+
+---
+
+## 6. 进一步阅读
+
+- push-server（投递消费者）：`server/PUSH_SERVER_README.md`
+- 总览（Webhook + worker + push）：`docs/DELIVERY_WEBHOOK_PUSH_SERVER_OVERVIEW.md`
+- webhook-receiver（事件入库）：`pages/mall/delivery/webhook-server/README.md`
--- a/server/PUSH_SERVER_README.md
+++ b/server/PUSH_SERVER_README.md
@@ -3,6 +3,10 @@
 简要说明
 - 该文档记录对 `server/push-server.js` 的修改、运行所需的环境变量、表结构依赖、以及如何把 Supabase 的 cid 与通知通过云函数（`CLOUD_FUNC_URL`）下发的端到端操作步骤。

+组件补充（新增）
+- `server/notify-worker.js`：常驻 worker，从 `notify_queue` 生成/写入 `express_notifications`（消息入队）。
+- 说明文档：`server/NOTIFY_WORKER_README.md`（包含技术栈/依赖、配置项、监控与管理者摘要）。
+
 变更要点（代码修改摘要）
 - supaFetch: 默认仅发送 `apikey`；仅当 `SUPA_USE_BEARER=true` 时才发送 `Authorization: Bearer`。用于避免自托管 Supabase/Kong 场景下因 JWT_SECRET 不一致触发 `PGRST301`。
 - 新增 endpoint `/api/v1/notifications`：将通知写入 `express_notifications`（排队）；由消费者（轮询）读取待处理记录并 POST 到 `CLOUD_FUNC_URL`。
@@ -26,6 +30,20 @@
 - `public.push_devices`：用于存储设备 cid、user_id/merchant_id、is_active 等（见仓库迁移脚本 `20260224_add_push_devices_and_notifications.sql`）。
 - `public.express_notifications`：用于保存通知记录与状态（见迁移脚本）。

+字段语义（从 2026-03-09 起，重要变更）
+- `status_code`：物流/业务状态（如 `OUT_FOR_DELIVERY/DELIVERED/...`），由 `notify-worker` 写入。
+- `send_status`：投递处理状态（`null`=待发送，`processing/retrying/success/failed/no-targets`），由 push-server consumer 读写。
+
+为什么需要这个变更：引入 `notify-worker` 后，`status_code` 不再适合作为“投递状态”，否则 consumer 会捞不到记录（或把物流状态误当投递状态）。
+
+迁移脚本（必需执行一次）：
+- `pages/mall/delivery/doc/需求文档/20260309_add_express_notifications_send_status.sql`
+
+消息生成（可选但推荐）
+- `public.notify_queue`：轨迹事件入队表（由 DB trigger 写入）。
+- 迁移脚本：`pages/mall/delivery/doc/需求文档/20260309_add_notify_queue_and_trigger.sql`（创建 `notify_queue` + `AFTER INSERT` trigger）。
+- 常驻消费者：`server/notify-worker.js`（从 `notify_queue` 生成 `express_notifications`；push-server consumer 再负责下发）。
+
 关键环境变量（示例与说明）
 - SUPA_URL — Supabase REST（PostgREST）地址（内部建议 `http://rest:3000`）。
 - SERVICE_ROLE_KEY 或 SUPA_KEY — 用作 `apikey` 向 PostgREST 请求（不要把明文放到 Authorization 除非该值确为 JWT）。
@@ -34,6 +52,7 @@
 - CONSUMER_POLL_MS — 轮询间隔（毫秒）。
 - CLOUD_FUNC_URL — 云函数外网调用 URL（每个目标 cid 会对该 URL 发 POST）。
 - PUSH_TOKEN — (可选) 云函数鉴权 token（会在 POST body 的 token 字段中透传）。
+- PUSH_SERVER_PORT — (可选) push-server 监听端口，默认 7301（推荐用这个，便于与 webhook-receiver 共享同一份 `server/config.json`）。

 <!--
 UNI_PUSH_URL / UNI_PUSH_APPID / UNI_PUSH_SECRET / PUSH_PROXY_URL / PUSH_PROXY_TOKEN：仅在非云函数模式下使用；当前已不再使用。
@@ -101,7 +120,7 @@ Invoke-RestMethod -Uri 'http://localhost:7301/api/v1/notifications' -Method POST
 - 在 Supabase 中查询最近记录：

 ```sql
-select id, message_id, status_code, retry_count, last_error, updated_at
+select id, message_id, send_status, status_code, retry_count, last_error, updated_at
 from public.express_notifications
 order by created_at desc
 limit 10;
--- a/server/README.md
+++ b/server/README.md
@@ -84,10 +84,13 @@ Invoke-RestMethod -Uri 'http://localhost:7301/api/v1/notifications' -Method POST
 # 3) 等待 2 秒（CONSUMER_POLL_MS=2000），观察 push-server 控制台：应出现对 CLOUD_FUNC_URL 的 POST
 ```

- 如何确认是否已处理：在 Supabase 查询最近记录的 `status_code` / `last_error`：
+-- 如何确认是否已处理：在 Supabase 查询最近记录的 `send_status` / `last_error`：
+
+> 说明：`status_code` 是物流/业务状态（由 notify-worker 写入）；consumer 的投递状态使用 `send_status`。
+> 若你的数据库还没有该字段，请先执行迁移：`pages/mall/delivery/doc/需求文档/20260309_add_express_notifications_send_status.sql`。

 ```sql
-select id, message_id, status_code, retry_count, last_error, updated_at
+select id, message_id, send_status, status_code, retry_count, last_error, updated_at
 from public.express_notifications
 order by created_at desc
 limit 10;
--- a/server/config.json
+++ b/server/config.json
@@ -9,5 +9,6 @@
  "RETRY_INITIAL_MS": "5000",
  "RETRY_FACTOR": "2",
  "RETRY_MAX_MS": "3600000",
-  "PORT": "7301"
+  "PUSH_SERVER_PORT": "7301",
+  "WEBHOOK_PORT": "7201"
 }
--- a/server/notify-worker.js
+++ b/server/notify-worker.js
@@ -0,0 +1,255 @@
+// notify-worker.js
+// 常驻队列消费者：从 notify_queue 读取待处理事件，写入 express_notifications。
+// push-server consumer 将继续轮询 express_notifications 并调用 CLOUD_FUNC_URL 实际下发。
+//
+// 配置加载优先级：
+// 1) 系统环境变量
+// 2) CONFIG_FILE/CONFIG_PATH
+// 3) 同目录 notify-worker.config.json
+// 4) server/.env / server/config.json via server/load-config.js
+
+const fs = require('fs')
+const path = require('path')
+
+const localConfigPath = path.join(__dirname, 'notify-worker.config.json')
+if (!process.env.CONFIG_FILE && !process.env.CONFIG_PATH && fs.existsSync(localConfigPath)) {
+  process.env.CONFIG_FILE = localConfigPath
+}
+
+require('./load-config')
+
+const crypto = require('crypto')
+
+const fetchImpl = (globalThis.fetch ? globalThis.fetch.bind(globalThis) : (() => {
+  try {
+    return require('node-fetch')
+  } catch (e) {
+    throw new Error("No fetch implementation found. Use Node.js 18+ or install 'node-fetch'.")
+  }
+})())
+
+const SUPA_URL = (process.env.SUPA_URL || process.env.SUPA_URL_OVERRIDE || '').replace(/\/$/, '')
+// Prefer service role key when present (server-side), to avoid RLS issues.
+const SUPA_KEY = process.env.SERVICE_ROLE_KEY || process.env.SUPA_KEY || ''
+const SUPA_USE_BEARER = (process.env.SUPA_USE_BEARER === 'true')
+
+const POLL_MS = Number(process.env.NOTIFY_WORKER_POLL_MS || process.env.WORKER_POLL_MS || 2000)
+const BATCH_SIZE = Number(process.env.NOTIFY_WORKER_BATCH_SIZE || process.env.WORKER_BATCH_SIZE || 20)
+const RUN_ONCE = (process.env.RUN_ONCE === 'true' || process.env.RUN_ONCE === '1')
+
+function supaFetch(restPath, opts = {}) {
+  const url = `${SUPA_URL}/rest/v1/${restPath}`
+  const headers = Object.assign({}, opts.headers || {}, {
+    apikey: SUPA_KEY,
+    Accept: 'application/json'
+  })
+  if (SUPA_USE_BEARER) headers.Authorization = `Bearer ${SUPA_KEY}`
+  return fetchImpl(url, Object.assign({}, opts, { headers }))
+}
+
+function sha256Hex(input) {
+  return crypto.createHash('sha256').update(String(input)).digest('hex')
+}
+
+function buildMessageId({ aud, waybill_id, dedupe_key }) {
+  const h = sha256Hex(`${aud}|${waybill_id}|${dedupe_key}`)
+  return `evt_${aud}_${h.slice(0, 32)}`
+}
+
+async function fetchPendingQueue(limit) {
+  const q = `notify_queue?processed_at=is.null&order=created_at.asc&limit=${encodeURIComponent(String(limit))}`
+  const resp = await supaFetch(q, { method: 'GET' })
+  if (!resp.ok) {
+    const txt = await resp.text().catch(() => '')
+    throw new Error(`fetch notify_queue failed: HTTP ${resp.status} ${txt}`)
+  }
+  return await resp.json()
+}
+
+async function fetchWaybill(waybillId) {
+  const resp = await supaFetch(`platform_express_waybills?id=eq.${encodeURIComponent(waybillId)}&select=id,order_id,order_no,carrier,tracking_no`, { method: 'GET' })
+  if (!resp.ok) {
+    const txt = await resp.text().catch(() => '')
+    throw new Error(`fetch waybill failed: HTTP ${resp.status} ${txt}`)
+  }
+  const rows = await resp.json()
+  return rows && rows[0] ? rows[0] : null
+}
+
+async function fetchOrderById(orderId) {
+  const resp = await supaFetch(`ml_orders?id=eq.${encodeURIComponent(orderId)}&select=id,order_no,user_id,merchant_id`, { method: 'GET' })
+  if (!resp.ok) {
+    const txt = await resp.text().catch(() => '')
+    throw new Error(`fetch order (id) failed: HTTP ${resp.status} ${txt}`)
+  }
+  const rows = await resp.json()
+  return rows && rows[0] ? rows[0] : null
+}
+
+async function fetchOrderByNo(orderNo) {
+  const resp = await supaFetch(`ml_orders?order_no=eq.${encodeURIComponent(orderNo)}&select=id,order_no,user_id,merchant_id`, { method: 'GET' })
+  if (!resp.ok) {
+    const txt = await resp.text().catch(() => '')
+    throw new Error(`fetch order (order_no) failed: HTTP ${resp.status} ${txt}`)
+  }
+  const rows = await resp.json()
+  return rows && rows[0] ? rows[0] : null
+}
+
+async function upsertExpressNotification(body) {
+  const resp = await supaFetch('express_notifications?on_conflict=message_id', {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      Prefer: 'return=representation,resolution=merge-duplicates'
+    },
+    body: JSON.stringify(body)
+  })
+
+  if (!resp.ok) {
+    const txt = await resp.text().catch(() => '')
+    throw new Error(`insert express_notifications failed: HTTP ${resp.status} ${txt}`)
+  }
+
+  const rows = await resp.json().catch(() => null)
+  return Array.isArray(rows) ? rows[0] : rows
+}
+
+async function markQueueProcessed(id, patch) {
+  const resp = await supaFetch(`notify_queue?id=eq.${encodeURIComponent(id)}`, {
+    method: 'PATCH',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify(Object.assign({}, patch, { processed_at: new Date().toISOString() }))
+  })
+  if (!resp.ok) {
+    const txt = await resp.text().catch(() => '')
+    throw new Error(`patch notify_queue failed: HTTP ${resp.status} ${txt}`)
+  }
+}
+
+async function processOne(row) {
+  const waybillId = row.waybill_id
+  if (!waybillId) {
+    await markQueueProcessed(row.id, { process_status: 'skipped', last_error: 'missing waybill_id' })
+    return
+  }
+
+  const waybill = await fetchWaybill(waybillId)
+  if (!waybill) {
+    await markQueueProcessed(row.id, { process_status: 'skipped', last_error: 'waybill not found' })
+    return
+  }
+
+  let order = null
+  if (waybill.order_id) order = await fetchOrderById(waybill.order_id)
+  if (!order && waybill.order_no) order = await fetchOrderByNo(waybill.order_no)
+
+  if (!order) {
+    await markQueueProcessed(row.id, { process_status: 'skipped', last_error: 'order not found for waybill' })
+    return
+  }
+
+  const recipients = []
+  if (order.user_id) recipients.push({ aud: 'user', recipient_id: order.user_id })
+  if (order.merchant_id) recipients.push({ aud: 'merchant', recipient_id: order.merchant_id })
+
+  if (recipients.length === 0) {
+    await markQueueProcessed(row.id, { process_status: 'skipped', last_error: 'no recipient_id resolved from order' })
+    return
+  }
+
+  const payload = {
+    biz: 'express',
+    order_no: order.order_no || waybill.order_no || null,
+    order_id: order.id || waybill.order_id || null,
+    waybill_id: waybill.id,
+    carrier: waybill.carrier || row.carrier || null,
+    tracking_no: waybill.tracking_no || row.tracking_no || null,
+    status_code: row.status_code,
+    event_time: row.event_time,
+    event_id: row.event_id || null
+  }
+
+  // 为每个受众写入一条通知（message_id 按 aud 区分，确保幂等）
+  for (const r of recipients) {
+    const message_id = buildMessageId({ aud: r.aud, waybill_id: waybill.id, dedupe_key: row.dedupe_key })
+    const dedupe_key = `${waybill.id}|${r.aud}|${row.dedupe_key}`.slice(0, 256)
+
+    await upsertExpressNotification({
+      aud: r.aud,
+      recipient_id: r.recipient_id,
+      order_id: order.id || null,
+      waybill_id: waybill.id,
+      tracking_no: waybill.tracking_no || null,
+      carrier: waybill.carrier || null,
+      message_id,
+      event_text_safe: row.event_text || null,
+      status_code: row.status_code,
+      event_time: row.event_time || null,
+      payload,
+      dedupe_key
+    })
+  }
+
+  await markQueueProcessed(row.id, { process_status: 'queued', last_error: null })
+}
+
+async function loopOnce() {
+  const rows = await fetchPendingQueue(BATCH_SIZE)
+  if (!rows || rows.length === 0) return { processed: 0 }
+
+  let processed = 0
+  for (const row of rows) {
+    try {
+      await processOne(row)
+      processed += 1
+    } catch (e) {
+      const msg = (e && e.message) ? e.message : String(e)
+      try {
+        await markQueueProcessed(row.id, { process_status: 'failed', last_error: msg.substring(0, 2000) })
+      } catch (patchErr) {
+        console.warn('failed to mark queue row failed:', patchErr)
+      }
+      console.warn('processOne failed:', msg)
+      processed += 1
+    }
+  }
+
+  return { processed }
+}
+
+async function main() {
+  if (!SUPA_URL || !SUPA_KEY) {
+    console.error('SUPA_URL and SUPA_KEY must be set in env')
+    process.exit(1)
+  }
+
+  console.log('notify-worker starting...')
+  console.log('ENV: SUPA_URL=', SUPA_URL)
+  console.log('ENV: SUPA_USE_BEARER=', SUPA_USE_BEARER)
+  console.log('ENV: POLL_MS=', POLL_MS, 'BATCH_SIZE=', BATCH_SIZE, 'RUN_ONCE=', RUN_ONCE)
+
+  if (RUN_ONCE) {
+    const r = await loopOnce()
+    console.log('notify-worker done (run once):', r)
+    return
+  }
+
+  // 常驻轮询
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
+    try {
+      const r = await loopOnce()
+      if (r.processed > 0) console.log('notify-worker processed:', r.processed)
+    } catch (e) {
+      console.warn('notify-worker loop error:', e && e.message ? e.message : e)
+    }
+
+    await new Promise(resolve => setTimeout(resolve, POLL_MS))
+  }
+}
+
+main().catch(e => {
+  console.error('notify-worker fatal:', e)
+  process.exit(1)
+})
--- a/server/package.json
+++ b/server/package.json
@@ -4,7 +4,8 @@
  "description": "Local push backend for development (register devices, mock send)",
  "main": "push-server.js",
  "scripts": {
-    "start": "node push-server.js"
+    "start": "node push-server.js",
+    "worker": "node notify-worker.js"
  },
  "dependencies": {
    "archiver": "^7.0.1",
--- a/server/push-server.js
+++ b/server/push-server.js
@@ -12,12 +12,14 @@ const path = require('path')
 const fetch = require('node-fetch')
 const { spawn } = require('child_process')

-const PORT = process.env.PORT || 7301
+// 支持服务专用端口，避免与 webhook-receiver 共用 server/config.json 时发生端口冲突
+const PORT = process.env.PUSH_SERVER_PORT || process.env.PORT || 7301
 const DATA_DIR = path.join(__dirname, 'data')
 const DEVICES_FILE = path.join(DATA_DIR, 'push_devices.json')
 const SUPA_URL = process.env.SUPA_URL || ''
 // 支持两种环境变量名：SUPA_KEY（旧）和 SERVICE_ROLE_KEY（常用于自托管 .env）
-const SUPA_KEY = process.env.SUPA_KEY || process.env.SERVICE_ROLE_KEY || ''
+// Prefer service role key when present (server-side), to avoid RLS issues.
+const SUPA_KEY = process.env.SERVICE_ROLE_KEY || process.env.SUPA_KEY || ''
 const SUPA_SCHEMA = process.env.SUPA_SCHEMA || 'public'

 // Consumer 配置
@@ -31,6 +33,8 @@ const RETRY_INITIAL_MS = parseInt(process.env.RETRY_INITIAL_MS || '5000', 10)
 const RETRY_FACTOR = parseInt(process.env.RETRY_FACTOR || '2', 10)
 const RETRY_MAX_MS = parseInt(process.env.RETRY_MAX_MS || '3600000', 10) // 1 hour

+let warnedMissingSendStatus = false
+
 async function ensureDataDir() {
  try {
    await fs.mkdir(DATA_DIR, { recursive: true })
@@ -144,12 +148,18 @@ async function start() {
    if (!SUPA_URL || !SUPA_KEY) return []
    try {
      // 查询：
-      // - pending: status_code IS NULL
-      // - retrying: status_code = 'retrying' 且 (next_attempt_at IS NULL 或 next_attempt_at <= now)
+      // - pending: send_status IS NULL
+      // - retrying: send_status = 'retrying' 且 (next_attempt_at IS NULL 或 next_attempt_at <= now)
      const now = new Date().toISOString()
-      const q = `express_notifications?or=(status_code.is.null,and(status_code.eq.retrying,or(next_attempt_at.is.null,next_attempt_at.lte.${encodeURIComponent(now)})))&order=created_at.asc&limit=${limit}`
+      const q = `express_notifications?or=(send_status.is.null,and(send_status.eq.retrying,or(next_attempt_at.is.null,next_attempt_at.lte.${encodeURIComponent(now)})))&order=created_at.asc&limit=${limit}`
      const resp = await supaFetch(q, { method: 'GET' })
-      if (!resp.ok) return []
+      if (!resp.ok) {
+        if (!warnedMissingSendStatus && resp.status === 400) {
+          warnedMissingSendStatus = true
+          console.warn('Consumer query failed. If this mentions missing column "send_status", run migration: pages/mall/delivery/doc/需求文档/20260309_add_express_notifications_send_status.sql')
+        }
+        return []
+      }
      const data = await resp.json().catch(() => [])
      return Array.isArray(data) ? data : []
    } catch (e) {
@@ -160,19 +170,23 @@ async function start() {

  async function claimNotification(id) {
    try {
-      const body = { status_code: 'processing', updated_at: new Date().toISOString() }
+      const body = { send_status: 'processing', updated_at: new Date().toISOString() }
      // 仅当当前仍为 pending 或 retrying 且到期时，才抢占。
      // 为了避免在 PATCH 中使用复杂 or= 逻辑树导致匹配失败，这里拆成两次尝试。
      const now = new Date().toISOString()
      const attempts = [
-        // pending: status_code IS NULL
-        `express_notifications?id=eq.${encodeURIComponent(id)}&status_code=is.null`,
-        // retrying: status_code='retrying' and next_attempt_at is null or <= now
-        `express_notifications?id=eq.${encodeURIComponent(id)}&status_code=eq.retrying&or=(next_attempt_at.is.null,next_attempt_at.lte.${encodeURIComponent(now)})`
+        // pending: send_status IS NULL
+        `express_notifications?id=eq.${encodeURIComponent(id)}&send_status=is.null`,
+        // retrying: send_status='retrying' and next_attempt_at is null or <= now
+        `express_notifications?id=eq.${encodeURIComponent(id)}&send_status=eq.retrying&or=(next_attempt_at.is.null,next_attempt_at.lte.${encodeURIComponent(now)})`
      ]

      for (const path of attempts) {
        const resp = await supaFetch(path, { method: 'PATCH', headers: { 'Content-Type': 'application/json', Prefer: 'return=representation' }, body: JSON.stringify(body) })
+        if (!resp.ok && !warnedMissingSendStatus && resp.status === 400) {
+          warnedMissingSendStatus = true
+          console.warn('Consumer claim failed. If this mentions missing column "send_status", run migration: pages/mall/delivery/doc/需求文档/20260309_add_express_notifications_send_status.sql')
+        }
        if (!resp.ok) continue
        const j = await resp.json().catch(() => null)
        if (Array.isArray(j)) {
@@ -193,9 +207,13 @@ async function start() {
  async function updateNotificationStatus(id, status, note) {
    try {
      if (!id) return
-      const body = { status_code: String(status), updated_at: new Date().toISOString() }
-      if (note) body.event_text_safe = String(note).substring(0, 2000)
-      await supaFetch(`express_notifications?id=eq.${encodeURIComponent(id)}`, { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) })
+      const body = { send_status: String(status), updated_at: new Date().toISOString() }
+      if (note) body.last_error = String(note).substring(0, 2000)
+      const resp = await supaFetch(`express_notifications?id=eq.${encodeURIComponent(id)}`, { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) })
+      if (!resp.ok && !warnedMissingSendStatus && resp.status === 400) {
+        warnedMissingSendStatus = true
+        console.warn('Consumer status update failed. If this mentions missing column "send_status", run migration: pages/mall/delivery/doc/需求文档/20260309_add_express_notifications_send_status.sql')
+      }
    } catch (e) { console.warn('updateNotificationStatus error', e) }
  }

@@ -233,6 +251,12 @@ async function start() {
    const notification = claimed.payload && claimed.payload.notification ? claimed.payload.notification : (claimed.event_text_safe ? { title: claimed.event_text_safe, body: '' } : {})
    const payload = claimed.payload || {}

+    // Some push providers (and client notification renderers) may not display a system notification
+    // when content/body is empty, even if the push API returns success.
+    const titleText = (notification && (notification.title || notification.event_text || notification.text)) || claimed.event_text_safe || ''
+    const bodyText = (notification && (notification.body || notification.content)) || ''
+    const contentText = bodyText || titleText || ''
+
    // build targets similar to /api/v1/notifications
    let targets = []
    try {
@@ -261,7 +285,7 @@ async function start() {
    let allOk = true
    let lastNote = ''
    if (CLOUD_FUNC_URL) {
-      const calls = await Promise.all(targets.map(cid => invokeCloudFuncForCid(CLOUD_FUNC_URL, PUSH_TOKEN, cid, notification && notification.title, notification && notification.body, payload)))
+      const calls = await Promise.all(targets.map(cid => invokeCloudFuncForCid(CLOUD_FUNC_URL, PUSH_TOKEN, cid, titleText, contentText, payload)))
      for (const c of calls) {
        if (!c.ok) {
          allOk = false
@@ -285,7 +309,7 @@ async function start() {
      if (nextRetry >= MAX_RETRIES) {
        // mark failed
        try {
-          await supaFetch(`express_notifications?id=eq.${encodeURIComponent(claimed.id)}`, { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ status_code: 'failed', retry_count: nextRetry, last_error: String(lastNote).substring(0,2000), updated_at: new Date().toISOString() }) })
+          await supaFetch(`express_notifications?id=eq.${encodeURIComponent(claimed.id)}`, { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ send_status: 'failed', retry_count: nextRetry, last_error: String(lastNote).substring(0,2000), updated_at: new Date().toISOString() }) })
        } catch (e) { console.warn('mark failed error', e) }
      } else {
        // compute exponential backoff
@@ -293,7 +317,7 @@ async function start() {
        if (delay > RETRY_MAX_MS) delay = RETRY_MAX_MS
        const nextAt = new Date(Date.now() + delay).toISOString()
        try {
-          await supaFetch(`express_notifications?id=eq.${encodeURIComponent(claimed.id)}`, { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ status_code: 'retrying', retry_count: nextRetry, last_error: String(lastNote).substring(0,2000), next_attempt_at: nextAt, updated_at: new Date().toISOString() }) })
+          await supaFetch(`express_notifications?id=eq.${encodeURIComponent(claimed.id)}`, { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ send_status: 'retrying', retry_count: nextRetry, last_error: String(lastNote).substring(0,2000), next_attempt_at: nextAt, updated_at: new Date().toISOString() }) })
        } catch (e) { console.warn('schedule retry error', e) }
      }
    }
--- a/server/scripts/start-delivery-backend.ps1
+++ b/server/scripts/start-delivery-backend.ps1
@@ -0,0 +1,59 @@
+param(
+  [switch]$NoKill
+)
+
+$ErrorActionPreference = 'Stop'
+
+function Ensure-Dir($p) {
+  if (-not (Test-Path $p)) { New-Item -ItemType Directory -Path $p -Force | Out-Null }
+}
+
+function Stop-ListeningPort($port) {
+  $lines = netstat -ano | Select-String (":$port ")
+  foreach ($l in $lines) {
+    $m = [regex]::Match($l.ToString(), "\sLISTENING\s+(\d+)$")
+    if ($m.Success) {
+      $procId = [int]$m.Groups[1].Value
+      try {
+        Stop-Process -Id $procId -Force -ErrorAction Stop
+        Write-Host "Stopped PID $procId on port $port"
+      } catch {
+        Write-Host "Failed to stop PID $procId on port $($port): $($_.Exception.Message)"
+      }
+    }
+  }
+}
+
+# Repo root is two levels up from this script: server\scripts\*.ps1
+$root = Resolve-Path (Join-Path $PSScriptRoot "..\..")
+$configPath = Resolve-Path (Join-Path $root "server\config.json")
+
+if (-not $NoKill) {
+  Stop-ListeningPort 7201
+  Stop-ListeningPort 7301
+}
+
+Ensure-Dir (Join-Path $root "server\scripts")
+
+$env:CONFIG_FILE = $configPath.Path
+
+$services = @(
+  @{ Name = 'webhook-receiver'; Script = (Join-Path $root 'pages\mall\delivery\webhook-server\webhook-receiver.js'); Out = (Join-Path $root 'webhook-receiver.log'); Err = (Join-Path $root 'webhook-receiver.err.log') },
+  @{ Name = 'notify-worker';    Script = (Join-Path $root 'server\notify-worker.js');                          Out = (Join-Path $root 'notify-worker.log');    Err = (Join-Path $root 'notify-worker.err.log') },
+  @{ Name = 'push-server';      Script = (Join-Path $root 'server\push-server.js');                            Out = (Join-Path $root 'push-server.log');      Err = (Join-Path $root 'push-server.err.log') }
+)
+
+$pids = @{}
+foreach ($s in $services) {
+  if (-not (Test-Path $s.Script)) { throw "Missing script: $($s.Script)" }
+  Write-Host "Starting $($s.Name)..."
+  $p = Start-Process -FilePath "node" -ArgumentList @($s.Script) -WorkingDirectory $root.Path -WindowStyle Hidden -RedirectStandardOutput $s.Out -RedirectStandardError $s.Err -PassThru
+  $pids[$s.Name] = $p.Id
+  Write-Host "  PID=$($p.Id)  out=$($s.Out)  err=$($s.Err)"
+}
+
+Ensure-Dir (Join-Path $root 'server\.runtime')
+$pidsFile = Join-Path $root 'server\.runtime\delivery-backend.pids.json'
+$pids | ConvertTo-Json -Depth 5 | Set-Content -Path $pidsFile -Encoding UTF8
+Write-Host "Saved pids: $pidsFile"
+Write-Host "Done. Use test sender: node .\pages\mall\delivery\webhook-server\test-send.js"
--- a/server/scripts/stop-delivery-backend.ps1
+++ b/server/scripts/stop-delivery-backend.ps1
@@ -0,0 +1,27 @@
+$ErrorActionPreference = 'SilentlyContinue'
+
+$root = Resolve-Path (Join-Path $PSScriptRoot "..\..")
+$pidsFile = Join-Path $root 'server\.runtime\delivery-backend.pids.json'
+
+if (Test-Path $pidsFile) {
+  try {
+    $pids = Get-Content $pidsFile -Raw | ConvertFrom-Json
+    foreach ($name in $pids.PSObject.Properties.Name) {
+      $pid = [int]$pids.$name
+      if ($pid -gt 0) {
+        try {
+          Stop-Process -Id $pid -Force
+          Write-Host "Stopped $name PID=$pid"
+        } catch {
+          Write-Host "Failed to stop $name PID=$pid"
+        }
+      }
+    }
+  } catch {
+    Write-Host "Failed to parse $pidsFile"
+  }
+} else {
+  Write-Host "No pid file found: $pidsFile"
+}
+
+Write-Host "If ports still occupied, run: netstat -ano | Select-String ':7201 ' / ':7301 '"