feat(webhooks): implement outbound webhook management endpoints and remove scheduled messages feature
ZhenYi committed
This change is very large, so some diffs are not shown. Clone the repository to see the full change.
Modified docs/业务架构.md docs/业务架构.md +15 −9 View docs/业务架构.md
@@ -12,7 +12,7 @@
1212| 能力线 | 对标产品 | 承载 crate | 完成度 |
1313|---|---|---|---|
1414| Git 仓库托管（HTTP/SSH/LFS/保护规则/代码浏览与分析） | Gitea / GitLab | `lib/git` | 主流程完整 |
15| 团队即时通讯（频道/线程/DM/表情/附件/在场/定时与提醒） | Slack / Discord | `lib/chat` | 主流程完整 |
15| 团队即时通讯（频道/线程/DM/表情/附件/在场/提醒） | Slack / Discord | `lib/chat` | 主流程完整 |
1616| 项目与工单（看板/标签/里程碑/依赖/工时/事件线） | Linear / Jira | `lib/issues` | 主流程完整 |
1717| 组织与租户（空间/组/成员/目录树/覆写/邀请/封禁/加入规则） | 自研模型 | `lib/space` | 完整 |
1818| 身份与账号安全（口令/邮箱验证/2FA/恢复码/PAT/SSH Key/PoW/注销） | 自研模型 | `lib/auth` `lib/user` | 完整 |
@@ -283,18 +283,25 @@
283283### 5.5 聊天消息链路
284284
285285```
286发送: 幂等 nonce(房间作用域) → 慢发言模式检查 → [事务] chat_room_seq 取号 → chat_message
286发送: 幂等 nonce(房间+作者作用域) → 慢发言模式检查 → [事务] chat_room_seq 取号 → chat_message
287287 → 事件发布到 Redis chat:room/{uuid} → 订阅该房间的 WS 连接收到 event{room_id, type:message_published}
288288 → @提及 额外写 chat:user/{uid}（mention）+ user_notifications + 邮件（尊重静音/通知级别，tokio::spawn 异步）
289 → @everyone 需管理权限；异步扇出到成员（尊重 suppress_everyone/静音/通知级别），编辑消息会重解析提及
289290线程: 根消息 → chat_thread(root_message_id 唯一) + chat_thread_member(notify_level, last_read_seq)
290291附件: POST /chat/attachments 拿 presign PUT（≤100MiB，危险 content-type 拒绝，
291292 待处理附件 ≤200/人）→ 客户端直传 S3 → POST .../finalize（服务端 head() 取权威 size/type）
292293 → send_message 时校验“属于本人 + 未绑定 + 已上传”的数量吻合后绑定 message_id/room_id
294 → 转发复制附件行（共享 S3 对象，删除时校验共享）；worker 每小时清理过期孤儿附件与 S3 对象
293295下载: GET /chat/attachments/{id}/file → 服务端流式代理 + nosniff + inline 白名单
294296已读: PUT /chat/rooms/{id}/read-state 单调推进 → 触发 read_marker 用户事件
295297在场: PUT /chat/presence → Redis EX 300 写穿 + chat_presence 落库；读时 Redis 优先，回落 DB 视为 Offline
296定时/提醒: 202 受理 → user_scheduled_message(chat payload) / chat_message_reminder → apihost worker 到点投递
297Inbound Webhook: POST /chat/hooks/{token}（SHA-256(token) 查房 + 30/60s 限流）→ 以机器人身份落消息
298提醒: chat_message_reminder → apihost worker 到点投递（失败回退 Pending 重试）+ WS 用户事件
299链接卡片: 发送后异步抓取 http(s) 链接（≤3 条，1MiB/10s，拒绝私网地址）→ og:/title 元数据
300 → chat_link_unfurl 缓存(7 天) → 回填消息 embeds → 广播 message_edited
301Inbound Webhook: POST /chat/hooks/{token}（SHA-256(token) 查房 + 30/60s 限流）→ 以机器人身份落消息（可指定 thread_id）
302Outbound Webhook: 房间事件（message_published/edited/deleted）→ chat_webhook_delivery 队列
303 → worker 投递（HMAC-SHA256 签名、SSRF 防护、60s×attempts 退避、5 次后 Failed）
304DM: 1:1（dm:{lo}:{hi} 幂等键）与群 DM（≤10 人，成员集 SHA-256 幂等键）
298305```
299306
300307### 5.6 代码浏览（前端如何拿到页面）
@@ -307,7 +314,7 @@
307314
308315## 6. 业务能力清单（按对外 API 面）
309316
310355 个 operation / 246 个 path（`openapi.json`）。
317407 个 operation / 281 个 path（`openapi.json`）。
311318
312319| 域 | 能力 |
313320|---|---|
@@ -315,7 +322,7 @@
315322| **自助**（`/me`，28） | 资料/用户名/头像（上传-读-删）/偏好（主题、通知位标志、DM 与提及受众）、多邮箱与设主、会话列表与吊销、设备管理、通知与未读数、推送订阅、近期提及、收藏消息、表情包(meme)、举报、注销申请/查看/撤销、停权记录、我的空间 |
316323| **用户公开面**（`/users`，5） | 列表/按 ID/按用户名/头像/公开主页详情（唯一公开端点） |
317324| **空间**（`/spaces`，26） | 空间 CRUD、成员增删改查与自改、组 CRUD 与加入退出、目录树读写/排序/父级修复、组与用户覆写、邀请生成/接受/撤销、封禁与解禁、加入规则、四类空间图片（头像/横幅/邀请图/启动图） |
318| **聊天**（`/chat`，53） | 房间列表/建/改/删/加入/离开/成员管理/转让、邀请、消息收发/编辑/删除/批量删除/转发/置顶/历史、线程、反应、已读、DM、附件、表情、书签、草稿、在场、搜索（Postgres FTS）、提醒、房间 Webhook、WS 与 ticket |
325| **聊天**（`/chat`，55） | 房间列表/建/改/删/加入/离开/成员管理/转让、邀请、消息收发/编辑/删除/批量删除/转发/置顶/历史、线程、反应、已读、DM（含群 DM）、附件、表情、书签、草稿、在场、搜索（FTS + CJK 子串回退）、提醒、链接卡片、房间入站/出站 Webhook、WS 与 ticket |
319326| **工单**（`/issues`，39） | 看板 CRUD、工单 CRUD + 批量状态 + 关/重开、评论（含树）、标签与绑定、里程碑、反应、关注、成员（负责人/提及/参与者）、置顶、依赖与反向依赖、事件线、内容历史、计时器与工时、仓库关联、已读 |
320327| **Git**（`/repos` 等，67 + `/graphql`） | 仓库 CRUD/归档、分支（建改删与默认分支）、标签、提交与状态、树/对象/路径历史、差异与补丁、blame、子模块、签名校验、冲突三方检测、远程导入/拉/推、LFS 锁与清理、Hooks 脚本执行、Pipelines 触发/排程/查询/上报/取消、语言与贡献者与活跃度分析、归档与 bundle、star/watch/topic、分支与标签保护白名单 |
321328| **内部**（`/api/v1/internal/*`，不计文档） | refs 强列/强改/解析、提交同步、patch apply、pack 统计/入库/repack、maintenance prune/fsck、LFS 清理、内容原语写入、license 写入、pipeline cancel/report |
@@ -357,7 +364,7 @@
357364
358365## 9. 已实现 / 预留 / 未接通（路线图基线）
359366
360**完整主流程**：账号与凭据体系、空间与组与成员、目录树与覆写、聊天全域（含 WS/附件/线程/搜索/提醒/定时消息）、工单全域、仓库托管（HTTP+SSH+LFS+保护规则+分析+归档+浏览片段）、push 后同步、GraphQL 只读查询。
367**完整主流程**：账号与凭据体系、空间与组与成员、目录树与覆写、聊天全域（含 WS/附件/线程/搜索/提醒）、工单全域、仓库托管（HTTP+SSH+LFS+保护规则+分析+归档+浏览片段）、push 后同步、GraphQL 只读查询。
361368
362369**数据模型已就位、业务未接通**（按投入产出比可作为下一步）：
363370
@@ -369,8 +376,7 @@
369376| Deploy Key | `repo_deploy_keys.rs` 存在但未在 `repo/mod.rs` 声明（不参与编译）；SSH 侧无接入 |
370377| 流水线执行 | API 面完整（trigger/schedule/report/cancel/list）+ `pipeline_run` 状态机；**Runner 与触发编排不在本仓** |
371378| 系统管理面 | `system_settings/roles/permissions/announcements/tasks/rate_limits/signing_keys/security_keys` 7 表零读写；WebAuthn/passkey 无实现 |
372| 用户关系 | 关注/粉丝、拉黑、日程消息：有表有实体，**无 service/handler/路由** |
373| 链接卡片 | `chat_link_unfurl` 表与状态枚举存在，无 worker，消息 `embeds` 恒为 `[]` |
379| 用户关系 | 关注/粉丝、拉黑：有表有实体，**无 service/handler/路由**（定时消息功能已整体移除） |
374380| 空间策略 | `allow_join/allow_invite/allow_join_level`、`space_warn`、封禁细分豁免位：字段在、行为无 |
375381
376382**已实现但与预期不符的细节**（业务侧需知晓）：工单描述/正文/优先级/截止日变更不进事件线（只在内容历史）；置顶与仓库关联不产生事件；工单邮件不发用户偏好过滤（聊天会过滤）；注销不清理空间成员/仓库/聊天/头像对象；默认安装 `email.enabled = false` 下验证邮件不发；邮件内链接指向未注册的路由。
Modified docs/技术架构.md docs/技术架构.md +7 −8 View docs/技术架构.md
@@ -234,18 +234,18 @@
234234 - 复合索引；
235235 - `DO $$ ... pg_constraint` 保护的 CHECK 约束（`comment_target_xor`、`issue_reaction_target_exactly_one`、`issue_watch_target_xor`、`issue_event_target_xor` …）；
236236 - **Postgres FTS**：给 `chat_message` 追加 `search_vector tsvector` + GIN 索引 + `BEFORE INSERT OR UPDATE OF content` 触发器（`schema_extras.sql:172-186`）。该列不在实体里，`chat::services::search::search_messages` 用裸 SQL 读写（`chat/src/services/search.rs:77-115`）。
237- 已知冲突：`idx_chat_message_room_nonce` 在 extras 里被声明两次不同键集（`:121` 与 `:165`），`IF NOT EXISTS` 让先出现的 `(room_id, nonce)` 生效，而代码幂等预检按 `(room_id, author_id, nonce)`（`chat/src/services/messages.rs:345-350`）。
237- 已知冲突已修复：`idx_chat_message_room_nonce` 现仅有一份定义 `(room_id, author_id, nonce)`（`schema_extras.sql`），与代码幂等预检一致（`chat/src/services/messages.rs`）。
238238
239239### 5.3 表的领域分布
240240
241241| 领域前缀 | 代表表 |
242242|---|---|
243| identity | `users` `user_email` `user_profile` `user_preferences` `user_session` `user_devices` `user_access_key` `user_ssh_key` `user_recovery_codes` `user_agreement_acceptances` `user_email_verification_codes` `user_email_change_ticket` `user_deletion_requests` `user_suspensions` `user_reports` `user_notifications` `user_followers` `user_blocks` `user_push_subscription` `user_saved_message` `user_scheduled_message` `user_recent_mention` `user_favorite_meme` |
243| identity | `users` `user_email` `user_profile` `user_preferences` `user_session` `user_devices` `user_access_key` `user_ssh_key` `user_recovery_codes` `user_agreement_acceptances` `user_email_verification_codes` `user_email_change_ticket` `user_deletion_requests` `user_suspensions` `user_reports` `user_notifications` `user_followers` `user_blocks` `user_push_subscription` `user_saved_message` `user_recent_mention` `user_favorite_meme` |
244244| tenancy | `spaces` `space_members` `space_groups` `space_group_permissions` `space_member_groups` `space_invite` `space_invite_groups` `space_bans` `space_warn` `space_join_rule` |
245245| 通用树 | `dir` `dir_overwrite_group` `dir_overwrite_user` |
246246| git | `repos` `repo_collaboration` `repo_star` `repo_watch` `repo_topic` `repo_relation_topic` `repo_license` `repo_release` `repo_branch_protection_policy/whitelist` `repo_tag_protection_policy/whitelist` `pipeline_run` `git_branch` `git_commit` `git_commit_status` `git_tag` `git_language` `git_lfs` `git_lfs_lock` |
247247| issue | `issues`(看板) `issue` `comment` `issue_label` `issue_relation_label` `issue_dependency` `issue_relation_repo` `issue_relation_pull` `issue_member` `issue_user` `issue_watch` `issue_pin` `issue_reaction` `issue_event_line` `issue_content_history` `milestone` `tracked_time` `stopwatch` |
248| chat | `rooms` `room_members` `room_invites` `room_bans` `chat_message` `chat_room_seq` `chat_thread(_member)` `chat_reaction` `chat_read_state` `chat_room_user_setting` `chat_attachment` `chat_emoji` `chat_room_bookmark` `chat_user_draft` `chat_room_webhook` `chat_message_edit` `chat_message_reminder` `chat_presence` |
248| chat | `rooms` `room_members` `room_invites` `room_bans` `chat_message` `chat_room_seq` `chat_thread(_member)` `chat_reaction` `chat_read_state` `chat_room_user_setting` `chat_attachment` `chat_emoji` `chat_room_bookmark` `chat_user_draft` `chat_room_webhook` `chat_outbound_webhook` `chat_webhook_delivery` `chat_link_unfurl` `chat_message_edit` `chat_message_reminder` `chat_presence` |
249249| 保留 | `products` `product_versions` `product_installs` `product_reviews` `pull_request(_comment/_review/_event_line)` `webhook` `repo_webhook` `webhook_delivery` `system_settings` `system_roles` `system_permissions` `system_announcements` `system_tasks` `system_rate_limits` `system_signing_keys` `system_security_keys` |
250250
251251### 5.4 ID / 时间约定
@@ -507,13 +507,13 @@
507507
508508| 循环 | 所在进程 | 周期 | 职责 |
509509|---|---|---|---|
510| `chat::worker::run_background_tasks` | **apihost** | 30s tick，每 120 tick(1h) 清理提及 | 到期提醒(≤200)、定时消息(≤100)、剪除 90 天前提及；卡死状态 `Delivering/Sending → Pending`（10 分钟） |
510| `chat::worker::run_background_tasks` | **apihost** | 30s tick，每 120 tick(1h) 清理提及与孤儿附件 | 到期提醒(≤200)、出站 Webhook 投递(≤50)、剪除 90 天前提及、清理过期未用附件；卡死状态 `Delivering → Pending`（10 分钟） |
511511| `auth::worker::run_deletion_executor` | **apihost** | 3600s | 到期注销执行（脱敏而非删行） |
512512| `SyncWorker` | synchost | BRPOP 阻塞 | push 后元数据同步（§8.5） |
513513| `email subscribers` | mails | 常驻 | 订阅 `email.send/batch/render` |
514514
515515两个 apihost 内的循环都依赖 **DB 原子状态抢占**（`update_many ... WHERE status = Pending`）保证多副本正确性，而非分布式锁。
516`is_rate_limited()` 的错误会被重新排队为 `Pending`，其他错误置 `Failed`（`chat/src/worker.rs:70-82`）。
516出站 Webhook 投递失败按 `60s × attempts` 退避重试，5 次后置 `Failed`；请求带 `X-Gitdata-Signature`（HMAC-SHA256）并拒绝私网/环回地址。
517517
518518**邮件队列是 Redis Pub/Sub，不是 Stream**（`lib/email/src/queue.rs`）：`subscribe` 收到的 `group_id` 立即 `let _ = group_id;` 丢弃（`:122`），配置里的 `subscriber_group_id` 无效。后果：无持久化、无 ack、无消费组、无死信、无重放；**多实例是扇出复制而非竞争消费 ⇒ 水平扩容会重复发信**。重试是进程内包裹 handler（`run_with_retries:267-311`，超过 `max_retries` 直接丢弃）。集群 URL 时 pub/sub 只连第一个节点，该节点故障 ⇒ 投递静默停止。
519519TLS 模式仅按端口推断（465 隐式 TLS / 587 STARTTLS / 其他 `builder_dangerous()` 明文，`email/src/transport.rs:74-104`）；`response_timeout_ms` 从未应用；凭据存在时强制 `Mechanism::Plain`。
@@ -556,7 +556,7 @@
556556| 前缀 | path 数 | 域 |
557557|---|---:|---|
558558| `/repos` | 64 | 仓库 REST + 20 个 HTML 片段 + pipelines + LFS 锁 |
559| `/chat` | 53 | 55 条路由注册（房间/消息/线程/表情/附件/提醒/草稿/书签/在场/搜索/Webhook/WS） |
559| `/chat` | 55 | 57 条路由注册（房间/消息/线程/表情/附件/提醒/草稿/书签/在场/搜索/入站与出站 Webhook/WS） |
560560| `/issues` | 39 | 看板/工单/评论/标签/里程碑/工时/依赖/关联 |
561561| `/me` | 28 | 当前用户自助 |
562562| `/spaces` | 26 | 租户/成员/组/目录/覆写/邀请/封禁/加入规则 |
@@ -597,12 +597,11 @@
597597| 8 | `verify_access_token`（auth）与 `verify_access_key`（git）两套 PAT 校验，前者零调用 | `auth/src/services/access_keys.rs:226`、`git/src/share/access.rs:46` | 过期/作用域语义只在 git 侧生效 |
598598| 9 | 邮件内链接与注册路由不匹配（`/auth/verify-email` 等，实际只有 `/pages/*`），落地页还要求邮件不带 `code` | `auth/src/services/email.rs:600-610`、`handlers/verify_email_page.rs:13-18`、`email/templates/*` | 邮件点击链路 404 |
599599| 10 | 目录可见性有两份实现且兜底不同（`true` vs `false`） | `space/src/services/dir.rs:155-161`、`share/dir_visibility.rs:40` | 越权可见风险不一致 |
600| 11 | 大量“有字段无行为”：`space.allow_join/allow_invite/allow_join_level`、`space_warn`、`ban.allow_*` 四列、`chat_link_unfurl`、`chat.typing_ttl_secs`、`milestone.nums_*`、`AttachmentStatus::Ready/Failed`、`label.exclusive` | 各文件 | 配置与产品预期不符 |
600| 11 | 大量“有字段无行为”：`space.allow_join/allow_invite/allow_join_level`、`space_warn`、`ban.allow_*` 四列、`milestone.nums_*`、`label.exclusive`、`space_members.suppress_roles/mobile_push` | 各文件 | 配置与产品预期不符 |
601601| 12 | `context()` 按 `repos.lower_name` 全表取回后线性上溯目录找空间，且每次调用新建 `CacheClient` | `git/src/service/context.rs:15` | 仓库数增长后的热路径放大 |
602602| 13 | `is_dir_visible` 单次检查加载整空间节点 + 两张覆写表 | `share/dir_visibility.rs:66-77` | 目录规模大时放大 |
603603| 14 | 默认 `cache.l1_ttl_secs=0 / default_ttl_secs=0` | `config/default.toml:21-23` | 出厂缓存近似空转 |
604604| 15 | `paginate()` 内存分页；`/me/avatar` 的 6 MiB 被外层 2 MiB 截断 | `export/src/pagination.rs:39-48`、`user/src/routes.rs:45`、`webapi/src/lib.rs:12` | 深翻成本、头像尺寸 |
605| 16 | `idx_chat_message_room_nonce` 双份定义，实际生效 `(room_id, nonce)` 与代码预检不一致 | `schema_extras.sql:121,165` | 跨用户 nonce 撞库 |
606605| 17 | 注销脱敏的 11 步为逐条语句、非事务；不级联清理空间/仓库/聊天/S3 | `auth/src/services/deletion.rs:20-128` | 半完成状态与残留数据 |
607606| 18 | Issue 号段靠 board 行 `lock_exclusive()` 串行 `max+1`，非 Postgres sequence | `issues/src/services/issues.rs:405-415` | 高并发建单串行化 |
608607| 19 | Issue 写邮件同步阻塞请求（mentions/reply/assignment 三处 inline），且不查用户偏好 | `issues/src/services/{issues,comments,members}.rs` | SMTP 慢直接拖高 RT；无免打扰 |
Modified lib/chat/src/handlers/messages.rs lib/chat/src/handlers/messages.rs +2 −10 View lib/chat/src/handlers/messages.rs
@@ -8,7 +8,7 @@
88use crate::errors::ApiError;
99use crate::services::messages::{
1010 BulkDeleteCmd, ForwardMessageCmd, ListMessageQuery, MessageDto, MessageEditDto, MessagePage,
11 MessagePath, PinCmd, ScheduledMessageDto, SendMessageCmd, UpdateMessageCmd,
11 MessagePath, PinCmd, SendMessageCmd, UpdateMessageCmd,
1212};
1313use crate::services::service::ChatService;
1414
@@ -16,12 +16,11 @@
1616 post,
1717 path = "/chat/rooms/{room_id}/messages",
1818 tag = "chat",
19 description = "向房间即时发送消息，或按未来 send_at 时间预约发送。",
19 description = "向房间发送消息。",
2020 params(("room_id" = Uuid, Path, description = "room id")),
2121 request_body = SendMessageCmd,
2222 responses(
2323 (status = 200, description = "message sent", body = AppResponseData<MessageDto>),
24 (status = 202, description = "message scheduled", body = AppResponseData<ScheduledMessageDto>),
2524 (status = 400, description = "bad request", body = AppResponseData<NoData>),
2625 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
2726 (status = 409, description = "resource already exists", body = AppResponseData<NoData>),
@@ -35,13 +34,6 @@
3534 Json(cmd): Json<SendMessageCmd>,
3635) -> Result<axum::response::Response, ApiError> {
3736 let uid = require_uid(&session)?;
38 if cmd.send_at.is_some() {
39 let dto = service.schedule_message(uid, room_id, cmd).await?;
40 let mut envelope = AppResponseData::ok(dto);
41 envelope.code = 202;
42 envelope.message = "scheduled".to_string();
43 return Ok(envelope.into_response());
44 }
4537 let dto = service.send_message(uid, room_id, cmd).await?;
4638 Ok(AppResponseData::ok(dto).into_response())
4739}
Modified lib/chat/src/handlers/webhooks.rs lib/chat/src/handlers/webhooks.rs +103 −0 View lib/chat/src/handlers/webhooks.rs
@@ -130,3 +130,106 @@
130130 let dto = service.update_webhook(uid, room_id, webhook_id, cmd).await?;
131131 Ok(AppResponseData::ok(dto))
132132}
133
134use crate::services::outbound::{
135 CreateOutboundWebhookCmd, OutboundWebhookDto, OutboundWebhookSecretDto,
136 UpdateOutboundWebhookCmd,
137};
138
139#[utoipa::path(
140 post,
141 path = "/chat/rooms/{room_id}/outbound-webhooks",
142 tag = "chat",
143 description = "创建出站 Webhook（房间消息事件回调），一次性返回签名密钥。",
144 params(("room_id" = Uuid, Path, description = "room id")),
145 request_body = CreateOutboundWebhookCmd,
146 responses(
147 (status = 200, description = "outbound webhook created", body = AppResponseData<OutboundWebhookSecretDto>),
148 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
149 (status = 400, description = "request failed validation", body = AppResponseData<NoData>),
150 (status = 401, description = "authentication required", body = AppResponseData<NoData>)
151 )
152)]
153pub async fn create_outbound_webhook(
154 State(service): State<ChatService>,
155 session: Session,
156 Path(room_id): Path<Uuid>,
157 Json(cmd): Json<CreateOutboundWebhookCmd>,
158) -> Result<AppResponseData<OutboundWebhookSecretDto>, ApiError> {
159 let uid = require_uid(&session)?;
160 let dto = service.create_outbound_webhook(uid, room_id, cmd).await?;
161 Ok(AppResponseData::ok(dto))
162}
163
164#[utoipa::path(
165 get,
166 path = "/chat/rooms/{room_id}/outbound-webhooks",
167 tag = "chat",
168 description = "列出房间的出站 Webhook（不返回密钥）。",
169 params(("room_id" = Uuid, Path, description = "room id")),
170 responses(
171 (status = 200, description = "outbound webhook list", body = AppResponseData<Vec<OutboundWebhookDto>>),
172 (status = 403, description = "forbidden", body = AppResponseData<NoData>),
173 (status = 401, description = "authentication required", body = AppResponseData<NoData>)
174 )
175)]
176pub async fn list_outbound_webhooks(
177 State(service): State<ChatService>,
178 session: Session,
179 Path(room_id): Path<Uuid>,
180) -> Result<AppResponseData<Vec<OutboundWebhookDto>>, ApiError> {
181 let uid = require_uid(&session)?;
182 let rows = service.list_outbound_webhooks(uid, room_id).await?;
183 Ok(AppResponseData::ok(rows))
184}
185
186#[utoipa::path(
187 patch,
188 path = "/chat/rooms/{room_id}/outbound-webhooks/{webhook_id}",
189 tag = "chat",
190 description = "更新出站 Webhook 的名称、地址或启用状态。",
191 params(("room_id" = Uuid, Path), ("webhook_id" = Uuid, Path)),
192 request_body = UpdateOutboundWebhookCmd,
193 responses(
194 (status = 200, description = "outbound webhook updated", body = AppResponseData<OutboundWebhookDto>),
195 (status = 404, description = "not found", body = AppResponseData<NoData>),
196 (status = 400, description = "request failed validation", body = AppResponseData<NoData>),
197 (status = 401, description = "authentication required", body = AppResponseData<NoData>)
198 )
199)]
200pub async fn update_outbound_webhook(
201 State(service): State<ChatService>,
202 session: Session,
203 Path((room_id, webhook_id)): Path<(Uuid, Uuid)>,
204 Json(cmd): Json<UpdateOutboundWebhookCmd>,
205) -> Result<AppResponseData<OutboundWebhookDto>, ApiError> {
206 let uid = require_uid(&session)?;
207 let dto = service
208 .update_outbound_webhook(uid, room_id, webhook_id, cmd)
209 .await?;
210 Ok(AppResponseData::ok(dto))
211}
212
213#[utoipa::path(
214 delete,
215 path = "/chat/rooms/{room_id}/outbound-webhooks/{webhook_id}",
216 tag = "chat",
217 description = "删除出站 Webhook 并清理待投递事件。",
218 params(("room_id" = Uuid, Path), ("webhook_id" = Uuid, Path)),
219 responses(
220 (status = 200, description = "outbound webhook deleted", body = AppResponseData<NoData>),
221 (status = 404, description = "not found", body = AppResponseData<NoData>),
222 (status = 401, description = "authentication required", body = AppResponseData<NoData>)
223 )
224)]
225pub async fn delete_outbound_webhook(
226 State(service): State<ChatService>,
227 session: Session,
228 Path((room_id, webhook_id)): Path<(Uuid, Uuid)>,
229) -> Result<AppResponseData<NoData>, ApiError> {
230 let uid = require_uid(&session)?;
231 service
232 .delete_outbound_webhook(uid, room_id, webhook_id)
233 .await?;
234 Ok(AppResponseData::empty())
235}
Modified lib/chat/src/services/attachments.rs lib/chat/src/services/attachments.rs +44 −1 View lib/chat/src/services/attachments.rs
@@ -1,7 +1,10 @@
11use axum::body::Body;
22use axum::http::header;
33use models::chat::attachments::{self, AttachmentStatus};
4use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, Set};
4use sea_orm::{
5 ActiveModelTrait, ColumnTrait, Condition, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect,
6 Set,
7};
58use serde::{Deserialize, Serialize};
69use time::OffsetDateTime;
710use utoipa::ToSchema;
@@ -250,6 +253,46 @@
250253 Ok(AttachmentDto::from(updated))
251254 }
252255
256 pub async fn cleanup_stale_attachments(&self) -> Result<u64, ChatError> {
257 let now = OffsetDateTime::now_utc();
258 let pending_cutoff = now - time::Duration::hours(24);
259 let uploaded_cutoff = now - time::Duration::days(7);
260 let stale = attachments::Entity::find()
261 .filter(
262 Condition::any()
263 .add(
264 Condition::all()
265 .add(attachments::Column::Status.eq(AttachmentStatus::Pending))
266 .add(attachments::Column::CreatedAt.lt(pending_cutoff)),
267 )
268 .add(
269 Condition::all()
270 .add(attachments::Column::Status.eq(AttachmentStatus::Uploaded))
271 .add(attachments::Column::MessageId.is_null())
272 .add(attachments::Column::CreatedAt.lt(uploaded_cutoff)),
273 ),
274 )
275 .limit(200)
276 .all(self.db().write())
277 .await?;
278 let count = stale.len() as u64;
279 for row in stale {
280 let shared = attachments::Entity::find()
281 .filter(attachments::Column::S3Key.eq(row.s3_key.clone()))
282 .filter(attachments::Column::Id.ne(row.id))
283 .count(self.db().read())
284 .await
285 .unwrap_or(1);
286 let _ = attachments::Entity::delete_by_id(row.id)
287 .exec(self.db().write())
288 .await;
289 if shared == 0 {
290 let _ = self.storage().client().delete(&row.s3_key).await;
291 }
292 }
293 Ok(count)
294 }
295
253296 pub async fn get_attachment_meta(
254297 &self,
255298 caller: Uuid,
Modified lib/chat/src/services/dms.rs lib/chat/src/services/dms.rs +177 −26 View lib/chat/src/services/dms.rs
@@ -5,6 +5,7 @@
55use models::users::user;
66use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
77use serde::{Deserialize, Serialize};
8use sha2::{Digest, Sha256};
89use space::services::spaces::UserDto;
910use time::OffsetDateTime;
1011use utoipa::ToSchema;
@@ -14,12 +15,16 @@
1415use crate::services::service::ChatService;
1516use crate::share::realtime::RoomEvent;
1617
18const MAX_GROUP_DM_MEMBERS: usize = 10;
19
1720#[derive(Debug, Clone, Serialize, ToSchema)]
1821pub struct DmDto {
1922 pub id: Uuid,
2023 pub kind: RoomKind,
2124 pub name: String,
2225 pub peer: Option<UserDto>,
26 #[serde(default)]
27 pub members: Vec<UserDto>,
2328 pub last_message_id: Option<Uuid>,
2429 pub last_read_seq: i64,
2530 pub tip_seq: i64,
@@ -29,7 +34,10 @@
2934
3035#[derive(Debug, Clone, Deserialize, ToSchema)]
3136pub struct OpenDmCmd {
32 pub peer_user_id: Uuid,
37 #[serde(default)]
38 pub peer_user_id: Option<Uuid>,
39 #[serde(default)]
40 pub peer_user_ids: Option<Vec<Uuid>>,
3341}
3442
3543fn dm_pair_key(a: Uuid, b: Uuid) -> String {
@@ -37,8 +45,47 @@
3745 format!("dm:{lo}:{hi}")
3846}
3947
48fn group_dm_key(mut uids: Vec<Uuid>) -> String {
49 uids.sort();
50 let mut hasher = Sha256::new();
51 for uid in &uids {
52 hasher.update(uid.as_bytes());
53 }
54 let hex: String = hasher
55 .finalize()
56 .iter()
57 .map(|b| format!("{b:02x}"))
58 .collect();
59 format!("gdm:{hex}")
60}
61
4062impl ChatService {
4163 pub async fn open_dm(&self, caller: Uuid, cmd: OpenDmCmd) -> Result<DmDto, ChatError> {
64 let mut peers: Vec<Uuid> = Vec::new();
65 if let Some(peer) = cmd.peer_user_id {
66 peers.push(peer);
67 }
68 if let Some(list) = cmd.peer_user_ids {
69 peers.extend(list);
70 }
71 let mut uniq: Vec<Uuid> = Vec::new();
72 for peer in peers {
73 if peer != caller && !uniq.contains(&peer) {
74 uniq.push(peer);
75 }
76 }
77 if uniq.is_empty() {
78 return Err(ChatError::BadRequest(
79 "peer_user_id or peer_user_ids is required".to_string(),
80 ));
81 }
82 if uniq.len() == 1 {
83 return self.open_direct_dm(caller, uniq[0]).await;
84 }
85 self.open_group_dm(caller, uniq).await
86 }
87
88 async fn check_dm_quota(&self, caller: Uuid) -> Result<(), ChatError> {
4289 if !self
4390 .state
4491 .redis
@@ -53,12 +100,12 @@
53100 "DM creation quota exceeded, retry tomorrow".to_string(),
54101 ));
55102 }
56 if cmd.peer_user_id == caller {
57 return Err(ChatError::BadRequest(
58 "cannot open DM with self".to_string(),
59 ));
60 }
61 let peer = user::Entity::find_by_id(cmd.peer_user_id)
103 Ok(())
104 }
105
106 async fn open_direct_dm(&self, caller: Uuid, peer_user_id: Uuid) -> Result<DmDto, ChatError> {
107 self.check_dm_quota(caller).await?;
108 let peer = user::Entity::find_by_id(peer_user_id)
62109 .one(self.db().read())
63110 .await?
64111 .filter(|u| u.is_active)
@@ -73,14 +120,7 @@
73120
74121 let room = match existing {
75122 Some(room) if room.is_active => room,
76 Some(mut room) => {
77 room.is_active = true;
78 let mut am: rooms::ActiveModel = room.clone().into();
79 am.is_active = Set(true);
80 am.updated_at = Set(OffsetDateTime::now_utc());
81 am.update(self.db().write()).await?;
82 room
83 }
123 Some(room) => self.reactivate_room(room).await?,
84124 None => {
85125 let now = OffsetDateTime::now_utc();
86126 let room_id = Uuid::now_v7();
@@ -109,9 +149,105 @@
109149 }
110150 };
111151
152 self.ensure_dm_members(room.id, caller, &[caller, peer.uid])
153 .await?;
154
155 self.build_dm_dto(caller, room, Some(peer.into()), Vec::new())
156 .await
157 }
158
159 async fn open_group_dm(&self, caller: Uuid, peers: Vec<Uuid>) -> Result<DmDto, ChatError> {
160 self.check_dm_quota(caller).await?;
161 if peers.len() + 1 > MAX_GROUP_DM_MEMBERS {
162 return Err(ChatError::BadRequest(format!(
163 "group DM supports at most {MAX_GROUP_DM_MEMBERS} members"
164 )));
165 }
166 let users = user::Entity::find()
167 .filter(user::Column::Uid.is_in(peers.clone()))
168 .all(self.db().read())
169 .await?;
170 if users.len() != peers.len() || users.iter().any(|u| !u.is_active) {
171 return Err(ChatError::NotFound);
172 }
173
174 let mut member_uids = peers.clone();
175 member_uids.push(caller);
176 let key = group_dm_key(member_uids.clone());
177
178 let existing = rooms::Entity::find()
179 .filter(rooms::Column::LowerName.eq(&key))
180 .filter(rooms::Column::Kind.eq(RoomKind::GroupDm))
181 .one(self.db().read())
182 .await?;
183
184 let room = match existing {
185 Some(room) if room.is_active => room,
186 Some(room) => self.reactivate_room(room).await?,
187 None => {
188 let now = OffsetDateTime::now_utc();
189 let room_id = Uuid::now_v7();
190 let name: String = users
191 .iter()
192 .map(|u| u.username.clone())
193 .collect::<Vec<_>>()
194 .join(", ")
195 .chars()
196 .take(100)
197 .collect();
198 rooms::ActiveModel {
199 id: Set(room_id),
200 space_id: Set(None),
201 kind: Set(RoomKind::GroupDm),
202 lower_name: Set(key.clone()),
203 name: Set(name),
204 topic: Set(None),
205 position: Set(0),
206 parent_id: Set(None),
207 owner_id: Set(caller),
208 is_private: Set(true),
209 nsfw: Set(false),
210 user_limit: Set(Some(member_uids.len() as i32)),
211 rate_limit_per_user: Set(None),
212 bitrate: Set(None),
213 last_message_id: Set(None),
214 is_active: Set(true),
215 created_at: Set(now),
216 updated_at: Set(now),
217 }
218 .insert(self.db().write())
219 .await?
220 }
221 };
222
223 self.ensure_dm_members(room.id, caller, &member_uids).await?;
224
225 let members = users
226 .into_iter()
227 .map(UserDto::from)
228 .collect::<Vec<_>>();
229 self.build_dm_dto(caller, room, None, members).await
230 }
231
232 async fn reactivate_room(&self, room: rooms::Model) -> Result<rooms::Model, ChatError> {
233 let mut room = room;
234 room.is_active = true;
235 let mut am: rooms::ActiveModel = room.clone().into();
236 am.is_active = Set(true);
237 am.updated_at = Set(OffsetDateTime::now_utc());
238 am.update(self.db().write()).await?;
239 Ok(room)
240 }
241
242 async fn ensure_dm_members(
243 &self,
244 room_id: Uuid,
245 caller: Uuid,
246 uids: &[Uuid],
247 ) -> Result<(), ChatError> {
112248 let now = OffsetDateTime::now_utc();
113 for uid in [caller, peer.uid] {
114 let existing_member = room_members::Entity::find_by_id((room.id, uid))
249 for &uid in uids {
250 let existing_member = room_members::Entity::find_by_id((room_id, uid))
115251 .one(self.db().write())
116252 .await?;
117253 match existing_member {
@@ -124,9 +260,13 @@
124260 Some(_) => {}
125261 None => {
126262 room_members::ActiveModel {
127 room_id: Set(room.id),
263 room_id: Set(room_id),
128264 user_id: Set(uid),
129 role: Set(models::room::room_members::RoomMemberRole::Member),
265 role: Set(if uid == caller {
266 models::room::room_members::RoomMemberRole::Owner
267 } else {
268 models::room::room_members::RoomMemberRole::Member
269 }),
130270 nick: Set(None),
131271 joined_at: Set(now),
132272 left_at: Set(None),
@@ -138,8 +278,7 @@
138278 }
139279 }
140280 }
141
142 self.build_dm_dto(caller, room, Some(peer.into())).await
281 Ok(())
143282 }
144283
145284 pub async fn list_dms(&self, caller: Uuid) -> Result<Vec<DmDto>, ChatError> {
@@ -168,12 +307,12 @@
168307 .await?;
169308 let peer_uids: Vec<Uuid> = peer_rows.iter().map(|m| m.user_id).collect();
170309 let user_map = self.load_user_briefs(&peer_uids).await?;
171 let mut peer_by_room: std::collections::HashMap<Uuid, Option<UserDto>> =
310 let mut peers_by_room: std::collections::HashMap<Uuid, Vec<UserDto>> =
172311 std::collections::HashMap::new();
173312 for m in peer_rows {
174 peer_by_room
175 .entry(m.room_id)
176 .or_insert_with(|| user_map.get(&m.user_id).cloned());
313 if let Some(dto) = user_map.get(&m.user_id).cloned() {
314 peers_by_room.entry(m.room_id).or_default().push(dto);
315 }
177316 }
178317
179318 let tips = room_seq::Entity::find()
@@ -196,11 +335,21 @@
196335 for room in rooms {
197336 let tip = tip_map.get(&room.id).copied().unwrap_or(0);
198337 let last_read = read_map.get(&room.id).copied().unwrap_or(0);
338 let mut members = peers_by_room.remove(&room.id).unwrap_or_default();
339 let peer = if room.kind == RoomKind::DirectMessage {
340 members.first().cloned()
341 } else {
342 None
343 };
344 if room.kind == RoomKind::DirectMessage {
345 members.clear();
346 }
199347 result.push(DmDto {
200348 id: room.id,
201349 kind: room.kind,
202350 name: room.name,
203 peer: peer_by_room.remove(&room.id).flatten(),
351 peer,
352 members,
204353 last_message_id: room.last_message_id,
205354 last_read_seq: last_read,
206355 tip_seq: tip,
@@ -217,6 +366,7 @@
217366 caller: Uuid,
218367 room: rooms::Model,
219368 peer: Option<UserDto>,
369 members: Vec<UserDto>,
220370 ) -> Result<DmDto, ChatError> {
221371 let tip = room_seq::Entity::find_by_id(room.id)
222372 .one(self.db().read())
@@ -233,6 +383,7 @@
233383 kind: room.kind,
234384 name: room.name,
235385 peer,
386 members,
236387 last_message_id: room.last_message_id,
237388 last_read_seq: last_read,
238389 tip_seq: tip,
Modified lib/chat/src/services/messages.rs lib/chat/src/services/messages.rs +187 −63 View lib/chat/src/services/messages.rs
@@ -26,8 +26,9 @@
2626use crate::share::lookup::{RoomContext, require_room_readable, require_room_writable};
2727use crate::share::mention::parse_mentions;
2828use crate::share::notify::{
29 NotifyDecision, advance_read_state, current_room_seq, mention_email_allowed, next_room_seq,
30 notification_decision, notify_user, publish_message_event, record_mentions,
29 NotifyDecision, advance_read_state, current_room_seq, fanout_everyone_mentions,
30 mention_email_allowed, next_room_seq, notification_decision, notify_user,
31 publish_message_event, record_mentions,
3132};
3233use crate::share::realtime::RoomEvent;
3334
@@ -90,16 +91,6 @@
9091 pub urgent: bool,
9192 #[serde(default)]
9293 pub forwarded_from: Option<Uuid>,
93 #[serde(default)]
94 pub send_at: Option<OffsetDateTime>,
95}
96
97#[derive(Debug, Clone, Serialize, ToSchema)]
98pub struct ScheduledMessageDto {
99 pub id: Uuid,
100 pub room_id: Uuid,
101 pub scheduled_at: OffsetDateTime,
102 pub status: models::users::user_scheduled_messages::ScheduledMessageStatus,
10394}
10495
10596#[derive(Debug, Clone, Deserialize, ToSchema)]
@@ -266,47 +257,22 @@
266257 Ok(())
267258 }
268259
269 pub async fn schedule_message(
260 pub async fn send_message(
270261 &self,
271262 caller: Uuid,
272263 room_id: Uuid,
273 mut cmd: SendMessageCmd,
274 ) -> Result<ScheduledMessageDto, ChatError> {
275 let Some(send_at) = cmd.send_at else {
276 return Err(ChatError::BadRequest("send_at is required".to_string()));
277 };
278 if send_at <= OffsetDateTime::now_utc() {
279 return Err(ChatError::BadRequest("send_at must be in the future".to_string()));
280 }
281 require_room_writable(self, room_id, caller).await?;
282 cmd.send_at = None;
283 let payload = serde_json::to_value(&cmd)?;
284 let id = Uuid::now_v7();
285 let now = OffsetDateTime::now_utc();
286 let row = models::users::user_scheduled_messages::ActiveModel {
287 id: Set(id),
288 user_id: Set(caller),
289 target_channel_id: Set(room_id),
290 payload: Set(payload),
291 scheduled_at: Set(send_at),
292 status: Set(models::users::user_scheduled_messages::ScheduledMessageStatus::Pending),
293 created_at: Set(now),
294 }
295 .insert(self.db().write())
296 .await?;
297 Ok(ScheduledMessageDto {
298 id: row.id,
299 room_id,
300 scheduled_at: row.scheduled_at,
301 status: row.status,
302 })
264 cmd: SendMessageCmd,
265 ) -> Result<MessageDto, ChatError> {
266 self.send_message_internal(caller, room_id, cmd, Vec::new())
267 .await
303268 }
304269
305 pub async fn send_message(
270 pub(crate) async fn send_message_internal(
306271 &self,
307272 caller: Uuid,
308273 room_id: Uuid,
309274 cmd: SendMessageCmd,
275 attachment_copies: Vec<attachment_rows::Model>,
310276 ) -> Result<MessageDto, ChatError> {
311277 let ctx = require_room_writable(self, room_id, caller).await?;
312278 self.check_slow_mode(&ctx, caller).await?;
@@ -401,9 +367,22 @@
401367 let content = cmd.content.clone();
402368 let mentions_json = json!(parsed.users);
403369 let groups_json = json!(parsed.groups);
404 let attachments_json = json!(cmd.attachments);
370 let copy_ids: Vec<(Uuid, attachment_rows::Model)> = attachment_copies
371 .into_iter()
372 .map(|row| (Uuid::now_v7(), row))
373 .collect();
374 let mut attachment_ids = cmd.attachments.clone();
375 attachment_ids.extend(copy_ids.iter().map(|(id, _)| *id));
376 let attachments_json = json!(attachment_ids);
405377 let embeds_json = json!([]);
406 let mention_everyone = parsed.everyone;
378 let mention_everyone = if parsed.everyone {
379 match &ctx.space {
380 Some(space) => can_manage_space(self, space, caller).await?,
381 None => true,
382 }
383 } else {
384 false
385 };
407386 let flags = if cmd.urgent {
408387 models::chat::messages::FLAG_URGENT
409388 } else {
@@ -420,6 +399,7 @@
420399 let thread_clone = thread.clone();
421400 let mention_uids_for_txn = mention_uids.clone();
422401 let content_for_email = content.clone();
402 let copy_ids_for_txn = copy_ids.clone();
423403 let svc = self.clone();
424404 let space_id = room.space_id;
425405 let nonce_retry = cmd.nonce.clone();
@@ -480,6 +460,30 @@
480460 .await?;
481461 }
482462
463 for (new_id, src) in ©_ids_for_txn {
464 attachment_rows::ActiveModel {
465 id: Set(*new_id),
466 user_id: Set(caller),
467 room_id: Set(Some(room.id)),
468 message_id: Set(Some(message_id)),
469 filename: Set(src.filename.clone()),
470 description: Set(src.description.clone()),
471 content_type: Set(src.content_type.clone()),
472 size: Set(src.size),
473 bucket: Set(src.bucket.clone()),
474 s3_key: Set(src.s3_key.clone()),
475 status: Set(attachment_rows::AttachmentStatus::Uploaded),
476 width: Set(src.width),
477 height: Set(src.height),
478 duration_secs: Set(src.duration_secs),
479 variants: Set(src.variants.clone()),
480 created_at: Set(now),
481 updated_at: Set(now),
482 }
483 .insert(txn)
484 .await?;
485 }
486
483487 rooms::Entity::update_many()
484488 .filter(rooms::Column::Id.eq(room.id))
485489 .col_expr(rooms::Column::LastMessageId, Expr::value(message_id))
@@ -505,7 +509,7 @@
505509 content.clone()
506510 };
507511 for &uid in &mention_uids_for_txn {
508 if notification_decision(&svc, room.id, space_id, uid, true).await?
512 if notification_decision(&svc, room.id, space_id, uid, true, false).await?
509513 == NotifyDecision::Skip
510514 {
511515 continue;
@@ -559,6 +563,25 @@
559563 .sender_mark_read(caller, room_id, message_id, seq)
560564 .await;
561565
566 if mention_everyone {
567 let svc = self.clone();
568 let body = if content_for_email.chars().count() > 120 {
569 content_for_email.chars().take(120).collect::<String>()
570 } else {
571 content_for_email.clone()
572 };
573 let explicit = mention_uids.clone();
574 let is_private = ctx.room.is_private;
575 let space_id = ctx.room.space_id;
576 tokio::spawn(async move {
577 fanout_everyone_mentions(
578 &svc, room_id, space_id, is_private, message_id, body, caller, &explicit,
579 )
580 .await;
581 });
582 }
583 crate::services::unfurl::spawn_unfurl(self, room_id, message_id, content_for_email.clone());
584
562585 let recipients: Vec<Uuid> = mention_uids
563586 .iter()
564587 .copied()
@@ -741,7 +764,31 @@
741764 let old_attachments = row.attachments.clone();
742765 let now = OffsetDateTime::now_utc();
743766
767 let parsed = parse_mentions(self, &ctx, &content, caller).await?;
768 let mention_everyone = if parsed.everyone {
769 match &ctx.space {
770 Some(space) => can_manage_space(self, space, caller).await?,
771 None => true,
772 }
773 } else {
774 false
775 };
776 let old_mentions: Vec<Uuid> =
777 serde_json::from_value(row.mentions.clone()).unwrap_or_default();
778 let new_mentions: Vec<Uuid> = parsed
779 .users
780 .iter()
781 .copied()
782 .filter(|u| !old_mentions.contains(u))
783 .collect();
784 let mentions_json = json!(parsed.users);
785 let groups_json = json!(parsed.groups);
786 let new_mentions_txn = new_mentions.clone();
787 let space_id = ctx.space.as_ref().map(|s| s.id);
788 let content_for_notify = content.clone();
789
744790 let this = self.clone();
791 let this_txn = self.clone();
745792 self.db()
746793 .primary()
747794 .transaction::<_, (), ChatError>(move |txn| {
@@ -760,10 +807,50 @@
760807 messages::Entity::update_many()
761808 .filter(messages::Column::Id.eq(message_id))
762809 .col_expr(messages::Column::Content, Expr::value(content))
810 .col_expr(messages::Column::Mentions, Expr::value(mentions_json))
811 .col_expr(
812 messages::Column::MentionEveryone,
813 Expr::value(mention_everyone),
814 )
815 .col_expr(
816 messages::Column::MentionGroupIds,
817 Expr::value(groups_json),
818 )
763819 .col_expr(messages::Column::EditedAt, Expr::value(now))
764820 .exec(txn)
765821 .await?;
766822
823 if !new_mentions_txn.is_empty() {
824 record_mentions(txn, room_id, message_id, &new_mentions_txn).await?;
825 let body = if content_for_notify.chars().count() > 120 {
826 content_for_notify.chars().take(120).collect::<String>()
827 } else {
828 content_for_notify.clone()
829 };
830 let data = json!({
831 "room_id": room_id,
832 "message_id": message_id,
833 });
834 for &uid in &new_mentions_txn {
835 if notification_decision(&this_txn, room_id, space_id, uid, true, false)
836 .await?
837 == NotifyDecision::Skip
838 {
839 continue;
840 }
841 notify_user(
842 txn,
843 uid,
844 NotificationType::Mention,
845 "You were mentioned in a message".to_string(),
846 Some(body.clone()),
847 Some(data.clone()),
848 Some(caller),
849 )
850 .await?;
851 }
852 }
853
767854 Ok(())
768855 })
769856 })
@@ -783,6 +870,23 @@
783870 message: dto_value,
784871 };
785872 this.realtime.publish_room(&event).await;
873 crate::services::outbound::enqueue_outbound_event(
874 self,
875 room_id,
876 "message_edited",
877 &serde_json::to_value(&event)?,
878 )
879 .await;
880
881 for &uid in &new_mentions {
882 self.realtime
883 .publish_user(
884 uid,
885 &json!({"kind": "mention", "room_id": room_id, "message_id": message_id}),
886 )
887 .await;
888 }
889 crate::services::unfurl::spawn_unfurl(self, room_id, message_id, dto.content.clone());
786890
787891 Ok(dto)
788892 }
@@ -878,6 +982,25 @@
878982 })
879983 .await?;
880984
985 let attached = attachment_rows::Entity::find()
986 .filter(attachment_rows::Column::MessageId.is_in(row_ids.clone()))
987 .all(self.db().read())
988 .await?;
989 for attachment in attached {
990 let shared = attachment_rows::Entity::find()
991 .filter(attachment_rows::Column::S3Key.eq(attachment.s3_key.clone()))
992 .filter(attachment_rows::Column::Id.ne(attachment.id))
993 .count(self.db().read())
994 .await
995 .unwrap_or(1);
996 let _ = attachment_rows::Entity::delete_by_id(attachment.id)
997 .exec(self.db().write())
998 .await;
999 if shared == 0 {
1000 let _ = self.storage().client().delete(&attachment.s3_key).await;
1001 }
1002 }
1003
8811004 let seq = current_room_seq(self.db().read(), room_id).await;
8821005
8831006 let event = if single {
@@ -894,6 +1017,17 @@
8941017 }
8951018 };
8961019 self.realtime.publish_room(&event).await;
1020 crate::services::outbound::enqueue_outbound_event(
1021 self,
1022 room_id,
1023 if single {
1024 "message_deleted"
1025 } else {
1026 "messages_deleted"
1027 },
1028 &serde_json::to_value(&event)?,
1029 )
1030 .await;
8971031 let _ = ctx;
8981032
8991033 Ok(())
@@ -994,32 +1128,22 @@
9941128 .filter(|m| m.deleted_at.is_none())
9951129 .ok_or(ChatError::NotFound)?;
9961130
997 let forwarded_attachments = if original.author_id == caller {
998 original
999 .attachments
1000 .as_array()
1001 .map(|arr| {
1002 arr.iter()
1003 .filter_map(|v| v.as_str().and_then(|s| Uuid::parse_str(s).ok()))
1004 .collect()
1005 })
1006 .unwrap_or_default()
1007 } else {
1008 Vec::new()
1009 };
1131 let original_attachments = attachment_rows::Entity::find()
1132 .filter(attachment_rows::Column::MessageId.eq(original.id))
1133 .all(self.db().read())
1134 .await?;
10101135 let send_cmd = SendMessageCmd {
10111136 content: cmd.comment.unwrap_or_default(),
10121137 reply_to_id: None,
10131138 thread_id: None,
10141139 thread_broadcast: false,
1015 attachments: forwarded_attachments,
1140 attachments: Vec::new(),
10161141 nonce: None,
10171142 urgent: false,
10181143 forwarded_from: Some(original.id),
1019 send_at: None,
10201144 };
10211145 let dto = self
1022 .send_message(caller, cmd.target_room_id, send_cmd)
1146 .send_message_internal(caller, cmd.target_room_id, send_cmd, original_attachments)
10231147 .await?;
10241148 Ok(dto)
10251149 }
Modified lib/chat/src/services/mod.rs lib/chat/src/services/mod.rs +2 −0 View lib/chat/src/services/mod.rs
@@ -4,6 +4,7 @@
44pub mod drafts;
55pub mod emojis;
66pub mod messages;
7pub mod outbound;
78pub mod presence;
89pub mod reactions;
910pub mod read_state;
@@ -14,4 +15,5 @@
1415pub mod service;
1516pub mod settings;
1617pub mod threads;
18pub mod unfurl;
1719pub mod webhooks;
Added lib/chat/src/services/outbound.rs lib/chat/src/services/outbound.rs +451 −0 View lib/chat/src/services/outbound.rs
@@ -0,0 +1,451 @@
1use models::chat::outbound_webhooks;
2use models::chat::webhook_deliveries::{self, WebhookDeliveryStatus};
3use sea_orm::sea_query::Expr;
4use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};
5use serde::{Deserialize, Serialize};
6use time::OffsetDateTime;
7use utoipa::ToSchema;
8use uuid::Uuid;
9
10use crate::errors::ChatError;
11use crate::services::service::ChatService;
12use crate::share::authz::can_manage_space;
13use crate::share::lookup::require_room_readable;
14
15const MAX_DELIVERY_ATTEMPTS: i32 = 5;
16
17#[derive(Debug, Clone, Serialize, ToSchema)]
18pub struct OutboundWebhookDto {
19 pub id: Uuid,
20 pub room_id: Uuid,
21 pub name: String,
22 pub url: String,
23 pub disabled: bool,
24 pub last_status: Option<String>,
25 pub last_used_at: Option<OffsetDateTime>,
26 pub created_at: OffsetDateTime,
27}
28
29#[derive(Debug, Clone, Serialize, ToSchema)]
30pub struct OutboundWebhookSecretDto {
31 pub id: Uuid,
32 pub room_id: Uuid,
33 pub name: String,
34 pub url: String,
35 pub secret: String,
36 pub created_at: OffsetDateTime,
37}
38
39#[derive(Debug, Clone, Deserialize, ToSchema)]
40pub struct CreateOutboundWebhookCmd {
41 pub name: String,
42 pub url: String,
43}
44
45#[derive(Debug, Clone, Deserialize, ToSchema)]
46pub struct UpdateOutboundWebhookCmd {
47 #[serde(default)]
48 pub name: Option<String>,
49 #[serde(default)]
50 pub url: Option<String>,
51 #[serde(default)]
52 pub disabled: Option<bool>,
53}
54
55impl From<outbound_webhooks::Model> for OutboundWebhookDto {
56 fn from(w: outbound_webhooks::Model) -> Self {
57 Self {
58 id: w.id,
59 room_id: w.room_id,
60 name: w.name,
61 url: w.url,
62 disabled: w.disabled,
63 last_status: w.last_status,
64 last_used_at: w.last_used_at,
65 created_at: w.created_at,
66 }
67 }
68}
69
70pub(crate) fn validate_outbound_url(url: &str) -> Result<http::Uri, ChatError> {
71 let uri: http::Uri = url
72 .parse()
73 .map_err(|_| ChatError::BadRequest("invalid url".to_string()))?;
74 let scheme = uri.scheme_str().unwrap_or("");
75 if scheme != "http" && scheme != "https" {
76 return Err(ChatError::BadRequest(
77 "url scheme must be http or https".to_string(),
78 ));
79 }
80 let host = uri
81 .host()
82 .ok_or_else(|| ChatError::BadRequest("url host is required".to_string()))?;
83 if host.is_empty() {
84 return Err(ChatError::BadRequest("url host is required".to_string()));
85 }
86 if let Ok(ip) = host.parse::<std::net::IpAddr>()
87 && !is_public_ip(&ip)
88 {
89 return Err(ChatError::BadRequest(
90 "url must not point at a private or loopback address".to_string(),
91 ));
92 }
93 Ok(uri)
94}
95
96pub(crate) fn is_public_ip(ip: &std::net::IpAddr) -> bool {
97 match ip {
98 std::net::IpAddr::V4(v4) => {
99 !v4.is_loopback()
100 && !v4.is_private()
101 && !v4.is_link_local()
102 && !v4.is_broadcast()
103 && !v4.is_documentation()
104 && !v4.is_unspecified()
105 && !(v4.octets()[0] == 100 && (v4.octets()[1] & 0b1100_0000) == 0b0100_0000)
106 }
107 std::net::IpAddr::V6(v6) => {
108 !v6.is_loopback() && !v6.is_unspecified() && !v6.is_multicast() && {
109 let segments = v6.segments();
110 // 64:ff9b::/96 (NAT64), fc00::/7 (ULA), fe80::/10 (link-local),
111 // ::ffff:0:0/96 (IPv4-mapped) handled via to_ipv4 below.
112 !(segments[0] == 0x0064 && segments[1] == 0xff9b)
113 && (segments[0] & 0xfe00) != 0xfc00
114 && (segments[0] & 0xffc0) != 0xfe80
115 } && match v6.to_ipv4_mapped() {
116 Some(v4) => is_public_ip(&std::net::IpAddr::V4(v4)),
117 None => true,
118 }
119 }
120 }
121}
122
123async fn require_room_manager(
124 service: &ChatService,
125 room_id: Uuid,
126 caller: Uuid,
127) -> Result<(), ChatError> {
128 let ctx = require_room_readable(service, room_id, caller).await?;
129 match &ctx.space {
130 Some(space) => {
131 if !can_manage_space(service, space, caller).await? {
132 return Err(ChatError::Forbidden);
133 }
134 }
135 None => return Err(ChatError::Forbidden),
136 }
137 Ok(())
138}
139
140fn generate_secret() -> String {
141 use rand::Rng;
142 const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
143 let mut rng = rand::thread_rng();
144 (0..48)
145 .map(|_| CHARSET[rng.gen_range(0..CHARSET.len())] as char)
146 .collect()
147}
148
149impl ChatService {
150 pub async fn create_outbound_webhook(
151 &self,
152 caller: Uuid,
153 room_id: Uuid,
154 cmd: CreateOutboundWebhookCmd,
155 ) -> Result<OutboundWebhookSecretDto, ChatError> {
156 require_room_manager(self, room_id, caller).await?;
157 let name = cmd.name.trim();
158 if name.is_empty() || name.chars().count() > self.config().chat.max_webhook_name_len {
159 return Err(ChatError::BadRequest("invalid webhook name".to_string()));
160 }
161 validate_outbound_url(cmd.url.trim())?;
162 let now = OffsetDateTime::now_utc();
163 let row = outbound_webhooks::ActiveModel {
164 id: Set(Uuid::now_v7()),
165 room_id: Set(room_id),
166 user_id: Set(caller),
167 name: Set(name.to_string()),
168 url: Set(cmd.url.trim().to_string()),
169 secret: Set(generate_secret()),
170 disabled: Set(false),
171 last_status: Set(None),
172 last_used_at: Set(None),
173 created_at: Set(now),
174 updated_at: Set(now),
175 }
176 .insert(self.db().write())
177 .await?;
178 Ok(OutboundWebhookSecretDto {
179 id: row.id,
180 room_id: row.room_id,
181 name: row.name,
182 url: row.url,
183 secret: row.secret,
184 created_at: row.created_at,
185 })
186 }
187
188 pub async fn list_outbound_webhooks(
189 &self,
190 caller: Uuid,
191 room_id: Uuid,
192 ) -> Result<Vec<OutboundWebhookDto>, ChatError> {
193 require_room_manager(self, room_id, caller).await?;
194 let rows = outbound_webhooks::Entity::find()
195 .filter(outbound_webhooks::Column::RoomId.eq(room_id))
196 .all(self.db().read())
197 .await?;
198 Ok(rows.into_iter().map(OutboundWebhookDto::from).collect())
199 }
200
201 pub async fn update_outbound_webhook(
202 &self,
203 caller: Uuid,
204 room_id: Uuid,
205 webhook_id: Uuid,
206 cmd: UpdateOutboundWebhookCmd,
207 ) -> Result<OutboundWebhookDto, ChatError> {
208 require_room_manager(self, room_id, caller).await?;
209 let webhook = outbound_webhooks::Entity::find_by_id(webhook_id)
210 .filter(outbound_webhooks::Column::RoomId.eq(room_id))
211 .one(self.db().write())
212 .await?
213 .ok_or(ChatError::NotFound)?;
214 let mut am: outbound_webhooks::ActiveModel = webhook.into();
215 if let Some(name) = cmd.name {
216 let name = name.trim();
217 if name.is_empty() || name.chars().count() > self.config().chat.max_webhook_name_len {
218 return Err(ChatError::BadRequest("invalid webhook name".to_string()));
219 }
220 am.name = Set(name.to_string());
221 }
222 if let Some(url) = cmd.url {
223 validate_outbound_url(url.trim())?;
224 am.url = Set(url.trim().to_string());
225 }
226 if let Some(disabled) = cmd.disabled {
227 am.disabled = Set(disabled);
228 }
229 am.updated_at = Set(OffsetDateTime::now_utc());
230 let updated = am.update(self.db().write()).await?;
231 Ok(OutboundWebhookDto::from(updated))
232 }
233
234 pub async fn delete_outbound_webhook(
235 &self,
236 caller: Uuid,
237 room_id: Uuid,
238 webhook_id: Uuid,
239 ) -> Result<(), ChatError> {
240 require_room_manager(self, room_id, caller).await?;
241 let deleted = outbound_webhooks::Entity::delete_many()
242 .filter(outbound_webhooks::Column::Id.eq(webhook_id))
243 .filter(outbound_webhooks::Column::RoomId.eq(room_id))
244 .exec(self.db().write())
245 .await?;
246 if deleted.rows_affected == 0 {
247 return Err(ChatError::NotFound);
248 }
249 webhook_deliveries::Entity::delete_many()
250 .filter(webhook_deliveries::Column::WebhookId.eq(webhook_id))
251 .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
252 .exec(self.db().write())
253 .await?;
254 Ok(())
255 }
256
257 pub async fn deliver_pending_webhook_events(&self) -> Result<u64, ChatError> {
258 let now = OffsetDateTime::now_utc();
259
260 let _ = webhook_deliveries::Entity::update_many()
261 .col_expr(
262 webhook_deliveries::Column::Status,
263 Expr::value(WebhookDeliveryStatus::Pending),
264 )
265 .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Delivering))
266 .filter(
267 webhook_deliveries::Column::NextRetryAt
268 .lt(now - time::Duration::minutes(crate::worker::STUCK_CLAIM_MINUTES)),
269 )
270 .exec(self.db().write())
271 .await?;
272
273 let due = webhook_deliveries::Entity::find()
274 .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
275 .filter(webhook_deliveries::Column::NextRetryAt.lte(now))
276 .limit(50)
277 .all(self.db().write())
278 .await?;
279 let count = due.len() as u64;
280 for delivery in due {
281 let claimed = webhook_deliveries::Entity::update_many()
282 .col_expr(
283 webhook_deliveries::Column::Status,
284 Expr::value(WebhookDeliveryStatus::Delivering),
285 )
286 .filter(webhook_deliveries::Column::Id.eq(delivery.id))
287 .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
288 .exec(self.db().write())
289 .await?;
290 if claimed.rows_affected == 0 {
291 continue;
292 }
293
294 let Some(webhook) = outbound_webhooks::Entity::find_by_id(delivery.webhook_id)
295 .one(self.db().read())
296 .await?
297 else {
298 self.finish_delivery(delivery.id, WebhookDeliveryStatus::Failed, Some("webhook deleted".to_string()), delivery.attempts)
299 .await;
300 continue;
301 };
302 let result = if webhook.disabled {
303 Err("webhook disabled".to_string())
304 } else {
305 self.post_webhook_payload(&webhook, &delivery).await
306 };
307 let (status, error) = match result {
308 Ok(status_line) => {
309 let mut am: outbound_webhooks::ActiveModel = webhook.into();
310 am.last_status = Set(Some(status_line.clone()));
311 am.last_used_at = Set(Some(OffsetDateTime::now_utc()));
312 am.updated_at = Set(OffsetDateTime::now_utc());
313 let _ = am.update(self.db().write()).await;
314 (WebhookDeliveryStatus::Done, None)
315 }
316 Err(error) => {
317 let attempts = delivery.attempts + 1;
318 if attempts >= MAX_DELIVERY_ATTEMPTS {
319 (WebhookDeliveryStatus::Failed, Some(error))
320 } else {
321 let mut am: webhook_deliveries::ActiveModel = delivery.clone().into();
322 am.status = Set(WebhookDeliveryStatus::Pending);
323 am.attempts = Set(attempts);
324 am.next_retry_at =
325 Set(OffsetDateTime::now_utc() + time::Duration::seconds(60 * attempts as i64));
326 am.last_error = Set(Some(error.clone()));
327 am.updated_at = Set(OffsetDateTime::now_utc());
328 let _ = am.update(self.db().write()).await;
329 continue;
330 }
331 }
332 };
333 self.finish_delivery(delivery.id, status, error, delivery.attempts)
334 .await;
335 }
336 Ok(count)
337 }
338
339 async fn finish_delivery(
340 &self,
341 id: Uuid,
342 status: WebhookDeliveryStatus,
343 error: Option<String>,
344 attempts: i32,
345 ) {
346 let am = webhook_deliveries::ActiveModel {
347 id: Set(id),
348 status: Set(status),
349 attempts: Set(attempts + 1),
350 last_error: Set(error),
351 updated_at: Set(OffsetDateTime::now_utc()),
352 ..Default::default()
353 };
354 let _ = am.update(self.db().write()).await;
355 }
356
357 async fn post_webhook_payload(
358 &self,
359 webhook: &outbound_webhooks::Model,
360 delivery: &webhook_deliveries::Model,
361 ) -> Result<String, String> {
362 use hmac::{Hmac, KeyInit, Mac};
363 use sha2::Sha256;
364
365 let uri = validate_outbound_url(&webhook.url).map_err(|e| e.to_string())?;
366 let host = uri.host().unwrap_or("").to_string();
367 let port = uri.port_u16();
368
369 let addrs = tokio::net::lookup_host((host.as_str(), port.unwrap_or(443)))
370 .await
371 .map_err(|e| format!("dns resolution failed: {e}"))?
372 .collect::<Vec<_>>();
373 if addrs.is_empty() {
374 return Err("dns resolution returned no address".to_string());
375 }
376 if addrs.iter().any(|addr| !is_public_ip(&addr.ip())) {
377 return Err("url resolves to a private or loopback address".to_string());
378 }
379
380 let body = serde_json::to_vec(&delivery.payload).map_err(|e| e.to_string())?;
381 let mut mac =
382 Hmac::<Sha256>::new_from_slice(webhook.secret.as_bytes()).map_err(|e| e.to_string())?;
383 mac.update(&body);
384 let signature = format!("sha256={}", hex_lower(&mac.finalize().into_bytes()));
385
386 let mut client_builder = reqwest::Client::builder()
387 .timeout(std::time::Duration::from_secs(10))
388 .redirect(reqwest::redirect::Policy::none());
389 for addr in &addrs {
390 client_builder = client_builder.resolve(&host, *addr);
391 }
392 let client = client_builder.build().map_err(|e| e.to_string())?;
393
394 let response = client
395 .post(&webhook.url)
396 .header("content-type", "application/json")
397 .header("x-gitdata-event", &delivery.event_type)
398 .header("x-gitdata-delivery", delivery.id.to_string())
399 .header("x-gitdata-signature", &signature)
400 .body(body)
401 .send()
402 .await
403 .map_err(|e| format!("request failed: {e}"))?;
404 let status = response.status().as_u16();
405 if (200..300).contains(&status) {
406 Ok(format!("delivered {status}"))
407 } else {
408 Err(format!("endpoint returned {status}"))
409 }
410 }
411}
412
413fn hex_lower(bytes: &[u8]) -> String {
414 bytes.iter().map(|b| format!("{b:02x}")).collect()
415}
416
417pub async fn enqueue_outbound_event(
418 service: &ChatService,
419 room_id: Uuid,
420 event_type: &str,
421 payload: &serde_json::Value,
422) {
423 let hooks = outbound_webhooks::Entity::find()
424 .filter(outbound_webhooks::Column::RoomId.eq(room_id))
425 .filter(outbound_webhooks::Column::Disabled.eq(false))
426 .all(service.db().read())
427 .await
428 .unwrap_or_default();
429 if hooks.is_empty() {
430 return;
431 }
432 let now = OffsetDateTime::now_utc();
433 for hook in hooks {
434 let row = webhook_deliveries::ActiveModel {
435 id: Set(Uuid::now_v7()),
436 webhook_id: Set(hook.id),
437 room_id: Set(room_id),
438 event_type: Set(event_type.to_string()),
439 payload: Set(payload.clone()),
440 status: Set(WebhookDeliveryStatus::Pending),
441 attempts: Set(0),
442 next_retry_at: Set(now),
443 last_error: Set(None),
444 created_at: Set(now),
445 updated_at: Set(now),
446 };
447 if let Err(error) = row.insert(service.db().write()).await {
448 tracing::warn!(%error, "failed to enqueue outbound webhook delivery");
449 }
450 }
451}
Modified lib/chat/src/services/reminders.rs lib/chat/src/services/reminders.rs +10 −0 View lib/chat/src/services/reminders.rs
@@ -228,6 +228,16 @@
228228 .await
229229 {
230230 tracing::warn!(%error, "failed to write reminder notification");
231 message_reminders::Entity::update_many()
232 .col_expr(
233 message_reminders::Column::Status,
234 Expr::value(ReminderStatus::Pending),
235 )
236 .filter(message_reminders::Column::Id.eq(reminder.id))
237 .filter(message_reminders::Column::Status.eq(ReminderStatus::Delivering))
238 .exec(self.db().write())
239 .await?;
240 continue;
231241 }
232242 message_reminders::Entity::update_many()
233243 .col_expr(
Modified lib/chat/src/services/rooms.rs lib/chat/src/services/rooms.rs +8 −1 View lib/chat/src/services/rooms.rs
@@ -143,7 +143,7 @@
143143 space_name: &str,
144144 ) -> Result<Vec<RoomBriefDto>, ChatError> {
145145 let space = lookup_space_by_name(self, space_name).await?;
146 require_space_member(self, &space, caller).await?;
146 let member = require_space_member(self, &space, caller).await?;
147147
148148 let rooms = rooms::Entity::find()
149149 .filter(rooms::Column::SpaceId.eq(space.id))
@@ -238,6 +238,13 @@
238238 })
239239 .unwrap_or(0);
240240 let setting = settings_map.get(&room.id);
241 if member.hide_muted_channels
242 && setting.is_some_and(|s| {
243 crate::share::notify::mute_active(s.muted, s.mute_until)
244 })
245 {
246 continue;
247 }
241248 result.push(RoomBriefDto {
242249 id: room.id,
243250 owner_id: room.owner_id,
Modified lib/chat/src/services/search.rs lib/chat/src/services/search.rs +49 −7 View lib/chat/src/services/search.rs
@@ -73,20 +73,36 @@
7373 .clamp(0, export::pagination::MAX_OFFSET);
7474
7575 // Private rooms must be restricted to their members; public space
76 // rooms are readable by every space member.
77 const FILTERS: &str = r#"
76 // rooms are readable by every space member. CJK text is not tokenized
77 // by Postgres FTS ('simple' config), so CJK queries fall back to
78 // substring ILIKE matching.
79 let cjk = query.chars().any(is_cjk_char);
80 let text_predicate = if cjk {
81 "AND m.content ILIKE $8"
82 } else {
83 "AND m.search_vector @@ plainto_tsquery('simple', $2)"
84 };
85 let order_clause = if cjk {
86 "ORDER BY m.created_at DESC"
87 } else {
88 "ORDER BY ts_rank(m.search_vector, plainto_tsquery('simple', $2)) DESC, m.created_at DESC"
89 };
90 let like_pattern = format!("%{}%", escape_like(query));
91 let filters = format!(
92 r#"
7893 AND r.space_id = $1
7994 AND m.deleted_at IS NULL
80 AND m.search_vector @@ plainto_tsquery('simple', $2)
95 {text_predicate}
8196 AND ($5::uuid IS NULL OR m.room_id = $5)
8297 AND ($6::uuid IS NULL OR m.author_id = $6)
8398 AND (r.is_private = false OR r.owner_id = $7
8499 OR EXISTS (SELECT 1 FROM room_members rm
85100 WHERE rm.room_id = m.room_id AND rm.user_id = $7
86 AND rm.left_at IS NULL))"#;
101 AND rm.left_at IS NULL))"#
102 );
87103
88104 let count_sql = format!(
89 "SELECT count(*)::bigint AS total FROM chat_message m JOIN rooms r ON r.id = m.room_id WHERE 1=1{FILTERS}"
105 "SELECT count(*)::bigint AS total FROM chat_message m JOIN rooms r ON r.id = m.room_id WHERE 1=1{filters}"
90106 );
91107 let count_row = CountRow::find_by_statement(Statement::from_sql_and_values(
92108 DatabaseBackend::Postgres,
@@ -99,6 +115,7 @@
99115 q.room_id.into(),
100116 q.author_id.into(),
101117 caller.into(),
118 like_pattern.clone().into(),
102119 ],
103120 ))
104121 .one(self.db().read())
@@ -109,8 +126,8 @@
109126 r#"SELECT m.id AS message_id, m.room_id, m.author_id, m.seq, m.content, m.created_at
110127 FROM chat_message m
111128 JOIN rooms r ON r.id = m.room_id
112 WHERE 1=1{FILTERS}
113 ORDER BY ts_rank(m.search_vector, plainto_tsquery('simple', $2)) DESC, m.created_at DESC
129 WHERE 1=1{filters}
130 {order_clause}
114131 LIMIT $3 OFFSET $4"#
115132 );
116133 let rows = SearchRow::find_by_statement(Statement::from_sql_and_values(
@@ -124,6 +141,7 @@
124141 q.room_id.into(),
125142 q.author_id.into(),
126143 caller.into(),
144 like_pattern.into(),
127145 ],
128146 ))
129147 .all(self.db().read())
@@ -149,3 +167,27 @@
149167 })
150168 }
151169}
170
171fn is_cjk_char(c: char) -> bool {
172 matches!(c as u32,
173 0x4E00..=0x9FFF
174 | 0x3400..=0x4DBF
175 | 0x20000..=0x2A6DF
176 | 0xF900..=0xFAFF
177 | 0x3040..=0x30FF
178 | 0x31F0..=0x31FF
179 | 0xAC00..=0xD7AF
180 | 0x1100..=0x11FF
181 | 0x3130..=0x318F)
182}
183
184fn escape_like(q: &str) -> String {
185 let mut out = String::with_capacity(q.len());
186 for c in q.chars() {
187 if matches!(c, '%' | '_' | '\\') {
188 out.push('\\');
189 }
190 out.push(c);
191 }
192 out
193}
Added lib/chat/src/services/unfurl.rs lib/chat/src/services/unfurl.rs +345 −0 View lib/chat/src/services/unfurl.rs
@@ -0,0 +1,345 @@
1use models::chat::link_unfurls::{self, UnfurlStatus};
2use models::chat::messages;
3use sea_orm::{ActiveModelTrait, EntityTrait, Set};
4use sha2::{Digest, Sha256};
5use time::OffsetDateTime;
6use uuid::Uuid;
7
8use crate::services::outbound::is_public_ip;
9use crate::services::service::ChatService;
10use crate::share::notify::current_room_seq;
11use crate::share::realtime::RoomEvent;
12
13const MAX_URLS_PER_MESSAGE: usize = 3;
14const MAX_FETCH_BYTES: usize = 1_048_576;
15const CACHE_FRESH_DAYS: i64 = 7;
16const FAILED_RETRY_MINUTES: i64 = 60;
17
18fn url_hash(url: &str) -> String {
19 let mut hasher = Sha256::new();
20 hasher.update(url.as_bytes());
21 hasher
22 .finalize()
23 .iter()
24 .map(|b| format!("{b:02x}"))
25 .collect()
26}
27
28fn extract_urls(content: &str) -> Vec<String> {
29 let mut urls = Vec::new();
30 for token in content.split_whitespace() {
31 let trimmed = token.trim_matches(|c: char| {
32 !c.is_alphanumeric() && !matches!(c, ':' | '/' | '.' | '-' | '_' | '?' | '&' | '=' | '%' | '~' | '+' | '#' | '@')
33 });
34 if !(trimmed.starts_with("http://") || trimmed.starts_with("https://")) {
35 continue;
36 }
37 let cleaned = trimmed.trim_end_matches(|c: char| matches!(c, '.' | ',' | ')' | ']' | '>' | '"' | '\''));
38 if cleaned.len() < 12 || cleaned.len() > 2048 {
39 continue;
40 }
41 if !urls.iter().any(|u: &String| u == cleaned) {
42 urls.push(cleaned.to_string());
43 }
44 if urls.len() >= MAX_URLS_PER_MESSAGE {
45 break;
46 }
47 }
48 urls
49}
50
51fn meta_content(head: &str, property: &str) -> Option<String> {
52 let lower = head.to_ascii_lowercase();
53 let mut search_from = 0;
54 loop {
55 let idx = lower[search_from..].find("<meta")? + search_from;
56 let end = lower[idx..].find('>')? + idx;
57 let tag = &head[idx..=end];
58 let tag_lower = &lower[idx..=end];
59 search_from = end + 1;
60 let matches = tag_lower.contains(&format!("property=\"{property}\""))
61 || tag_lower.contains(&format!("property='{property}'"))
62 || tag_lower.contains(&format!("name=\"{property}\""))
63 || tag_lower.contains(&format!("name='{property}'"));
64 if !matches {
65 continue;
66 }
67 for attr in ["content=\"", "content='"] {
68 if let Some(start) = tag.find(attr) {
69 let value_start = start + attr.len();
70 let quote = attr.chars().last()?;
71 if let Some(value_end) = tag[value_start..].find(quote) {
72 let value = tag[value_start..value_start + value_end].trim();
73 if !value.is_empty() {
74 return Some(decode_entities(value).chars().take(300).collect());
75 }
76 }
77 }
78 }
79 }
80}
81
82fn page_title(head: &str) -> Option<String> {
83 let lower = head.to_ascii_lowercase();
84 let start = lower.find("<title")?;
85 let content_start = lower[start..].find('>')? + start + 1;
86 let end = lower[content_start..].find("</title>")? + content_start;
87 let title = head[content_start..end].trim();
88 (!title.is_empty()).then(|| decode_entities(title).chars().take(200).collect())
89}
90
91fn decode_entities(value: &str) -> String {
92 value
93 .replace("&", "&")
94 .replace("<", "<")
95 .replace(">", ">")
96 .replace(""", "\"")
97 .replace("'", "'")
98 .replace("'", "'")
99 .replace(" ", " ")
100}
101
102struct FetchedMeta {
103 title: Option<String>,
104 description: Option<String>,
105 image_url: Option<String>,
106 site_name: Option<String>,
107}
108
109async fn fetch_meta(url: &str) -> Result<FetchedMeta, String> {
110 use crate::services::outbound::validate_outbound_url;
111
112 let uri = validate_outbound_url(url).map_err(|e| e.to_string())?;
113 let host = uri.host().unwrap_or("").to_string();
114 let port = uri.port_u16();
115 let addrs = tokio::net::lookup_host((host.as_str(), port.unwrap_or(443)))
116 .await
117 .map_err(|e| format!("dns resolution failed: {e}"))?
118 .collect::<Vec<_>>();
119 if addrs.is_empty() {
120 return Err("dns resolution returned no address".to_string());
121 }
122 if addrs.iter().any(|addr| !is_public_ip(&addr.ip())) {
123 return Err("url resolves to a private or loopback address".to_string());
124 }
125 let mut client_builder = reqwest::Client::builder()
126 .timeout(std::time::Duration::from_secs(10))
127 .redirect(reqwest::redirect::Policy::none())
128 .user_agent("gitdata-chat-unfurl/1.0");
129 for addr in &addrs {
130 client_builder = client_builder.resolve(&host, *addr);
131 }
132 let client = client_builder.build().map_err(|e| e.to_string())?;
133
134 let response = client
135 .get(url)
136 .header("accept", "text/html")
137 .send()
138 .await
139 .map_err(|e| format!("request failed: {e}"))?;
140 if !response.status().is_success() {
141 return Err(format!("endpoint returned {}", response.status().as_u16()));
142 }
143 let content_type = response
144 .headers()
145 .get("content-type")
146 .and_then(|v| v.to_str().ok())
147 .unwrap_or("")
148 .to_lowercase();
149 if !content_type.is_empty() && !content_type.contains("text/html") {
150 return Err("not an html document".to_string());
151 }
152
153 let mut body: Vec<u8> = Vec::new();
154 let mut stream = response;
155 while let Some(chunk) = stream.chunk().await.map_err(|e| e.to_string())? {
156 body.extend_from_slice(&chunk);
157 if body.len() >= MAX_FETCH_BYTES {
158 break;
159 }
160 }
161 let text = String::from_utf8_lossy(&body);
162 let head = match text.find("</head>") {
163 Some(idx) => &text[..idx],
164 None => &text[..text.len().min(MAX_FETCH_BYTES)],
165 };
166
167 Ok(FetchedMeta {
168 title: meta_content(head, "og:title").or_else(|| page_title(head)),
169 description: meta_content(head, "og:description").or_else(|| meta_content(head, "description")),
170 image_url: meta_content(head, "og:image"),
171 site_name: meta_content(head, "og:site_name"),
172 })
173}
174
175fn embed_json(url: &str, row: &link_unfurls::Model) -> serde_json::Value {
176 serde_json::json!({
177 "type": "link",
178 "url": url,
179 "title": row.title,
180 "description": row.description,
181 "image_url": row.image_url,
182 "site_name": row.site_name,
183 })
184}
185
186impl ChatService {
187 async fn resolve_unfurl(&self, url: &str) -> Option<serde_json::Value> {
188 let hash = url_hash(url);
189 let now = OffsetDateTime::now_utc();
190 let cached = link_unfurls::Entity::find_by_id(hash.clone())
191 .one(self.db().read())
192 .await
193 .ok()
194 .flatten();
195 if let Some(row) = &cached {
196 let fresh = row.status == UnfurlStatus::Ready
197 && row.fetched_at > now - time::Duration::days(CACHE_FRESH_DAYS);
198 let retryable_failed = row.status == UnfurlStatus::Failed
199 && row.fetched_at < now - time::Duration::minutes(FAILED_RETRY_MINUTES);
200 if fresh {
201 return Some(embed_json(url, row));
202 }
203 if row.status == UnfurlStatus::Failed && !retryable_failed {
204 return None;
205 }
206 }
207
208 let meta = match fetch_meta(url).await {
209 Ok(meta) => meta,
210 Err(_) => {
211 self.upsert_unfurl_row(&hash, url, None, now).await;
212 return None;
213 }
214 };
215 if meta.title.is_none() && meta.description.is_none() && meta.image_url.is_none() {
216 self.upsert_unfurl_row(&hash, url, None, now).await;
217 return None;
218 }
219 let row = self
220 .upsert_unfurl_row(
221 &hash,
222 url,
223 Some((
224 meta.title.clone(),
225 meta.description.clone(),
226 meta.image_url.clone(),
227 meta.site_name.clone(),
228 )),
229 now,
230 )
231 .await?;
232 Some(embed_json(url, &row))
233 }
234
235 async fn upsert_unfurl_row(
236 &self,
237 hash: &str,
238 url: &str,
239 meta: Option<(Option<String>, Option<String>, Option<String>, Option<String>)>,
240 now: OffsetDateTime,
241 ) -> Option<link_unfurls::Model> {
242 let (status, title, description, image_url, site_name) = match &meta {
243 Some((t, d, i, s)) => (UnfurlStatus::Ready, t.clone(), d.clone(), i.clone(), s.clone()),
244 None => (UnfurlStatus::Failed, None, None, None, None),
245 };
246 let row = link_unfurls::ActiveModel {
247 url_hash: Set(hash.to_string()),
248 url: Set(url.chars().take(2048).collect()),
249 title: Set(title.clone()),
250 description: Set(description.clone()),
251 image_url: Set(image_url.clone()),
252 site_name: Set(site_name.clone()),
253 status: Set(status),
254 fetched_at: Set(now),
255 };
256 match row.insert(self.db().write()).await {
257 Ok(inserted) => Some(inserted),
258 Err(_) => {
259 let existing = link_unfurls::Entity::find_by_id(hash.to_string())
260 .one(self.db().write())
261 .await
262 .ok()
263 .flatten()?;
264 let mut am: link_unfurls::ActiveModel = existing.into();
265 am.url = Set(url.chars().take(2048).collect());
266 am.title = Set(title);
267 am.description = Set(description);
268 am.image_url = Set(image_url);
269 am.site_name = Set(site_name);
270 am.status = Set(status);
271 am.fetched_at = Set(now);
272 am.update(self.db().write()).await.ok()
273 }
274 }
275 }
276
277 pub async fn unfurl_message(&self, room_id: Uuid, message_id: Uuid, content: String) {
278 let urls = extract_urls(&content);
279 if urls.is_empty() {
280 return;
281 }
282 let mut embeds = Vec::new();
283 for url in urls {
284 if let Some(embed) = self.resolve_unfurl(&url).await {
285 embeds.push(embed);
286 }
287 }
288 if embeds.is_empty() {
289 return;
290 }
291 let Some(row) = messages::Entity::find_by_id(message_id)
292 .one(self.db().write())
293 .await
294 .ok()
295 .flatten()
296 else {
297 return;
298 };
299 if row.deleted_at.is_some() {
300 return;
301 }
302 if row.embeds.as_array().is_some_and(|arr| !arr.is_empty()) {
303 return;
304 }
305 let mut am: messages::ActiveModel = row.into();
306 am.embeds = Set(serde_json::json!(embeds));
307 if am.update(self.db().write()).await.is_err() {
308 return;
309 }
310 let Some(updated) = messages::Entity::find_by_id(message_id)
311 .one(self.db().read())
312 .await
313 .ok()
314 .flatten()
315 else {
316 return;
317 };
318 let Ok(items) = self.build_message_dto(vec![updated], Uuid::nil()).await else {
319 return;
320 };
321 let Some(dto) = items.into_iter().next() else {
322 return;
323 };
324 let Ok(payload) = serde_json::to_value(&dto) else {
325 return;
326 };
327 let seq = current_room_seq(self.db().read(), room_id).await;
328 let event = RoomEvent::MessageEdited {
329 room_id,
330 seq,
331 message: payload,
332 };
333 self.realtime.publish_room(&event).await;
334 }
335}
336
337pub fn spawn_unfurl(service: &ChatService, room_id: Uuid, message_id: Uuid, content: String) {
338 if !content.contains("http://") && !content.contains("https://") {
339 return;
340 }
341 let service = service.clone();
342 tokio::spawn(async move {
343 service.unfurl_message(room_id, message_id, content).await;
344 });
345}
Modified lib/chat/src/services/webhooks.rs lib/chat/src/services/webhooks.rs +27 −3 View lib/chat/src/services/webhooks.rs
@@ -52,6 +52,8 @@
5252 pub content: String,
5353 #[serde(default)]
5454 pub nonce: Option<String>,
55 #[serde(default)]
56 pub thread_id: Option<Uuid>,
5557}
5658
5759fn webhook_token() -> String {
@@ -274,6 +276,17 @@
274276 am.last_used_at = Set(Some(now));
275277 am.update(self.db().write()).await?;
276278
279 if let Some(thread_id) = cmd.thread_id {
280 let thread = models::chat::threads::Entity::find_by_id(thread_id)
281 .one(self.db().read())
282 .await?
283 .filter(|t| t.room_id == webhook.room_id)
284 .ok_or(ChatError::NotFound)?;
285 if thread.archived || thread.locked {
286 return Err(ChatError::Forbidden);
287 }
288 }
289
277290 let message = self
278291 .post_webhook_message(
279292 webhook.room_id,
@@ -281,13 +294,12 @@
281294 SendMessageCmd {
282295 content: cmd.content,
283296 reply_to_id: None,
284 thread_id: None,
297 thread_id: cmd.thread_id,
285298 thread_broadcast: false,
286299 attachments: Vec::new(),
287300 nonce: cmd.nonce,
288301 urgent: false,
289302 forwarded_from: None,
290 send_at: None,
291303 },
292304 )
293305 .await?;
@@ -333,6 +345,7 @@
333345 let content_clone = content.clone();
334346 let webhook_id = webhook.id;
335347 let owner = webhook.user_id;
348 let thread_id = cmd.thread_id;
336349
337350 let seq = self
338351 .db()
@@ -346,7 +359,7 @@
346359 author_id: Set(owner),
347360 author_type: Set(AuthorType::Webhook),
348361 webhook_id: Set(Some(webhook_id)),
349 thread_id: Set(None),
362 thread_id: Set(thread_id),
350363 thread_broadcast: Set(false),
351364 reply_to_id: Set(None),
352365 forwarded_from_id: Set(None),
@@ -369,6 +382,16 @@
369382 }
370383 .insert(txn)
371384 .await?;
385 if let Some(tid) = thread_id {
386 if let Some(t) =
387 models::chat::threads::Entity::find_by_id(tid).one(txn).await?
388 {
389 crate::services::threads::maintain_thread_on_reply(
390 txn, &t, message_id, owner, now,
391 )
392 .await?;
393 }
394 }
372395 rooms::Entity::update_many()
373396 .filter(rooms::Column::Id.eq(room_id))
374397 .col_expr(rooms::Column::LastMessageId, Expr::value(message_id))
@@ -388,6 +411,7 @@
388411 let dto = items.into_iter().next().unwrap();
389412 let value = serde_json::to_value(&dto)?;
390413 crate::share::notify::publish_message_event(self, room_id, seq, value).await;
414 crate::services::unfurl::spawn_unfurl(self, room_id, message_id, dto.content.clone());
391415 Ok(dto)
392416 }
393417}
Modified lib/chat/src/share/notify.rs lib/chat/src/share/notify.rs +103 −7 View lib/chat/src/share/notify.rs
@@ -1,9 +1,12 @@
11use models::chat::room_seq;
22use models::chat::room_user_settings::{self, RoomNotificationLevel};
3use models::space::space_members::{self, MessageNotifications};
3use models::space::space_members::{self, MemberStatus, MessageNotifications};
44use models::users::user_notifications::{self, NotificationType};
55use models::users::user_recent_mentions;
6use sea_orm::{ActiveModelTrait, ConnectionTrait, DatabaseBackend, EntityTrait, Set, Statement};
6use sea_orm::{
7 ActiveModelTrait, ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, QueryFilter, Set,
8 Statement,
9};
710use time::OffsetDateTime;
811use uuid::Uuid;
912
@@ -93,7 +96,7 @@
9396 Skip,
9497}
9598
96fn mute_active(muted: bool, mute_until: Option<OffsetDateTime>) -> bool {
99pub(crate) fn mute_active(muted: bool, mute_until: Option<OffsetDateTime>) -> bool {
97100 muted && mute_until.is_none_or(|until| until > OffsetDateTime::now_utc())
98101}
99102
@@ -103,13 +106,14 @@
103106 space_id: Option<Uuid>,
104107 uid: Uuid,
105108 mentioned: bool,
109 everyone: bool,
106110) -> Result<NotifyDecision, ChatError> {
107111 if let Some(setting) = room_user_settings::Entity::find_by_id((uid, room_id))
108112 .one(service.db().read())
109113 .await?
110114 {
111115 if mute_active(setting.muted, setting.mute_until) {
112 return Ok(if mentioned {
116 return Ok(if mentioned && !everyone {
113117 NotifyDecision::Notify
114118 } else {
115119 NotifyDecision::Skip
@@ -118,7 +122,7 @@
118122 match setting.notification_level {
119123 RoomNotificationLevel::All => return Ok(NotifyDecision::Notify),
120124 RoomNotificationLevel::Mentions | RoomNotificationLevel::Nothing => {
121 return Ok(if mentioned {
125 return Ok(if mentioned && !everyone {
122126 NotifyDecision::Notify
123127 } else {
124128 NotifyDecision::Skip
@@ -133,13 +137,13 @@
133137 .one(service.db().read())
134138 .await?
135139 {
136 if mute_active(member.muted, member.mute_until) && !mentioned {
140 if mute_active(member.muted, member.mute_until) && !(mentioned && !everyone) {
137141 return Ok(NotifyDecision::Skip);
138142 }
139143 return Ok(match member.message_notifications {
140144 MessageNotifications::All => NotifyDecision::Notify,
141145 MessageNotifications::Mentions | MessageNotifications::Nothing => {
142 if mentioned {
146 if mentioned && (!everyone || !member.suppress_everyone) {
143147 NotifyDecision::Notify
144148 } else {
145149 NotifyDecision::Skip
@@ -151,6 +155,89 @@
151155 Ok(NotifyDecision::Notify)
152156}
153157
158/// Fan out an @everyone mention to room/space members that were not explicitly
159/// mentioned. Runs best-effort after commit: writes notifications and recent
160/// mentions, then pushes a WS user event.
161pub async fn fanout_everyone_mentions(
162 service: &ChatService,
163 room_id: Uuid,
164 space_id: Option<Uuid>,
165 is_private: bool,
166 message_id: Uuid,
167 body: String,
168 actor: Uuid,
169 explicit: &[Uuid],
170) {
171 use models::room::room_members;
172
173 let candidates: Vec<Uuid> = if is_private || space_id.is_none() {
174 room_members::Entity::find()
175 .filter(room_members::Column::RoomId.eq(room_id))
176 .filter(room_members::Column::LeftAt.is_null())
177 .all(service.db().read())
178 .await
179 .unwrap_or_default()
180 .into_iter()
181 .map(|m| m.user_id)
182 .collect()
183 } else if let Some(space_id) = space_id {
184 space_members::Entity::find()
185 .filter(space_members::Column::SpaceId.eq(space_id))
186 .filter(space_members::Column::Status.eq(MemberStatus::Active))
187 .all(service.db().read())
188 .await
189 .unwrap_or_default()
190 .into_iter()
191 .filter(|m| !m.suppress_everyone)
192 .map(|m| m.user_id)
193 .collect()
194 } else {
195 Vec::new()
196 };
197
198 let data = serde_json::json!({
199 "room_id": room_id,
200 "message_id": message_id,
201 "everyone": true,
202 });
203 for uid in candidates {
204 if uid == actor || explicit.contains(&uid) {
205 continue;
206 }
207 match notification_decision(service, room_id, space_id, uid, true, true).await {
208 Ok(NotifyDecision::Notify) => {}
209 _ => continue,
210 }
211 let now = OffsetDateTime::now_utc();
212 let _ = user_recent_mentions::ActiveModel {
213 id: Set(Uuid::now_v7()),
214 user_id: Set(uid),
215 channel_id: Set(room_id),
216 message_id: Set(message_id),
217 created_at: Set(now),
218 }
219 .insert(service.db().write())
220 .await;
221 let _ = notify_user(
222 service.db().write(),
223 uid,
224 NotificationType::Mention,
225 "You were mentioned in a message".to_string(),
226 Some(body.clone()),
227 Some(data.clone()),
228 Some(actor),
229 )
230 .await;
231 service
232 .realtime
233 .publish_user(
234 uid,
235 &serde_json::json!({"kind": "mention", "room_id": room_id, "message_id": message_id, "everyone": true}),
236 )
237 .await;
238 }
239}
240
154241/// Mention e-mails honour per-room / per-space mute and notification level:
155242/// Nothing, muted (even mid-window) and blocked settings opt out.
156243pub async fn mention_email_allowed(
@@ -200,6 +287,15 @@
200287 message,
201288 };
202289 service.realtime.publish_room(&event).await;
290 if let Ok(payload) = serde_json::to_value(&event) {
291 crate::services::outbound::enqueue_outbound_event(
292 service,
293 room_id,
294 "message_published",
295 &payload,
296 )
297 .await;
298 }
203299}
204300
205301pub async fn advance_read_state(
Modified lib/chat/src/share/realtime.rs lib/chat/src/share/realtime.rs +1 −0 View lib/chat/src/share/realtime.rs
@@ -54,6 +54,7 @@
5454 room_id: Uuid,
5555 user_id: Uuid,
5656 thread_id: Option<Uuid>,
57 ttl_secs: u64,
5758 },
5859 RoomUpdated {
5960 room_id: Uuid,
Modified lib/chat/src/ws/mod.rs lib/chat/src/ws/mod.rs +1 −0 View lib/chat/src/ws/mod.rs
@@ -542,6 +542,7 @@
542542 room_id,
543543 user_id: uid,
544544 thread_id,
545 ttl_secs: service.config().chat.typing_ttl_secs.max(1),
545546 };
546547 service.realtime.publish_room(&event).await;
547548 }
Modified lib/chat/src/errors.rs lib/chat/src/errors.rs +0 −6 View lib/chat/src/errors.rs
@@ -18,12 +18,6 @@
1818 Internal(String),
1919}
2020
21impl ChatError {
22 pub fn is_rate_limited(&self) -> bool {
23 matches!(self, Self::RateLimited(_))
24 }
25}
26
2721impl fmt::Display for ChatError {
2822 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2923 match self {
Modified lib/chat/src/openapi.rs lib/chat/src/openapi.rs +8 −2 View lib/chat/src/openapi.rs
@@ -83,6 +83,10 @@
8383 crate::handlers::webhooks::delete_webhook,
8484 crate::handlers::webhooks::update_webhook,
8585 crate::handlers::webhooks::post_via_webhook,
86 crate::handlers::webhooks::create_outbound_webhook,
87 crate::handlers::webhooks::list_outbound_webhooks,
88 crate::handlers::webhooks::update_outbound_webhook,
89 crate::handlers::webhooks::delete_outbound_webhook,
8690 crate::handlers::ws::issue_ws_ticket,
8791 ),
8892 components(schemas(
@@ -109,12 +113,10 @@
109113 models::chat::message_reminders::ReminderStatus,
110114 models::room::rooms::RoomKind,
111115 models::room::room_members::RoomMemberRole,
112 models::users::user_scheduled_messages::ScheduledMessageStatus,
113116 crate::services::messages::MessageDto,
114117 crate::services::messages::MessagePage,
115118 crate::services::messages::MessageEditDto,
116119 crate::services::messages::SendMessageCmd,
117 crate::services::messages::ScheduledMessageDto,
118120 crate::services::messages::UpdateMessageCmd,
119121 crate::services::messages::BulkDeleteCmd,
120122 crate::services::messages::PinCmd,
@@ -174,6 +176,10 @@
174176 crate::services::webhooks::WebhookSecretDto,
175177 crate::services::webhooks::CreateWebhookCmd,
176178 crate::services::webhooks::WebhookPostCmd,
179 crate::services::outbound::OutboundWebhookDto,
180 crate::services::outbound::OutboundWebhookSecretDto,
181 crate::services::outbound::CreateOutboundWebhookCmd,
182 crate::services::outbound::UpdateOutboundWebhookCmd,
177183 crate::handlers::ws::WsTicketDto,
178184 ))
179185)]
Modified lib/chat/src/routes.rs lib/chat/src/routes.rs +10 −0 View lib/chat/src/routes.rs
@@ -219,6 +219,16 @@
219219 "/chat/rooms/{room_id}/webhooks/{webhook_id}",
220220 axum::routing::delete(delete_webhook).patch(crate::handlers::webhooks::update_webhook),
221221 )
222 .route(
223 "/chat/rooms/{room_id}/outbound-webhooks",
224 get(crate::handlers::webhooks::list_outbound_webhooks)
225 .post(crate::handlers::webhooks::create_outbound_webhook),
226 )
227 .route(
228 "/chat/rooms/{room_id}/outbound-webhooks/{webhook_id}",
229 axum::routing::patch(crate::handlers::webhooks::update_outbound_webhook)
230 .delete(crate::handlers::webhooks::delete_outbound_webhook),
231 )
222232 .route("/chat/ws/ticket", post(issue_ws_ticket))
223233 .route_layer(axum::middleware::from_fn_with_state(
224234 service.clone(),
Modified lib/chat/src/worker.rs lib/chat/src/worker.rs +10 −88 View lib/chat/src/worker.rs
@@ -2,97 +2,16 @@
22
33use export::AppState;
44use models::users::user_recent_mentions;
5use models::users::user_scheduled_messages::{self, ScheduledMessageStatus};
6use sea_orm::sea_query::Expr;
7use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};
5use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
86use time::OffsetDateTime;
97
108use crate::errors::ChatError;
11use crate::services::messages::SendMessageCmd;
129use crate::services::service::ChatService;
1310
1411pub(crate) const STUCK_CLAIM_MINUTES: i64 = 10;
1512const MENTION_RETENTION_DAYS: i64 = 90;
1613
1714impl ChatService {
18 pub async fn deliver_due_scheduled_messages(&self) -> Result<u64, ChatError> {
19 let now = OffsetDateTime::now_utc();
20
21 // Recover claims abandoned by a crashed instance.
22 let _ = user_scheduled_messages::Entity::update_many()
23 .col_expr(
24 user_scheduled_messages::Column::Status,
25 Expr::value(ScheduledMessageStatus::Pending),
26 )
27 .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Sending))
28 .filter(
29 user_scheduled_messages::Column::ScheduledAt
30 .lt(now - time::Duration::minutes(STUCK_CLAIM_MINUTES)),
31 )
32 .exec(self.db().write())
33 .await?;
34
35 let due = user_scheduled_messages::Entity::find()
36 .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Pending))
37 .filter(user_scheduled_messages::Column::ScheduledAt.lte(now))
38 .limit(100)
39 .all(self.db().write())
40 .await?;
41 let count = due.len() as u64;
42 for scheduled in due {
43 // Atomic claim so concurrent replicas deliver each row at most once.
44 let claimed = user_scheduled_messages::Entity::update_many()
45 .col_expr(
46 user_scheduled_messages::Column::Status,
47 Expr::value(ScheduledMessageStatus::Sending),
48 )
49 .filter(user_scheduled_messages::Column::Id.eq(scheduled.id))
50 .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Pending))
51 .exec(self.db().write())
52 .await?;
53 if claimed.rows_affected == 0 {
54 continue;
55 }
56
57 let id = scheduled.id;
58 let user_id = scheduled.user_id;
59 let channel_id = scheduled.target_channel_id;
60 let cmd: SendMessageCmd = match serde_json::from_value(scheduled.payload.clone()) {
61 Ok(cmd) => cmd,
62 Err(error) => {
63 tracing::warn!(%error, %id, "invalid scheduled message payload");
64 let mut am: user_scheduled_messages::ActiveModel = scheduled.into();
65 am.status = Set(ScheduledMessageStatus::Cancelled);
66 am.update(self.db().write()).await?;
67 continue;
68 }
69 };
70 let next_status = match self.send_message(user_id, channel_id, cmd).await {
71 Ok(_) => ScheduledMessageStatus::Sent,
72 Err(error) => {
73 tracing::warn!(%error, %id, "scheduled message delivery failed");
74 // Authz/validation failures are permanent; transient slow
75 // mode retries later.
76 if error.is_rate_limited() {
77 ScheduledMessageStatus::Pending
78 } else {
79 ScheduledMessageStatus::Failed
80 }
81 }
82 };
83 user_scheduled_messages::Entity::update_many()
84 .col_expr(
85 user_scheduled_messages::Column::Status,
86 Expr::value(next_status),
87 )
88 .filter(user_scheduled_messages::Column::Id.eq(id))
89 .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Sending))
90 .exec(self.db().write())
91 .await?;
92 }
93 Ok(count)
94 }
95
9615 pub async fn prune_old_mentions(&self) -> Result<(), ChatError> {
9716 let cutoff = OffsetDateTime::now_utc() - time::Duration::days(MENTION_RETENTION_DAYS);
9817 user_recent_mentions::Entity::delete_many()
@@ -113,14 +32,17 @@
11332 if let Err(error) = service.deliver_due_reminders().await {
11433 tracing::warn!(%error, "chat reminder delivery failed");
11534 }
116 if let Err(error) = service.deliver_due_scheduled_messages().await {
117 tracing::warn!(%error, "chat scheduled message delivery failed");
35 if let Err(error) = service.deliver_pending_webhook_events().await {
36 tracing::warn!(%error, "chat outbound webhook delivery failed");
11837 }
11938 tick = tick.wrapping_add(1);
120 if tick % 120 == 0
121 && let Err(error) = service.prune_old_mentions().await
122 {
123 tracing::warn!(%error, "chat mention prune failed");
39 if tick % 120 == 0 {
40 if let Err(error) = service.prune_old_mentions().await {
41 tracing::warn!(%error, "chat mention prune failed");
42 }
43 if let Err(error) = service.cleanup_stale_attachments().await {
44 tracing::warn!(%error, "chat attachment cleanup failed");
45 }
12446 }
12547 }
12648}
Modified lib/chat/Cargo.toml lib/chat/Cargo.toml +2 −1 View lib/chat/Cargo.toml
@@ -14,6 +14,7 @@
1414axum = { workspace = true }
1515sea-orm = { workspace = true }
1616redis = { workspace = true }
17reqwest = { workspace = true }
1718serde = { workspace = true, features = ["derive"] }
1819serde_json = { workspace = true }
1920utoipa = { workspace = true }
@@ -24,7 +25,7 @@
2425sha2 = { workspace = true }
2526hmac = { workspace = true }
2627tracing = { workspace = true }
27tokio = { workspace = true }
28tokio = { workspace = true, features = ["net"] }
2829tokio-util = { workspace = true }
2930futures = { workspace = true }
3031bytes = { workspace = true }
Modified lib/models/src/chat/attachments.rs lib/models/src/chat/attachments.rs +0 −4 View lib/models/src/chat/attachments.rs
@@ -24,10 +24,6 @@
2424 Pending,
2525 #[sea_orm(num_value = 1)]
2626 Uploaded,
27 #[sea_orm(num_value = 2)]
28 Ready,
29 #[sea_orm(num_value = 3)]
30 Failed,
3127}
3228
3329#[sea_orm::model]
Modified lib/models/src/chat/messages.rs lib/models/src/chat/messages.rs +0 −2 View lib/models/src/chat/messages.rs
@@ -60,8 +60,6 @@
6060 ChannelPinnedItem,
6161 #[sea_orm(num_value = 7)]
6262 Reminder,
63 #[sea_orm(num_value = 8)]
64 Scheduled,
6563 #[sea_orm(num_value = 9)]
6664 System,
6765}
Modified lib/models/src/chat/mod.rs lib/models/src/chat/mod.rs +2 −0 View lib/models/src/chat/mod.rs
@@ -4,6 +4,7 @@
44pub mod message_edits;
55pub mod message_reminders;
66pub mod messages;
7pub mod outbound_webhooks;
78pub mod presence_states;
89pub mod reactions;
910pub mod read_states;
@@ -14,5 +15,6 @@
1415pub mod thread_members;
1516pub mod threads;
1617pub mod user_drafts;
18pub mod webhook_deliveries;
1719
1820pub use messages::*;
Added lib/models/src/chat/outbound_webhooks.rs lib/models/src/chat/outbound_webhooks.rs +25 −0 View lib/models/src/chat/outbound_webhooks.rs
@@ -0,0 +1,25 @@
1use sea_orm::entity::prelude::*;
2use time::OffsetDateTime;
3use uuid::Uuid;
4
5#[sea_orm::model]
6#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
7#[sea_orm(table_name = "chat_outbound_webhook")]
8pub struct Model {
9 #[sea_orm(primary_key, auto_increment = false)]
10 pub id: Uuid,
11 #[sea_orm(indexed)]
12 pub room_id: Uuid,
13 pub user_id: Uuid,
14 pub name: String,
15 pub url: String,
16 pub secret: String,
17 #[sea_orm(default_value = "false")]
18 pub disabled: bool,
19 pub last_status: Option<String>,
20 pub last_used_at: Option<OffsetDateTime>,
21 pub created_at: OffsetDateTime,
22 pub updated_at: OffsetDateTime,
23}
24
25impl ActiveModelBehavior for ActiveModel {}
Added lib/models/src/chat/webhook_deliveries.rs lib/models/src/chat/webhook_deliveries.rs +54 −0 View lib/models/src/chat/webhook_deliveries.rs
@@ -0,0 +1,54 @@
1use sea_orm::entity::prelude::*;
2use serde::{Deserialize, Serialize};
3use time::OffsetDateTime;
4use uuid::Uuid;
5
6#[derive(
7 Debug,
8 Clone,
9 Copy,
10 PartialEq,
11 Eq,
12 Default,
13 EnumIter,
14 DeriveActiveEnum,
15 Serialize,
16 Deserialize,
17 utoipa::ToSchema,
18)]
19#[sea_orm(rs_type = "i16", db_type = "SmallInteger")]
20pub enum WebhookDeliveryStatus {
21 #[default]
22 #[sea_orm(num_value = 0)]
23 Pending,
24 #[sea_orm(num_value = 1)]
25 Delivering,
26 #[sea_orm(num_value = 2)]
27 Done,
28 #[sea_orm(num_value = 3)]
29 Failed,
30}
31
32#[sea_orm::model]
33#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
34#[sea_orm(table_name = "chat_webhook_delivery")]
35pub struct Model {
36 #[sea_orm(primary_key, auto_increment = false)]
37 pub id: Uuid,
38 #[sea_orm(indexed)]
39 pub webhook_id: Uuid,
40 pub room_id: Uuid,
41 pub event_type: String,
42 #[sea_orm(column_type = "JsonBinary")]
43 pub payload: sea_orm::JsonValue,
44 #[sea_orm(default_value = 0)]
45 pub status: WebhookDeliveryStatus,
46 #[sea_orm(default_value = 0)]
47 pub attempts: i32,
48 pub next_retry_at: OffsetDateTime,
49 pub last_error: Option<String>,
50 pub created_at: OffsetDateTime,
51 pub updated_at: OffsetDateTime,
52}
53
54impl ActiveModelBehavior for ActiveModel {}
Modified lib/models/src/users/mod.rs lib/models/src/users/mod.rs +0 −1 View lib/models/src/users/mod.rs
@@ -17,7 +17,6 @@
1717pub mod user_recovery_codes;
1818pub mod user_reports;
1919pub mod user_saved_messages;
20pub mod user_scheduled_messages;
2120pub mod user_sessions;
2221pub mod user_ssh_keys;
2322pub mod user_suspensions;
Deleted lib/models/src/users/user_scheduled_messages.rs lib/models/src/users/user_scheduled_messages.rs +0 −51
@@ -1,51 +0,0 @@
1use sea_orm::JsonValue;
2use sea_orm::entity::prelude::*;
3use serde::{Deserialize, Serialize};
4use time::OffsetDateTime;
5use uuid::Uuid;
6
7#[derive(
8 Debug,
9 Clone,
10 Copy,
11 PartialEq,
12 Eq,
13 Default,
14 EnumIter,
15 DeriveActiveEnum,
16 Serialize,
17 Deserialize,
18 utoipa::ToSchema,
19)]
20#[sea_orm(rs_type = "String", db_type = "String(StringLen::N(16))")]
21pub enum ScheduledMessageStatus {
22 #[default]
23 #[sea_orm(string_value = "pending")]
24 Pending,
25 #[sea_orm(string_value = "sent")]
26 Sent,
27 #[sea_orm(string_value = "cancelled")]
28 Cancelled,
29 #[sea_orm(string_value = "sending")]
30 Sending,
31 #[sea_orm(string_value = "failed")]
32 Failed,
33}
34
35#[sea_orm::model]
36#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
37#[sea_orm(table_name = "user_scheduled_message")]
38pub struct Model {
39 #[sea_orm(primary_key, auto_increment = false)]
40 pub id: Uuid,
41 pub user_id: Uuid,
42 pub target_channel_id: Uuid,
43 pub payload: JsonValue,
44 pub scheduled_at: OffsetDateTime,
45 pub status: ScheduledMessageStatus,
46 pub created_at: OffsetDateTime,
47 #[sea_orm(belongs_to, from = "user_id", to = "uid")]
48 pub user: BelongsTo<super::user::Entity>,
49}
50
51impl ActiveModelBehavior for ActiveModel {}
Modified lib/models/schema_extras.sql lib/models/schema_extras.sql +9 −2 View lib/models/schema_extras.sql
@@ -119,8 +119,6 @@
119119-- Chat domain (lib/chat): message ordering, idempotency, FTS, and composite
120120-- indexes that entity attributes cannot express.
121121-- ---------------------------------------------------------------------------
122CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_message_room_nonce
123 ON chat_message (room_id, nonce) WHERE nonce IS NOT NULL;
124122CREATE INDEX IF NOT EXISTS idx_chat_message_thread_seq
125123 ON chat_message (thread_id, seq) WHERE thread_id IS NOT NULL;
126124CREATE INDEX IF NOT EXISTS idx_chat_message_author_created
@@ -166,9 +164,18 @@
166164CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_dm_pair
167165 ON rooms (lower_name) WHERE kind = 15 AND space_id IS NULL;
168166
167CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_gdm_key
168 ON rooms (lower_name) WHERE kind = 1 AND space_id IS NULL;
169
169170CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_message_room_nonce
170171 ON chat_message (room_id, author_id, nonce) WHERE nonce IS NOT NULL;
171172
173CREATE INDEX IF NOT EXISTS idx_chat_outbound_webhook_room
174 ON chat_outbound_webhook (room_id);
175
176CREATE INDEX IF NOT EXISTS idx_chat_webhook_delivery_due
177 ON chat_webhook_delivery (status, next_retry_at);
178
172179ALTER TABLE chat_message ADD COLUMN IF NOT EXISTS search_vector tsvector;
173180CREATE INDEX IF NOT EXISTS idx_chat_message_fts
174181 ON chat_message USING gin (search_vector);
Modified lib/models/schema_pre_sync.sql lib/models/schema_pre_sync.sql +4 −0 View lib/models/schema_pre_sync.sql
@@ -14,8 +14,12 @@
1414DROP INDEX IF EXISTS idx_branch_protection_whitelist_group;
1515DROP INDEX IF EXISTS idx_chat_message_room_nonce;
1616DROP INDEX IF EXISTS idx_chat_dm_pair;
17DROP INDEX IF EXISTS idx_chat_gdm_key;
1718DROP INDEX IF EXISTS idx_chat_draft_unique;
1819
20-- scheduled messages feature removed; drop the legacy table before sync
21DROP TABLE IF EXISTS user_scheduled_message;
22
1923-- idempotent column adds for phase-2 model changes (sync layer cannot
2024-- express NOT NULL DEFAULT additions on populated tables).
2125-- Guarded with to_regclass: on a fresh database the tables do not exist yet
Modified lib/space/src/services/rooms.rs lib/space/src/services/rooms.rs +9 −0 View lib/space/src/services/rooms.rs
@@ -109,6 +109,15 @@
109109 if !space.is_active {
110110 return Err(SpaceError::NotFound);
111111 }
112 if !matches!(
113 cmd.kind,
114 RoomKind::GuildRoom | RoomKind::Text | RoomKind::Category | RoomKind::News
115 ) {
116 return Err(SpaceError::BadRequest(
117 "unsupported room kind: only guild_room, text, category and news can be created"
118 .to_string(),
119 ));
120 }
112121
113122 let name = cmd.name.trim();
114123 if name.is_empty() || name.chars().count() > 100 {
Modified Cargo.lock Cargo.lock +137 −2 View Cargo.lock
@@ -1106,6 +1106,7 @@
11061106 "models",
11071107 "rand 0.8.7",
11081108 "redis",
1109 "reqwest 0.12.28",
11091110 "sea-orm",
11101111 "serde",
11111112 "serde_json",
@@ -2547,8 +2548,10 @@
25472548checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0"
25482549dependencies = [
25492550 "cfg-if",
2551 "js-sys",
25502552 "libc",
25512553 "wasi 0.11.1+wasi-snapshot-preview1",
2554 "wasm-bindgen",
25522555]
25532556
25542557[[package]]
@@ -3964,6 +3967,22 @@
39643967]
39653968
39663969[[package]]
3970name = "hyper-rustls"
3971version = "0.27.9"
3972source = "registry+https://github.com/rust-lang/crates.io-index"
3973checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f"
3974dependencies = [
3975 "http 1.5.0",
3976 "hyper 1.11.0",
3977 "hyper-util",
3978 "rustls",
3979 "tokio",
3980 "tokio-rustls",
3981 "tower-service",
3982 "webpki-roots",
3983]
3984
3985[[package]]
39673986name = "hyper-tls"
39683987version = "0.5.0"
39693988source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -4880,6 +4899,12 @@
48804899checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"
48814900
48824901[[package]]
4902name = "lru-slab"
4903version = "0.1.2"
4904source = "registry+https://github.com/rust-lang/crates.io-index"
4905checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
4906
4907[[package]]
48834908name = "lz4"
48844909version = "1.28.1"
48854910source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -5462,7 +5487,7 @@
54625487 "bytes",
54635488 "http 1.5.0",
54645489 "opentelemetry",
5465 "reqwest",
5490 "reqwest 0.13.4",
54665491]
54675492
54685493[[package]]
@@ -5477,7 +5502,7 @@
54775502 "opentelemetry-proto",
54785503 "opentelemetry_sdk",
54795504 "prost",
5480 "reqwest",
5505 "reqwest 0.13.4",
54815506 "thiserror 2.0.20",
54825507]
54835508
@@ -6238,6 +6263,62 @@
62386263]
62396264
62406265[[package]]
6266name = "quinn"
6267version = "0.11.11"
6268source = "registry+https://github.com/rust-lang/crates.io-index"
6269checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8"
6270dependencies = [
6271 "bytes",
6272 "cfg_aliases",
6273 "pin-project-lite",
6274 "quinn-proto",
6275 "quinn-udp",
6276 "rustc-hash",
6277 "rustls",
6278 "socket2 0.6.5",
6279 "thiserror 2.0.20",
6280 "tokio",
6281 "tracing",
6282 "web-time",
6283]
6284
6285[[package]]
6286name = "quinn-proto"
6287version = "0.11.17"
6288source = "registry+https://github.com/rust-lang/crates.io-index"
6289checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83"
6290dependencies = [
6291 "bytes",
6292 "getrandom 0.4.3",
6293 "lru-slab",
6294 "rand 0.10.2",
6295 "rand_pcg",
6296 "ring",
6297 "rustc-hash",
6298 "rustls",
6299 "rustls-pki-types",
6300 "slab",
6301 "thiserror 2.0.20",
6302 "tinyvec",
6303 "tracing",
6304 "web-time",
6305]
6306
6307[[package]]
6308name = "quinn-udp"
6309version = "0.5.15"
6310source = "registry+https://github.com/rust-lang/crates.io-index"
6311checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694"
6312dependencies = [
6313 "cfg_aliases",
6314 "libc",
6315 "once_cell",
6316 "socket2 0.6.5",
6317 "tracing",
6318 "windows-sys 0.61.2",
6319]
6320
6321[[package]]
62416322name = "quote"
62426323version = "1.0.47"
62436324source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -6388,6 +6469,15 @@
63886469]
63896470
63906471[[package]]
6472name = "rand_pcg"
6473version = "0.10.2"
6474source = "registry+https://github.com/rust-lang/crates.io-index"
6475checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a"
6476dependencies = [
6477 "rand_core 0.10.1",
6478]
6479
6480[[package]]
63916481name = "redis"
63926482version = "1.6.0"
63936483source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -6478,6 +6568,44 @@
64786568
64796569[[package]]
64806570name = "reqwest"
6571version = "0.12.28"
6572source = "registry+https://github.com/rust-lang/crates.io-index"
6573checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147"
6574dependencies = [
6575 "base64 0.22.1",
6576 "bytes",
6577 "futures-core",
6578 "http 1.5.0",
6579 "http-body 1.1.0",
6580 "http-body-util",
6581 "hyper 1.11.0",
6582 "hyper-rustls",
6583 "hyper-util",
6584 "js-sys",
6585 "log 0.4.33",
6586 "percent-encoding 2.3.2",
6587 "pin-project-lite",
6588 "quinn",
6589 "rustls",
6590 "rustls-pki-types",
6591 "serde",
6592 "serde_json",
6593 "serde_urlencoded",
6594 "sync_wrapper",
6595 "tokio",
6596 "tokio-rustls",
6597 "tower",
6598 "tower-http 0.6.11",
6599 "tower-service",
6600 "url 2.5.8",
6601 "wasm-bindgen",
6602 "wasm-bindgen-futures",
6603 "web-sys",
6604 "webpki-roots",
6605]
6606
6607[[package]]
6608name = "reqwest"
64816609version = "0.13.4"
64826610source = "registry+https://github.com/rust-lang/crates.io-index"
64836611checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3"
@@ -6805,6 +6933,12 @@
68056933]
68066934
68076935[[package]]
6936name = "rustc-hash"
6937version = "2.1.3"
6938source = "registry+https://github.com/rust-lang/crates.io-index"
6939checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d"
6940
6941[[package]]
68086942name = "rustc_version"
68096943version = "0.4.1"
68106944source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -6859,6 +6993,7 @@
68596993source = "registry+https://github.com/rust-lang/crates.io-index"
68606994checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96"
68616995dependencies = [
6996 "web-time",
68626997 "zeroize",
68636998]
68646999
Modified Cargo.toml Cargo.toml +1 −0 View Cargo.toml
@@ -105,6 +105,7 @@
105105pulldown-cmark = { version = "0.13", default-features = false, features = ["html"] }
106106
107107lettre = { version = "0.11", default-features = false, features = ["smtp-transport", "tokio1-rustls", "ring", "rustls-native-certs", "builder"] }
108reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
108109anyhow = "1"
109110tracing-subscriber = { version = "0.3", features = ["env-filter"] }
110111
Modified openapi.json openapi.json View openapi.json
Diff too large to display.