编程语言构建一个即时消息应用

本文是该系列的第三篇。

• 第一篇：模式
• 第二篇：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))

这三个端点都需要进行身份验证，所以我们将会使用 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. }

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

创建对话

1. type Conversationstruct{
2. ID string `json:"id"`
3. OtherParticipant*User`json:"otherParticipant"`
4. LastMessage*Message`json:"lastMessage"`
5. HasUnreadMessagesbool`json:"hasUnreadMessages"`
6. }

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

1. type Messagestruct{
2. ID string `json:"id"`
3. Content string `json:"content"`
4. UserID string `json:"-"`
5. ConversationID string `json:"conversationID,omitempty"`
6. CreatedAttime.Time`json:"createdAt"`
7. Minebool`json:"mine"`
8. ReceiverID string `json:"-"`
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. }elseif 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. }elseif 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. }

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

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

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

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

1. {
2. "errors":{
3. "username":"Username required"
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. }

该处理程序仅对数据库进行查询。它通过一些联接来查询对话表……首先，从消息表中获取最后一条消息。然后依据“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. }elseif 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. }

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

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

编程语言往往使程序员能够比使用机器语言更准确地表达他们所想表达的目的。对那些从事计算机科学的人来说，懂得程序设计语言是十分重要的，因为在当今所有的计算都需要程序设计语言才能完成。