構(gòu)建一個即時消息應(yīng)用(三):對話
本文是該系列的第三篇。
在我們的即時消息應(yīng)用中,消息表現(xiàn)為兩個參與者對話的堆疊。如果你想要開始一場對話,就應(yīng)該向應(yīng)用提供你想要交談的用戶,而當對話創(chuàng)建后(如果該對話此前并不存在),就可以向該對話發(fā)送消息。
就前端而言,我們可能想要顯示一份近期對話列表。并在此處顯示對話的最后一條消息以及另一個參與者的姓名和頭像。
在這篇帖子中,我們將會編寫一些端點來完成像“創(chuàng)建對話”、“獲取對話列表”以及“找到單個對話”這樣的任務(wù)。
首先,要在主函數(shù) main() 中添加下面的路由。
router.HandleFunc("POST", "/api/conversations", requireJSON(guard(createConversation)))router.HandleFunc("GET", "/api/conversations", guard(getConversations))router.HandleFunc("GET", "/api/conversations/:conversationID", guard(getConversation))
這三個端點都需要進行身份驗證,所以我們將會使用 guard() 中間件。我們也會構(gòu)建一個新的中間件,用于檢查請求內(nèi)容是否為 JSON 格式。
JSON 請求檢查中間件
func requireJSON(handler http.HandlerFunc) http.HandlerFunc {return func(w http.ResponseWriter, r *http.Request) {if ct := r.Header.Get("Content-Type"); !strings.HasPrefix(ct, "application/json") {http.Error(w, "Content type of application/json required", http.StatusUnsupportedMediaType)return}handler(w, r)}}
如果請求不是 JSON 格式,那么它會返回 415 Unsupported Media Type(不支持的媒體類型)錯誤。
創(chuàng)建對話
type Conversation struct {ID string `json:"id"`OtherParticipant *User `json:"otherParticipant"`LastMessage *Message `json:"lastMessage"`HasUnreadMessages bool `json:"hasUnreadMessages"`}
就像上面的代碼那樣,對話中保持對另一個參與者和最后一條消息的引用,還有一個 bool 類型的字段,用來告知是否有未讀消息。
type Message struct {ID string `json:"id"`Content string `json:"content"`UserID string `json:"-"`ConversationID string `json:"conversationID,omitempty"`CreatedAt time.Time `json:"createdAt"`Mine bool `json:"mine"`ReceiverID string `json:"-"`}
我們會在下一篇文章介紹與消息相關(guān)的內(nèi)容,但由于我們這里也需要用到它,所以先定義了 Message 結(jié)構(gòu)體。其中大多數(shù)字段與數(shù)據(jù)庫表一致。我們需要使用 Mine 來斷定消息是否屬于當前已驗證用戶所有。一旦加入實時功能,ReceiverID 可以幫助我們過濾消息。
接下來讓我們編寫 HTTP 處理程序。盡管它有些長,但也沒什么好怕的。
func createConversation(w http.ResponseWriter, r *http.Request) {var input struct {Username string `json:"username"`}defer r.Body.Close()if err := json.NewDecoder(r.Body).Decode(&input); err != nil {http.Error(w, err.Error(), http.StatusBadRequest)return}input.Username = strings.TrimSpace(input.Username)if input.Username == "" {respond(w, Errors{map[string]string{"username": "Username required",}}, http.StatusUnprocessableEntity)return}ctx := r.Context()authUserID := ctx.Value(keyAuthUserID).(string)tx, err := db.BeginTx(ctx, nil)if err != nil {respondError(w, fmt.Errorf("could not begin tx: %v", err))return}defer tx.Rollback()var otherParticipant Userif err := tx.QueryRowContext(ctx, `SELECT id, avatar_url FROM users WHERE username = $1`, input.Username).Scan(&otherParticipant.ID,&otherParticipant.AvatarURL,); err == sql.ErrNoRows {http.Error(w, "User not found", http.StatusNotFound)return} else if err != nil {respondError(w, fmt.Errorf("could not query other participant: %v", err))return}otherParticipant.Username = input.Usernameif otherParticipant.ID == authUserID {http.Error(w, "Try start a conversation with someone else", http.StatusForbidden)return}var conversationID stringif err := tx.QueryRowContext(ctx, `SELECT conversation_id FROM participants WHERE user_id = $1INTERSECTSELECT conversation_id FROM participants WHERE user_id = $2`, authUserID, otherParticipant.ID).Scan(&conversationID); err != nil && err != sql.ErrNoRows {respondError(w, fmt.Errorf("could not query common conversation id: %v", err))return} else if err == nil {http.Redirect(w, r, "/api/conversations/"+conversationID, http.StatusFound)return}var conversation Conversationif err = tx.QueryRowContext(ctx, `INSERT INTO conversations DEFAULT VALUESRETURNING id`).Scan(&conversation.ID); err != nil {respondError(w, fmt.Errorf("could not insert conversation: %v", err))return}if _, err = tx.ExecContext(ctx, `INSERT INTO participants (user_id, conversation_id) VALUES($1, $2),($3, $2)`, authUserID, conversation.ID, otherParticipant.ID); err != nil {respondError(w, fmt.Errorf("could not insert participants: %v", err))return}if err = tx.Commit(); err != nil {respondError(w, fmt.Errorf("could not commit tx to create conversation: %v", err))return}conversation.OtherParticipant = &otherParticipantrespond(w, conversation, http.StatusCreated)}
在此端點,你會向 /api/conversations 發(fā)送 POST 請求,請求的 JSON 主體中包含要對話的用戶的用戶名。
因此,首先需要將請求主體解析成包含用戶名的結(jié)構(gòu)。然后,校驗用戶名不能為空。
type Errors struct {Errors map[string]string `json:"errors"`}
這是錯誤消息的結(jié)構(gòu)體 Errors,它僅僅是一個映射。如果輸入空用戶名,你就會得到一段帶有 422 Unprocessable Entity(無法處理的實體)錯誤消息的 JSON 。
{"errors": {"username": "Username required"}}
然后,我們開始執(zhí)行 SQL 事務(wù)。收到的僅僅是用戶名,但事實上,我們需要知道實際的用戶 ID 。因此,事務(wù)的第一項內(nèi)容是查詢另一個參與者的 ID 和頭像。如果找不到該用戶,我們將會返回 404 Not Found(未找到) 錯誤。另外,如果找到的用戶恰好和“當前已驗證用戶”相同,我們應(yīng)該返回 403 Forbidden(拒絕處理)錯誤。這是由于對話只應(yīng)當在兩個不同的用戶之間發(fā)起,而不能是同一個。
然后,我們試圖找到這兩個用戶所共有的對話,所以需要使用 INTERSECT 語句。如果存在,只需要通過 /api/conversations/{conversationID} 重定向到該對話并將其返回。
如果未找到共有的對話,我們需要創(chuàng)建一個新的對話并添加指定的兩個參與者。最后,我們 COMMIT 該事務(wù)并使用新創(chuàng)建的對話進行響應(yīng)。
獲取對話列表
端點 /api/conversations 將獲取當前已驗證用戶的所有對話。
func getConversations(w http.ResponseWriter, r *http.Request) {ctx := r.Context()authUserID := ctx.Value(keyAuthUserID).(string)rows, err := db.QueryContext(ctx, `SELECTconversations.id,auth_user.messages_read_at < messages.created_at AS has_unread_messages,messages.id,messages.content,messages.created_at,messages.user_id = $1 AS mine,other_users.id,other_users.username,other_users.avatar_urlFROM conversationsINNER JOIN messages ON conversations.last_message_id = messages.idINNER JOIN participants other_participantsON other_participants.conversation_id = conversations.idAND other_participants.user_id != $1INNER JOIN users other_users ON other_participants.user_id = other_users.idINNER JOIN participants auth_userON auth_user.conversation_id = conversations.idAND auth_user.user_id = $1ORDER BY messages.created_at DESC`, authUserID)if err != nil {respondError(w, fmt.Errorf("could not query conversations: %v", err))return}defer rows.Close()conversations := make([]Conversation, 0)for rows.Next() {var conversation Conversationvar lastMessage Messagevar otherParticipant Userif err = rows.Scan(&conversation.ID,&conversation.HasUnreadMessages,&lastMessage.ID,&lastMessage.Content,&lastMessage.CreatedAt,&lastMessage.Mine,&otherParticipant.ID,&otherParticipant.Username,&otherParticipant.AvatarURL,); err != nil {respondError(w, fmt.Errorf("could not scan conversation: %v", err))return}conversation.LastMessage = &lastMessageconversation.OtherParticipant = &otherParticipantconversations = append(conversations, conversation)}if err = rows.Err(); err != nil {respondError(w, fmt.Errorf("could not iterate over conversations: %v", err))return}respond(w, conversations, http.StatusOK)}
該處理程序僅對數(shù)據(jù)庫進行查詢。它通過一些聯(lián)接來查詢對話表……首先,從消息表中獲取最后一條消息。然后依據(jù)“ID  與當前已驗證用戶不同”的條件,從參與者表找到對話的另一個參與者。然后聯(lián)接到用戶表以獲取該用戶的用戶名和頭像。最后,再次聯(lián)接參與者表,并以相反的條件從該表中找出參與對話的另一個用戶,其實就是當前已驗證用戶。我們會對比消息中的  messages_read_at 和 created_at 兩個字段,以確定對話中是否存在未讀消息。然后,我們通過 user_id 字段來判定該消息是否屬于“我”(指當前已驗證用戶)。
注意,此查詢過程假定對話中只有兩個用戶參與,它也僅僅適用于這種情況。另外,該設(shè)計也不很適用于需要顯示未讀消息數(shù)量的情況。如果需要顯示未讀消息的數(shù)量,我認為可以在 participants 表上添加一個unread_messages_count INT 字段,并在每次創(chuàng)建新消息的時候遞增它,如果用戶已讀則重置該字段。
接下來需要遍歷每一條記錄,通過掃描每一個存在的對話來建立一個對話切片并在最后進行響應(yīng)。
找到單個對話
端點 /api/conversations/{conversationID} 會根據(jù) ID 對單個對話進行響應(yīng)。
func getConversation(w http.ResponseWriter, r *http.Request) {ctx := r.Context()authUserID := ctx.Value(keyAuthUserID).(string)conversationID := way.Param(ctx, "conversationID")var conversation Conversationvar otherParticipant Userif err := db.QueryRowContext(ctx, `SELECTIFNULL(auth_user.messages_read_at < messages.created_at, false) AS has_unread_messages,other_users.id,other_users.username,other_users.avatar_urlFROM conversationsLEFT JOIN messages ON conversations.last_message_id = messages.idINNER JOIN participants other_participantsON other_participants.conversation_id = conversations.idAND other_participants.user_id != $1INNER JOIN users other_users ON other_participants.user_id = other_users.idINNER JOIN participants auth_userON auth_user.conversation_id = conversations.idAND auth_user.user_id = $1WHERE conversations.id = $2`, authUserID, conversationID).Scan(&conversation.HasUnreadMessages,&otherParticipant.ID,&otherParticipant.Username,&otherParticipant.AvatarURL,); err == sql.ErrNoRows {http.Error(w, "Conversation not found", http.StatusNotFound)return} else if err != nil {respondError(w, fmt.Errorf("could not query conversation: %v", err))return}conversation.ID = conversationIDconversation.OtherParticipant = &otherParticipantrespond(w, conversation, http.StatusOK)}
這里的查詢與之前有點類似。盡管我們并不關(guān)心最后一條消息的顯示問題,并因此忽略了與之相關(guān)的一些字段,但是我們需要根據(jù)這條消息來判斷對話中是否存在未讀消息。此時,我們使用 LEFT JOIN 來代替 INNER JOIN,因為 last_message_id 字段是 NULLABLE(可以為空)的;而其他情況下,我們無法得到任何記錄?;谕瑯拥睦碛桑覀冊?has_unread_messages 的比較中使用了 IFNULL 語句。最后,我們按 ID 進行過濾。
如果查詢沒有返回任何記錄,我們的響應(yīng)會返回 404 Not Found 錯誤,否則響應(yīng)將會返回 200 OK 以及找到的對話。
本篇帖子以創(chuàng)建了一些對話端點結(jié)束。
在下一篇帖子中,我們將會看到如何創(chuàng)建并列出消息。















 
 
 






 
 
 
 