深圳幻海软件技术有限公司 欢迎您！
更多

构建一个即时消息应用（三）：对话

2023-02-27

http json err
本文是该系列的第三篇。第一篇：模式第二篇：OAuth在我们的即时消息应用中，消息表现为两个参与者对话的堆叠。如果你想要开始一场对话，就应该向应用提供你想要交谈的用户，而当对话创建后（如果该对话此前并不存在），就可以向该对话发送消息。就前端而言，我们可能想要显示一份近期对话列表。并在此处显示对话的最后

本文是该系列的第三篇。

  • 第一篇：模式
  • 第二篇：OAuth

在我们的即时消息应用中，消息表现为两个参与者对话的堆叠。如果你想要开始一场对话，就应该向应用提供你想要交谈的用户，而当对话创建后（如果该对话此前并不存在），就可以向该对话发送消息。

就前端而言，我们可能想要显示一份近期对话列表。并在此处显示对话的最后一条消息以及另一个参与者的姓名和头像。

在这篇帖子中，我们将会编写一些端点endpoint来完成像“创建对话”、“获取对话列表”以及“找到单个对话”这样的任务。

首先，要在主函数 main() 中添加下面的路由。

  1. router.HandleFunc("POST", "/api/conversations", requireJSON(guard(createConversation)))
  2. router.HandleFunc("GET", "/api/conversations", guard(getConversations))
  3. router.HandleFunc("GET", "/api/conversations/:conversationID", guard(getConversation))
  • 1.
  • 2.
  • 3.

这三个端点都需要进行身份验证，所以我们将会使用 guard() 中间件。我们也会构建一个新的中间件，用于检查请求内容是否为 JSON 格式。

JSON 请求检查中间件

  1. func requireJSON(handler http.HandlerFunc) http.HandlerFunc {
  2.     return func(w http.ResponseWriter, r *http.Request) {
  3.         if ct := r.Header.Get("Content-Type"); !strings.HasPrefix(ct, "application/json") {
  4.             http.Error(w, "Content type of application/json required", http.StatusUnsupportedMediaType)
  5.             return
  6.         }
  7.         handler(w, r)
  8.     }
  9. }
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.

如果请求request不是 JSON 格式，那么它会返回 415 Unsupported Media Type（不支持的媒体类型）错误。

创建对话

  1. type Conversation struct {
  2.     ID                string   `json:"id"`
  3.     OtherParticipant  *User    `json:"otherParticipant"`
  4.     LastMessage       *Message `json:"lastMessage"`
  5.     HasUnreadMessages bool     `json:"hasUnreadMessages"`
  6. }
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.

就像上面的代码那样，对话中保持对另一个参与者和最后一条消息的引用，还有一个 bool 类型的字段，用来告知是否有未读消息。

  1. type Message struct {
  2.     ID             string    `json:"id"`
  3.     Content        string    `json:"content"`
  4.     UserID         string    `json:"-"`
  5.     ConversationID string    `json:"conversationID,omitempty"`
  6.     CreatedAt      time.Time `json:"createdAt"`
  7.     Mine           bool      `json:"mine"`
  8.     ReceiverID     string    `json:"-"`
  9. }
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.

我们会在下一篇文章介绍与消息相关的内容，但由于我们这里也需要用到它，所以先定义了 Message 结构体。其中大多数字段与数据库表一致。我们需要使用 Mine 来断定消息是否属于当前已验证用户所有。一旦加入实时功能，ReceiverID 可以帮助我们过滤消息。

