Acknowledge websocket POSTED events when they should be (#722)

* WIP

* Use a goroutine for the ACK

* Only ACK other users posts

* Only ack if the prop is true

* Change the ack name

* Use non-blocking call to buffered channel

loadtest/control/simulcontroller/websocket.go

@@ -36,6 +36,13 @@ func (c *SimulController) wsEventHandler(wg *sync.WaitGroup) {
 				break
 			}
 
+			if ack, ok := ev.GetData()["should_ack"]; ok && ack.(bool) {
+				if err := c.user.PostedAck(post.Id, "success", "", ""); err != nil {
+					c.status <- c.newErrorStatus(err)
+					break
+				}
+			}
+
 			cm, _ := c.user.Store().ChannelMember(post.ChannelId, c.user.Store().Id())
 			if cm.UserId != "" {
 				break

loadtest/user/user.go

@@ -46,6 +46,7 @@ type User interface {
 	UpdateActiveChannel(channelId string) error
 	UpdateActiveThread(channelId string) error
 	UpdateActiveTeam(teamId string) error
+	PostedAck(postId string, result string, reason string, postedData string) error
 
 	//server
 	// GetConfig fetches and stores the server's configuration.

loadtest/user/userentity/user.go

@@ -77,6 +77,13 @@ type teamPresenceMsg struct {
 	teamId string
 }
 
+type postedAckMsg struct {
+	postId     string
+	result     string
+	reason     string
+	postedData string
+}
+
 type ueTransport struct {
 	transport http.RoundTripper
 	ue        *UserEntity
@@ -153,7 +160,7 @@ func (ue *UserEntity) Connect() (<-chan error, error) {
 	}
 
 	ue.wsEventChan = make(chan *model.WebSocketEvent)
-	ue.dataChan = make(chan any)
+	ue.dataChan = make(chan any, 10)
 	go ue.listen(ue.wsErrorChan)
 	ue.connected = true
 	return ue.wsErrorChan, nil

loadtest/user/userentity/websocket.go

@@ -211,7 +211,12 @@ start:
 					if err := client.UpdateActiveTeam(v.teamId); err != nil {
 						errChan <- fmt.Errorf("userentity: error in client.UpdateActiveTeam: %w", err)
 					}
+				case postedAckMsg:
+					if err := client.PostedAck(v.postId, v.result, v.reason, v.postedData); err != nil {
+						errChan <- fmt.Errorf("userentity: error in client.PostedAck: %w", err)
+					}
 				}
+
 			}
 			if chanClosed {
 				client.Close()
@@ -298,3 +303,21 @@ func (ue *UserEntity) UpdateActiveTeam(teamId string) error {
 	}
 	return nil
 }
+
+func (ue *UserEntity) PostedAck(postId string, result string, reason string, postedData string) error {
+	if !ue.connected {
+		return errors.New("user is not connected")
+	}
+	select {
+	case ue.dataChan <- postedAckMsg{
+		postId,
+		result,
+		reason,
+		postedData,
+	}:
+	default:
+		return errors.New("failed to send posted ACK")
+	}
+
+	return nil
+}

loadtest/user/websocket/client.go

@@ -204,3 +204,15 @@ func (c *Client) GetStatusesByIds(userIds []string) error {
 	}
 	return c.SendMessage("get_statuses_by_ids", data)
 }
+
+func (c *Client) PostedAck(postId string, result string, reason string, postedData string) error {
+	data := map[string]interface{}{
+		"post_id":    postId,
+		"user_agent": "LoadTest",
+		"result":     result,
+		"reason":     reason,
+		"data":       postedData,
+	}
+
+	return c.SendMessage("posted_notify_ack", data)
+}