POC of queue to store events from MSTeams

server/api.go

@@ -186,6 +186,9 @@ func (a *API) processActivity(w http.ResponseWriter, req *http.Request) {
 
 	requireEncryptedContent := a.p.getConfiguration().CertificateKey != ""
 	errors := ""
+	activitiesToHandle := make([]msteams.Activity, 0, len(activities.Value))
+	activitiesToHandleData := make([]string, 0, len(activities.Value))
+	activitiesToHandleIds := make([]string, 0, len(activities.Value))
 	for _, activity := range activities.Value {
 		if activity.EncryptedContent != nil {
 			content, err := a.processEncryptedContent(*activity.EncryptedContent)
@@ -203,17 +206,31 @@ func (a *API) processActivity(w http.ResponseWriter, req *http.Request) {
 			errors += "Invalid webhook secret"
 			continue
 		}
-
-		if err := a.p.activityHandler.Handle(activity); err != nil {
-			a.p.API.LogWarn("Unable to process created activity", "activity", activity, "error", err.Error())
+		activity.ID = model.NewId()
+		activitiesToHandleIds = append(activitiesToHandleIds, activity.ID)
+		activitiesToHandle = append(activitiesToHandle, activity)
+		data, err := json.Marshal(activity)
+		if err != nil {
 			errors += err.Error() + "\n"
+			continue
 		}
+		activitiesToHandleData = append(activitiesToHandleData, string(data))
 	}
 	if errors != "" {
 		http.Error(w, errors, http.StatusBadRequest)
 		return
 	}
 
+	if err := a.store.EnqueueActivities(activitiesToHandleIds, activitiesToHandleData); err != nil {
+		a.p.API.LogWarn("Unable to enqueue activities", "error", err.Error())
+	}
+	for _, activity := range activitiesToHandle {
+		if err := a.p.activityHandler.Handle(activity); err != nil {
+			a.p.API.LogWarn("Unable to process created activity", "activity", activity, "error", err.Error())
+			errors += err.Error() + "\n"
+		}
+	}
+
 	w.WriteHeader(http.StatusAccepted)
 }
 

server/handlers/handlers.go

@@ -236,6 +236,8 @@ func (ah *ActivityHandler) handleActivity(activity msteams.Activity) {
 				ah.plugin.GetAPI().LogWarn("Unable to unmarshal activity message", "activity", activity, "error", err)
 			}
 		}
+		ah.plugin.GetAPI().LogWarn("Receiving activity and waiting", "activity", activity)
+		time.Sleep(5 * time.Second)
 		discardedReason = ah.handleCreatedActivity(msg, activity.SubscriptionID, activityIds)
 	case "updated":
 		var msg *clientmodels.Message
@@ -254,6 +256,24 @@ func (ah *ActivityHandler) handleActivity(activity msteams.Activity) {
 		ah.plugin.GetAPI().LogWarn("Unsupported change type", "change_type", activity.ChangeType)
 	}
 
+	if discardedReason == metrics.DiscardedReasonNone ||
+		discardedReason == metrics.DiscardedReasonAlreadyAppliedChange ||
+		discardedReason == metrics.DiscardedReasonDuplicatedPost ||
+		discardedReason == metrics.DiscardedReasonIsBotUser ||
+		discardedReason == metrics.DiscardedReasonSelectiveSync ||
+		discardedReason == metrics.DiscardedReasonInvalidChangeType ||
+		discardedReason == metrics.DiscardedReasonNotUserEvent ||
+		discardedReason == metrics.DiscardedReasonDirectMessagesDisabled ||
+		discardedReason == metrics.DiscardedReasonLinkedChannelsDisabled ||
+		discardedReason == metrics.DiscardedReasonInactiveUser ||
+		discardedReason == metrics.DiscardedReasonInvalidWebhookSecret ||
+		discardedReason == metrics.DiscardedReasonFileLimitReached ||
+		discardedReason == metrics.DiscardedReasonEmptyFileID ||
+		discardedReason == metrics.DiscardedReasonMaxFileSizeExceeded ||
+		discardedReason == metrics.DiscardedReasonExpiredSubscription {
+		ah.plugin.GetStore().DequeueActivity(activity.ID)
+	}
+
 	ah.plugin.GetMetrics().ObserveChangeEvent(activity.ChangeType, discardedReason)
 }
 

server/msteams/client.go