接下来让我们编写 HTTP 处理程序。尽管它有些长，但也没什么好怕的。

  1. func createConversation(w http.ResponseWriter, r *http.Request) {
  2.     var input struct {
  3.         Username string `json:"username"`
  4.     }
  5.     defer r.Body.Close()
  6.     if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
  7.         http.Error(w, err.Error(), http.StatusBadRequest)
  8.         return
  9.     }
  10.  
  11.     input.Username = strings.TrimSpace(input.Username)
  12.     if input.Username == "" {
  13.         respond(w, Errors{map[string]string{
  14.             "username": "Username required",
  15.         }}, http.StatusUnprocessableEntity)
  16.         return
  17.     }
  18.  
  19.     ctx := r.Context()
  20.     authUserID := ctx.Value(keyAuthUserID).(string)
  21.  
  22.     tx, err := db.BeginTx(ctx, nil)
  23.     if err != nil {
  24.         respondError(w, fmt.Errorf("could not begin tx: %v", err))
  25.         return
  26.     }
  27.     defer tx.Rollback()
  28.  
  29.     var otherParticipant User
  30.     if err := tx.QueryRowContext(ctx, `
  31.         SELECT id, avatar_url FROM users WHERE username = $1
  32.     `, input.Username).Scan(
  33.         &otherParticipant.ID,
  34.         &otherParticipant.AvatarURL,
  35.     ); err == sql.ErrNoRows {
  36.         http.Error(w, "User not found", http.StatusNotFound)
  37.         return
  38.     } else if err != nil {
  39.         respondError(w, fmt.Errorf("could not query other participant: %v", err))
  40.         return
  41.     }
  42.  
  43.     otherParticipant.Username = input.Username
  44.  
  45.     if otherParticipant.ID == authUserID {
  46.         http.Error(w, "Try start a conversation with someone else", http.StatusForbidden)
  47.         return
  48.     }
  49.  
  50.     var conversationID string
  51.     if err := tx.QueryRowContext(ctx, `
  52.         SELECT conversation_id FROM participants WHERE user_id = $1
  53.         INTERSECT
  54.         SELECT conversation_id FROM participants WHERE user_id = $2
  55.     `, authUserID, otherParticipant.ID).Scan(&conversationID); err != nil && err != sql.ErrNoRows {
  56.         respondError(w, fmt.Errorf("could not query common conversation id: %v", err))
  57.         return
  58.     } else if err == nil {
  59.         http.Redirect(w, r, "/api/conversations/"+conversationID, http.StatusFound)
  60.         return
  61.     }
  62.  
  63.     var conversation Conversation
  64.     if err = tx.QueryRowContext(ctx, `
  65.         INSERT INTO conversations DEFAULT VALUES
  66.         RETURNING id
  67.     `).Scan(&conversation.ID); err != nil {
  68.         respondError(w, fmt.Errorf("could not insert conversation: %v", err))
  69.         return
  70.     }
  71.  
  72.     if _, err = tx.ExecContext(ctx, `
  73.         INSERT INTO participants (user_id, conversation_id) VALUES
  74.             ($1, $2),
  75.             ($3, $2)
  76.     `, authUserID, conversation.ID, otherParticipant.ID); err != nil {
  77.         respondError(w, fmt.Errorf("could not insert participants: %v", err))
  78.         return
  79.     }
  80.  
  81.     if err = tx.Commit(); err != nil {
  82.         respondError(w, fmt.Errorf("could not commit tx to create conversation: %v", err))
  83.         return
  84.     }
  85.  
  86.     conversation.OtherParticipant = &otherParticipant
  87.  
  88.     respond(w, conversation, http.StatusCreated)
  89. }
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.
  • 35.
  • 36.
  • 37.
  • 38.
  • 39.
  • 40.
  • 41.
  • 42.
  • 43.
  • 44.
  • 45.
  • 46.
  • 47.
  • 48.
  • 49.
  • 50.
  • 51.
  • 52.
  • 53.
  • 54.
  • 55.
  • 56.
  • 57.
  • 58.
  • 59.
  • 60.
  • 61.
  • 62.
  • 63.
  • 64.
  • 65.
  • 66.
  • 67.
  • 68.
  • 69.
  • 70.
  • 71.
  • 72.
  • 73.
  • 74.
  • 75.
  • 76.
  • 77.
  • 78.
  • 79.
  • 80.
  • 81.
  • 82.
  • 83.
  • 84.
  • 85.
  • 86.
  • 87.
  • 88.
  • 89.

在此端点，你会向 /api/conversations 发送 POST 请求，请求的 JSON 主体中包含要对话的用户的用户名。

因此，首先需要将请求主体解析成包含用户名的结构。然后，校验用户名不能为空。

  1. type Errors struct {
  2.     Errors map[string]string `json:"errors"`
  3. }
  • 1.
  • 2.
  • 3.

