diff --git a/docs/IM-前后端对接文档.md b/docs/IM-前后端对接文档.md index a1be708..c2fdaec 100644 --- a/docs/IM-前后端对接文档.md +++ b/docs/IM-前后端对接文档.md @@ -636,8 +636,14 @@ X-MeetingIM-APIKey: <预置的普通 API Key> | WebSocket `login basic` | `{"login":{"id":"login-1","scheme":"basic","secret":"base64(username:password)"}}` | `{"ctrl":{"id":"login-1","code":200,"params":{"user":"usr...","token":"token...","expires":"2026-09-03T09:00:00Z","authlvl":"auth"}}}` | | WebSocket `acc new` | `{"acc":{"id":"acc-1","user":"new","scheme":"basic","username":"alice","phone":"15500000000","registration_type":1,"secret":"base64(alice:password)","login":true,"cred":[{"meth":"tel","val":"15500000000","resp":"verification_token","params":{"challenge_id":"chg_123"}}]}}` | `{"ctrl":{"id":"acc-1","code":200,"params":{"user":"usr...","username":"alice","token":"token...","expires":"2026-09-03T09:00:00Z"}}}` | | `POST /v0/channels/lp` | Long Polling 首包 `{"hi":{"id":"hi-1","ver":"0.25","tenant":"default"}}` | `{"ctrl":{"id":"hi-1","code":201,"params":{"sid":"lp_sid_123"}}}` | -| `POST /v0/file/u/` | Multipart 文件;Header `X-MeetingIM-Auth: token `;业务表单可带 `purpose=chat_file&topic=grp...` | `{"ctrl":{"code":200,"params":{"url":"/v0/file/s/file123","name":"a.pdf","size":1024,"mime":"application/pdf"}}}` | -| `GET /v0/file/s/{file_id}` | Header `X-MeetingIM-Auth: token `,可带 `Range: bytes=0-` | 对象存储模式下鉴权通过后返回 302 到短期签名 URL;客户端跟随 Location 获取文件 | +| `POST /v0/file/u/` | Multipart 文件;Header `X-MeetingIM-Auth: token `;聊天附件必须带 `topic=<消息所在会话 topic>`(`p2p.../grp...`),可选 `purpose=chat_file`、`id=<幂等ID>` | `{"ctrl":{"code":200,"params":{"expires":"2026-10-13T06:21:20Z","file_id":"s6yhMyiOfaU","url":"/v0/file/s/s6yhMyiOfaU.txt"}}}` | +| `GET /v0/file/s/{file_id}` | Header `X-MeetingIM-Auth: token `,可带 `Range: bytes=0-`;URL 可带文件扩展名(服务端自动剥离) | 对象存储模式下鉴权通过后返回 302 到短期签名 URL;客户端跟随 Location 获取文件 | + +**文件上传/下载鉴权与回收规则(已实现,客户端必须遵守):** + +1. 上传聊天附件(图片/视频/语音/文件)必须携带 `topic=<消息将要发送到的会话 topic>`。服务端校验上传者是该 `p2p`/`grp` 会话的订阅成员(`modewant`/`modegiven` 含读权限),不是成员返回 `403 permission denied`;校验通过则在服务端建立「文件-会话」引用,该会话全体成员凭有效 Token 即可下载。 +2. 不带 `topic` 的上传只有上传者本人可下载(可用于头像等场景)。任何文件若一直未被任何会话/消息引用,服务端在创建 1 小时后由后台任务物理回收;上传后必须及时发送消息或携带 `topic`,否则对端将拿到 404。 +3. 下载放行范围:上传者本人、文件所属会话成员、消息级引用可见者(未撤回未删除且未被执行删除)、群成员、好友关系、收藏引用。非成员返回 `403 permission denied`(即 `FILE_FORBIDDEN`),客户端不自动重试。 ### 0.6 Topic 与产品消息 @@ -1153,11 +1159,12 @@ X-MeetingIM-APIKey: <预置的普通 API Key> 双方沟通状态 -**状态:上传基础已实现;媒体消息 Schema、ACL 和恢复需改造。** +**状态:上传带 `topic` 与下载按会话授权已在服务端实现;客户端按本节接入。** -- 选择媒体后本地生成 `client_mid` 和稳定占位;调用文件上传,得到 `file_id` 后再 `pub x-im-type=image|video|voice`。 +- 选择媒体后本地生成 `client_mid` 和稳定占位;调用文件上传(**必须带 `topic=<该会话 topic>`**,服务端校验成员资格并自动建立文件-会话引用),得到 `file_id` 后再 `pub x-im-type=image|video|voice`。 - `content` 至少含 `file_id,name,size,mime,width,height,duration_ms,thumbnail_file_id` 中适用字段;上传进度是客户端任务状态,不写成聊天消息。 -- 服务端发布成功后关联文件与 Topic/消息;文件下载必须校验消息可见性。 +- 服务端在上传阶段即建立文件与会话的引用(带 `topic`);消息发布成功后按消息附件引用建立消息级 ACL;文件下载校验会话成员身份与消息可见性。 +- 媒体文件(缩略图、原图、视频、语音)一律走同一次上传流程并带同一个 `topic`;缩略图也必须挂到会话,否则对端只能看到占位图。 - 上传失败(网络中断、超时、5xx)按**整文件重传**处理:重新调用 `POST /v0/file/u/` 上传完整文件,不做分片/断点续传;每次重传得到新 `file_id`,旧 `file_id` 直接放弃、由服务端 TTL 回收,客户端不得再引用。 - 上传成功但 `pub` 失败时不重新上传,复用已上传 `file_id` 和原 `client_mid` 重发消息,服务端按 `client_mid` 幂等去重。 @@ -2331,10 +2338,11 @@ GET /v1/im/topics/{topic}/message-mutations?since_version={localMutationVersion} 文件-下载 -**状态:底层下载已实现租户校验;资源 ACL 需改造。** +**状态:下载授权已实现(按上传者/会话成员/消息引用/群成员/好友/收藏放行)。** -- 客户端使用 `GET /v0/file/s/{file_id}`,Header 传 API Key 与 `X-MeetingIM-Auth: token `;禁止 `?auth=&secret=`。 -- 服务端除 Token 租户外,还必须校验该用户通过消息、群文件或收藏拥有有效引用,并支持 `Range`、`ETag`、内容长度和安全文件名。 +- 客户端使用 `GET /v0/file/s/{file_id}`,Header 传 API Key 与 `X-MeetingIM-Auth: token `;禁止 `?auth=&secret=`。URL 可带扩展名,服务端自动剥离。 +- 服务端按文件 ACL 放行:上传者本人、文件所属 `p2p`/`grp` 会话成员、消息级引用可见者(未撤回未删除且未被执行删除)、群成员、好友关系、收藏引用;非成员返回 `403 permission denied`(`FILE_FORBIDDEN`)。对象存储模式下鉴权通过返回 302 短期签名 URL。 +- 前提:发送方上传附件时带了 `topic`(见 0.5.5.1)。若对端下载一直 404,说明该文件从未关联会话/消息且已被 1 小时回收任务删除,应让发送方重新上传并携带 `topic`。 - 下载前客户端校验本地空间和网络策略;`FILE_FORBIDDEN` 不自动重试,`FILE_UNAVAILABLE` 显示失效。 ### 7.10 文件-下载中.png diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index 7975c70..f371009 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -159,6 +159,7 @@ type kafkaMessagePersistItem struct { envelope kafkaEnvelope tenantID types.TenantID payload kafkaMessagePersistPayload + consume *types.KafkaConsumeRecord } type kafkaMessagePersistPreparer func(context.Context, kafka.Message, string) (*kafkaMessagePersistItem, bool, error) @@ -259,7 +260,7 @@ func prepareKafkaMessagePersistItem(ctx context.Context, message kafka.Message, } return nil, true, nil } - item := &kafkaMessagePersistItem{envelope: envelope, tenantID: tenantID} + item := &kafkaMessagePersistItem{envelope: envelope, tenantID: tenantID, consume: consume} if err = json.Unmarshal(envelope.Payload, &item.payload); err != nil { return nil, true, markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error()) } @@ -308,14 +309,23 @@ func persistKafkaMessageIndividually(ctx context.Context, item *kafkaMessagePers } err := runKafkaMessageSingleRetries(ctx, kafkaBus.config.MessageRetries, operation, markFailed, waitKafkaRetry) if err != nil { - // Message persistence failures are recoverable infrastructure failures. Returning - // the error keeps the Kafka offset uncommitted so the outer consumer retries the - // message until it is durable. Only malformed payloads are dead-lettered while - // preparing the item. immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "retry_exhausted").Inc() immetrics.MessagePersistDuration.WithLabelValues(kind, "retry_exhausted").Observe(time.Since(started).Seconds()) logs.Err.Printf("kafka message persist retry exhausted: tenant=%d event=%s topic=%s seq=%d kind=%s err=%v", item.tenantID, item.envelope.EventID, item.payload.Message.Topic, item.payload.Message.SeqId, kind, err) + // Dead-letter the message instead of leaving the offset uncommitted: an + // unresolvable failure (e.g. an attachment file already reclaimed by GC) + // would otherwise be retried forever and stall every message behind it + // on the partition. The dead letter keeps the envelope for replay once + // the underlying issue is fixed. + if deadErr := markKafkaDeadLetter(ctx, item.tenantID, item.envelope, item.consume, "PERSIST_RETRY_EXHAUSTED", err.Error()); deadErr != nil { + logs.Err.Printf("kafka message persist dead letter failed, keeping offset uncommitted: tenant=%d event=%s topic=%s seq=%d err=%v", + item.tenantID, item.envelope.EventID, item.payload.Message.Topic, item.payload.Message.SeqId, deadErr) + return err + } + logs.Warn.Printf("kafka message persist dead-lettered after retries: tenant=%d event=%s topic=%s seq=%d kind=%s err=%v", + item.tenantID, item.envelope.EventID, item.payload.Message.Topic, item.payload.Message.SeqId, kind, err) + return nil } return err }