@@ -79,14 +79,15 @@ type ClientImpl struct {
 }
 
 type Activity struct {
-	Resource                       string
-	ClientState                    string
-	ChangeType                     string
-	LifecycleEvent                 string
-	SubscriptionExpirationDateTime time.Time
-	SubscriptionID                 string
-	EncryptedContent               *EncryptedContent
-	Content                        []byte
+	ID                             string            `json:"id"`
+	Resource                       string            `json:"resource"`
+	ClientState                    string            `json:"clientState"`
+	ChangeType                     string            `json:"changeType"`
+	LifecycleEvent                 string            `json:"lifecycleEvent"`
+	SubscriptionExpirationDateTime time.Time         `json:"subscriptionExpirationDateTime"`
+	SubscriptionID                 string            `json:"subscriptionId"`
+	EncryptedContent               *EncryptedContent `json:"encryptedContent"`
+	Content                        []byte            `json:"content"`
 }
 
 type EncryptedContent struct {

server/store/sqlstore/store.go

@@ -29,6 +29,7 @@ const (
 	usersTableName                  = "msteamssync_users"
 	linksTableName                  = "msteamssync_links"
 	postsTableName                  = "msteamssync_posts"
+	activityQueueTableName          = "msteamssync_activity_queue"
 	subscriptionsTableName          = "msteamssync_subscriptions"
 	whitelistedUsersLegacyTableName = "msteamssync_whitelisted_users" // LEGACY-UNUSED
 	whitelistTableName              = "msteamssync_whitelist"
@@ -58,6 +59,10 @@ func (s *SQLStore) Init(remoteID string) error {
 		return err
 	}
 
+	if err := s.createTable(activityQueueTableName, "id VARCHAR(255) NOT NULL PRIMARY KEY, createdAt TIMESTAMP DEFAULT NOW(), activity TEXT, tries INT"); err != nil {
+		return err
+	}
+
 	if err := s.createTable(linksTableName, "mmChannelID VARCHAR(255) PRIMARY KEY, mmTeamID VARCHAR(255), msTeamsChannelID VARCHAR(255), msTeamsTeamID VARCHAR(255), creator VARCHAR(255)"); err != nil {
 		return err
 	}
@@ -1157,6 +1162,67 @@ func (s *SQLStore) DeleteUserInvite(mmUserID string) error {
 	return nil
 }
 
+func (s *SQLStore) EnqueueActivities(activitiesIDs, activities []string) error {
+	query := s.getQueryBuilder().Insert(activityQueueTableName).Columns("id", "activity", "tries")
+	for idx, activity := range activities {
+		query = query.Values(activitiesIDs[idx], activity, 0)
+	}
+	if _, err := query.Exec(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *SQLStore) DequeueActivity(activityId string) error {
+	query := s.getQueryBuilder().Delete(activityQueueTableName).Where(sq.Eq{"id": activityId})
+	if _, err := query.Exec(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *SQLStore) ProcessQueuedActivity(activityCallback func(activity string) error) error {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_ = tx.Rollback()
+	}()
+
+	query := "DELETE FROM " + activityQueueTableName + " USING (SELECT * FROM " + activityQueueTableName + " LIMIT 1 FOR UPDATE SKIP LOCKED) q WHERE q.id =  " + activityQueueTableName + ".id RETURNING " + activityQueueTableName + ".activity"
+
+	row := tx.QueryRow(query)
+
+	var data string
+	if err := row.Scan(&data); err != nil {
+		return err
+	}
+
+	if err := activityCallback(data); err != nil {
+		_ = tx.Rollback()
+
+		query := s.getQueryBuilder().Update(activityQueueTableName).Set("tries", "INC(tries)")
+		if _, err := query.Exec(); err != nil {
+			return err
+		}
+
+		return err
+	}
+
+	if err := tx.Commit(); err != nil {
+		query := s.getQueryBuilder().Update(activityQueueTableName).Set("tries", "INC(tries)")
+		if _, err := query.Exec(); err != nil {
+			return err
+		}
+
+		return err
+	}
+	return nil
+}
+
 func (s *SQLStore) GetInvitedCount() (int, error) {
 	query := s.getQueryBuilder().Select("count(*)").From(invitedUsersTableName)
 	rows, err := query.Query()

server/store/store.go

@@ -78,4 +78,7 @@ type Store interface {
 	GetSubscriptionType(subscriptionID string) (string, error)
 	UpdateSubscriptionLastActivityAt(subscriptionID string, lastActivityAt time.Time) error
 	GetSubscriptionsLastActivityAt() (map[string]time.Time, error)
+	EnqueueActivities(activitiesIDs, activities []string) error
+	DequeueActivity(activityId string) error
+	ProcessQueuedActivity(func(activity string) error) error
 }