这是错误消息的结构体 Errors，它仅仅是一个映射。如果输入空用户名，你就会得到一段带有 422 Unprocessable Entity（无法处理的实体）错误消息的 JSON 。

  1. {
  2.     "errors": {
  3.         "username": "Username required"
  4.     }
  5. }
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.

然后，我们开始执行 SQL 事务。收到的仅仅是用户名，但事实上，我们需要知道实际的用户 ID 。因此，事务的第一项内容是查询另一个参与者的 ID 和头像。如果找不到该用户，我们将会返回 404 Not Found（未找到） 错误。另外，如果找到的用户恰好和“当前已验证用户”相同，我们应该返回 403 Forbidden（拒绝处理）错误。这是由于对话只应当在两个不同的用户之间发起，而不能是同一个。

然后，我们试图找到这两个用户所共有的对话，所以需要使用 INTERSECT 语句。如果存在，只需要通过 /api/conversations/{conversationID} 重定向到该对话并将其返回。

如果未找到共有的对话，我们需要创建一个新的对话并添加指定的两个参与者。最后，我们 COMMIT 该事务并使用新创建的对话进行响应。

获取对话列表

端点 /api/conversations 将获取当前已验证用户的所有对话。

  1. func getConversations(w http.ResponseWriter, r *http.Request) {
  2.     ctx := r.Context()
  3.     authUserID := ctx.Value(keyAuthUserID).(string)
  4.  
  5.     rows, err := db.QueryContext(ctx, `
  6.         SELECT
  7.             conversations.id,
  8.             auth_user.messages_read_at < messages.created_at AS has_unread_messages,
  9.             messages.id,
  10.             messages.content,
  11.             messages.created_at,
  12.             messages.user_id = $1 AS mine,
  13.             other_users.id,
  14.             other_users.username,
  15.             other_users.avatar_url
  16.         FROM conversations
  17.         INNER JOIN messages ON conversations.last_message_id = messages.id
  18.         INNER JOIN participants other_participants
  19.             ON other_participants.conversation_id = conversations.id
  20.                 AND other_participants.user_id != $1
  21.         INNER JOIN users other_users ON other_participants.user_id = other_users.id
  22.         INNER JOIN participants auth_user
  23.             ON auth_user.conversation_id = conversations.id
  24.                 AND auth_user.user_id = $1
  25.         ORDER BY messages.created_at DESC
  26.     `, authUserID)
  27.     if err != nil {
  28.         respondError(w, fmt.Errorf("could not query conversations: %v", err))
  29.         return
  30.     }
  31.     defer rows.Close()
  32.  
  33.     conversations := make([]Conversation, 0)
  34.     for rows.Next() {
  35.         var conversation Conversation
  36.         var lastMessage Message
  37.         var otherParticipant User
  38.         if err = rows.Scan(
  39.             &conversation.ID,
  40.             &conversation.HasUnreadMessages,
  41.             &lastMessage.ID,
  42.             &lastMessage.Content,
  43.             &lastMessage.CreatedAt,
  44.             &lastMessage.Mine,
  45.             &otherParticipant.ID,
  46.             &otherParticipant.Username,
  47.             &otherParticipant.AvatarURL,
  48.         ); err != nil {
  49.             respondError(w, fmt.Errorf("could not scan conversation: %v", err))
  50.             return
  51.         }
  52.  
  53.         conversation.LastMessage = &lastMessage
  54.         conversation.OtherParticipant = &otherParticipant
  55.         conversations = append(conversations, conversation)
  56.     }
  57.  
  58.     if err = rows.Err(); err != nil {
  59.         respondError(w, fmt.Errorf("could not iterate over conversations: %v", err))
  60.         return
  61.     }
  62.  
  63.     respond(w, conversations, http.StatusOK)
  64. }
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.
  • 35.
  • 36.
  • 37.
  • 38.
  • 39.
  • 40.
  • 41.
  • 42.
  • 43.
  • 44.
  • 45.
  • 46.
  • 47.
  • 48.
  • 49.
  • 50.
  • 51.
  • 52.
  • 53.
  • 54.
  • 55.
  • 56.
  • 57.
  • 58.
  • 59.
  • 60.
  • 61.
  • 62.
  • 63.
  • 64.

