diff --git a/api4/post.go b/api4/post.go index c5a8c5f0b3..024bcdc873 100644 --- a/api4/post.go +++ b/api4/post.go @@ -34,6 +34,8 @@ func (api *API) InitPost() { api.BaseRoutes.Post.Handle("", api.APISessionRequired(updatePost)).Methods("PUT") api.BaseRoutes.Post.Handle("/patch", api.APISessionRequired(patchPost)).Methods("PUT") api.BaseRoutes.PostForUser.Handle("/set_unread", api.APISessionRequired(setPostUnread)).Methods("POST") + api.BaseRoutes.PostForUser.Handle("/reminder", api.APISessionRequired(setPostReminder)).Methods("POST") + api.BaseRoutes.Post.Handle("/pin", api.APISessionRequired(pinPost)).Methods("POST") api.BaseRoutes.Post.Handle("/unpin", api.APISessionRequired(unpinPost)).Methods("POST") } @@ -842,6 +844,36 @@ func setPostUnread(c *Context, w http.ResponseWriter, r *http.Request) { } } +func setPostReminder(c *Context, w http.ResponseWriter, r *http.Request) { + c.RequirePostId().RequireUserId() + if c.Err != nil { + return + } + + if c.AppContext.Session().UserId != c.Params.UserId && !c.App.SessionHasPermissionToUser(*c.AppContext.Session(), c.Params.UserId) { + c.SetPermissionError(model.PermissionEditOtherUsers) + return + } + if !c.App.SessionHasPermissionToChannelByPost(*c.AppContext.Session(), c.Params.PostId, model.PermissionReadChannel) { + c.SetPermissionError(model.PermissionReadChannel) + return + } + + var reminder model.PostReminder + if jsonErr := json.NewDecoder(r.Body).Decode(&reminder); jsonErr != nil { + c.SetInvalidParam("target_time") + return + } + + appErr := c.App.SetPostReminder(c.Params.PostId, c.Params.UserId, reminder.TargetTime) + if appErr != nil { + c.Err = appErr + return + } + + ReturnStatusOK(w) +} + func saveIsPinnedPost(c *Context, w http.ResponseWriter, isPinned bool) { c.RequirePostId() if c.Err != nil { diff --git a/api4/post_test.go b/api4/post_test.go index 22006fb812..a7550edffc 100644 --- a/api4/post_test.go +++ b/api4/post_test.go @@ -3164,3 +3164,63 @@ func TestGetPostStripActionIntegrations(t *testing.T) { // integration must be omitted require.Nil(t, action["integration"]) } + +func TestPostReminder(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + client := th.Client + userWSClient, err := th.CreateWebSocketClient() + require.NoError(t, err) + defer userWSClient.Close() + userWSClient.Listen() + + targetTime := time.Now().UTC().Unix() + resp, err := client.SetPostReminder(&model.PostReminder{ + TargetTime: targetTime, + PostId: th.BasicPost.Id, + UserId: th.BasicUser.Id, + }) + require.NoError(t, err) + CheckOKStatus(t, resp) + + post, _, err := client.GetPost(th.BasicPost.Id, "") + require.NoError(t, err) + + user, _, err := client.GetUser(post.UserId, "") + require.NoError(t, err) + + var caught bool + func() { + for { + select { + case ev := <-userWSClient.EventChannel: + if ev.EventType() == model.WebsocketEventEphemeralMessage { + caught = true + data := ev.GetData() + + post, ok := data["post"].(string) + require.True(t, ok) + + var parsedPost model.Post + err := json.Unmarshal([]byte(post), &parsedPost) + require.NoError(t, err) + + assert.Equal(t, model.PostTypeEphemeral, parsedPost.Type) + assert.Equal(t, th.BasicUser.Id, parsedPost.UserId) + assert.Equal(t, th.BasicPost.Id, parsedPost.RootId) + + require.Equal(t, float64(targetTime), parsedPost.GetProp("target_time").(float64)) + require.Equal(t, th.BasicPost.Id, parsedPost.GetProp("post_id").(string)) + require.Equal(t, user.Username, parsedPost.GetProp("username").(string)) + require.Equal(t, th.BasicTeam.Name, parsedPost.GetProp("team_name").(string)) + return + } + case <-time.After(1 * time.Second): + return + } + } + }() + + require.Truef(t, caught, "User should have received %s event", model.WebsocketEventEphemeralMessage) +} diff --git a/app/app_iface.go b/app/app_iface.go index fcce9b66e3..49f298ab40 100644 --- a/app/app_iface.go +++ b/app/app_iface.go @@ -446,6 +446,7 @@ type AppIface interface { CheckIntegrity() <-chan model.IntegrityCheckResult CheckMandatoryS3Fields(settings *model.FileSettings) *model.AppError CheckPasswordAndAllCriteria(user *model.User, password string, mfaToken string) *model.AppError + CheckPostReminders() CheckRolesExist(roleNames []string) *model.AppError CheckUserAllAuthenticationCriteria(user *model.User, mfaToken string) *model.AppError CheckUserMfa(user *model.User, token string) *model.AppError @@ -1041,6 +1042,7 @@ type AppIface interface { SetPluginKey(pluginID string, key string, value []byte) *model.AppError SetPluginKeyWithExpiry(pluginID string, key string, value []byte, expireInSeconds int64) *model.AppError SetPluginKeyWithOptions(pluginID string, key string, value []byte, options model.PluginKVSetOptions) (bool, *model.AppError) + SetPostReminder(postID, userID string, targetTime int64) *model.AppError SetProfileImage(userID string, imageData *multipart.FileHeader) *model.AppError SetProfileImageFromFile(userID string, file io.Reader) *model.AppError SetProfileImageFromMultiPartFile(userID string, file multipart.File) *model.AppError diff --git a/app/channels.go b/app/channels.go index 24a2f460d1..48aba3297c 100644 --- a/app/channels.go +++ b/app/channels.go @@ -86,6 +86,9 @@ type Channels struct { dndTaskMut sync.Mutex dndTask *model.ScheduledTask + + postReminderMut sync.Mutex + postReminderTask *model.ScheduledTask } func init() { diff --git a/app/opentracing/opentracing_layer.go b/app/opentracing/opentracing_layer.go index 28a25fdcd6..c8eaca45fe 100644 --- a/app/opentracing/opentracing_layer.go +++ b/app/opentracing/opentracing_layer.go @@ -1310,6 +1310,21 @@ func (a *OpenTracingAppLayer) CheckPasswordAndAllCriteria(user *model.User, pass return resultVar0 } +func (a *OpenTracingAppLayer) CheckPostReminders() { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CheckPostReminders") + + a.ctx = newCtx + a.app.Srv().Store.SetContext(newCtx) + defer func() { + a.app.Srv().Store.SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + a.app.CheckPostReminders() +} + func (a *OpenTracingAppLayer) CheckProviderAttributes(user *model.User, patch *model.UserPatch) string { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.CheckProviderAttributes") @@ -15592,6 +15607,28 @@ func (a *OpenTracingAppLayer) SetPluginKeyWithOptions(pluginID string, key strin return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) SetPostReminder(postID string, userID string, targetTime int64) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SetPostReminder") + + a.ctx = newCtx + a.app.Srv().Store.SetContext(newCtx) + defer func() { + a.app.Srv().Store.SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.SetPostReminder(postID, userID, targetTime) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) SetProfileImage(userID string, imageData *multipart.FileHeader) *model.AppError { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SetProfileImage") diff --git a/app/post.go b/app/post.go index adfa7302ec..797f1b6861 100644 --- a/app/post.go +++ b/app/post.go @@ -1933,6 +1933,126 @@ func (a *App) GetTopThreadsForUserSince(c request.CTX, teamID, userID string, op return topThreadsWithEmbedAndImage, nil } +func (a *App) SetPostReminder(postID, userID string, targetTime int64) *model.AppError { + // Store the reminder in the DB + reminder := &model.PostReminder{ + PostId: postID, + UserId: userID, + TargetTime: targetTime, + } + err := a.Srv().Store.Post().SetPostReminder(reminder) + if err != nil { + return model.NewAppError("SetPostReminder", "app.post_reminder.app_error", nil, err.Error(), http.StatusInternalServerError) + } + + metadata, err := a.Srv().Store.Post().GetPostReminderMetadata(postID) + if err != nil { + return model.NewAppError("SetPostReminder", "app.post_reminder.app_error", nil, err.Error(), http.StatusInternalServerError) + } + + parsed := time.Unix(targetTime, 0).UTC().Format(time.RFC822) + siteURL := *a.Config().ServiceSettings.SiteURL + // Send an ack message. + ephemeralPost := &model.Post{ + Type: model.PostTypeEphemeral, + Id: model.NewId(), + CreateAt: model.GetMillis(), + UserId: userID, + RootId: postID, + ChannelId: metadata.ChannelId, + // It's okay to keep this non-translated. This is just a fallback. + // The webapp will parse the timestamp and show that in user's local timezone. + Message: fmt.Sprintf("You will be reminded about %s/%s/pl/%s by @%s at %s", siteURL, metadata.TeamName, postID, metadata.Username, parsed), + Props: model.StringInterface{ + "target_time": targetTime, + "team_name": metadata.TeamName, + "post_id": postID, + "username": metadata.Username, + "type": model.PostTypeReminder, + }, + } + + message := model.NewWebSocketEvent(model.WebsocketEventEphemeralMessage, "", ephemeralPost.ChannelId, userID, nil) + ephemeralPost = a.PreparePostForClientWithEmbedsAndImages(request.EmptyContext(a.Log()), ephemeralPost, true, false) + ephemeralPost = model.AddPostActionCookies(ephemeralPost, a.PostActionCookieSecret()) + + postJSON, jsonErr := ephemeralPost.ToJSON() + if jsonErr != nil { + mlog.Warn("Failed to encode post to JSON", mlog.Err(jsonErr)) + } + message.Add("post", postJSON) + a.Publish(message) + + return nil +} + +func (a *App) CheckPostReminders() { + systemBot, appErr := a.GetSystemBot() + if appErr != nil { + mlog.Error("Failed to get system bot", mlog.Err(appErr)) + return + } + + // This will return the reminders and also delete them from the DB. + // In case, any of the next steps fail, those reminders would be lost. + // Alternatively, if we delete those reminders _after_ it has been sent, + // then in case of any temporary failure, they would get sent in the next batch. + // MM-45595. + reminders, err := a.Srv().Store.Post().GetPostReminders(time.Now().UTC().Unix()) + if err != nil { + mlog.Error("Failed to get post reminders", mlog.Err(err)) + return + } + + // We group multiple reminders for a single user. + groupedReminders := make(map[string][]string) + for _, r := range reminders { + if groupedReminders[r.UserId] == nil { + groupedReminders[r.UserId] = []string{r.PostId} + } else { + groupedReminders[r.UserId] = append(groupedReminders[r.UserId], r.PostId) + } + } + + siteURL := *a.Config().ServiceSettings.SiteURL + for userID, postIDs := range groupedReminders { + ch, appErr := a.GetOrCreateDirectChannel(request.EmptyContext(a.Log()), userID, systemBot.UserId) + if appErr != nil { + mlog.Error("Failed to get direct channel", mlog.Err(appErr)) + return + } + + for _, postID := range postIDs { + metadata, err := a.Srv().Store.Post().GetPostReminderMetadata(postID) + if err != nil { + mlog.Error("Failed to get post reminder metadata", mlog.Err(err)) + continue + } + + T := i18n.GetUserTranslations(metadata.UserLocale) + dm := &model.Post{ + ChannelId: ch.Id, + Message: T("app.post_reminder_dm", model.StringInterface{ + "SiteURL": siteURL, + "TeamName": metadata.TeamName, + "PostId": postID, + "Username": metadata.Username, + }), + Type: model.PostTypeDefault, + UserId: systemBot.UserId, + Props: model.StringInterface{ + "username": systemBot.Username, + }, + } + + if _, err := a.CreatePost(request.EmptyContext(a.Log()), dm, ch, false, true); err != nil { + mlog.Error("Failed to post reminder message", mlog.Err(err)) + } + } + } + +} + func includeEmbedsAndImages(a *App, c request.CTX, topThreadList *model.TopThreadList, userID string) (*model.TopThreadList, error) { for _, topThread := range topThreadList.Items { topThread.Post = a.PreparePostForClientWithEmbedsAndImages(c, topThread.Post, false, false) diff --git a/app/server.go b/app/server.go index 3066ecc858..fddb9ef1c5 100644 --- a/app/server.go +++ b/app/server.go @@ -681,6 +681,7 @@ func NewServer(options ...Option) (*Server, error) { s.runLicenseExpirationCheckJob() s.runInactivityCheckJob() runDNDStatusExpireJob(appInstance) + runPostReminderJob(appInstance) }) s.runJobs() } @@ -2066,31 +2067,53 @@ func (s *Server) ReadFile(path string) ([]byte, *model.AppError) { return result, nil } -func createDNDStatusExpirationRecurringTask(a *App) { - a.ch.dndTaskMut.Lock() - a.ch.dndTask = model.CreateRecurringTaskFromNextIntervalTime("Unset DND Statuses", a.UpdateDNDStatusOfUsers, 5*time.Minute) - a.ch.dndTaskMut.Unlock() +func withMut(mut *sync.Mutex, f func()) { + mut.Lock() + defer mut.Unlock() + f() } -func cancelDNDStatusExpirationRecurringTask(a *App) { - a.ch.dndTaskMut.Lock() - if a.ch.dndTask != nil { - a.ch.dndTask.Cancel() - a.ch.dndTask = nil +func cancelTask(mut *sync.Mutex, task *model.ScheduledTask) { + mut.Lock() + defer mut.Unlock() + if task != nil { + task.Cancel() + task = nil } - a.ch.dndTaskMut.Unlock() } func runDNDStatusExpireJob(a *App) { if a.IsLeader() { - createDNDStatusExpirationRecurringTask(a) + withMut(&a.ch.dndTaskMut, func() { + a.ch.dndTask = model.CreateRecurringTaskFromNextIntervalTime("Unset DND Statuses", a.UpdateDNDStatusOfUsers, 5*time.Minute) + }) } a.ch.srv.AddClusterLeaderChangedListener(func() { mlog.Info("Cluster leader changed. Determining if unset DNS status task should be running", mlog.Bool("isLeader", a.IsLeader())) if a.IsLeader() { - createDNDStatusExpirationRecurringTask(a) + withMut(&a.ch.dndTaskMut, func() { + a.ch.dndTask = model.CreateRecurringTaskFromNextIntervalTime("Unset DND Statuses", a.UpdateDNDStatusOfUsers, 5*time.Minute) + }) } else { - cancelDNDStatusExpirationRecurringTask(a) + cancelTask(&a.ch.dndTaskMut, a.ch.dndTask) + } + }) +} + +func runPostReminderJob(a *App) { + if a.IsLeader() { + withMut(&a.ch.postReminderMut, func() { + a.ch.postReminderTask = model.CreateRecurringTaskFromNextIntervalTime("Check Post reminders", a.CheckPostReminders, 5*time.Minute) + }) + } + a.ch.srv.AddClusterLeaderChangedListener(func() { + mlog.Info("Cluster leader changed. Determining if post reminder task should be running", mlog.Bool("isLeader", a.IsLeader())) + if a.IsLeader() { + withMut(&a.ch.postReminderMut, func() { + a.ch.postReminderTask = model.CreateRecurringTaskFromNextIntervalTime("Check Post reminders", a.CheckPostReminders, 5*time.Minute) + }) + } else { + cancelTask(&a.ch.postReminderMut, a.ch.postReminderTask) } }) } diff --git a/db/migrations/migrations.list b/db/migrations/migrations.list index 6f8dd5ac80..1caa7a23a7 100644 --- a/db/migrations/migrations.list +++ b/db/migrations/migrations.list @@ -180,6 +180,8 @@ db/migrations/mysql/000089_add-channelid-to-reaction.down.sql db/migrations/mysql/000089_add-channelid-to-reaction.up.sql db/migrations/mysql/000090_create_enums.down.sql db/migrations/mysql/000090_create_enums.up.sql +db/migrations/mysql/000091_create_post_reminder.down.sql +db/migrations/mysql/000091_create_post_reminder.up.sql db/migrations/postgres/000001_create_teams.down.sql db/migrations/postgres/000001_create_teams.up.sql db/migrations/postgres/000002_create_team_members.down.sql @@ -360,3 +362,5 @@ db/migrations/postgres/000089_add-channelid-to-reaction.down.sql db/migrations/postgres/000089_add-channelid-to-reaction.up.sql db/migrations/postgres/000090_create_enums.down.sql db/migrations/postgres/000090_create_enums.up.sql +db/migrations/postgres/000091_create_post_reminder.down.sql +db/migrations/postgres/000091_create_post_reminder.up.sql diff --git a/db/migrations/mysql/000091_create_post_reminder.down.sql b/db/migrations/mysql/000091_create_post_reminder.down.sql new file mode 100644 index 0000000000..40aeec0bf9 --- /dev/null +++ b/db/migrations/mysql/000091_create_post_reminder.down.sql @@ -0,0 +1,16 @@ +SET @preparedStatement = (SELECT IF( + ( + SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS + WHERE table_name = 'PostReminders' + AND table_schema = DATABASE() + AND index_name = 'idx_postreminders_targettime' + ) > 0, + 'DROP INDEX idx_postreminders_targettime ON PostReminders;', + 'SELECT 1' +)); + +PREPARE removeIndexIfExists FROM @preparedStatement; +EXECUTE removeIndexIfExists; +DEALLOCATE PREPARE removeIndexIfExists; + +DROP TABLE IF EXISTS PostReminders; \ No newline at end of file diff --git a/db/migrations/mysql/000091_create_post_reminder.up.sql b/db/migrations/mysql/000091_create_post_reminder.up.sql new file mode 100644 index 0000000000..7cea8025b0 --- /dev/null +++ b/db/migrations/mysql/000091_create_post_reminder.up.sql @@ -0,0 +1,21 @@ +CREATE TABLE IF NOT EXISTS PostReminders ( + PostId varchar(26) NOT NULL, + UserId varchar(26) NOT NULL, + TargetTime bigint, + PRIMARY KEY (PostId, UserId) +); + +SET @preparedStatement = (SELECT IF( + ( + SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS + WHERE table_name = 'PostReminders' + AND table_schema = DATABASE() + AND index_name = 'idx_postreminders_targettime' + ) > 0, + 'SELECT 1', + 'CREATE INDEX idx_postreminders_targettime ON PostReminders(TargetTime);' +)); + +PREPARE createIndexIfNotExists FROM @preparedStatement; +EXECUTE createIndexIfNotExists; +DEALLOCATE PREPARE createIndexIfNotExists; \ No newline at end of file diff --git a/db/migrations/postgres/000091_create_post_reminder.down.sql b/db/migrations/postgres/000091_create_post_reminder.down.sql new file mode 100644 index 0000000000..268d3908e5 --- /dev/null +++ b/db/migrations/postgres/000091_create_post_reminder.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_postreminders_targettime; + +DROP TABLE IF EXISTS postreminders; \ No newline at end of file diff --git a/db/migrations/postgres/000091_create_post_reminder.up.sql b/db/migrations/postgres/000091_create_post_reminder.up.sql new file mode 100644 index 0000000000..cdc87f66de --- /dev/null +++ b/db/migrations/postgres/000091_create_post_reminder.up.sql @@ -0,0 +1,8 @@ +CREATE TABLE IF NOT EXISTS postreminders ( + postid varchar(26) NOT NULL, + userid varchar(26) NOT NULL, + targettime bigint, + PRIMARY KEY (postid, userid) +); + +CREATE INDEX IF NOT EXISTS idx_postreminders_targettime ON postreminders(targettime); \ No newline at end of file diff --git a/i18n/en.json b/i18n/en.json index a86209588a..5ad9f74c1d 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -5911,6 +5911,14 @@ "id": "app.post.update.app_error", "translation": "Unable to update the Post." }, + { + "id": "app.post_reminder.app_error", + "translation": " " + }, + { + "id": "app.post_reminder_dm", + "translation": "Hi there, you asked me to remind you about {{.SiteURL}}/{{.TeamName}}/pl/{{.PostId}} by @{{.Username}}" + }, { "id": "app.preference.delete.app_error", "translation": "We encountered an error while deleting preferences." diff --git a/model/client4.go b/model/client4.go index 1dc5f0aa8d..98cd3c739e 100644 --- a/model/client4.go +++ b/model/client4.go @@ -3743,6 +3743,23 @@ func (c *Client4) SetPostUnread(userId string, postId string, collapsedThreadsSu return BuildResponse(r), nil } +// SetPostReminder creates a post reminder for a given post at a specified time. +// The time needs to be in UTC epoch in seconds. It is always truncated to a +// 5 minute resolution minimum. +func (c *Client4) SetPostReminder(reminder *PostReminder) (*Response, error) { + b, err := json.Marshal(reminder) + if err != nil { + return nil, NewAppError("SetPostReminder", "api.marshal_error", nil, err.Error(), http.StatusInternalServerError) + } + + r, err := c.DoAPIPostBytes(c.userRoute(reminder.UserId)+c.postRoute(reminder.PostId)+"/reminder", b) + if err != nil { + return BuildResponse(r), err + } + defer closeBody(r) + return BuildResponse(r), nil +} + // PinPost pin a post based on provided post id string. func (c *Client4) PinPost(postId string) (*Response, error) { r, err := c.DoAPIPost(c.postRoute(postId)+"/pin", "") diff --git a/model/post.go b/model/post.go index f8e3240154..f0ebcfbea2 100644 --- a/model/post.go +++ b/model/post.go @@ -48,6 +48,7 @@ const ( PostTypeSystemWarnMetricStatus = "warn_metric_status" PostTypeMe = "me" PostCustomTypePrefix = "custom_" + PostTypeReminder = "reminder" PostFileidsMaxRunes = 300 PostFilenamesMaxRunes = 4000 @@ -149,6 +150,13 @@ type PostPatch struct { HasReactions *bool `json:"has_reactions"` } +type PostReminder struct { + TargetTime int64 `json:"target_time"` + // These fields are only used internally for interacting with DB. + PostId string `json:",omitempty"` + UserId string `json:",omitempty"` +} + type SearchParameter struct { Terms *string `json:"terms"` IsOrSearch *bool `json:"is_or_search"` diff --git a/store/layer_generators/main.go b/store/layer_generators/main.go index 3f9089246e..28b015ac0c 100644 --- a/store/layer_generators/main.go +++ b/store/layer_generators/main.go @@ -221,10 +221,20 @@ func generateLayer(name, templateFile string) ([]byte, error) { if len(results) == 0 { return "" } - if len(results) == 1 { - return strings.Join(results, ", ") + returns := []string{} + for _, result := range results { + switch result { + case "*PostReminderMetadata": + returns = append(returns, fmt.Sprintf("*store.%s", strings.TrimPrefix(result, "*"))) + default: + returns = append(returns, result) + } } - return fmt.Sprintf("(%s)", strings.Join(results, ", ")) + + if len(returns) == 1 { + return strings.Join(returns, ", ") + } + return fmt.Sprintf("(%s)", strings.Join(returns, ", ")) }, "genResultsVars": func(results []string, withNilError bool) string { vars := []string{} diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index 85c4829f58..ee89078792 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -5809,6 +5809,42 @@ func (s *OpenTracingLayerPostStore) GetPostIdBeforeTime(channelID string, timest return result, err } +func (s *OpenTracingLayerPostStore) GetPostReminderMetadata(postID string) (*store.PostReminderMetadata, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.GetPostReminderMetadata") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.PostStore.GetPostReminderMetadata(postID) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + +func (s *OpenTracingLayerPostStore) GetPostReminders(now int64) ([]*model.PostReminder, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.GetPostReminders") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.PostStore.GetPostReminders(now) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + func (s *OpenTracingLayerPostStore) GetPosts(options model.GetPostsOptions, allowFromCache bool, sanitizeOptions map[string]bool) (*model.PostList, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.GetPosts") @@ -6236,6 +6272,24 @@ func (s *OpenTracingLayerPostStore) SearchPostsForUser(paramsList []*model.Searc return result, err } +func (s *OpenTracingLayerPostStore) SetPostReminder(reminder *model.PostReminder) error { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.SetPostReminder") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + err := s.PostStore.SetPostReminder(reminder) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return err +} + func (s *OpenTracingLayerPostStore) Update(newPost *model.Post, oldPost *model.Post) (*model.Post, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.Update") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 5094c41fcb..ebd66004f5 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -6575,6 +6575,48 @@ func (s *RetryLayerPostStore) GetPostIdBeforeTime(channelID string, timestamp in } +func (s *RetryLayerPostStore) GetPostReminderMetadata(postID string) (*store.PostReminderMetadata, error) { + + tries := 0 + for { + result, err := s.PostStore.GetPostReminderMetadata(postID) + if err == nil { + return result, nil + } + if !isRepeatableError(err) { + return result, err + } + tries++ + if tries >= 3 { + err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") + return result, err + } + timepkg.Sleep(100 * timepkg.Millisecond) + } + +} + +func (s *RetryLayerPostStore) GetPostReminders(now int64) ([]*model.PostReminder, error) { + + tries := 0 + for { + result, err := s.PostStore.GetPostReminders(now) + if err == nil { + return result, nil + } + if !isRepeatableError(err) { + return result, err + } + tries++ + if tries >= 3 { + err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") + return result, err + } + timepkg.Sleep(100 * timepkg.Millisecond) + } + +} + func (s *RetryLayerPostStore) GetPosts(options model.GetPostsOptions, allowFromCache bool, sanitizeOptions map[string]bool) (*model.PostList, error) { tries := 0 @@ -7064,6 +7106,27 @@ func (s *RetryLayerPostStore) SearchPostsForUser(paramsList []*model.SearchParam } +func (s *RetryLayerPostStore) SetPostReminder(reminder *model.PostReminder) error { + + tries := 0 + for { + err := s.PostStore.SetPostReminder(reminder) + if err == nil { + return nil + } + if !isRepeatableError(err) { + return err + } + tries++ + if tries >= 3 { + err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") + return err + } + timepkg.Sleep(100 * timepkg.Millisecond) + } + +} + func (s *RetryLayerPostStore) Update(newPost *model.Post, oldPost *model.Post) (*model.Post, error) { tries := 0 diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index af7b11ca01..03ee58888e 100644 --- a/store/sqlstore/post_store.go +++ b/store/sqlstore/post_store.go @@ -2963,3 +2963,97 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts } return nil } + +func (s *SqlPostStore) SetPostReminder(reminder *model.PostReminder) error { + transaction, err := s.GetMasterX().Beginx() + if err != nil { + return errors.Wrap(err, "begin_transaction") + } + defer finalizeTransactionX(transaction) + + sql := `SELECT EXISTS (SELECT 1 FROM Posts WHERE Id=?)` + var exist bool + err = transaction.Get(&exist, sql, reminder.PostId) + if err != nil { + return errors.Wrap(err, "failed to check for post") + } + if !exist { + return store.NewErrNotFound("Post", reminder.PostId) + } + + query := s.getQueryBuilder(). + Insert("PostReminders"). + Columns("PostId", "UserId", "TargetTime"). + Values(reminder.PostId, reminder.UserId, reminder.TargetTime) + + if s.DriverName() == model.DatabaseDriverMysql { + query = query.SuffixExpr(sq.Expr("ON DUPLICATE KEY UPDATE TargetTime = ?", reminder.TargetTime)) + } else { + query = query.SuffixExpr(sq.Expr("ON CONFLICT (postid, userid) DO UPDATE SET TargetTime = ?", reminder.TargetTime)) + } + + sql, args, err := query.ToSql() + if err != nil { + return errors.Wrap(err, "setPostReminder_tosql") + } + if _, err2 := transaction.Exec(sql, args...); err2 != nil { + return errors.Wrap(err2, "failed to insert post reminder") + } + if err = transaction.Commit(); err != nil { + return errors.Wrap(err, "commit_transaction") + } + return nil +} + +func (s *SqlPostStore) GetPostReminders(now int64) ([]*model.PostReminder, error) { + reminders := []*model.PostReminder{} + + transaction, err := s.GetMasterX().Beginx() + if err != nil { + return nil, errors.Wrap(err, "begin_transaction") + } + defer finalizeTransactionX(transaction) + + err = transaction.Select(&reminders, `SELECT PostId, UserId + FROM PostReminders + WHERE TargetTime < ?`, now) + if err != nil && err != sql.ErrNoRows { + return nil, errors.Wrap(err, "failed to get post reminders") + } + + if err == sql.ErrNoRows { + // No need to execute delete statement if there's nothing to delete. + return reminders, nil + } + + // Postgres supports RETURNING * in a DELETE statement, but MySQL doesn't. + // So we are stuck with 2 queries. Not taking separate paths for Postgres + // and MySQL for simplicity. + _, err = transaction.Exec(`DELETE from PostReminders WHERE TargetTime < ?`, now) + if err != nil { + return nil, errors.Wrap(err, "failed to delete post reminders") + } + + if err = transaction.Commit(); err != nil { + return nil, errors.Wrap(err, "commit_transaction") + } + + return reminders, nil +} + +func (s *SqlPostStore) GetPostReminderMetadata(postID string) (*store.PostReminderMetadata, error) { + meta := &store.PostReminderMetadata{} + err := s.GetReplicaX().Get(meta, `SELECT c.id as ChannelId, + t.name as TeamName, + u.locale as UserLocale, u.username as Username + FROM Posts p, Channels c, Teams t, Users u + WHERE p.ChannelId=c.Id + AND c.TeamId=t.Id + AND p.UserId=u.Id + AND p.Id=?`, postID) + if err != nil { + return nil, errors.Wrap(err, "failed to get post reminder metadata") + } + + return meta, nil +} diff --git a/store/store.go b/store/store.go index 370dad74dc..911c06191f 100644 --- a/store/store.go +++ b/store/store.go @@ -386,6 +386,9 @@ type PostStore interface { GetOldestEntityCreationTime() (int64, error) HasAutoResponsePostByUserSince(options model.GetPostsSinceOptions, userId string) (bool, error) GetPostsSinceForSync(options model.GetPostsSinceForSyncOptions, cursor model.GetPostsSinceForSyncCursor, limit int) ([]*model.Post, model.GetPostsSinceForSyncCursor, error) + SetPostReminder(reminder *model.PostReminder) error + GetPostReminders(now int64) ([]*model.PostReminder, error) + GetPostReminderMetadata(postID string) (*PostReminderMetadata, error) // GetNthRecentPostTime returns the CreateAt time of the nth most recent post. GetNthRecentPostTime(n int64) (int64, error) } @@ -1019,6 +1022,15 @@ type ChannelMemberGraphQLSearchOpts struct { ExcludeTeam bool } +// PostReminderMetadata contains some info needed to send +// the reminder message to the user. +type PostReminderMetadata struct { + ChannelId string + TeamName string + UserLocale string + Username string +} + // SidebarCategorySearchOpts contains the options for a graphQL query // to get the sidebar categories. type SidebarCategorySearchOpts struct { diff --git a/store/storetest/mocks/PostStore.go b/store/storetest/mocks/PostStore.go index 0f24d48db0..42977fb9e3 100644 --- a/store/storetest/mocks/PostStore.go +++ b/store/storetest/mocks/PostStore.go @@ -9,6 +9,8 @@ import ( model "github.com/mattermost/mattermost-server/v6/model" mock "github.com/stretchr/testify/mock" + + store "github.com/mattermost/mattermost-server/v6/store" ) // PostStore is an autogenerated mock type for the PostStore type @@ -440,6 +442,52 @@ func (_m *PostStore) GetPostIdBeforeTime(channelID string, timestamp int64, coll return r0, r1 } +// GetPostReminderMetadata provides a mock function with given fields: postID +func (_m *PostStore) GetPostReminderMetadata(postID string) (*store.PostReminderMetadata, error) { + ret := _m.Called(postID) + + var r0 *store.PostReminderMetadata + if rf, ok := ret.Get(0).(func(string) *store.PostReminderMetadata); ok { + r0 = rf(postID) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*store.PostReminderMetadata) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func(string) error); ok { + r1 = rf(postID) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// GetPostReminders provides a mock function with given fields: now +func (_m *PostStore) GetPostReminders(now int64) ([]*model.PostReminder, error) { + ret := _m.Called(now) + + var r0 []*model.PostReminder + if rf, ok := ret.Get(0).(func(int64) []*model.PostReminder); ok { + r0 = rf(now) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*model.PostReminder) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func(int64) error); ok { + r1 = rf(now) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // GetPosts provides a mock function with given fields: options, allowFromCache, sanitizeOptions func (_m *PostStore) GetPosts(options model.GetPostsOptions, allowFromCache bool, sanitizeOptions map[string]bool) (*model.PostList, error) { ret := _m.Called(options, allowFromCache, sanitizeOptions) @@ -969,6 +1017,20 @@ func (_m *PostStore) SearchPostsForUser(paramsList []*model.SearchParams, userID return r0, r1 } +// SetPostReminder provides a mock function with given fields: reminder +func (_m *PostStore) SetPostReminder(reminder *model.PostReminder) error { + ret := _m.Called(reminder) + + var r0 error + if rf, ok := ret.Get(0).(func(*model.PostReminder) error); ok { + r0 = rf(reminder) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // Update provides a mock function with given fields: newPost, oldPost func (_m *PostStore) Update(newPost *model.Post, oldPost *model.Post) (*model.Post, error) { ret := _m.Called(newPost, oldPost) diff --git a/store/storetest/post_store.go b/store/storetest/post_store.go index 15147cbfb6..98efff5321 100644 --- a/store/storetest/post_store.go +++ b/store/storetest/post_store.go @@ -5,6 +5,7 @@ package storetest import ( "context" + "errors" "fmt" "sort" "strings" @@ -57,6 +58,9 @@ func TestPostStore(t *testing.T, ss store.Store, s SqlStore) { t.Run("GetForThread", func(t *testing.T) { testPostStoreGetForThread(t, ss) }) t.Run("HasAutoResponsePostByUserSince", func(t *testing.T) { testHasAutoResponsePostByUserSince(t, ss) }) t.Run("GetPostsSinceForSync", func(t *testing.T) { testGetPostsSinceForSync(t, ss, s) }) + t.Run("SetPostReminder", func(t *testing.T) { testSetPostReminder(t, ss, s) }) + t.Run("GetPostReminders", func(t *testing.T) { testGetPostReminders(t, ss, s) }) + t.Run("GetPostReminderMetadata", func(t *testing.T) { testGetPostReminderMetadata(t, ss, s) }) t.Run("GetNthRecentPostTime", func(t *testing.T) { testGetNthRecentPostTime(t, ss) }) } @@ -3756,6 +3760,130 @@ func testGetPostsSinceForSync(t *testing.T, ss store.Store, s SqlStore) { }) } +func testSetPostReminder(t *testing.T, ss store.Store, s SqlStore) { + // Basic + userID := NewTestId() + + p1 := &model.Post{ + UserId: userID, + ChannelId: NewTestId(), + Message: "hi there", + Type: model.PostTypeDefault, + } + p1, err := ss.Post().Save(p1) + require.NoError(t, err) + + reminder := &model.PostReminder{ + TargetTime: 1234, + PostId: p1.Id, + UserId: userID, + } + + require.NoError(t, ss.Post().SetPostReminder(reminder)) + + out := model.PostReminder{} + require.NoError(t, s.GetMasterX().Get(&out, `SELECT PostId, UserId, TargetTime FROM PostReminders WHERE PostId=? AND UserId=?`, reminder.PostId, reminder.UserId)) + assert.Equal(t, reminder, &out) + + reminder.PostId = "notfound" + err = ss.Post().SetPostReminder(reminder) + var nfErr *store.ErrNotFound + require.True(t, errors.As(err, &nfErr)) + + // Upsert + reminder = &model.PostReminder{ + TargetTime: 12345, + PostId: p1.Id, + UserId: userID, + } + + require.NoError(t, ss.Post().SetPostReminder(reminder)) + require.NoError(t, s.GetMasterX().Get(&out, `SELECT PostId, UserId, TargetTime FROM PostReminders WHERE PostId=? AND UserId=?`, reminder.PostId, reminder.UserId)) + assert.Equal(t, reminder, &out) +} + +func testGetPostReminders(t *testing.T, ss store.Store, s SqlStore) { + times := []int64{100, 101, 102} + for _, tt := range times { + userID := NewTestId() + + p1 := &model.Post{ + UserId: userID, + ChannelId: NewTestId(), + Message: "hi there", + Type: model.PostTypeDefault, + } + p1, err := ss.Post().Save(p1) + require.NoError(t, err) + + reminder := &model.PostReminder{ + TargetTime: tt, + PostId: p1.Id, + UserId: userID, + } + + require.NoError(t, ss.Post().SetPostReminder(reminder)) + } + + reminders, err := ss.Post().GetPostReminders(102) + require.NoError(t, err) + require.Len(t, reminders, 2) + + // assert one reminder is left + reminders, err = ss.Post().GetPostReminders(103) + require.NoError(t, err) + require.Len(t, reminders, 1) + + // assert everything is deleted. + reminders, err = ss.Post().GetPostReminders(103) + require.NoError(t, err) + require.Len(t, reminders, 0) +} + +func testGetPostReminderMetadata(t *testing.T, ss store.Store, s SqlStore) { + team := &model.Team{ + Name: "teamname", + DisplayName: "display", + Type: model.TeamOpen, + } + team, err := ss.Team().Save(team) + require.NoError(t, err) + + ch := &model.Channel{ + TeamId: team.Id, + DisplayName: "channeldisplay", + Name: NewTestId(), + Type: model.ChannelTypeOpen, + } + ch, err = ss.Channel().Save(ch, -1) + require.NoError(t, err) + + u1 := &model.User{ + Email: MakeEmail(), + Username: model.NewId(), + Locale: "es", + } + + u1, err = ss.User().Save(u1) + require.NoError(t, err) + + p1 := &model.Post{ + UserId: u1.Id, + ChannelId: ch.Id, + Message: "hi there", + Type: model.PostTypeDefault, + } + p1, err = ss.Post().Save(p1) + require.NoError(t, err) + + meta, err := ss.Post().GetPostReminderMetadata(p1.Id) + require.NoError(t, err) + assert.Equal(t, meta.ChannelId, ch.Id) + assert.Equal(t, meta.TeamName, team.Name) + assert.Equal(t, meta.Username, u1.Username) + assert.Equal(t, meta.UserLocale, u1.Locale) +} + func getPostIds(posts []*model.Post, morePosts ...*model.Post) []string { ids := make([]string, 0, len(posts)+len(morePosts)) for _, p := range posts { diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index eb920da051..3066b77559 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -5258,6 +5258,38 @@ func (s *TimerLayerPostStore) GetPostIdBeforeTime(channelID string, timestamp in return result, err } +func (s *TimerLayerPostStore) GetPostReminderMetadata(postID string) (*store.PostReminderMetadata, error) { + start := time.Now() + + result, err := s.PostStore.GetPostReminderMetadata(postID) + + elapsed := float64(time.Since(start)) / float64(time.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("PostStore.GetPostReminderMetadata", success, elapsed) + } + return result, err +} + +func (s *TimerLayerPostStore) GetPostReminders(now int64) ([]*model.PostReminder, error) { + start := time.Now() + + result, err := s.PostStore.GetPostReminders(now) + + elapsed := float64(time.Since(start)) / float64(time.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("PostStore.GetPostReminders", success, elapsed) + } + return result, err +} + func (s *TimerLayerPostStore) GetPosts(options model.GetPostsOptions, allowFromCache bool, sanitizeOptions map[string]bool) (*model.PostList, error) { start := time.Now() @@ -5641,6 +5673,22 @@ func (s *TimerLayerPostStore) SearchPostsForUser(paramsList []*model.SearchParam return result, err } +func (s *TimerLayerPostStore) SetPostReminder(reminder *model.PostReminder) error { + start := time.Now() + + err := s.PostStore.SetPostReminder(reminder) + + elapsed := float64(time.Since(start)) / float64(time.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("PostStore.SetPostReminder", success, elapsed) + } + return err +} + func (s *TimerLayerPostStore) Update(newPost *model.Post, oldPost *model.Post) (*model.Post, error) { start := time.Now()