From f54df7081125d005aefcd908af8d3c5f184c3a84 Mon Sep 17 00:00:00 2001 From: Steven Date: Tue, 10 Nov 2026 13:47:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(notify):=20=E7=AB=99=E5=86=85=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=E8=A1=A5=E9=BD=90=E5=AE=8C=E6=88=90=E3=80=81=E5=88=B0?= =?UTF-8?q?=E6=9C=9F=E9=80=BE=E6=9C=9F=E4=B8=8E=E7=9F=A5=E8=AF=86=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../main/java/com/dadao/task/MainActivity.kt | 21 +- .../src/main/java/com/dadao/task/data/Api.kt | 2 + .../com/dadao/task/ui/NotificationsScreen.kt | 54 ++++- .../versions/0017_notification_reminders.py | 60 +++++ backend/app/api/items.py | 68 +++++- backend/app/api/knowledge.py | 58 +++++ backend/app/api/tickets.py | 20 +- backend/app/config.py | 8 + backend/app/main.py | 4 + backend/app/models_ticket.py | 14 ++ backend/app/services/notify.py | 52 ++++ backend/app/services/reminders.py | 215 +++++++++++++++++ backend/tests/test_notifications.py | 223 ++++++++++++++++++ web/src/lib/types.ts | 2 + web/src/pages/Notifications.tsx | 22 +- 15 files changed, 791 insertions(+), 32 deletions(-) create mode 100644 backend/alembic/versions/0017_notification_reminders.py create mode 100644 backend/app/services/reminders.py diff --git a/android/app/src/main/java/com/dadao/task/MainActivity.kt b/android/app/src/main/java/com/dadao/task/MainActivity.kt index 26a0242..abea94c 100644 --- a/android/app/src/main/java/com/dadao/task/MainActivity.kt +++ b/android/app/src/main/java/com/dadao/task/MainActivity.kt @@ -36,6 +36,7 @@ import com.dadao.task.ui.TicketsScreen import com.dadao.task.ui.UpdateDialog import com.dadao.task.ui.WorkspaceScaffold import com.dadao.task.ui.WorkspaceTab +import kotlinx.coroutines.delay class MainActivity : ComponentActivity() { @@ -114,10 +115,14 @@ private fun AppRoot(joinUri: String?) { .onFailure { pendingJoin = null } } - // 未读数：进工作区、切页面、或通知页标了已读（unreadRefresh）时刷新 + // 未读数：进工作区、切页面、或通知页标了已读（unreadRefresh）时刷新； + // 停留不动时每 30 秒再拉一次，这样"任务被指派/快到期/已逾期"不用切页面也能看到角标。 LaunchedEffect(current?.id, tab, unreadRefresh) { val ws = current ?: return@LaunchedEffect - unread = api.unreadCount(ws.id) + while (true) { + unread = api.unreadCount(ws.id) + delay(30_000) + } } // 版本检查：登录成功后查一次就够，查不到（网络、没发布过包）就当没有新版 @@ -234,11 +239,15 @@ private fun AppRoot(joinUri: String?) { WorkspaceTab.Notifications -> NotificationsScreen( api = api, workspaceId = ws.id, - onOpenItem = { itemId, isTicket -> - // 通知点开后跳到对应列表并展开那一条；通知本身不带数据，页面自己回源 + onOpenItem = { itemId, itemKind -> + // 通知点开后跳到对应页签并展开那一条；通知本身不带数据，页面自己回源 reopenItem = itemId - reopenKnowledge = false - tab = if (isTicket) WorkspaceTab.Tickets else WorkspaceTab.Tasks + reopenKnowledge = itemKind == "knowledge" + tab = when (itemKind) { + "knowledge" -> WorkspaceTab.Knowledge + "ticket" -> WorkspaceTab.Tickets + else -> WorkspaceTab.Tasks + } }, onUnreadChanged = { unreadRefresh += 1 }, ) diff --git a/android/app/src/main/java/com/dadao/task/data/Api.kt b/android/app/src/main/java/com/dadao/task/data/Api.kt index 0e14210..5239082 100644 --- a/android/app/src/main/java/com/dadao/task/data/Api.kt +++ b/android/app/src/main/java/com/dadao/task/data/Api.kt @@ -328,6 +328,8 @@ data class AppNotification( val id: String, val kind: String = "", @SerialName("item_id") val itemId: String? = null, + /** task / ticket / knowledge：决定点开跳到哪个页签。 */ + @SerialName("item_kind") val itemKind: String? = null, @SerialName("item_title") val itemTitle: String? = null, val title: String = "", val body: String = "", diff --git a/android/app/src/main/java/com/dadao/task/ui/NotificationsScreen.kt b/android/app/src/main/java/com/dadao/task/ui/NotificationsScreen.kt index 96f19fb..cbc8f1f 100644 --- a/android/app/src/main/java/com/dadao/task/ui/NotificationsScreen.kt +++ b/android/app/src/main/java/com/dadao/task/ui/NotificationsScreen.kt @@ -32,6 +32,7 @@ import androidx.compose.ui.unit.sp import com.dadao.task.data.Api import com.dadao.task.data.ApiException import com.dadao.task.data.AppNotification +import kotlinx.coroutines.delay import kotlinx.coroutines.launch /** @@ -44,7 +45,7 @@ import kotlinx.coroutines.launch fun NotificationsScreen( api: Api, workspaceId: String, - onOpenItem: (itemId: String, isTicket: Boolean) -> Unit, + onOpenItem: (itemId: String, itemKind: String) -> Unit, onUnreadChanged: () -> Unit, refreshKey: Int = 0, ) { @@ -54,17 +55,25 @@ fun NotificationsScreen( var error by remember { mutableStateOf(null) } val scope = rememberCoroutineScope() - suspend fun reload() { - loading = true + suspend fun reload(silent: Boolean = false) { + if (!silent) loading = true error = null runCatching { api.notifications(workspaceId, unreadOnly) } .onSuccess { rows = it.notifications } .onFailure { error = verifyStatusCode((it as? ApiException)?.code ?: 0) } - loading = false + if (!silent) loading = false } LaunchedEffect(workspaceId, unreadOnly, refreshKey) { reload() } + // 停在通知页时也要能看到新动静：每 30 秒静默拉一次，不闪加载态。 + LaunchedEffect(workspaceId, unreadOnly) { + while (true) { + delay(30_000) + reload(silent = true) + } + } + Column(Modifier.fillMaxSize().padding(horizontal = 16.dp)) { Row( modifier = Modifier.fillMaxWidth(), @@ -85,6 +94,26 @@ fun NotificationsScreen( bg = if (unreadOnly) PaperSunken else PrimarySoft, modifier = Modifier.clickable { unreadOnly = false }, ) + Box(Modifier.weight(1f)) + SoftPill( + text = "刷新", + fg = InkMuted, + bg = PaperSunken, + modifier = Modifier.clickable { scope.launch { reload() } }, + ) + SoftPill( + text = "全部已读", + fg = InkMuted, + bg = PaperSunken, + modifier = Modifier.clickable { + scope.launch { + runCatching { api.markAllNotificationsRead(workspaceId) } + rows = rows.map { if (it.readAt == null) it.copy(readAt = "read") else it } + onUnreadChanged() + reload(silent = true) + } + }, + ) } Text( @@ -121,7 +150,14 @@ fun NotificationsScreen( contentPadding = androidx.compose.foundation.layout.PaddingValues(bottom = 20.dp), ) { items(rows, key = { it.id }) { n -> - val isTicket = n.kind.startsWith("ticket_") + // 知识/工单/任务三种落点：优先看后端给的 item_kind，老数据再退回 kind 前缀 + val isKnowledge = n.kind.startsWith("knowledge_") || n.itemKind == "knowledge" + val isTicket = !isKnowledge && (n.kind.startsWith("ticket_") || n.itemKind == "ticket") + val targetKind = when { + isKnowledge -> "knowledge" + isTicket -> "ticket" + else -> "task" + } SoftCard( Modifier.fillMaxWidth(), onClick = { @@ -131,7 +167,7 @@ fun NotificationsScreen( rows = rows.map { if (it.id == n.id) it.copy(readAt = "read") else it } onUnreadChanged() } - n.itemId?.let { onOpenItem(it, isTicket) } + n.itemId?.let { onOpenItem(it, targetKind) } } }, ) { @@ -143,7 +179,11 @@ fun NotificationsScreen( contentAlignment = Alignment.Center, ) { Text( - text = if (isTicket) "工" else "记", + text = when { + isKnowledge -> "知" + isTicket -> "工" + else -> "记" + }, fontSize = 11.sp, fontWeight = FontWeight.SemiBold, color = if (n.readAt == null) Primary else InkMuted, diff --git a/backend/alembic/versions/0017_notification_reminders.py b/backend/alembic/versions/0017_notification_reminders.py new file mode 100644 index 0000000..02faddd --- /dev/null +++ b/backend/alembic/versions/0017_notification_reminders.py @@ -0,0 +1,60 @@ +"""通知提醒：幂等键 + 到期扫描的只读口子 + +Revision ID: 0017 +Revises: 0016 +Create Date: 2026-11-10 + +两件事： + +1. **提醒的幂等键**。到期/逾期提醒由后台循环每几分钟扫一次，扫描本身有 + advisory lock 保证只有一个进程在跑，但进程重启、时钟抖动、人工重跑都可能 + 让同一条提醒再写一次。`dedupe_key` + 部分唯一索引把"同一件事只提醒一次" + 交给数据库保证，调用方用 `ON CONFLICT DO NOTHING` 即可。普通通知的 + `dedupe_key` 为空，不参与唯一约束。 + +2. **到期扫描要跨租户列出待办**。提醒是平台级定时任务，扫的时候没有"当前工作区"， + `item` 的隔离策略会让它一行都看不到。这里照 0016 的做法加一条只在 + `app.reminder_list = 'on'` 时生效的 **SELECT-only** 策略：扫描器能列出 + 全平台的截止时间，但写通知仍然必须逐个工作区绑定租户，改不了别人的数据。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +from app.roles import APP_ROLE + +revision: str = "0017" +down_revision: str = "0016" +branch_labels: Sequence[str] | None = None +depends_on: Sequence[str] | None = None + +CREATE_REMINDER_POLICY = """ +CREATE POLICY item_reminder_list ON item + FOR SELECT + USING (current_setting('app.reminder_list', true) = 'on') +""" + + +def upgrade() -> None: + op.add_column("notification", sa.Column("dedupe_key", sa.String(120), nullable=True)) + op.create_index( + "uq_notification_dedupe", + "notification", + ["workspace_id", "user_id", "dedupe_key"], + unique=True, + postgresql_where=sa.text("dedupe_key IS NOT NULL"), + ) + + op.execute("DROP POLICY IF EXISTS item_reminder_list ON item") + op.execute(CREATE_REMINDER_POLICY) + op.execute(f"GRANT SELECT ON item TO {APP_ROLE}") + + +def downgrade() -> None: + op.execute("DROP POLICY IF EXISTS item_reminder_list ON item") + op.drop_index("uq_notification_dedupe", table_name="notification") + op.drop_column("notification", "dedupe_key") diff --git a/backend/app/api/items.py b/backend/app/api/items.py index 59831d5..9e1859c 100644 --- a/backend/app/api/items.py +++ b/backend/app/api/items.py @@ -205,6 +205,48 @@ async def _apply_update( return item +async def _notify_status_change( + session: AsyncSession, + *, + ws_id: uuid.UUID, + item: Item, + previous: str, + to_status: str, + actor_id: uuid.UUID, + actor_name: str, +) -> None: + """状态变化的通知。 + + "完成"单独一种 kind：收件人一眼能看出是结项，而不是又一次普通的状态流转； + 其余状态仍走 `task_status_changed`，正文里写清从什么改成了什么。 + """ + if to_status == "done": + await notify( + session, + ws_id=ws_id, + user_ids=(item.creator_id, item.assignee_id), + kind="task_completed", + item_id=item.id, + title=f"「{item.title}」已完成", + body=f"{actor_name} 把它标记为完成。", + actor_id=actor_id, + ) + return + await notify( + session, + ws_id=ws_id, + user_ids=(item.creator_id, item.assignee_id), + kind="task_status_changed", + item_id=item.id, + title=f"「{item.title}」变成{TASK_STATUS_CN.get(to_status, to_status)}了", + body=( + f"{actor_name} 把它从 {TASK_STATUS_CN.get(previous, previous)} " + f"改成了 {TASK_STATUS_CN.get(to_status, to_status)}" + ), + actor_id=actor_id, + ) + + # ── 列表 ──────────────────────────────────────────────────────────────── def _apply_filters( stmt: Select, @@ -581,6 +623,17 @@ async def batch_status(body: BatchStatusIn, ctx: MemberDep, session: SessionDep) } ) continue + # 批量改状态以前是"悄悄改完"：被指派人/创建人收不到任何通知， + # 和单条改状态的行为不一致。这里补上同一条通知。 + await _notify_status_change( + session, + ws_id=ctx.ws_id, + item=item, + previous=previous, + to_status=body.status, + actor_id=ctx.user.id, + actor_name=ctx.user.nickname or "同事", + ) updated.append({"id": str(item.id), "version": item.version, "unchanged": False}) return ok( @@ -682,19 +735,14 @@ async def change_status( payload={"from": previous, "to": body.status}, ) tags = await _tags_of(session, ctx.ws_id, [item.id]) - await notify( + await _notify_status_change( session, ws_id=ctx.ws_id, - user_ids=(item.creator_id, item.assignee_id), - kind="task_status_changed", - item_id=item.id, - title=f"「{item.title}」变成{TASK_STATUS_CN.get(body.status, body.status)}了", - body=( - f"{ctx.user.nickname or '同事'} 把它从 " - f"{TASK_STATUS_CN.get(previous, previous)} 改成了 " - f"{TASK_STATUS_CN.get(body.status, body.status)}" - ), + item=item, + previous=previous, + to_status=body.status, actor_id=ctx.user.id, + actor_name=ctx.user.nickname or "同事", ) return ok(_item_out(item, tags.get(item.id)), message="状态已更新") diff --git a/backend/app/api/knowledge.py b/backend/app/api/knowledge.py index 2ab87c3..7daa464 100644 --- a/backend/app/api/knowledge.py +++ b/backend/app/api/knowledge.py @@ -45,6 +45,7 @@ from ..schemas import clamp_limit, decode_cursor, encode_cursor, ok from ..serializers import item_out, tags_of from ..services.knowledge import create_draft_from_source, find_doc_by_source from ..services.knowledge_index import reindex_document, search_knowledge +from ..services.notify import notify, workspace_admins from .attachments import count_for_items, list_for_item router = APIRouter(prefix="/workspaces/{ws_id}", tags=["知识库"]) @@ -1025,6 +1026,17 @@ async def create_knowledge(body: KnowledgeIn, ctx: MemberDep, session: SessionDe await session.flush() await _log(session, item, ctx.user.id, "knowledge_created", {"generated_by": "human"}) await _set_tags_for(session, ws_id, item, body.tag_names) + # 知识是团队资产：新建草稿先告诉能发布的人，别让草稿在列表里躺到没人发现。 + await notify( + session, + ws_id=ws_id, + user_ids=await workspace_admins(session, ws_id), + kind="knowledge_draft_created", + item_id=item.id, + title=f"新的知识草稿：{item.title}", + body=f"{ctx.user.nickname or '同事'} 新建了一条知识，核对后可以发布。", + actor_id=ctx.user.id, + ) await audit_record( session, workspace_id=ws_id, @@ -1204,6 +1216,42 @@ async def change_status( target_id=str(item.id), payload={"to": to_status}, ) + # 知识的状态变化要让相关的人知道：提交审议找管理员核对， + # 发布/归档则回告作者——否则作者不知道自己的草稿已经被团队采纳。 + actor_name = ctx.user.nickname or "同事" + if body.action == "submit_review": + await notify( + session, + ws_id=ws_id, + user_ids=await workspace_admins(session, ws_id), + kind="knowledge_review_requested", + item_id=item.id, + title=f"待核对：{item.title}", + body=f"{actor_name} 提交了一份知识，核对后可以发布。", + actor_id=ctx.user.id, + ) + elif body.action == "publish": + await notify( + session, + ws_id=ws_id, + user_ids=(doc.author_id,), + kind="knowledge_published", + item_id=item.id, + title=f"「{item.title}」已发布", + body=f"{actor_name} 发布了这份知识，现在可以被团队检索到。", + actor_id=ctx.user.id, + ) + elif body.action == "archive": + await notify( + session, + ws_id=ws_id, + user_ids=(doc.author_id,), + kind="knowledge_archived", + item_id=item.id, + title=f"「{item.title}」已归档", + body=f"{actor_name} 把它归档了，检索不再返回它。", + actor_id=ctx.user.id, + ) await session.flush() if body.action == "publish": # 发布后立刻建索引：检索只认 published，不在这里建就要等定时任务， @@ -1270,6 +1318,16 @@ async def promote_to_knowledge( await _log( session, doc_item, ctx.user.id, "knowledge_created", {"source_item_id": str(source.id)} ) + await notify( + session, + ws_id=ws_id, + user_ids=await workspace_admins(session, ws_id), + kind="knowledge_draft_created", + item_id=doc_item.id, + title=f"从事项沉淀了知识：{doc_item.title}", + body=f"{ctx.user.nickname or '同事'} 把「{source.title}」沉淀成草稿，核对后可以发布。", + actor_id=ctx.user.id, + ) await audit_record( session, workspace_id=ws_id, diff --git a/backend/app/api/tickets.py b/backend/app/api/tickets.py index 86b0c9f..aa69475 100644 --- a/backend/app/api/tickets.py +++ b/backend/app/api/tickets.py @@ -767,11 +767,14 @@ async def delete_link( # -- 通知 --------------------------------------------------------------- -def _notification_out(n: Notification, item_title: str | None = None) -> dict[str, Any]: +def _notification_out( + n: Notification, item_title: str | None = None, item_kind: str | None = None +) -> dict[str, Any]: return { "id": str(n.id), "kind": n.kind, "item_id": str(n.item_id) if n.item_id else None, + "item_kind": item_kind, "item_title": item_title, "title": n.title, "body": n.body, @@ -813,18 +816,27 @@ async def list_notifications( rows = list(rows[:limit]) titles: dict[uuid.UUID, str] = {} + kinds: dict[uuid.UUID, str] = {} item_ids = [n.item_id for n in rows if n.item_id] if item_ids: - for item_id, title in ( - await session.execute(select(Item.id, Item.title).where(Item.id.in_(item_ids))) + for item_id, title, kind in ( + await session.execute( + select(Item.id, Item.title, Item.kind).where(Item.id.in_(item_ids)) + ) ).all(): titles[item_id] = title + kinds[item_id] = kind next_cursor = encode_cursor(rows[-1].created_at, rows[-1].id) if has_more and rows else None return ok( { "notifications": [ - _notification_out(n, titles.get(n.item_id) if n.item_id else None) for n in rows + _notification_out( + n, + titles.get(n.item_id) if n.item_id else None, + kinds.get(n.item_id) if n.item_id else None, + ) + for n in rows ], "next_cursor": next_cursor, "has_more": has_more, diff --git a/backend/app/config.py b/backend/app/config.py index 128f9c0..f2d7abf 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -112,6 +112,14 @@ class Settings(BaseSettings): # 30 分钟够一次阅读/编辑，过期后刷新页面会重新签发。 attachment_url_ttl_seconds: int = 1800 + # ── 站内提醒 ────────────────────────────────────────── + # 到期/逾期提醒是后台轮询，不引入额外组件（见 services/reminders.py）。 + # 间隔决定"最晚多久能收到提醒"，5 分钟对截止时间以小时计的任务足够。 + reminders_enabled: bool = True + reminder_interval_seconds: int = 300 + # 提前多久提醒"快到期"。默认 24 小时；改成 0 等于只提醒逾期。 + reminder_due_soon_hours: int = 24 + # ── 知识检索（M3）────────────────────────────────────── # 切片大小与重叠：中文一段 600 字大约 400~600 token，够表达一个完整片段； # 重叠 80 字是为了不把跨段的话切断（检索命中时前后文还读得懂）。 diff --git a/backend/app/main.py b/backend/app/main.py index 7a0264b..60d3599 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -32,6 +32,7 @@ from .db import SessionFactory, current_app_role, engine from .errors import AppError, ErrorCode, http_status_for from .logging_setup import get_logger, request_id_ctx, setup_logging from .middleware import IdempotencyMiddleware, RequestContextMiddleware, error_body +from .services import reminders from .services.wechat_bot import manager as wechat_manager settings = get_settings() @@ -53,7 +54,10 @@ async def lifespan(app: FastAPI): ) # 已登录的微信通道在进程启动时接上，不然重启之后消息就静默收不到了。 await wechat_manager.start_all() + # 到期/逾期提醒：多 worker 下靠 advisory lock 保证只有一个进程在扫。 + reminders.start() yield + await reminders.shutdown() await wechat_manager.shutdown() await engine.dispose() diff --git a/backend/app/models_ticket.py b/backend/app/models_ticket.py index be9a59d..1362529 100644 --- a/backend/app/models_ticket.py +++ b/backend/app/models_ticket.py @@ -26,6 +26,7 @@ from sqlalchemy import ( Text, UniqueConstraint, func, + text, ) from sqlalchemy.dialects.postgresql import UUID as PgUUID from sqlalchemy.orm import Mapped, mapped_column @@ -121,6 +122,16 @@ class Notification(Base, Timestamps): __table_args__ = ( Index("ix_notification_ws_user_read", "workspace_id", "user_id", "read_at"), Index("ix_notification_ws_created", "workspace_id", "created_at"), + # 到期/逾期提醒由后台循环生成，可能被重复触发；带 dedupe_key 的写入用 + # ON CONFLICT DO NOTHING，把"同一件事只提醒一次"交给数据库保证。 + Index( + "uq_notification_dedupe", + "workspace_id", + "user_id", + "dedupe_key", + unique=True, + postgresql_where=text("dedupe_key IS NOT NULL"), + ), ) id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid) @@ -137,6 +148,9 @@ class Notification(Base, Timestamps): title: Mapped[str] = mapped_column(String(120), nullable=False) body: Mapped[str] = mapped_column(String(300), nullable=False, default="") read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + # 提醒类通知的幂等键（如 task_due_soon::）。 + # 普通通知为空，不参与唯一约束。 + dedupe_key: Mapped[str | None] = mapped_column(String(120)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) diff --git a/backend/app/services/notify.py b/backend/app/services/notify.py index 2dbfe2c..c9d856f 100644 --- a/backend/app/services/notify.py +++ b/backend/app/services/notify.py @@ -15,8 +15,11 @@ from __future__ import annotations import uuid from collections.abc import Iterable +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession +from ..models import Member from ..models_ticket import Notification TICKET_STATUS_CN = { @@ -48,6 +51,7 @@ async def notify( title: str, body: str = "", actor_id: uuid.UUID | None = None, + dedupe_key: str | None = None, ) -> int: """给若干人各写一条通知，返回实际写入条数。 @@ -67,7 +71,55 @@ async def notify( item_id=item_id, title=title[:120], body=body[:300], + dedupe_key=dedupe_key, ) ) count += 1 return count + + +async def notify_once( + session: AsyncSession, + *, + ws_id: uuid.UUID, + user_id: uuid.UUID | None, + kind: str, + item_id: uuid.UUID | None, + title: str, + body: str = "", + dedupe_key: str, + actor_id: uuid.UUID | None = None, +) -> bool: + """写一条带幂等键的通知；已经写过就跳过，返回是否真的写入。 + + 后台提醒循环会反复扫同一批任务，靠 `dedupe_key` 的唯一索引把重复挡在数据库层， + 比"先查再插"少一次竞态。提醒本身不是业务事实，丢了不致命，重复才烦人。 + """ + if user_id is None or user_id == actor_id: + return False + result = await session.execute( + pg_insert(Notification) + .values( + workspace_id=ws_id, + user_id=user_id, + kind=kind, + item_id=item_id, + title=title[:120], + body=body[:300], + dedupe_key=dedupe_key, + ) + .on_conflict_do_nothing() + ) + return result.rowcount > 0 + + +async def workspace_admins(session: AsyncSession, ws_id: uuid.UUID) -> list[uuid.UUID]: + """工作区里能"发布/把关"的人：owner + admin，只算正常成员。""" + rows = await session.execute( + select(Member.user_id).where( + Member.workspace_id == ws_id, + Member.status == "active", + Member.role.in_(("owner", "admin")), + ) + ) + return [row[0] for row in rows] diff --git a/backend/app/services/reminders.py b/backend/app/services/reminders.py new file mode 100644 index 0000000..df15086 --- /dev/null +++ b/backend/app/services/reminders.py @@ -0,0 +1,215 @@ +"""到期与逾期提醒。 + +为什么是"轮询 + 数据库租约"而不是引入 Celery/APScheduler： + +1. 生产是 gunicorn 多 worker，进程内定时器会各跑一遍，通知就重复了。 + `pg_try_advisory_lock` 是现成的单实例租约，和微信机器人用的是同一套做法， + 不额外增加运维组件。 +2. 提醒的准确度要求是"几分钟内"，不是秒级；一轮扫描就是一条按 + `(workspace_id, status, due_at)` 索引的查询，代价可以忽略。 +3. 重复触发用 `notification.dedupe_key` 兜底：扫描器重启、时钟抖动、 + 人工补跑都不会把同一条提醒写两遍。 + +扫描本身跨租户（平台级任务），所以先在 `app.reminder_list = 'on'` 的只读口子下 +把候选任务列出来，再逐个工作区绑定租户写通知——读可以跨，写必须归位。 +""" + +from __future__ import annotations + +import asyncio +import math +import uuid +from contextlib import suppress +from datetime import datetime, timedelta, timezone +from typing import Any +from zoneinfo import ZoneInfo + +from sqlalchemy import select, text + +from ..config import get_settings +from ..db import SessionFactory, bind_tenant, engine +from ..logging_setup import get_logger +from ..models_task import Item +from .notify import notify_once + +logger = get_logger("app.reminders") + +_LOCK_NAMESPACE = "dadao.reminders" +# 一轮最多处理这么多条：正常量级远小于它；真到了这个量级，说明该拆任务而不是加锁。 +_SCAN_LIMIT = 500 + +_task: asyncio.Task[None] | None = None +_stopping = asyncio.Event() + + +async def _candidates() -> list[dict[str, Any]]: + """列出所有"还没完成、有截止时间、已经进入提醒窗口"的事项。""" + settings = get_settings() + now = datetime.now(timezone.utc) + horizon = now + timedelta(hours=max(1, settings.reminder_due_soon_hours)) + async with SessionFactory() as session: + await session.execute(text("SELECT set_config('app.reminder_list', 'on', true)")) + rows = ( + await session.execute( + select( + Item.id, + Item.workspace_id, + Item.title, + Item.status, + Item.due_at, + Item.creator_id, + Item.assignee_id, + ) + .where( + Item.deleted_at.is_(None), + Item.kind != "knowledge", + Item.status.not_in(("done", "canceled")), + Item.due_at.is_not(None), + Item.due_at <= horizon, + ) + .order_by(Item.due_at) + .limit(_SCAN_LIMIT) + ) + ).all() + return [dict(row._mapping) for row in rows] + + +def _local(dt: datetime) -> datetime: + return dt.astimezone(ZoneInfo(get_settings().default_timezone)) + + +def _due_soon_message(row: dict[str, Any], now: datetime) -> tuple[str, str]: + due_at = row["due_at"] + hours = max(1, math.ceil((due_at - now).total_seconds() / 3600)) + return ( + f"「{row['title']}」{hours} 小时内到期", + f"截止时间 {_local(due_at):%m-%d %H:%M}，请尽快处理。", + ) + + +def _overdue_message(row: dict[str, Any], now: datetime) -> tuple[str, str]: + due_at = row["due_at"] + late = now - due_at + days = late.days + if days >= 1: + tail = f"已经逾期 {days} 天" + else: + tail = f"已经逾期 {max(1, int(late.total_seconds() // 3600))} 小时" + return ( + f"「{row['title']}」已逾期", + f"原定 {_local(due_at):%m-%d %H:%M} 完成，{tail}。", + ) + + +async def _remind_workspace(ws_id: uuid.UUID, rows: list[dict[str, Any]], now: datetime) -> int: + sent = 0 + async with SessionFactory() as session, session.begin(): + await bind_tenant(session, ws_id, None) + for row in rows: + recipient = row["assignee_id"] or row["creator_id"] + if recipient is None: + continue + due_at: datetime = row["due_at"] + if due_at > now: + title, body = _due_soon_message(row, now) + kind = "task_due_soon" + dedupe = f"task_due_soon:{row['id']}:{due_at.isoformat()}" + else: + title, body = _overdue_message(row, now) + kind = "task_overdue" + today = _local(now).date().isoformat() + dedupe = f"task_overdue:{row['id']}:{due_at.isoformat()}:{today}" + if await notify_once( + session, + ws_id=ws_id, + user_id=recipient, + kind=kind, + item_id=row["id"], + title=title, + body=body, + dedupe_key=dedupe, + ): + sent += 1 + return sent + + +async def tick() -> dict[str, int]: + """扫一轮。返回本轮的统计，供日志与测试使用。""" + settings = get_settings() + if not settings.reminders_enabled: + return {"scanned": 0, "sent": 0, "skipped": 1} + + lock_conn = await engine.connect() + try: + got = ( + await lock_conn.execute( + text("SELECT pg_try_advisory_lock(hashtext(:ns))"), {"ns": _LOCK_NAMESPACE} + ) + ).scalar() + if not got: + # 另一个 worker 正在扫，这一轮让给它。 + return {"scanned": 0, "sent": 0, "skipped": 1} + + now = datetime.now(timezone.utc) + rows = await _candidates() + by_ws: dict[uuid.UUID, list[dict[str, Any]]] = {} + for row in rows: + by_ws.setdefault(row["workspace_id"], []).append(row) + + sent = 0 + for ws_id, group in by_ws.items(): + sent += await _remind_workspace(ws_id, group, now) + if sent: + logger.info( + "到期提醒已发送", + extra={"extra_fields": {"scanned": len(rows), "sent": sent}}, + ) + return {"scanned": len(rows), "sent": sent, "skipped": 0} + finally: + # session 级 advisory lock 挂在连接上；解锁后连接才能安全回到池里。 + # 连接已断时解锁失败无需处理，连接本身也回不了池。 + with suppress(Exception): + await lock_conn.execute( + text("SELECT pg_advisory_unlock(hashtext(:ns))"), {"ns": _LOCK_NAMESPACE} + ) + await lock_conn.close() + + +async def _loop() -> None: + settings = get_settings() + while not _stopping.is_set(): + try: + await tick() + except asyncio.CancelledError: + raise + except Exception: # pragma: no cover - 单轮失败不该把循环打死 + logger.exception("到期提醒扫描失败") + try: + await asyncio.wait_for( + _stopping.wait(), timeout=max(5, settings.reminder_interval_seconds) + ) + except asyncio.TimeoutError: + continue + + +def start() -> None: + """启动后台提醒循环。测试环境与关闭开关时直接不启动。""" + global _task + settings = get_settings() + if settings.env == "test" or not settings.reminders_enabled: + return + if _task is not None and not _task.done(): + return + _stopping.clear() + _task = asyncio.create_task(_loop(), name="dadao-reminders") + + +async def shutdown() -> None: + global _task + _stopping.set() + if _task is None: + return + _task.cancel() + with suppress(asyncio.CancelledError, Exception): + await _task + _task = None diff --git a/backend/tests/test_notifications.py b/backend/tests/test_notifications.py index 756f2c5..52edb97 100644 --- a/backend/tests/test_notifications.py +++ b/backend/tests/test_notifications.py @@ -11,11 +11,14 @@ from __future__ import annotations +from datetime import datetime, timedelta, timezone from typing import Any import pytest from httpx import AsyncClient +from app.services import reminders + from .conftest import create_workspace, login, unwrap V1 = "/api/v1" @@ -269,3 +272,223 @@ async def test_unread_count_and_read_all(client: AsyncClient) -> None: ) ) assert count["unread"] == 0 + + +async def _set_status( + client: AsyncClient, user: dict[str, Any], ws_id: str, item: dict, status: str +) -> dict: + return unwrap( + await client.post( + f"{V1}/workspaces/{ws_id}/items/{item['id']}/status", + json={"status": status, "version": item["version"]}, + headers=user["headers"], + ) + ) + + +@pytest.mark.asyncio +async def test_task_completion_notifies_the_creator(client: AsyncClient) -> None: + """派出去的活被干完了，派活的人要知道——但干活的人不该收到"你完成了"。""" + owner = await login(client) + ws = await create_workspace(client, owner, "完成通知") + helper = await _member(client, owner, ws, "member") + item = await _task_assigned_to(client, owner, ws["id"], helper["user_id"], "季度盘点") + + await _set_status(client, helper, ws["id"], item, "done") + + owner_inbox = await _inbox(client, owner, ws["id"]) + completed = [n for n in owner_inbox if n["kind"] == "task_completed"] + assert len(completed) == 1 + assert completed[0]["item_id"] == item["id"] + assert completed[0]["item_kind"] == "task" + assert "已完成" in completed[0]["title"] + assert not [n for n in await _inbox(client, helper, ws["id"]) if n["kind"] == "task_completed"] + + +@pytest.mark.asyncio +async def test_batch_status_sends_the_same_notifications(client: AsyncClient) -> None: + """看板批量拖拽不能静默：拖 3 条完成，创建人就该收到 3 条完成通知。""" + owner = await login(client) + ws = await create_workspace(client, owner, "批量完成通知") + helper = await _member(client, owner, ws, "member") + items = [ + await _task_assigned_to(client, owner, ws["id"], helper["user_id"], f"批量任务 {i}") + for i in range(3) + ] + + unwrap( + await client.post( + f"{V1}/workspaces/{ws['id']}/items/batch-status", + json={ + "status": "done", + "items": [{"id": item["id"], "version": item["version"]} for item in items], + }, + headers=helper["headers"], + ) + ) + + owner_inbox = await _inbox(client, owner, ws["id"]) + completed = {n["item_id"] for n in owner_inbox if n["kind"] == "task_completed"} + assert completed == {item["id"] for item in items} + assert not [n for n in await _inbox(client, helper, ws["id"]) if n["kind"] == "task_completed"] + + +@pytest.mark.asyncio +async def test_knowledge_notifications_follow_the_flow(client: AsyncClient) -> None: + """知识库的四个节点都要有人知道：新草稿、待核对、已发布、已归档。""" + owner = await login(client) + ws = await create_workspace(client, owner, "知识通知") + admin = await _member(client, owner, ws, "admin") + + doc = unwrap( + await client.post( + f"{V1}/workspaces/{ws['id']}/knowledge", + json={"title": "报价流程", "body_md": "先核价，再报给客户"}, + headers=admin["headers"], + ) + ) + + # 管理员新建草稿 → 其他管理员（owner）收到 + created = [ + n + for n in await _inbox(client, owner, ws["id"]) + if n["kind"] == "knowledge_draft_created" and n["item_id"] == doc["id"] + ] + assert len(created) == 1 + assert created[0]["item_kind"] == "knowledge" + + # 提交核对 → owner 收到"待核对" + doc = unwrap( + await client.post( + f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status", + json={"action": "submit_review", "version": doc["version"]}, + headers=admin["headers"], + ) + ) + assert any( + n["kind"] == "knowledge_review_requested" and n["item_id"] == doc["id"] + for n in await _inbox(client, owner, ws["id"]) + ) + + # 发布 → 作者（admin）收到 + doc = unwrap( + await client.post( + f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status", + json={"action": "publish", "version": doc["version"]}, + headers=owner["headers"], + ) + ) + published = [ + n + for n in await _inbox(client, admin, ws["id"]) + if n["kind"] == "knowledge_published" and n["item_id"] == doc["id"] + ] + assert len(published) == 1 + assert published[0]["item_kind"] == "knowledge" + + # 归档 → 作者（admin）收到 + unwrap( + await client.post( + f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status", + json={"action": "archive", "version": doc["version"]}, + headers=owner["headers"], + ) + ) + assert any( + n["kind"] == "knowledge_archived" and n["item_id"] == doc["id"] + for n in await _inbox(client, admin, ws["id"]) + ) + + +async def _task_with_due( + client: AsyncClient, + owner: dict[str, Any], + ws_id: str, + assignee_id: str, + title: str, + due_at: datetime, + status: str = "todo", +) -> dict: + return unwrap( + await client.post( + f"{V1}/workspaces/{ws_id}/items", + json={ + "title": title, + "status": status, + "assignee_id": assignee_id, + "due_at": due_at.isoformat(), + }, + headers=owner["headers"], + ) + ) + + +@pytest.mark.asyncio +async def test_reminders_cover_due_soon_and_overdue_once(client: AsyncClient) -> None: + """扫描器只负责"该提醒的人收到且只收到一次"，不负责业务判断。""" + owner = await login(client) + ws = await create_workspace(client, owner, "到期提醒") + helper = await _member(client, owner, ws, "member") + now = datetime.now(timezone.utc) + + soon = await _task_with_due( + client, owner, ws["id"], helper["user_id"], "明早交付", now + timedelta(hours=3) + ) + late = await _task_with_due( + client, owner, ws["id"], helper["user_id"], "昨天该交", now - timedelta(days=1) + ) + await _task_with_due( + client, + owner, + ws["id"], + helper["user_id"], + "已经做完的", + now + timedelta(hours=1), + status="done", + ) + + first = await reminders.tick() + assert first["sent"] == 2, first + + inbox = await _inbox(client, helper, ws["id"]) + due_soon = [n for n in inbox if n["kind"] == "task_due_soon"] + overdue = [n for n in inbox if n["kind"] == "task_overdue"] + assert [n["item_id"] for n in due_soon] == [soon["id"]] + assert [n["item_id"] for n in overdue] == [late["id"]] + assert "截止时间" in due_soon[0]["body"] + assert "已逾期" in overdue[0]["title"] + + # 第二轮不能重复写：dedupe_key 兜底 + second = await reminders.tick() + assert second["sent"] == 0, second + inbox = await _inbox(client, helper, ws["id"]) + assert len([n for n in inbox if n["kind"] == "task_due_soon"]) == 1 + assert len([n for n in inbox if n["kind"] == "task_overdue"]) == 1 + + +@pytest.mark.asyncio +async def test_reminders_stay_inside_each_workspace(client: AsyncClient) -> None: + """提醒是平台级扫描，但写通知必须回到各自工作区，不能串租户。""" + owner_a = await login(client) + ws_a = await create_workspace(client, owner_a, "提醒甲") + helper_a = await _member(client, owner_a, ws_a, "member") + + owner_b = await login(client) + ws_b = await create_workspace(client, owner_b, "提醒乙") + helper_b = await _member(client, owner_b, ws_b, "member") + + now = datetime.now(timezone.utc) + item_a = await _task_with_due( + client, owner_a, ws_a["id"], helper_a["user_id"], "甲的逾期", now - timedelta(hours=2) + ) + item_b = await _task_with_due( + client, owner_b, ws_b["id"], helper_b["user_id"], "乙的逾期", now - timedelta(hours=2) + ) + + result = await reminders.tick() + assert result["sent"] == 2, result + + inbox_a = await _inbox(client, helper_a, ws_a["id"]) + inbox_b = await _inbox(client, helper_b, ws_b["id"]) + assert {n["item_id"] for n in inbox_a if n["kind"] == "task_overdue"} == {item_a["id"]} + assert {n["item_id"] for n in inbox_b if n["kind"] == "task_overdue"} == {item_b["id"]} diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 07e2741..8a102d2 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -224,6 +224,8 @@ export interface AppNotification { id: string; kind: string; item_id: string | null; + /** 事项类型：知识要跳阅读页，工单跳工单页，任务回任务页。 */ + item_kind: "task" | "ticket" | "knowledge" | null; item_title: string | null; title: string; body: string; diff --git a/web/src/pages/Notifications.tsx b/web/src/pages/Notifications.tsx index 9a970e1..b512573 100644 --- a/web/src/pages/Notifications.tsx +++ b/web/src/pages/Notifications.tsx @@ -2,21 +2,33 @@ import { useCallback, useEffect, useMemo, useState } from "react"; import { useNavigate, useParams } from "react-router-dom"; import { AppShell } from "../components/AppShell"; -import { Bell, Check, TicketIcon, Inbox, Clock } from "../components/icons"; +import { Bell, Book, Check, TicketIcon, Inbox, Clock, Flag } from "../components/icons"; import { Button, EmptyState, PageHeader, SkeletonRows } from "../components/ui"; import { api, ApiError } from "../lib/client"; import type { AppNotification } from "../lib/types"; const PAGE = 30; -/** 工单类通知点开跳到工单页，其余回任务页——都带 `?open=`，落地即展开那一条。 */ +/** + * 点开跳到哪：知识走独立阅读页（整屏），工单去工单页，其余回任务页—— + * 后两者都带 `?open=`，落地即展开那一条。判断以 `item_kind` 为准， + * 因为"知识被评论"的通知 kind 是 item_commented，光看 kind 会跳错页。 + */ function targetOf(wsId: string, n: AppNotification): string { if (!n.item_id) return `/w/${wsId}/notifications`; - const isTicket = n.kind.startsWith("ticket_"); - return isTicket ? `/w/${wsId}/tickets?open=${n.item_id}` : `/w/${wsId}?open=${n.item_id}`; + if (n.item_kind === "knowledge" || n.kind.startsWith("knowledge_")) { + return `/w/${wsId}/knowledge/${n.item_id}`; + } + if (n.item_kind === "ticket" || n.kind.startsWith("ticket_")) { + return `/w/${wsId}/tickets?open=${n.item_id}`; + } + return `/w/${wsId}?open=${n.item_id}`; } function iconOf(kind: string) { + if (kind.startsWith("knowledge_")) return ; + if (kind === "task_due_soon") return ; + if (kind === "task_overdue") return ; if (kind.startsWith("ticket_")) return ; if (kind === "item_commented") return ; return ; @@ -113,7 +125,7 @@ export function NotificationsPage() {