该处理程序仅对数据库进行查询。它通过一些联接来查询对话表……首先，从消息表中获取最后一条消息。然后依据“ID 与当前已验证用户不同”的条件，从参与者表找到对话的另一个参与者。然后联接到用户表以获取该用户的用户名和头像。最后，再次联接参与者表，并以相反的条件从该表中找出参与对话的另一个用户，其实就是当前已验证用户。我们会对比消息中的 messages_read_at 和 created_at 两个字段，以确定对话中是否存在未读消息。然后，我们通过 user_id 字段来判定该消息是否属于“我”（指当前已验证用户）。

注意，此查询过程假定对话中只有两个用户参与，它也仅仅适用于这种情况。另外,该设计也不很适用于需要显示未读消息数量的情况。如果需要显示未读消息的数量，我认为可以在 participants 表上添加一个unread_messages_count INT 字段，并在每次创建新消息的时候递增它，如果用户已读则重置该字段。

接下来需要遍历每一条记录，通过扫描每一个存在的对话来建立一个对话切片slice of conversations并在最后进行响应。

找到单个对话

端点 /api/conversations/{conversationID} 会根据 ID 对单个对话进行响应。

  1. func getConversation(w http.ResponseWriter, r *http.Request) {
  2.     ctx := r.Context()
  3.     authUserID := ctx.Value(keyAuthUserID).(string)
  4.     conversationID := way.Param(ctx, "conversationID")
  5.  
  6.     var conversation Conversation
  7.     var otherParticipant User
  8.     if err := db.QueryRowContext(ctx, `
  9.         SELECT
  10.             IFNULL(auth_user.messages_read_at < messages.created_at, false) AS has_unread_messages,
  11.             other_users.id,
  12.             other_users.username,
  13.             other_users.avatar_url
  14.         FROM conversations
  15.         LEFT JOIN messages ON conversations.last_message_id = messages.id
  16.         INNER JOIN participants other_participants
  17.             ON other_participants.conversation_id = conversations.id
  18.                 AND other_participants.user_id != $1
  19.         INNER JOIN users other_users ON other_participants.user_id = other_users.id
  20.         INNER JOIN participants auth_user
  21.             ON auth_user.conversation_id = conversations.id
  22.                 AND auth_user.user_id = $1
  23.         WHERE conversations.id = $2
  24.     `, authUserID, conversationID).Scan(
  25.         &conversation.HasUnreadMessages,
  26.         &otherParticipant.ID,
  27.         &otherParticipant.Username,
  28.         &otherParticipant.AvatarURL,
  29.     ); err == sql.ErrNoRows {
  30.         http.Error(w, "Conversation not found", http.StatusNotFound)
  31.         return
  32.     } else if err != nil {
  33.         respondError(w, fmt.Errorf("could not query conversation: %v", err))
  34.         return
  35.     }
  36.  
  37.     conversation.ID = conversationID
  38.     conversation.OtherParticipant = &otherParticipant
  39.  
  40.     respond(w, conversation, http.StatusOK)
  41. }
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.
  • 35.
  • 36.
  • 37.
  • 38.
  • 39.
  • 40.
  • 41.

这里的查询与之前有点类似。尽管我们并不关心最后一条消息的显示问题，并因此忽略了与之相关的一些字段，但是我们需要根据这条消息来判断对话中是否存在未读消息。此时，我们使用 LEFT JOIN 来代替 INNER JOIN，因为 last_message_id 字段是 NULLABLE（可以为空）的；而其他情况下，我们无法得到任何记录。基于同样的理由，我们在 has_unread_messages 的比较中使用了 IFNULL 语句。最后，我们按 ID 进行过滤。

如果查询没有返回任何记录，我们的响应会返回 404 Not Found 错误，否则响应将会返回 200 OK 以及找到的对话。

本篇帖子以创建了一些对话端点结束。

在下一篇帖子中，我们将会看到如何创建并列出消息。

  • 源代码

 

 

如何两天时间上线一款AI应用？
用自己的编程语言实现了一个网站（增强版）